#!/usr/bin/env python3 # ===================================================================== # Kresmion MCP server - standalone single-file build (stdio transport). # # Exposes Kresmion's read-only market-intelligence API to AI agents as the # same MCP tools the hosted server serves: prediction markets, on-chain crypto # flows, ETF flows, derivatives, equity/insider/congress/13F signals, short # interest, offerings, peer relative value, the macro regime, central banks, # money markets, Fed liquidity, Treasury curve and FRED history, and the # economic surprise index. Data is descriptive and event-framed. It is NOT # financial advice. # # GENERATED TOOLS: every tool and the server instructions below are copied # from the hosted server (backend/app/api/mcp_app.py) by # sdk/mcp/generate_stdio.py. Edit the hosted module and regenerate; do not # edit a tool here. # # INSTALL # Option 1 (no install, recommended): # uv run --with mcp --with httpx kresmion_mcp.py # Option 2 (pip): # pip install "mcp>=1.28" "httpx>=0.27" # python kresmion_mcp.py # # GET A KEY # Create a free Kresmion API key (krm_...) at https://kresmion.com/developers # and expose it to the server as the KRESMION_API_KEY environment variable. # # CLAUDE CODE / CLAUDE DESKTOP CONFIG (add to the mcpServers block) # { # "mcpServers": { # "kresmion": { # "command": "uv", # "args": ["run", "--with", "mcp", "--with", "httpx", # "/absolute/path/to/kresmion_mcp.py"], # "env": { "KRESMION_API_KEY": "krm_your_key_here" } # } # } # } # # Or, if you installed the deps with pip: # { # "mcpServers": { # "kresmion": { # "command": "python", # "args": ["/absolute/path/to/kresmion_mcp.py"], # "env": { "KRESMION_API_KEY": "krm_your_key_here" } # } # } # } # # The hosted / remote variant of this server reads each caller's key from the # request's Authorization header instead of the environment; this local file # uses KRESMION_API_KEY. Both wrap the same public https://kresmion.com/api/v1 # surface, so the same key and quota apply. # ===================================================================== """Kresmion MCP server (standalone stdio build).""" from __future__ import annotations import os from typing import Annotated, Any from urllib.parse import quote import httpx from pydantic import Field from mcp.types import Icon, ToolAnnotations from mcp.server.fastmcp import Context, FastMCP __all__ = ["create_mcp", "mcp"] # --------------------------------------------------------------------------- # Upstream configuration # --------------------------------------------------------------------------- # Standalone build talks to production: {base}/v1/... = https://kresmion.com/api/v1/... # Override with KRESMION_MCP_UPSTREAM if you run against a different host. _DEFAULT_UPSTREAM = os.environ.get("KRESMION_MCP_UPSTREAM", "https://kresmion.com/api") # 25s keeps a slow upstream from hanging an agent's tool call indefinitely. _TIMEOUT = httpx.Timeout(25.0, connect=10.0) _LIMITS = httpx.Limits(max_connections=20, max_keepalive_connections=10) # One pooled AsyncClient per base URL, created lazily on first use. _clients: dict[str, httpx.AsyncClient] = {} def _client(base: str) -> httpx.AsyncClient: c = _clients.get(base) if c is None or c.is_closed: c = httpx.AsyncClient( base_url=base, timeout=_TIMEOUT, limits=_LIMITS, headers={"User-Agent": "kresmion-mcp/1.0"}, ) _clients[base] = c return c # --------------------------------------------------------------------------- # Auth resolution + HTTP plumbing # --------------------------------------------------------------------------- def _resolve_api_key(ctx: Context | None) -> str | None: """Resolve the Kresmion API key: per-request header first, then env var. Header path (remote / mounted streamable-HTTP): the SDK threads the Starlette Request onto the Context, so we read ``Authorization: Bearer krm_...`` off the current request. Env path (local / stdio): ``KRESMION_API_KEY``. Returns None if neither is present.""" if ctx is not None: try: rc = ctx.request_context except (ValueError, LookupError): rc = None req = getattr(rc, "request", None) if rc is not None else None if req is not None: auth = req.headers.get("authorization") if auth: token = auth[7:].strip() if auth.lower().startswith("bearer ") else auth.strip() if token: return token key = os.environ.get("KRESMION_API_KEY") return key.strip() if key else None _MISSING_KEY = { "error": "missing_api_key", "hint": ( "No Kresmion API key available for this call. When using the hosted MCP " "server, send 'Authorization: Bearer krm_...' with the request. When " "running the local stdio server, set the KRESMION_API_KEY environment " "variable. Create a free key at https://kresmion.com/developers ." ), } def _clean(params: dict[str, Any]) -> dict[str, Any]: """Drop None-valued params so they are omitted from the query string.""" return {k: v for k, v in params.items() if v is not None} async def _request( base: str, path: str, params: dict[str, Any], key: str, ) -> dict[str, Any]: """One thin GET against the v1 surface. Every tool is a read. Returns the decoded envelope dict, or a structured error dict (never raises) so the agent can self-correct.""" headers = {"Authorization": f"Bearer {key}", "Accept": "application/json"} try: resp = await _client(base).get(path, params=params, headers=headers) except httpx.RequestError as exc: return { "error": "upstream_unreachable", "status": None, "retry_after": None, "detail": f"Could not reach the Kresmion API ({exc.__class__.__name__}): {exc}", } if resp.status_code >= 400: detail: Any = None try: body = resp.json() detail = body.get("detail") if isinstance(body, dict) else body except ValueError: detail = (resp.text or "")[:400] or None code = { 401: "unauthorized", 403: "forbidden", 404: "not_found", 422: "invalid_request", 429: "rate_limited", }.get(resp.status_code, "http_error") return { "error": code, "status": resp.status_code, "retry_after": resp.headers.get("retry-after"), "detail": detail, } try: return resp.json() except ValueError: return { "error": "bad_response", "status": resp.status_code, "retry_after": None, "detail": "Upstream returned a non-JSON body.", } async def _call( base: str, path: str, ctx: Context | None, params: dict[str, Any] | None = None, ) -> dict[str, Any]: key = _resolve_api_key(ctx) if not key: return dict(_MISSING_KEY) return await _request(base, path, _clean(params or {}), key) _SERVER_INSTRUCTIONS = ( "Kresmion is a financial-intelligence platform. CALL kresmion_status FIRST: " "it reports the freshness of every product family, so you know which " "numbers are current before you quote them. The tools return read-only, " "sourced market data: prediction markets (Polymarket only: search, " "detail, history, order book, execution cost, movers, calibration, " "upcoming closes), on-chain whale transfers, spot crypto-ETF flows (BTC " "and ETH session-day flows are withheld for licensing; SOL flows and " "fund AUM are served), perpetual derivatives, cross-exchange funding and " "liquidations, equity/insider/congress/13F signals, SEC Form 4 insider " "clusters, House congressional trade disclosures, FINRA short interest " "and short volume, modeled dealer gamma, per-ticker smart-money scores, " "the cross-asset macro regime, central bank policy rates, BIS statistics, " "CFTC Commitments of Traders, Treasury TIC foreign holdings, sovereign 10Y " "spreads vs the Bund and US Treasuries (a Kresmion calculation, not a " "credit rating), US money-market " "rates and the Fed corridor, Fed liquidity, the Treasury curve through " "history, curated FRED series history, the economic surprise index, IPOs " "and equity offerings from EDGAR, peer relative value, a published signal " "track record, cross-family confluence composites, and a delta feed for " "efficient polling. " "THE ECONOMIC CALENDAR IS NOT AVAILABLE: per-event rows, scheduled times " "and consensus figures are not served by any tool here, and only the " "derived economic surprise index crosses that boundary. " "Every tool is read-only: nothing here writes, pays or changes state. " "The data is event-framed and descriptive: it reports what happened " "with structural and historical context. It is NOT investment advice and " "contains no buy/sell recommendations. Every response is an envelope with " "'data', 'count' (lists), and 'as_of' (freshness); always check 'as_of' " "before acting on a number. Authenticate with a Kresmion API key " "(Authorization: Bearer krm_..., or the KRESMION_API_KEY env var for the local " "server)." ) # --------------------------------------------------------------------------- # Factory # --------------------------------------------------------------------------- _READ_ONLY_TOOL = ToolAnnotations( readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False) def create_mcp(base_url: str | None = None) -> FastMCP: """Build the configured Kresmion FastMCP server. ``base_url`` is the upstream root the tools call ("{base}/v1/..."); it defaults to KRESMION_MCP_UPSTREAM, then https://kresmion.com/api. Tools close over the resolved base.""" base = (base_url or _DEFAULT_UPSTREAM).rstrip("/") server = FastMCP( name="kresmion", instructions=_SERVER_INSTRUCTIONS, # Smithery/registry quality metadata: served in the initialize response. website_url="https://kresmion.com/developers", icons=[Icon(src="https://kresmion.com/favicon.svg", mimeType="image/svg+xml", sizes=["any"])], ) # ----- flagship: one-call cross-product digest --------------------------- @server.tool(annotations=_READ_ONLY_TOOL) async def kresmion_brief( ctx: Context, asset_class: Annotated[str, Field(pattern="^(crypto|equity)$", description="crypto or equity: which product stack to assemble the digest from.")], symbol: Annotated[str, Field(max_length=32, description="Symbol to profile: BTC / ETH / SOL for crypto, or a ticker like NVDA for equity.")], ) -> dict[str, Any]: """START HERE: one call returns the full cross-product Kresmion digest for a symbol. This is the flagship entry point. Instead of stitching together five or six separate tools, ``kresmion_brief`` assembles a single compound digest for one symbol from every relevant Kresmion product family at once: * crypto (asset_class='crypto', symbol BTC/ETH/SOL): perpetual derivatives + cross-exchange funding, recent liquidations, on-chain whale transfers, spot-ETF AUM and flows (BTC/ETH session flows are withheld, total_kind 'withheld', source not licensed), related prediction markets, and the prevailing macro regime. * equity (asset_class='equity', symbol a ticker): the equity signal feed, insider + congressional trades, 13F institutional positioning, modeled dealer gamma, and related prediction markets. Each sub-section carries its own ``as_of`` so you can see which legs are fresh and which are stale. It is a descriptive, event-framed digest with structural + historical context, never buy/sell advice. Reach for the family-specific tools (crypto_derivatives, crypto_funding, equity_gamma, prediction_markets, ...) only when you need to drill deeper than the digest returns.""" return await _call(base, f"/v1/brief/{quote(asset_class, safe='')}/{quote(symbol, safe='')}", ctx) # ----- prediction markets ------------------------------------------------ @server.tool(annotations=_READ_ONLY_TOOL) async def prediction_markets( ctx: Context, category: Annotated[str | None, Field(description="Exact category, case-insensitive, e.g. 'crypto', 'politics', 'economy'.")] = None, asset_class: Annotated[str | None, Field(description="Kresmion asset-class tag: crypto, equity, macro, forex, commodity.")] = None, q: Annotated[str | None, Field(max_length=80, description="Free-text match on the market question / event title.")] = None, min_volume: Annotated[float, Field(ge=0, description="Minimum lifetime USD volume.")] = 1000, limit: Annotated[int, Field(ge=1, le=200, description="Maximum rows to return.")] = 50, ) -> dict[str, Any]: """Search and filter LIVE prediction markets (Polymarket), volume-ranked. Reach for this to see what the crowd is currently pricing on a topic: each row carries the current YES probability, 24h/7d point change, USD volume, liquidity, event grouping, and the canonical Polymarket URL. Filter by category, asset class, a free-text question search, and a minimum volume floor. Only ACTIVE (unresolved, in-window) markets are returned; settled outcomes are a separate concept (calibration and resolutions live behind other tools). Probabilities are crowd prices, not forecasts or advice.""" return await _call(base, "/v1/prediction/markets", ctx, { "category": category, "asset_class": asset_class, "q": q, "min_volume": min_volume, "limit": limit, }) @server.tool(annotations=_READ_ONLY_TOOL) async def prediction_market_detail( ctx: Context, market_id: Annotated[str, Field(description="Polymarket market id (from prediction_markets rows).")], ) -> dict[str, Any]: """Full detail for ONE prediction market. Returns the Polymarket market row plus its sibling outcomes within the same event and the 30-day probability delta. Use it after prediction_markets to drill into a single question. The REST API also exposes detected algorithmic-order-flow flags for a market at ``GET /v1/prediction/markets/{id}/algo-flags`` (reach it via the Kresmion SDK's ``k.prediction.algo_flags(id)`` or the raw API) when you want to know whether a market's tape shows machine-driven flow.""" return await _call(base, f"/v1/prediction/markets/{quote(market_id, safe='')}", ctx) @server.tool(annotations=_READ_ONLY_TOOL) async def prediction_market_history( ctx: Context, market_id: Annotated[str, Field(description="Polymarket market id.")], hours: Annotated[int | None, Field(ge=1, le=8760, description="Look-back window in hours (mutually exclusive with days).")] = None, days: Annotated[int | None, Field(ge=1, le=365, description="Look-back window in days (mutually exclusive with hours).")] = None, limit: Annotated[int, Field(ge=1, le=2000, description="Maximum history points to return, oldest first.")] = 500, ) -> dict[str, Any]: """Time series of YES probability + volume for one market, oldest-first. Reach for this to chart how a market's odds moved, or to detect a repricing trend rather than a single snapshot. Bound the window with EITHER ``hours`` OR ``days`` (not both); the most recent points inside that window are returned, capped by ``limit``.""" return await _call(base, f"/v1/prediction/markets/{quote(market_id, safe='')}/history", ctx, { "hours": hours, "days": days, "limit": limit, }) @server.tool(annotations=_READ_ONLY_TOOL) async def prediction_order_book( ctx: Context, market_id: Annotated[str, Field(description="Polymarket market id.")], ) -> dict[str, Any]: """Live CLOB order-book microstructure for one market. Returns bid/ask depth, spread, and slippage-to-size for standard notionals. Reach for this to gauge how much size a market can absorb and what a fill would cost before treating a quoted probability as tradeable. The book is fetched live and cached ~60s server-side, so ``as_of`` is the fetch time. It describes the book; it never places an order. For how depth and spread EVOLVED (not just the current snapshot), the REST API exposes an order-book history at ``GET /v1/prediction/markets/{id}/book/history`` (SDK: ``k.prediction.book_history(id, hours=24)``). To model the fill cost of a specific USDC size, use the prediction_execution_cost tool.""" return await _call(base, f"/v1/prediction/markets/{quote(market_id, safe='')}/book", ctx) @server.tool(annotations=_READ_ONLY_TOOL) async def prediction_movers( ctx: Context, window: Annotated[str, Field(pattern="^(24h|7d)$", description="Look-back window: 24h or 7d.")] = "24h", min_volume: Annotated[float, Field(ge=0, description="Minimum lifetime USD volume.")] = 10000, include_sports: Annotated[bool, Field(description="Include sports markets (default off; finished games swing 0->100 and drown real repricing).")] = False, limit: Annotated[int, Field(ge=1, le=200, description="Maximum rows to return.")] = 30, ) -> dict[str, Any]: """Prediction markets with the biggest probability moves over 24h or 7d. Reach for this to find where the crowd changed its mind fastest, ranked by absolute probability move (gamma-native deltas). Sports are excluded by default because settled games produce meaningless 0->100 swings; opt in with ``include_sports`` only if you want them.""" return await _call(base, "/v1/prediction/movers", ctx, { "window": window, "min_volume": min_volume, "include_sports": include_sports, "limit": limit, }) # Phase 642b — prediction_divergence (Polymarket vs Kalshi pairs) was # REMOVED with /v1/prediction/divergence: Kalshi's terms bar # redistribution, and a cross-venue spread cannot stand on one venue. @server.tool(annotations=_READ_ONLY_TOOL) async def prediction_calibration(ctx: Context) -> dict[str, Any]: """How well-calibrated the prediction crowd has been (historical). Returns the Polymarket probability ~7 days before resolution bucketed against the actual outcome, an overall Brier score (0 = perfect, 0.25 = a coin flip), per-category Brier, and longshot / near-certain tail bias. Reach for this to judge how much weight to put on crowd probabilities in a given category. Historical framing only; it is not a forecast.""" return await _call(base, "/v1/prediction/calibration", ctx) @server.tool(annotations=_READ_ONLY_TOOL) async def prediction_execution_cost( ctx: Context, market_id: Annotated[str, Field(description="Polymarket market id.")], notionals: Annotated[list[float] | None, Field(description="USD sizes to price the fill for (default 100 / 1000 / 10000). Each is walked against one side of the live book.")] = None, ) -> dict[str, Any]: """Modeled slippage / cost to FILL given USDC sizes in one market. Walks the STORED top-5 order-book snapshots (no live CLOB call) to estimate the fill VWAP and slippage (in bps off mid) for each requested notional on both the buy and sell side, plus a window median / p90 summary, so you can see how far price moves against you as size grows and whether a divergence or mispriced-looking probability is actually tradeable. Where stored depth cannot cover a size the point is depth-exhausted (never extrapolated). CAVEAT: this is a ONE-LEG execution cost: taking one side of the book, NOT a round trip, and NOT inclusive of fees or of impact beyond the stored top-5 depth. Book coverage is the top ~150 live markets by 24h volume only, and the snapshot history is shallow (collection began 2026-07-16). It describes a hypothetical fill; it never places an order.""" return await _call(base, f"/v1/prediction/markets/{quote(market_id, safe='')}/execution-cost", ctx, { "notional": notionals if notionals else [100, 1000, 10000], }) @server.tool(annotations=_READ_ONLY_TOOL) async def prediction_calendar( ctx: Context, days: Annotated[int, Field(ge=1, le=60, description="Look-ahead window in days (max 60).")] = 14, ) -> dict[str, Any]: """Upcoming Polymarket market closes, grouped by date. Returns the Polymarket markets scheduled to close inside the window, grouped by calendar date, so you can see what the crowd is about to be graded on: earnings dates, elections, economic releases, expiries. Only still-contested markets (0.05 < YES < 0.95) are shown, and each day carries a ``contested_capital`` weight that up-weights liquidity resting on genuinely uncertain (~50/50) markets. CAVEAT: the date is the SCHEDULED close, NOT the actual resolution time. A market can settle early, resolve late, or have its end date pushed by the venue, so read the date as the planned close, not a guarantee. Descriptive schedule, not advice.""" return await _call(base, "/v1/prediction/calendar", ctx, {"days": days}) # Phase 642b — pm_options_divergence was REMOVED: its options leg is # built from Deribit option marks, and Deribit's terms bar publishing # derived data beyond personal use. /v1/prediction/options-divergence is # unmounted for the same reason. # ----- crypto ------------------------------------------------------------ @server.tool(annotations=_READ_ONLY_TOOL) async def crypto_whale_transfers( ctx: Context, blockchain: Annotated[str | None, Field(description="ETH | BTC | SOL. Omit for all chains.")] = None, direction: Annotated[str | None, Field(description="exchange_inflow | exchange_outflow | whale_transfer | exchange_transfer.")] = None, min_usd: Annotated[float, Field(ge=100000, description="USD size floor (hard minimum $100k).")] = 100000, hours: Annotated[int, Field(ge=1, le=720, description="Look-back window in hours (<= 720h / 30d).")] = 168, limit: Annotated[int, Field(ge=1, le=200, description="Maximum rows to return.")] = 50, ) -> dict[str, Any]: """Recent large on-chain transfers (whale / exchange-wallet flows, >= $100k). Reach for this to see notable on-chain movement: large deposits to and withdrawals from exchange wallets and whale-to-whale transfers, with USD sizing and labels. Filter by chain, direction, size, and a look-back window. IMPORTANT CAVEAT: do NOT read BTC exchange netflow from this feed. BTC rows are dominated by exchange-INTERNAL (self) transfers (hot/cold wallet reshuffles), so BTC inflow/outflow is an artifact, not genuine flow. Genuine, tradeable flow signal lives in the ETH and stablecoin rows. For scheduled SUPPLY rather than individual transfers, the REST API also exposes the upcoming token-unlock (vesting cliff) calendar at ``GET /v1/crypto/unlocks`` (SDK: ``k.crypto.unlocks(days=90)``: scheduled supply hitting the market).""" return await _call(base, "/v1/crypto/whales", ctx, { "blockchain": blockchain, "direction": direction, "min_usd": min_usd, "hours": hours, "limit": limit, }) @server.tool(annotations=_READ_ONLY_TOOL) async def crypto_etf_flows( ctx: Context, asset: Annotated[str, Field(pattern="^(BTC|ETH|SOL|ALL)$", description="BTC, ETH, SOL, or ALL for every asset.")] = "BTC", days: Annotated[int, Field(ge=1, le=365, description="Look-back window in days.")] = 30, period: Annotated[str | None, Field(pattern="^(1d|7d|30d)$", description="Optional headline aggregation window: 1d, 7d, or 30d.")] = None, ) -> dict[str, Any]: """Spot crypto-ETF net flows: aggregate + per-product breakdown. Reach for this to see whether money is flowing into or out of the SOL spot ETFs, and for the fund list and AUM of the BTC / ETH / SOL spot ETFs. SOL flows are derived from each fund's AUM/NAV change. BTC AND ETH SESSION FLOWS ARE WITHHELD. Farside Investors is not used by this API, and the only per-fund flow figures Kresmion holds for BTC/ETH trading sessions come from that compiler, which grants no redistribution licence. So every BTC/ETH NYSE-session day comes back with ``total_kind`` = ``withheld``, null flow values and ``withheld_reason`` = ``source_not_licensed``, and the window sums, the 7-day headline, ``latest_session`` and per-fund flows built from them are null. Days with no NYSE session are a true zero. Do not describe a withheld day as zero flow, and do not estimate one. Coverage caveat for served days: ``total_kind`` is ``total`` only when every fund in the registry is known; otherwise it is ``reported_subtotal``, the day's ``net_flow_usd`` is null and ``reported_subtotal_usd`` carries the partial sum with ``funds_known``/``funds_active``/``aum_coverage``/``missing_tickers``. Cite a subtotal only as "reported flows for N of M funds", never as the market total. Descriptive flow data, not a signal to buy or sell.""" return await _call(base, "/v1/crypto/etf/flows", ctx, { "asset": asset, "days": days, "period": period, }) @server.tool(annotations=_READ_ONLY_TOOL) async def crypto_derivatives( ctx: Context, symbol: Annotated[str, Field(description="BTC, ETH, or SOL.")], exchange: Annotated[str, Field(description="Exchange name (default Binance). Pass '' for the freshest row across exchanges.")] = "Binance", ) -> dict[str, Any]: """Latest perpetual-derivatives snapshot for a crypto symbol. Returns funding rate, open interest, long/short ratio, basis, and mark price for BTC / ETH / SOL. Reach for this to read positioning and leverage pressure (e.g. rich funding = crowded longs). Defaults to Binance; pass an empty ``exchange`` for the freshest row across venues. Descriptive market state, not advice.""" return await _call(base, f"/v1/crypto/derivatives/{quote(symbol, safe='')}", ctx, { "exchange": exchange, }) @server.tool(annotations=_READ_ONLY_TOOL) async def crypto_funding( ctx: Context, symbol: Annotated[str | None, Field(description="Optional BTC | ETH | SOL. Omit for the cross-exchange funding SURFACE (latest funding per symbol across venues); pass a symbol for its funding-rate HISTORY.")] = None, days: Annotated[int, Field(ge=1, le=90, description="History look-back in days (symbol only; retention is 90 days).")] = 30, exchange: Annotated[str | None, Field(description="History only: restrict to one venue, Binance | Bybit | OKX | Hyperliquid. Omit for every venue.")] = None, ) -> dict[str, Any]: """Cross-exchange perpetual FUNDING: the surface, or one symbol's history. With no ``symbol`` this returns the funding surface: the latest funding rate for each tracked symbol across every covered exchange, so you can see where funding is richest / cheapest and how the venues disagree. With a ``symbol`` it returns that symbol's funding-rate history for trend reading (persistently positive funding = crowded longs paying to hold; negative = crowded shorts). Funding is a positioning / leverage read, descriptive only, not advice. For the fuller perp snapshot (open interest, long/short, basis, mark) use crypto_derivatives; for forced deleveraging use crypto_liquidations. Example: symbol='ETH', days=14, exchange='OKX' returns two weeks of ETH funding prints from OKX only.""" if symbol: return await _call(base, f"/v1/crypto/funding/{quote(symbol, safe='')}", ctx, { "days": days, "exchange": exchange, }) return await _call(base, "/v1/crypto/funding", ctx) @server.tool(annotations=_READ_ONLY_TOOL) async def crypto_liquidations( ctx: Context, symbol: Annotated[str | None, Field(description="BTC | ETH | SOL | XRP | DOGE. Omit for all tracked symbols.")] = None, hours: Annotated[int, Field(ge=1, le=168, description="Look-back window in hours (<= 168h / 7d).")] = 24, ) -> dict[str, Any]: """Recent perpetual LIQUIDATIONS (forced deleveraging), long vs short. Returns liquidated notional over the window, split into long- and short-side liquidations. A spike marks a forced-deleveraging cascade (longs blown out on a drop, shorts on a squeeze), which often marks a local capitulation. Filter by symbol and a look-back window. COVERAGE CAVEAT: liquidation data is SINGLE-VENUE (OKX), over a small starter symbol set (BTC, ETH, SOL, XRP, DOGE). It is a single-venue sample, so read it as a representative proxy for market-wide forced flow, NOT a total; ``meta.coverage`` reports the venues + symbols actually seen and ``meta.cascades`` sums liquidated USD by side over the trailing 1h / 4h. Descriptive market state, not advice.""" return await _call(base, "/v1/crypto/liquidations", ctx, { "symbol": symbol, "hours": hours, }) # ----- equities ---------------------------------------------------------- @server.tool(annotations=_READ_ONLY_TOOL) async def equity_signals( ctx: Context, signal_type: Annotated[str | None, Field(max_length=64, description="Exact signal type filter, e.g. 'insider_cluster_sell'.")] = None, severity: Annotated[str | None, Field(description="critical | high | medium | low.")] = None, days: Annotated[int, Field(ge=1, le=365, description="Look-back window in days.")] = 7, limit: Annotated[int, Field(ge=1, le=200, description="Maximum rows to return.")] = 50, ) -> dict[str, Any]: """Cross-source equity signals from the Kresmion signal engine. Reach for this to see recent notable equity events: SEC filings, insider and congressional trades, technical breaks, credit and jobs signals, each with a symbol, a machine ``direction`` tag (bull/bear/neutral), a severity, a strength score, and a sourced headline. Filter by signal_type, severity, and a look-back window. ``direction`` is an event tag, NOT advice: it describes the nature of the event, not a recommendation.""" return await _call(base, "/v1/equities/signals", ctx, { "signal_type": signal_type, "severity": severity, "days": days, "limit": limit, }) @server.tool(annotations=_READ_ONLY_TOOL) async def institutional_holdings( ctx: Context, fund_cik: Annotated[str | None, Field(description="SEC CIK of the filing fund.")] = None, ticker: Annotated[str | None, Field(max_length=12, description="Resolve holdings of one ticker.")] = None, quarter: Annotated[str | None, Field(description="Report quarter, e.g. '2025Q4'.")] = None, limit: Annotated[int, Field(ge=1, le=500, description="Maximum holdings rows to return.")] = 50, ) -> dict[str, Any]: """SEC 13F-HR quarterly institutional holdings for tracked funds. Reach for this to see what large funds hold and how positions changed quarter over quarter (``qoq_signal``), sorted by position market value (in dollars). Filter by fund CIK, ticker, or quarter. CAVEAT: ticker resolves for only ~50-93% of rows because there is no free CUSIP->ticker map, so bonds, warrants, and some equities come back with ticker=null. Filter on ``cusip`` for completeness. Disclosure lags the quarter end (13F filings arrive up to ~45 days after quarter close).""" return await _call(base, "/v1/equities/institutional", ctx, { "fund_cik": fund_cik, "ticker": ticker, "quarter": quarter, "limit": limit, }) @server.tool(annotations=_READ_ONLY_TOOL) async def equity_short_interest( ctx: Context, symbol: Annotated[str | None, Field(max_length=12, description="One ticker's settlement history. Omit for the latest settlement's universe.")] = None, min_pct_float: Annotated[float | None, Field(ge=0, le=100, description="Only names at or above this percentage of float. Narrows to the covered universe, since a name with no float figure cannot satisfy a numeric comparison.")] = None, sort: Annotated[str, Field(description="short_pct_float | short_pct_shares_outstanding | days_to_cover | change_pct | settlement_date.")] = "short_pct_float", limit: Annotated[int, Field(ge=1, le=500, description="Maximum rows to return.")] = 100, ) -> dict[str, Any]: """FINRA consolidated short interest, with a derived percentage of float. Reach for this to answer "how much of the float is short?". This is the bi-monthly exchange short-interest STOCK, not the daily short-sale volume FLOW (that is a different dataset). Each row carries the short position, days-to-cover, the change since the prior settlement, and both percentages. HOW THE FLOAT IS BUILT: shares outstanding from SEC filings, less Section 16 affiliate holdings (officers, directors, ten-percent owners) and affiliate 13D/G blocks. Passive institutions are NOT subtracted - index managers like Vanguard and BlackRock file 13G and their shares ARE part of the float under SEC Rule 12b-2. THREE CAVEATS YOU MUST CARRY INTO ANY ANSWER: 1. ``short_pct_float`` is a LOWER BOUND, not a point estimate. Any affiliate not yet observed stays in the denominator, so the true percentage is at least this high. 2. It is published on only a MINORITY of shorted tickers (read `meta.coverage` on the response for the measured figure - do not quote a count from this text, it moves with every filing), because it needs Section 16 holdings many issuers have not filed since collection began. A NULL means UNKNOWN, never zero - ``float_source`` names the reason (no_affiliate_data, multi_class_unresolved, denominator_invalid, float_non_positive, not_latest_settlement). 3. ``short_pct_shares_outstanding`` is a DIFFERENT measure, always smaller because float is a subset of shares outstanding, and far more widely available. Never present one as the other. ATTRIBUTION, and reproduce it if you quote these figures: Source: FINRA. Short interest data owned by FINRA. (FINRA API ToS 2.3(a); the clause reaches the derived percentage of float too, not just the raw position. Canonical string: app/services/attribution.py.) ``days_to_cover`` NULL with ``days_to_cover_censored`` true means FINRA returned its 999.99 CEILING, and that covers TWO different facts: a ratio genuinely above the ceiling, and one that is UNDEFINED because average daily volume is zero. Measured at the time of writing, most sentinel rows were the undefined case. ``avg_daily_volume`` separates them where present; it is NULL on older settlements, where the two are NOT separable. Never describe a flagged row as ">999.99" unless avg_daily_volume is present and positive. Short interest is semi-monthly and ~11 days stale at publication; quote ``settlement_date`` when citing a figure.""" return await _call(base, "/v1/equities/short-interest", ctx, { "symbol": symbol, "min_pct_float": min_pct_float, "sort": sort, "limit": limit, }) @server.tool(annotations=_READ_ONLY_TOOL) async def equity_gamma( ctx: Context, ticker: Annotated[str | None, Field(max_length=12, description="Optional ticker. Omit for the gamma SURFACE (all tracked names); pass a ticker for its detail.")] = None, ) -> dict[str, Any]: """Modeled dealer GAMMA exposure: the surface, or one ticker's detail. With no ``ticker`` this returns the gamma surface across tracked names: each name's net dealer gamma, its gamma-flip level, and whether dealers sit long or short gamma. With a ``ticker`` it returns that name's detail: gamma by strike, the zero-gamma / flip price, and the nearest call/put walls. Long-dealer-gamma names tend to see moves DAMPENED (dealers sell rallies / buy dips to stay hedged); short-gamma names tend to see moves AMPLIFIED. CAVEAT: gamma here is MODELED from the listed options chain plus an assumed dealer-positioning convention: it is an ESTIMATE, not disclosed positioning. Descriptive structure, not advice.""" if ticker: return await _call(base, f"/v1/equities/gamma/{quote(ticker, safe='')}", ctx) return await _call(base, "/v1/equities/gamma", ctx) @server.tool(annotations=_READ_ONLY_TOOL) async def smart_money( ctx: Context, ticker: Annotated[str | None, Field(max_length=12, description="Optional ticker. Omit for the ranked scan; pass a ticker for its live composite + 30-snapshot trend.")] = None, min_families: Annotated[int, Field(ge=1, le=4, description="Minimum contributing legs for a scan row (default 2).")] = 2, direction: Annotated[str, Field(pattern="^(long|short|all)$", description="Filter scan rows by score sign.")] = "all", min_abs_score: Annotated[float, Field(ge=0, le=100, description="Minimum absolute composite score.")] = 0, limit: Annotated[int, Field(ge=1, le=200, description="Maximum scan rows.")] = 50, ) -> dict[str, Any]: """One per-ticker smart-money positioning score, with every leg shown. Collapses four public-filing families into a single -100..+100 composite per ticker: insider trade clusters (weight 0.35), congressional trades (0.20), 13F institutional changes (0.30), and FINRA short-volume trend (0.15). Each row carries the ``legs`` list with per-family direction, weight, contribution, and evidence, so you can judge the score rather than trust it. Absent legs redistribute their weight and are listed with a reason. CAVEATS: 13F data is quarterly with a roughly 45-day lag and congressional disclosures lag 30-45 days (each leg carries its own as_of); the short-volume leg covers only ~29 symbols and may be stale. This is a positioning summary of public filings, not advice. Use ``min_families`` >= 2 to require corroboration.""" if ticker: return await _call(base, f"/v1/equities/smart-money/{quote(ticker, safe='')}", ctx) return await _call(base, "/v1/equities/smart-money", ctx, { "min_families": min_families, "direction": direction, "min_abs_score": min_abs_score, "limit": limit, }) # ----- equities: filings-based feeds (Phase 643A) ------------------------ # # Wrappers over /v1 routes that already existed with no MCP door. Each # upstream was checked for licence: SEC Form 4 (public), House Clerk # periodic transaction reports (public), FINRA Reg SHO daily short sale # volume (FINRA API ToS 2.3(a): owner-and-source notice required). @server.tool(annotations=_READ_ONLY_TOOL) async def insider_clusters( ctx: Context, direction: Annotated[str | None, Field(pattern="^(buy|sell)$", description="buy or sell. Omit for both.")] = None, limit: Annotated[int, Field(ge=1, le=200, description="Maximum clusters to return, strongest first.")] = 50, ) -> dict[str, Any]: """Clusters of company insiders (SEC Form 4) trading the same stock in one window. Reach for this to answer "where are several insiders buying (or selling) their own company's stock at once?". Each row is one ticker and direction from the latest daily snapshot: how many insiders, how many trades, total USD value, whether the CEO or CFO took part (and which senior roles), the share of the insiders' own holdings sold or bought, the window dates and whether the cluster is new today. Ranked by ``strength_score`` (a Kresmion composite, not a recommendation). ``meta.by_direction`` counts buys and sells. Parameters: ``direction`` narrows to buys or sells; ``limit`` caps the rows. Example: direction='buy', limit=20 for the twenty strongest insider-buying clusters. WHAT THIS TOOL LEAVES OUT. Market capitalisation, sector, the cluster's share of market cap and the price change since the window come from a market-data vendor whose terms do not allow redistribution, so this tool removes those fields; ask for them elsewhere. Sells vastly outnumber buys (routine diversification, 10b5-1 plans: read ``has_10b5_1``), so a sell cluster is weaker evidence than a buy cluster of the same size. Form 4 is due two business days after the trade. Source: SEC EDGAR Form 4. Descriptive filings data, never advice.""" env = await _call(base, "/v1/equities/insider-clusters", ctx, { "direction": direction, "limit": limit, }) # Vendor-derived fields never leave through this door (Phase 643A). vendor_fields = ("sector", "market_cap", "pct_of_mcap", "price_since_window_start", "price_since_last_trade", "price_as_of") for row in env.get("data") or []: if isinstance(row, dict): for k in vendor_fields: row.pop(k, None) return env @server.tool(annotations=_READ_ONLY_TOOL) async def congress_trades( ctx: Context, ticker: Annotated[str | None, Field(max_length=12, description="Optional ticker, e.g. NVDA. Omit for every disclosed trade in the window.")] = None, days: Annotated[int, Field(ge=1, le=730, description="Look-back on the TRADE date, in days (max 730).")] = 90, limit: Annotated[int, Field(ge=1, le=200, description="Maximum rows, newest trade first.")] = 50, ) -> dict[str, Any]: """US congressional stock-trade disclosures (House periodic transaction reports). Reach for this to see which members of Congress disclosed trading a stock: member name, ticker, trade type, the disclosed AMOUNT RANGE, the trade date and the disclosure date. ``direction`` is a machine tag from the trade type (purchase = bull, sale = bear), not advice. Parameters: ``ticker`` scopes to one stock; ``days`` is the trade-date window; ``limit`` caps the rows. Example: ticker='NVDA', days=365 for a year of disclosed NVDA trades. READ BEFORE QUOTING. (1) Amounts are RANGES as filed (e.g. "$1,001 - $15,000"); there is no exact dollar figure, so never present a midpoint as the amount. (2) Disclosure lags the trade by up to ~45 days by law, so a fresh disclosure can describe an old trade: cite both dates. (3) Coverage: House periodic transaction reports, collected daily from the Office of the Clerk. Senate filings are not collected on an ongoing basis (older history carries some Senate rows from a one-off backfill), so an absent senator is not evidence of no trade. Descriptive disclosure data, never advice.""" return await _call(base, "/v1/equities/congress", ctx, { "ticker": ticker, "days": days, "limit": limit, }) @server.tool(annotations=_READ_ONLY_TOOL) async def short_volume( ctx: Context, symbol: Annotated[str | None, Field(max_length=12, description="Optional ticker for its own daily history, newest first. Omit for the latest trading day across the whole FINRA file.")] = None, days: Annotated[int, Field(ge=1, le=70, description="History look-back in days (symbol only; ~60 days retained).")] = 30, min_volume: Annotated[int, Field(ge=0, description="Off-exchange volume floor on total_volume. Use 100000 or more for a market-wide ranking, or thin names with a handful of shares dominate.")] = 0, limit: Annotated[int, Field(ge=1, le=1000, description="Rows per page (max 1000).")] = 100, offset: Annotated[int, Field(ge=0, description="Rows to skip; page with limit. meta.has_more says whether another page exists.")] = 0, ) -> dict[str, Any]: """FINRA daily short-sale VOLUME and the short-volume ratio, market wide. Reach for this to see how much of a stock's OFF-EXCHANGE volume was sold short on a given day (short_volume / total_volume), for one symbol through time or for every symbol on the latest trading day ranked by that ratio. Parameters: ``symbol`` switches to one ticker's history; ``days`` bounds that history; ``min_volume`` sets a volume floor for the ranking; ``limit``/``offset`` page it. Example: min_volume=1000000, limit=25 for the 25 highest ratios among names with at least a million shares of reported off-exchange volume. WHAT THIS IS NOT. It is a daily FLOW, NOT short interest (the bi-monthly stock of open short positions: use equity_short_interest for that). It counts only trades reported to a FINRA facility (TRF/ADF: dark pools, wholesalers, internalizers), it includes market-maker hedging, and a high ratio is routine for many names, so it is not a measure of directional positioning. ATTRIBUTION, and reproduce it whenever you quote these figures: Source: FINRA. Reg SHO Daily Short Sale Volume data owned by FINRA. (FINRA API ToS 2.3(a) requires naming FINRA as owner and source, and it reaches the derived ratio too. The response carries it in ``meta.attribution``.) Descriptive market data, never advice.""" return await _call(base, "/v1/equities/short-volume", ctx, { "symbol": symbol, "days": days, "min_volume": min_volume, "limit": limit, "offset": offset, }) # ----- macro ------------------------------------------------------------- @server.tool(annotations=_READ_ONLY_TOOL) async def macro_regime(ctx: Context) -> dict[str, Any]: """The current cross-asset Risk-On/Off macro regime. Returns the latest daily regime score (-3..+3, 5-day smoothed) with the four factor sub-scores (growth, liquidity, risk-appetite, volatility), the per-indicator z-scores, a conviction level, and an ``interpretation`` string. The numbers are deterministic engine output; the interpretation is LLM-written prose whose indicator direction words are fixed by the engine from each z-score's sign and vetoed when they contradict it (a rule-based template replaces a rejected draft). Quote the numbers, not the prose. Reach for this for a one-call read of the market's macro backdrop. ``data_as_of`` is the trading date the score is for; ``as_of`` / ``computed_at`` is when it was computed. Descriptive regime state, not a trade recommendation. For the regime score TIME SERIES (how the backdrop trended, regime transitions), the REST API exposes ``GET /v1/macro/regime/history`` (SDK: ``k.macro.regime_history(days=365)``). For cross-asset correlation BREAKS (pairs whose latest correlation has moved off its 12-month baseline), use ``GET /v1/macro/correlations`` (SDK: ``k.macro.correlations()``).""" return await _call(base, "/v1/macro/regime", ctx) # ----- macro: the Fed, rates and the curve (Phase 564, W7) --------------- # # The audit's §9.5 finding was that "the whole Fed and calendar surface is # app-only today, invisible to API and MCP consumers". These six tools are # the MCP door for the weekend's builds. Each description says what the # tool is NOT, because the failure mode for an agent is confident # over-reach, not a missing number. # # THE CALENDAR IS DELIBERATELY ABSENT. Its only live source is a # prohibited scrape, so the only thing derived from it that crosses this # boundary is the aggregate index in economic_surprise. @server.tool(annotations=_READ_ONLY_TOOL) async def central_banks( ctx: Context, view: Annotated[str, Field(pattern="^(rates|cycle|moves)$", description="rates = the current policy rate of every covered jurisdiction; cycle = who is hiking/cutting over a window; moves = the individual changes inside the daily window.")] = "rates", window_months: Annotated[int, Field(ge=1, le=120, description="Look-back in months, view='cycle' only.")] = 12, days: Annotated[int, Field(ge=1, le=400, description="Trailing daily window in days, view='moves' only.")] = 90, ) -> dict[str, Any]: """Central bank POLICY RATES for every jurisdiction Kresmion covers. view='rates' returns one row per bank: the instrument it sets, the level, the observation date and its age in days, the direction and size in basis points of the last change, and which source the number came from. view='cycle' reads the CUMULATIVE change over a window, so a bank that hiked then cut back reads as unchanged rather than as its last move. view='moves' lists the individual changes inside the daily window. WHAT THIS IS NOT. It is not a meeting calendar and it never will be: the decision-date source is prohibited, so no scheduled meeting dates are served here. It is not a real policy rate either (no clean cross-country CPI), and it is not a forecast of the next decision. FRESHNESS IS THE CAVEAT THAT MATTERS. The baseline source is BIS, which publishes policy rates with a lag of SEVERAL DAYS, so between a decision and BIS catching up a row can be one decision behind. Where a faster licence-clean official source exists, `rate_source` reads 'fred' or 'probe' and carries the level actually in force, while `superseded_rate_pct` carries what BIS still says. A rate whose source stopped publishing is WITHHELD (null, `withheld_stale` true) rather than served stale, because a policy rate reads as current whatever its age. Cite the BIS as the source of its statistics.""" if view == "cycle": return await _call(base, "/v1/macro/central-banks/cycle", ctx, {"window_months": window_months}) if view == "moves": return await _call(base, "/v1/macro/central-banks/moves", ctx, {"days": days}) return await _call(base, "/v1/macro/central-banks", ctx) @server.tool(annotations=_READ_ONLY_TOOL) async def money_markets( ctx: Context, view: Annotated[str, Field(pattern="^(rates|spread|stress)$", description="rates = today's rates and the Fed's corridor; spread = one spread through time; stress = the dispersion and volume behind an overnight print.")] = "rates", pair: Annotated[str | None, Field(max_length=48, description="Spread key for view='spread', e.g. 'sofr_iorb'. A bad key returns the valid list.")] = None, rate: Annotated[str, Field(max_length=24, description="Overnight rate key for view='stress': sofr, effr, obfr, bgcr, tgcr.")] = "sofr", ) -> dict[str, Any]: """US MONEY-MARKET RATES and the Fed's policy corridor around them. view='rates' returns the administered rates that define the corridor (the fed funds target range, IORB, the ON RRP award rate, the discount window), the five transaction-based overnight rates that trade inside it (EFFR and OBFR unsecured; SOFR, BGCR and TGCR secured), and the 4-week to 1-year Treasury bills. Each row carries the level, the day it describes, 1d/1w/1m moves in basis points, and where it sits inside the target range. view='spread' and view='stress' are the funding desk: one spread through time, and the 1st-to-99th percentile WIDTH of the transactions behind a print, which is what a median cannot show. WHAT THIS IS NOT. It is not a funding-stress forecast, it is not a repo market feed (no bilateral or tri-party detail beyond the NY Fed's published reference rates), and it is not an intraday surface: every rate here is one published daily print. THREE THINGS THAT WILL BITE AN UNWARY ANSWER. (1) Administered rates are dated by their EFFECTIVE day and run AHEAD of the market rates, so `as_of` is the newest date a MARKET rate carries and `administered_through` says how far the others reach. (2) `corridor_position_pct` is NOT clamped: above the ceiling reports over 100 and below the floor reports under 0, on purpose. (3) The Treasury bill rows are quoted on a DISCOUNT basis, systematically below the investment-basis constant-maturity yield, so do not subtract one from a policy rate without converting.""" if view == "spread": if not pair: return {"error": "invalid_request", "detail": "view='spread' requires 'pair' (e.g. 'sofr_iorb')."} return await _call(base, "/v1/macro/money-markets/spreads", ctx, {"pair": pair}) if view == "stress": return await _call(base, "/v1/macro/money-markets/stress", ctx, {"rate": rate}) return await _call(base, "/v1/macro/money-markets", ctx) @server.tool(annotations=_READ_ONLY_TOOL) async def yield_curve_history( ctx: Context, date: Annotated[str | None, Field(description="YYYY-MM-DD. The curve as it stood ON or LAST BEFORE this date. Omit for today.")] = None, ) -> dict[str, Any]: """The US Treasury constant-maturity curve AS IT STOOD on a past date. Returns the eleven tenors (1M to 30Y) in maturity order for the requested day, plus the 2s10s and 3m10s spreads and whether the curve was inverted. Reach for this to compare today's curve with a date that mattered, or to date an inversion. READ `meta.observation_date`, NOT the date you asked for. The resolved date is the newest day at or before your `date` carrying at least 6 of the 11 tenors, which is why a weekend or a holiday answers with the prior session. The two fields are separate precisely so you can see that happen. WHAT THIS IS NOT. It is not interpolated and it is not gap-filled: a PARTIAL CURVE IS SERVED AS PARTIAL. The tenors do not all exist for all time (the 1-month series begins 2001-07-31; the 20-year resumes 1993-10-01 after a six-year gap), and a date carries points only for the tenors it actually has. No tenor is ever carried forward from a neighbouring day. A date before `meta.available_from` returns an empty list with `meta.status` = 'no_data', never a fabricated curve.""" return await _call(base, "/v1/macro/yield-curve/history", ctx, {"date": date}) @server.tool(annotations=_READ_ONLY_TOOL) async def fred_history( ctx: Context, series_id: Annotated[str | None, Field(max_length=32, description="A curated FRED series id, e.g. DGS10, WALCL, M2SL. Omit to list the whole catalogue with each series' latest observation.")] = None, start: Annotated[str | None, Field(description="Inclusive start, YYYY-MM-DD.")] = None, end: Annotated[str | None, Field(description="Inclusive end, YYYY-MM-DD.")] = None, ) -> dict[str, Any]: """Observation history for a FRED series Kresmion tracks, oldest first. With no `series_id` this returns the CATALOGUE: every series the platform fetches, with its label, units and latest stored observation. With one, it returns that series' full stored history over an optional date window. WHAT THIS IS NOT, AND THE REASON IS LEGAL. It is NOT a FRED pass-through. FRED's terms prohibit an application that replicates the essential user experience of the FRED API, so ONLY the curated series behind Kresmion's own surfaces are served and any other id is a 404. Do not tell a user Kresmion can fetch an arbitrary FRED series: call this with no argument and answer from the catalogue instead. A series whose FRED copyright tier requires pre-approval (the ICE BofA index family) is flagged `restricted` in the catalogue and can be configured to return nothing at all; a null there is a licence boundary, not a data gap. `date` on a point is the OBSERVATION's own date, the period the number describes. `as_of` is when those rows were last confirmed against FRED. Those are two different facts and conflating them misdates the data by up to a full publication cycle. Attribution travels on every response in `meta.attribution` and must be reproduced verbatim if you quote a figure.""" if series_id: return await _call(base, f"/v1/macro/fred/{quote(series_id, safe='')}/history", ctx, { "start": start, "end": end, }) return await _call(base, "/v1/macro/fred", ctx) @server.tool(annotations=_READ_ONLY_TOOL) async def economic_surprise( ctx: Context, history: Annotated[bool, Field(description="False (default) returns the latest reading; True returns the daily series.")] = False, days: Annotated[int, Field(ge=1, le=3650, description="Look-back window in days, history=True only.")] = 730, country: Annotated[str, Field(max_length=8, description="Country the index is computed for.")] = "US", ) -> dict[str, Any]: """The Economic Surprise Index: how data is landing against consensus. One weighted number: each release contributes (actual minus forecast) over the absolute forecast, clipped at plus or minus five and weighted by importance. Positive means data has been BEATING expectations, negative means MISSING them. Reach for it to explain why rates or the dollar moved on a run of releases rather than on any single print. WHAT THIS IS NOT, AND IT IS THE MOST COMMON MISREADING. It is NOT a measure of how strong the economy is. A weak economy with even weaker forecasts prints a POSITIVE surprise index. It measures the gap to expectations and nothing else. It is also not a forecast, and it is not comparable across providers, who each weight and window differently. THE UNDERLYING CALENDAR IS NOT AVAILABLE THROUGH THIS API. Per-event rows, their scheduled times and their consensus figures are not served by any Kresmion tool or endpoint, for licensing reasons. Only this derived aggregate crosses the boundary. Do not tell a user Kresmion can give them the economic calendar; it cannot. Read `event_count` beside the value: it is how many releases carried BOTH an actual and a forecast that day, so a spike on a thin day is a small sample, not a macro turn.""" if history: return await _call(base, "/v1/macro/surprise/history", ctx, { "days": days, "country": country, }) return await _call(base, "/v1/macro/surprise", ctx) @server.tool(annotations=_READ_ONLY_TOOL) async def macro_liquidity(ctx: Context) -> dict[str, Any]: """Fed liquidity and monetary conditions, in one call. Returns the Fed balance sheet, the overnight reverse repo facility, reserve balances, M2, NET LIQUIDITY, the NFCI and STLFSI4 financial conditions indices, credit spreads, leading indicators (jobless claims, breakeven inflation, mortgage rate, consumer sentiment), and a composite liquidity regime score with its inputs named. NET LIQUIDITY IS A CONVENTION, NOT A PUBLISHED SERIES. It is the Fed balance sheet less the reverse repo facility less the Treasury General Account. Say so if you quote it; there is no official number to disagree with. WHAT THIS IS NOT. It is not a forecast of Fed policy, it is not a measure of market liquidity (bid-ask, depth, turnover: none of that is here), and the regime score is not calibrated against forward returns. TWO READING RULES. (1) Cadences differ by series, so a block only moved if its own date moved: the balance sheet is WEEKLY, the repo facility and the spreads are DAILY, M2 is MONTHLY. (2) `liquidity_regime.inputs_missing` names any leg that could not be read; the surviving weights renormalise, so the score is the mean of what was actually available and never a number with a silent zero in it. The two credit-spread legs are licence-gated ICE series and can be null by configuration rather than by outage.""" return await _call(base, "/v1/macro/liquidity", ctx) # ----- macro: positioning, flows, BIS and sovereign spreads (Phase 643A) -- @server.tool(annotations=_READ_ONLY_TOOL) async def cot_positioning( ctx: Context, market: Annotated[str | None, Field(max_length=80, description="Exact market name as returned in the rows' `market` field (e.g. 'GOLD'). Omit for the latest report of every tracked market.")] = None, weeks: Annotated[int, Field(ge=1, le=260, description="With market: how many weekly reports to return, newest first (max 260).")] = 52, ) -> dict[str, Any]: """CFTC Commitments of Traders: who is long and short futures, week by week. Reach for this to read speculative positioning in a futures market: each row carries open interest, commercial (hedger), non-commercial (large speculator) and retail long/short/net, the non-commercial net as a share of open interest, its 52-week z-score, and a descriptive extreme-positioning tag. Without ``market`` it returns the latest report for every tracked market (call that first to learn the exact market names); with ``market`` it returns that market's last ``weeks`` reports. Example: market='GOLD', weeks=26 for half a year of gold positioning. CAVEATS. Weekly data as of each Tuesday, released on roughly a three day lag, so it is never today's positioning. A positioning extreme is context, not a timing signal. Source: CFTC (US government, public). Descriptive, never advice.""" return await _call(base, "/v1/macro/cot", ctx, {"market": market, "weeks": weeks}) @server.tool(annotations=_READ_ONLY_TOOL) async def treasury_tic( ctx: Context, country: Annotated[str | None, Field(max_length=64, description="Exact country name as returned in the rows' `country` field (e.g. 'Japan'). Omit for the latest month of every holder.")] = None, months: Annotated[int, Field(ge=1, le=120, description="With country: how many monthly reports to return, newest first (max 120).")] = 24, ) -> dict[str, Any]: """Foreign holdings of US Treasury securities by country (Treasury TIC). Reach for this to answer "is Japan / China / the UK adding to or cutting its US Treasury holdings?". Each row is one country and month: holdings in USD billions, the prior month, the change in billions and percent, and whether the country is a major holder. Without ``country`` it returns every holder's latest month (call that first for the exact names); with ``country`` it returns that holder's history. Example: country='China, Mainland', months=36 for three years of China's holdings (use the exact name the list view returns). CAVEATS. TIC is published on roughly a SIX WEEK lag, so the newest month describes holdings from about six weeks earlier: quote ``report_month``. Holdings are by the custodian's country, so a holder that keeps Treasuries in Belgium or the Cayman Islands shows up there, not under its own name. Source: US Treasury (public). Descriptive, never advice.""" return await _call(base, "/v1/macro/tic", ctx, {"country": country, "months": months}) @server.tool(annotations=_READ_ONLY_TOOL) async def bis_policy_rates( ctx: Context, country: Annotated[str | None, Field(max_length=3, description="ISO-2 code (e.g. CH, BR, XM for the euro area). Omit for every jurisdiction.")] = None, dataflow: Annotated[str, Field(pattern="^(WS_CBPOL|WS_TC|WS_DSR|WS_CREDIT_GAP|WS_EER)$", description="BIS dataflow: WS_CBPOL policy rates (default), WS_TC total credit, WS_DSR debt-service ratios, WS_CREDIT_GAP credit-to-GDP gap, WS_EER effective exchange rates.")] = "WS_CBPOL", latest: Annotated[bool, Field(description="True (default) returns only the newest observation per country; False returns the stored series, newest first.")] = True, limit: Annotated[int, Field(ge=1, le=1000, description="Maximum rows.")] = 200, ) -> dict[str, Any]: """Bank for International Settlements statistics: policy rates for 36 jurisdictions, plus credit data. Reach for this for the BIS's own series: by default every covered central bank's latest policy rate as the BIS publishes it (36 jurisdictions, each row named in ``country_name``; XM is the euro area). Switch ``dataflow`` for total credit, debt-service ratios, the credit-to-GDP gap or effective exchange rates, and set ``latest`` false for a series through time. Example: country='CH', latest=False, limit=24 for the Swiss policy rate history. WHEN TO USE central_banks INSTEAD. For "what is the rate in force today" prefer central_banks: it joins faster official sources where the BIS lags and withholds a rate whose source went stale. This tool is the raw BIS series, which can be one decision behind for several days after a meeting. ``time_period`` is the BIS period (a month, a quarter or a day); quote it. Cite the BIS as the source of its statistics. Descriptive, never advice.""" return await _call(base, "/v1/macro/bis", ctx, { "dataflow": dataflow, "country": country, "latest": latest, "limit": limit, }) @server.tool(annotations=_READ_ONLY_TOOL) async def sovereign_spreads( ctx: Context, country: Annotated[str | None, Field(min_length=2, max_length=2, description="ISO-2 country code (e.g. IT, FR, GB, JP). Omit for every covered country's newest month.")] = None, months: Annotated[int, Field(ge=1, le=360, description="With country: months of history, oldest first (default 25).")] = 25, ) -> dict[str, Any]: """Sovereign 10Y bond spreads vs the German Bund and vs US Treasuries (a Kresmion calculation, not a rating). Reach for this for market-implied credit context: how much more (or less) a government pays to borrow for ten years than Germany and than the US. Each row carries the country's 10Y yield, the Bund and US legs, ``spread_bund_bp`` and ``spread_ust_bp`` in basis points, their 1-month and 12-month changes, and the source, series and month of every leg. Without ``country``: 40 countries (EU27, AU, CA, CH, CL, GB, IL, JP, KR, MX, NO, NZ, ZA, US), widest spread vs the Bund first. With ``country``: its monthly history plus ``meta.latest`` and ``meta.rank_vs_bund``. Example: country='IT', months=60 for five years of the BTP-Bund spread. WHAT THIS IS NOT. It is NOT a credit rating and no rating agency is involved: Kresmion subtracts one published yield from another. Sovereign ratings are not served by any tool. Every spread is MONTHLY (both legs the same calendar month; no daily spread). Outside the euro area the bond is in local currency, so the spread also carries currency and inflation expectations, not only credit. ATTRIBUTION, required by the data owners, reproduce it when you quote a figure: ``meta.attribution`` lists the lines (Source: ECB statistics; the ECB yields are available free of charge at data.ecb.europa.eu; the spreads are modified data computed by Kresmion; OECD Main Economic Indicators via FRED; Federal Reserve H.15 via FRED; the FRED API notice). Descriptive, never advice.""" return await _call(base, "/v1/macro/sovereign-spreads", ctx, { "country": country, "months": months if country else None, }) # ----- equities: offerings and relative value (Phase 564, W7) ------------ @server.tool(annotations=_READ_ONLY_TOOL) async def equity_offerings( ctx: Context, view: Annotated[str, Field(pattern="^(filings|pipeline)$", description="filings = the raw offering feed, one row per FILING; pipeline = IPOs in registration and recently priced, one row per DEAL.")] = "filings", type: Annotated[str | None, Field(description="filings only: ipo | follow_on | shelf_takedown | atm | preferred | warrants | units | rights | other.")] = None, q: Annotated[str | None, Field(max_length=64, description="filings only: company-name or ticker substring.")] = None, days: Annotated[int, Field(ge=1, le=365, description="filings only: filed-date look-back window in days.")] = 30, tier: Annotated[str, Field(pattern="^(all|exchange|otc|unknown)$", description="pipeline only: listing tier filter.")] = "all", limit: Annotated[int, Field(ge=1, le=200, description="filings only: maximum rows.")] = 50, ) -> dict[str, Any]: """IPOs and equity offerings read out of SEC EDGAR: the feed, or the pipeline. view='filings' is the raw feed: S-1, S-1/A, F-1, F-1/A, 424B4 and the 424B3/B5 supplements that offer stock rather than debt. view='pipeline' is the IPO view a human wants: what is still in registration and what has just priced. ONE ROW IS ONE FILING in the feed, NOT one deal. A single IPO appears as its registration, each amendment and its priced prospectus, tied together by `ipo_group_key`. Counting rows counts documents. Use view='pipeline', which collapses to one row per deal, before answering "how many IPOs". A NULL IS NEVER A ZERO. Every financial field is nullable and a null means the document did not state that figure; `parse_confidence` (0-1) reports how sure the classifier was. A price RANGE from an S-1/A stays a range and is never reported as its midpoint. WHAT THIS IS NOT. Debt offerings are not here (they are the bonds feed). First-day and since-IPO PERFORMANCE is not here at all: those are a signed-in-only surface, and first-day returns in particular are unreliable against an unadjusted price series, so do not compute one from an offer price and a later close. Underwriter league tables and lock-up expiry dates are likewise not served through this API.""" if view == "pipeline": return await _call(base, "/v1/intel/offerings/ipo-pipeline", ctx, {"tier": tier}) return await _call(base, "/v1/intel/offerings", ctx, { "type": type, "q": q, "days": days, "limit": limit, }) @server.tool(annotations=_READ_ONLY_TOOL) async def relative_value( ctx: Context, ticker: Annotated[str, Field(max_length=12, description="Equity ticker, case-insensitive.")], full_matrix: Annotated[bool, Field(description="False (default) returns eight headline metrics against the peer median; True returns every peer's row and every metric's distribution.")] = False, ) -> dict[str, Any]: """One company's fundamentals against its peer group, like for like. Returns revenue, revenue growth, gross/operating/net margin, return on equity, price/earnings and the one-year price return beside the PEER MEDIAN, the company's percentile in the peer distribution, and `n`, the number of peers that carried a comparable value. WHAT THIS IS NOT. A percentile is where a company sits in a population, NOT a rating: no metric here is scored as good or bad, and presenting one as a verdict misrepresents it. It is not a valuation model and it is not a screen. THE REFUSALS ARE THE PRODUCT. A null `value` ALWAYS carries a `refusal` code and a `refusal_text` saying why the figure could not be computed like for like. It never means zero. Quote the refusal text; do not substitute an estimate. PERIODS AND CURRENCY, the two traps. Every fundamental is a FULL FISCAL YEAR off ONE annual filing (10-K, 20-F, 40-F). Nothing is trailing twelve months and nothing is annualised from a quarter, because a 10-Q figure is year-to-date and Q4 is not a reported fact. Values are in the FILER'S OWN reporting currency with nothing converted, so a ratio is compared across filers and a LEVEL (revenue) is refused across currencies rather than silently mixed. A depositary filer's multiples are refused outright because its ADR ratio is not stored. Under `meta.rules.min_peers_for_stats` peers, no median, quartile or percentile is published at all.""" if full_matrix: return await _call(base, f"/v1/rv/{quote(ticker, safe='')}/matrix", ctx) return await _call(base, f"/v1/rv/{quote(ticker, safe='')}", ctx) # ----- cross-asset intelligence ------------------------------------------ @server.tool(annotations=_READ_ONLY_TOOL) async def signal_track_record( ctx: Context, signal_type: Annotated[str | None, Field(description="Optional signal_type. Omit for per-type aggregates; pass a type for its per-symbol breakdown plus the most recent individual occurrences with their returns.")] = None, asset_class: Annotated[str | None, Field(description="equity | crypto | forex | macro | commodity (aggregate view only).")] = None, symbol: Annotated[str | None, Field(description="Filter to one asset symbol.")] = None, direction: Annotated[str | None, Field(description="bull | bear | neutral (aggregate view only).")] = None, days: Annotated[int, Field(ge=1, le=1600, description="Look-back on signal detection time in days.")] = 365, min_n: Annotated[int, Field(ge=1, le=100, description="Minimum occurrences for a group to be published (aggregate view only).")] = 5, limit: Annotated[int, Field(ge=1, le=200, description="Maximum rows to return.")] = 50, ) -> dict[str, Any]: """Kresmion's published signal track record: forward returns after each signal, losses included. For every logged signal occurrence the platform computes the asset's actual forward return at 1, 7, and 30 days (entry = the next daily close AFTER the signal day, so there is no look-ahead) and publishes per-signal-type aggregates: median and mean return, win rate, direction hit rate, best and worst case, plus a naive same-symbol baseline for comparison. Losses are included and nothing is filtered on outcome. Only non-signals are left out: a withdrawn signal (retracted within an hour of detection) is not scored, and a claim written several times by a scraper re-run is scored once, on its first copy. Reach for this to calibrate how much weight a signal type and its strength deserve before acting on a live signal. CAVEATS: most detectors came online in 2026-04/05 so n is small for many types; stats are the live production history (thresholds evolved in-sample), descriptive rather than predictive; equity closes currently lag crypto by a few days. With ``signal_type`` set, only symbol, days, and limit apply.""" if signal_type: return await _call(base, f"/v1/track-record/{quote(signal_type, safe='')}", ctx, { "symbol": symbol, "days": days, "limit": limit, }) return await _call(base, "/v1/track-record", ctx, { "signal_type": signal_type, "asset_class": asset_class, "symbol": symbol, "direction": direction, "days": days, "min_n": min_n, "limit": limit, }) @server.tool(annotations=_READ_ONLY_TOOL) async def confluence( ctx: Context, symbol: Annotated[str | None, Field(description="Optional symbol. Omit for all active composites; pass a symbol for its current state + history.")] = None, conviction: Annotated[str | None, Field(description="HIGH | MEDIUM | FORMING.")] = None, min_score: Annotated[int, Field(ge=0, le=100, description="Minimum composite score.")] = 0, include_expired: Annotated[bool, Field(description="Include inactive/expired composites.")] = False, hours: Annotated[int, Field(ge=1, le=720, description="Look-back window in hours (max 30 days).")] = 48, limit: Annotated[int, Field(ge=1, le=200, description="Maximum rows to return.")] = 50, ) -> dict[str, Any]: """Independent signal families agreeing on ONE symbol inside a time window. Kresmion continuously scans independent families (asset signals, whale exchange flows, funding-rate anomalies, prediction-market repricing) and fires a composite when three or more distinct families align on the same symbol within 24h, with the contributing legs listed verbatim. A divergence variant fires when strong legs point in OPPOSITE directions. This is the cross-asset synthesis a human would otherwise do by eye across tabs. How to read it: independent sources agreeing is stronger context than any one source, but it is NOT additive proof of a move; the legs are listed so you can judge their independence yourself. Families differ in symbol coverage, so an absent family is not disagreement. Descriptive detection, never advice.""" if symbol: return await _call(base, f"/v1/confluence/{quote(symbol, safe='')}", ctx, { "hours": hours, "limit": limit, }) return await _call(base, "/v1/confluence", ctx, { "conviction": conviction, "min_score": min_score, "include_expired": include_expired, "hours": hours, "limit": limit, }) @server.tool(annotations=_READ_ONLY_TOOL) async def changes_since( ctx: Context, since: Annotated[str, Field(description="ISO-8601 timestamp cursor (required). Rows strictly newer are returned. Must be within the last 7 days. Pass meta.next_since from the previous call.")], families: Annotated[str | None, Field(description="Comma-separated families to poll. Valid: signals, whale, prediction, funding, etf_flow, gamma, confluence. Default: signals,whale,prediction,confluence. etf_flow rows carry total_kind; BTC/ETH session rows are 'withheld' with null flows (source not licensed).")] = None, symbol: Annotated[str | None, Field(description="Optional symbol filter, e.g. BTC or NVDA. Ignored for families with no symbol column.")] = None, limit_per_family: Annotated[int, Field(ge=1, le=500, description="Maximum rows per family; truncation is flagged in meta.")] = 200, ) -> dict[str, Any]: """Delta feed for polling agents: only what changed since your last call. Returns per-family arrays of rows newer than the ``since`` cursor, with compact columns (id, symbol, type, direction, strength, timestamp). Store ``meta.next_since`` and pass it back on the next call; empty arrays mean nothing changed. This endpoint exists to cut an agent's token and bandwidth cost versus re-fetching full family payloads on every poll. ``meta.truncated`` flags families that hit ``limit_per_family`` (page forward by calling again with the returned cursor). The cursor may be at most 7 days old; older cursors are rejected with a 400 telling you to do a full refetch instead. For push instead of poll, register a webhook at /settings/api-keys (any event carrying a symbol can be filtered to your watchlist symbols).""" return await _call(base, "/v1/changes", ctx, { "since": since, "families": families, "symbol": symbol, "limit_per_family": limit_per_family, }) # ----- status ------------------------------------------------------------ @server.tool(annotations=_READ_ONLY_TOOL) async def kresmion_status(ctx: Context) -> dict[str, Any]: """Data-freshness health check across every Kresmion product family. Returns one row per family (prediction markets, whales, ETF, derivatives, equity signals, insider clusters, congress, 13F, COT, TIC, macro regime, short volume, track record, confluence, smart money, and some internal collectors whose data no tool serves): the latest backing-row timestamp, the number of rows written in the last 24h, and an ``ok`` flag. Reach for this FIRST to confirm the data behind an endpoint is fresh before you act on it. Each family reports its own health independently. A row count is not the value: for ``crypto_etf`` ``ok`` also requires ``net_flow_usd`` on at least half the funds of the latest complete trading day, and the row says so in ``value_coverage`` / ``value_ok`` / ``reason``.""" return await _call(base, "/v1/status", ctx) return server # Module-level default instance for stdio. mcp = create_mcp() if __name__ == "__main__": # Run over stdio. Reads KRESMION_API_KEY from the environment. mcp.run()