Ajouter des Outils : Screening, Technicals, Fundamentals

Architecture providers : sources de données interchangeables

Un serveur MCP finance sérieux ne dépend jamais d'une seule source de données. Yahoo Finance peut être limité en rate, SEC EDGAR peut être lent, et un provider premium (Polygon, IEX) peut changer ses tarifs. La solution : une couche d'abstraction provider qui rend les sources interchangeables.

# providers/base.py — Interface commune
from abc import ABC, abstractmethod
from typing import Optional

class MarketDataProvider(ABC):
    """Interface commune pour toutes les sources de données marché."""

    @abstractmethod
    async def get_quote(self, ticker: str) -> dict:
        """Prix actuel, change%, volume, market cap."""

    @abstractmethod
    async def get_bars(self, ticker: str, period: str = "1y") -> list[dict]:
        """OHLCV historique."""

    @abstractmethod
    async def get_fundamentals(self, ticker: str) -> dict:
        """P/E, EV/EBITDA, margins, etc."""

    @abstractmethod
    async def get_earnings(self, ticker: str) -> list[dict]:
        """Résultats trimestriels (EPS actual vs estimate)."""

# providers/yahoo.py — Implémentation Yahoo Finance
import httpx

class YahooProvider(MarketDataProvider):
    BASE = "https://query1.finance.yahoo.com/v8/finance"

    def __init__(self):
        self.client = httpx.AsyncClient(
            timeout=10.0,
            headers={"User-Agent": "MCP-Finance/1.0"}
        )

    async def get_quote(self, ticker: str) -> dict:
        url = f"{self.BASE}/chart/{ticker}?interval=1d&range=1d"
        resp = await self.client.get(url)
        data = resp.json()
        result = data["chart"]["result"][0]
        meta = result["meta"]

        return {
            "ticker": ticker.upper(),
            "price": meta["regularMarketPrice"],
            "previous_close": meta["previousClose"],
            "change_pct": round(
                (meta["regularMarketPrice"] / meta["previousClose"] - 1) * 100, 2
            ),
            "volume": meta.get("regularMarketVolume", 0),
            "market_cap": meta.get("marketCap"),
            "currency": meta.get("currency", "USD"),
            "exchange": meta.get("exchangeName"),
            "source": "yahoo_finance",
            "as_of": meta.get("regularMarketTime")
        }

QueryData : le couteau suisse

QueryData est l'outil le plus appelé du serveur. Plutôt que d'avoir un outil par type de donnée (GetQuote, GetTechnicals, GetFundamentals...), on expose un point d'entrée unique avec un paramètre types qui spécifie ce qu'on veut :

# tools/query_data.py
from mcp.types import Tool

QUERY_DATA_TOOL = Tool(
    name="QueryData",
    description="""Retrieve financial data for a ticker symbol.
Supports multiple data types in a single call:
- quote: current price, change%, volume, market cap, 52-week range
- bars_daily: OHLCV history (default 1 year)
- technicals: RSI(14), EMA(20/50/200), MACD, Bollinger Bands, ATR(14)
- fundamentals: P/E, EV/EBITDA, P/S, margins, debt ratios, dividend yield
- earnings: last 8 quarterly EPS (actual vs estimate, beat/miss)

Use this tool whenever the user asks about a specific stock, ETF, or index.
Multiple types can be requested at once for efficiency.""",
    inputSchema={
        "type": "object",
        "properties": {
            "ticker": {
                "type": "string",
                "description": "Ticker symbol (AAPL, MSFT, BTC-USD, ^GSPC)"
            },
            "types": {
                "type": "array",
                "items": {
                    "type": "string",
                    "enum": ["quote", "bars_daily", "technicals",
                             "fundamentals", "earnings"]
                },
                "description": "Data types to retrieve"
            },
            "period": {
                "type": "string",
                "enum": ["1m", "3m", "6m", "1y", "2y", "5y"],
                "description": "History period for bars (default: 1y)"
            }
        },
        "required": ["ticker", "types"]
    }
)

async def handle_query_data(provider, arguments: dict) -> dict:
    ticker = arguments["ticker"]
    types = arguments["types"]
    period = arguments.get("period", "1y")
    result = {"ticker": ticker.upper()}

    if "quote" in types:
        result["quote"] = await provider.get_quote(ticker)

    if "bars_daily" in types:
        bars = await provider.get_bars(ticker, period)
        result["bars_daily"] = bars[-30:]  # Limiter à 30 jours pour le contexte
        result["bars_count"] = len(bars)

    if "technicals" in types:
        bars = await provider.get_bars(ticker, period)
        result["technicals"] = calculate_technicals(bars)

    if "fundamentals" in types:
        result["fundamentals"] = await provider.get_fundamentals(ticker)

    if "earnings" in types:
        result["earnings"] = await provider.get_earnings(ticker)

    return result

Calculer les indicateurs techniques

Plutôt que de dépendre d'une API externe pour les indicateurs, on les calcule côté serveur à partir des barres OHLCV. Ça nous donne le contrôle total et réduit les dépendances.

RSI — Relative Strength Index

def calculate_rsi(closes: list[float], period: int = 14) -> float:
    """RSI de Wilder — moyenne exponentielle des gains/pertes."""
    if len(closes) < period + 1:
        return 50.0  # Pas assez de données

    deltas = [closes[i] - closes[i-1] for i in range(1, len(closes))]
    gains = [d if d > 0 else 0 for d in deltas]
    losses = [-d if d < 0 else 0 for d in deltas]

    # Première moyenne = simple
    avg_gain = sum(gains[:period]) / period
    avg_loss = sum(losses[:period]) / period

    # Ensuite = exponentielle (Wilder)
    for i in range(period, len(gains)):
        avg_gain = (avg_gain * (period - 1) + gains[i]) / period
        avg_loss = (avg_loss * (period - 1) + losses[i]) / period

    if avg_loss == 0:
        return 100.0
    rs = avg_gain / avg_loss
    return round(100 - (100 / (1 + rs)), 2)

EMA — Exponential Moving Average

def calculate_ema(values: list[float], period: int) -> list[float]:
    """EMA avec multiplier = 2/(period+1)."""
    if len(values) < period:
        return []
    multiplier = 2 / (period + 1)
    ema = [sum(values[:period]) / period]  # SMA comme seed
    for v in values[period:]:
        ema.append(v * multiplier + ema[-1] * (1 - multiplier))
    return ema

Le résultat structuré

def calculate_technicals(bars: list[dict]) -> dict:
    closes = [b["close"] for b in bars]
    highs = [b["high"] for b in bars]
    lows = [b["low"] for b in bars]

    ema_20 = calculate_ema(closes, 20)
    ema_50 = calculate_ema(closes, 50)
    ema_200 = calculate_ema(closes, 200)

    return {
        "rsi_14": calculate_rsi(closes),
        "ema_20": round(ema_20[-1], 2) if ema_20 else None,
        "ema_50": round(ema_50[-1], 2) if ema_50 else None,
        "ema_200": round(ema_200[-1], 2) if ema_200 else None,
        "ema_20_slope": slope(ema_20, 5),  # pente sur 5 jours
        "atr_14": calculate_atr(highs, lows, closes, 14),
        "macd": calculate_macd(closes),
        "bollinger": calculate_bollinger(closes, 20),
        "week_52_high": max(highs[-252:]),
        "week_52_low": min(lows[-252:]),
        "distance_from_52w_high_pct": round(
            (closes[-1] / max(highs[-252:]) - 1) * 100, 2
        )
    }

Fondamentaux et earnings

Les données fondamentales sont les plus complexes à structurer. Chaque LLM interprète mieux quand le JSON est plat et explicite (pas de nesting profond) :

# Résultat idéal pour les fondamentaux
{
    "pe_trailing": 28.4,
    "pe_forward": 24.1,
    "ev_ebitda": 21.3,
    "ps_ratio": 7.2,
    "pb_ratio": 42.1,
    "gross_margin_pct": 46.2,
    "operating_margin_pct": 31.5,
    "net_margin_pct": 25.3,
    "roe_pct": 157.0,
    "debt_to_equity": 1.87,
    "current_ratio": 0.99,
    "free_cash_flow_b": 110.5,
    "dividend_yield_pct": 0.44,
    "payout_ratio_pct": 15.7,
    "revenue_growth_yoy_pct": 4.9,
    "sector": "Technology",
    "industry": "Consumer Electronics",
    "employees": 161000,
    "source": "yahoo_finance"
}

RunScreener : le DSL de filtrage

Le screening est la fonctionnalité qui rend un MCP finance vraiment puissant. Le LLM peut formuler des critères complexes en langage naturel, et votre outil les traduit en filtres :

SCREENER_TOOL = Tool(
    name="RunScreener",
    description="""Screen the US stock universe with custom filters.
Filters use a simple DSL: 'field operator value [AND field operator value]...'

Available fields:
- price, volume, market_cap, sector, industry
- rsi_14, ema_20_slope, distance_from_52w_high_pct
- pe_trailing, pe_forward, ev_ebitda, ps_ratio
- gross_margin_pct, net_margin_pct, roe_pct
- revenue_growth_yoy_pct, debt_to_equity

Operators: >, <, >=, <=, =, !=, BETWEEN, IN

Examples:
- 'rsi_14 < 30 AND volume > 1000000 AND market_cap > 2000000000'
- 'sector IN (Technology, Healthcare) AND pe_forward < 20'
- 'distance_from_52w_high_pct BETWEEN -15 AND -5'

Returns top 50 matches sorted by match score.""",
    inputSchema={
        "type": "object",
        "properties": {
            "filters": {
                "type": "string",
                "description": "Filter expression in DSL format"
            },
            "sort_by": {
                "type": "string",
                "description": "Field to sort results by (default: market_cap)"
            },
            "limit": {
                "type": "integer",
                "description": "Max results (default 50, max 200)"
            }
        },
        "required": ["filters"]
    }
)

Parser le DSL

import re

def parse_filter(expression: str) -> list[dict]:
    """Parse 'rsi_14 < 30 AND volume > 1M' → liste de conditions."""
    conditions = []
    parts = re.split(r'\s+AND\s+', expression, flags=re.IGNORECASE)

    for part in parts:
        # Handle BETWEEN
        m = re.match(r'(\w+)\s+BETWEEN\s+(.+?)\s+AND\s+(.+)', part, re.I)
        if m:
            conditions.append({
                "field": m.group(1),
                "op": "between",
                "low": parse_value(m.group(2)),
                "high": parse_value(m.group(3))
            })
            continue

        # Handle IN
        m = re.match(r'(\w+)\s+IN\s+\((.+?)\)', part, re.I)
        if m:
            values = [v.strip().strip("'") for v in m.group(2).split(",")]
            conditions.append({"field": m.group(1), "op": "in", "values": values})
            continue

        # Handle comparison operators
        m = re.match(r'(\w+)\s*(>=|<=|!=|>|<|=)\s*(.+)', part)
        if m:
            conditions.append({
                "field": m.group(1),
                "op": m.group(2),
                "value": parse_value(m.group(3).strip())
            })

    return conditions

GetMarketOverview : le snapshot global

GetMarketOverview est l'outil « sans paramètre » qui retourne un snapshot complet du marché. Le LLM l'appelle au début de chaque briefing pour calibrer le ton et le contexte :

# Résultat GetMarketOverview
{
    "market_us": {
        "sp500": {"price": 5480.21, "change_pct": 0.87},
        "nasdaq": {"price": 17892.45, "change_pct": 1.23},
        "dow": {"price": 42150.30, "change_pct": 0.45},
        "russell2000": {"price": 2180.55, "change_pct": -0.32}
    },
    "volatility": {
        "vix": 14.2,
        "vix_change_pct": -3.5,
        "regime": "risk_on"
    },
    "sectors": {
        "technology": 1.45, "healthcare": 0.82,
        "energy": -0.95, "financials": 0.67
    },
    "commodities": {
        "gold": {"price": 2345.80, "change_pct": -0.12},
        "oil_wti": {"price": 78.45, "change_pct": -1.30}
    },
    "crypto": {
        "btc": {"price": 67850, "change_pct_24h": 2.1},
        "eth": {"price": 3650, "change_pct_24h": 1.8}
    },
    "forex": {
        "eur_usd": 1.0845, "usd_jpy": 157.25
    },
    "bonds": {
        "us_10y": 4.28, "us_2y": 4.72,
        "spread_2_10": -0.44
    },
    "as_of": "2026-06-20T20:00:00Z",
    "source": "yahoo_finance+binance"
}

Gestion d'erreurs : ne jamais mentir

La règle #1 d'un serveur MCP finance : ne jamais retourner de données fausses. Si Yahoo est down, on dit « Yahoo est down ». On n'invente pas un prix.

async def call_tool(name: str, arguments: dict) -> CallToolResult:
    try:
        if name == "QueryData":
            result = await handle_query_data(provider, arguments)
            return CallToolResult(
                content=[TextContent(type="text", text=json.dumps(result))]
            )
    except httpx.TimeoutException:
        # Source de données timeout — le dire clairement
        return CallToolResult(
            content=[TextContent(
                type="text",
                text=json.dumps({
                    "error": "data_source_timeout",
                    "message": "Yahoo Finance did not respond within 10s",
                    "suggestion": "Try again in a few seconds or use a different ticker"
                })
            )],
            isError=True
        )
    except KeyError as e:
        # Ticker non trouvé
        return CallToolResult(
            content=[TextContent(
                type="text",
                text=json.dumps({
                    "error": "ticker_not_found",
                    "message": f"No data found for ticker '{arguments.get('ticker')}'",
                    "suggestion": "Check the ticker symbol — use AAPL, not Apple"
                })
            )],
            isError=True
        )