mcp_server.py

#!/usr/bin/env python3
"""
DeepBlue Trading MCP Server
Real trading signals from a live Polymarket bot.
Free tier: 5 calls/day. Paid: x402 USDC micropayments.

Run: python3 mcp_server.py                    # stdio (Claude Desktop)
     python3 mcp_server.py --sse --port 4300  # SSE (remote clients)
"""

import asyncio
import json
import os
import sys
import time
from collections import defaultdict

import httpx
from fastmcp import FastMCP

# ── Config ──────────────────────────────────────────────────────────────────────
API_BASE = "http://127.0.0.1:4200"
FREE_CALLS_PER_DAY = 5
RATE_LIMIT_STORE: dict[str, list[float]] = defaultdict(list)

mcp = FastMCP(
    name="DeepBlue Trading Intelligence",
    instructions=(
        "DeepBlue provides live crypto trading signals from an autonomous Polymarket bot "
        "with a verified track record. Tools include real-time BTC/ETH/SOL/XRP momentum signals, "
        "sentiment analysis, market intelligence, and performance data. "
        "Free tier: 5 calls/day. For unlimited access, use x402 micropayments (USDC on Base). "
        "More info: https://deepbluebase.xyz/pricing"
    ),
)


# ── Helpers ─────────────────────────────────────────────────────────────────────

def _check_rate_limit(tool_name: str) -> str | None:
    """Returns error string if rate limited, else None."""
    today = time.strftime("%Y-%m-%d")
    key = f"{today}"
    calls = RATE_LIMIT_STORE[key]
    # Prune old entries
    for k in list(RATE_LIMIT_STORE):
        if k != key:
            del RATE_LIMIT_STORE[k]
    if len(calls) >= FREE_CALLS_PER_DAY:
        return (
            f"Free tier limit reached ({FREE_CALLS_PER_DAY}/day). "
            f"For unlimited access, use the REST API with x402 payments: "
            f"https://deepbluebase.xyz/pricing"
        )
    calls.append(time.time())
    return None


async def _api_get(path: str, params: dict | None = None) -> dict:
    """Call our REST API."""
    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.get(f"{API_BASE}{path}", params=params)
        resp.raise_for_status()
        return resp.json()


def _fmt(data: dict) -> str:
    """Format response as readable JSON."""
    return json.dumps(data, indent=2, default=str)


# ── Tools ───────────────────────────────────────────────────────────────────────

@mcp.tool()
async def get_trading_signals() -> str:
    """Live BTC, ETH, SOL, XRP 5-minute directional signals with confidence scores.
    Generated by DeepBlue's MomentumSignalGenerator from real-time Binance websocket data.
    Includes: direction (UP/DOWN), confidence (0.50-0.78), regime, and indicator breakdown."""
    err = _check_rate_limit("signals")
    if err:
        return err
    data = await _api_get("/signals")
    return _fmt(data)


@mcp.tool()
async def get_market_snapshot() -> str:
    """All-coins snapshot: price, momentum, signals in one call.
    Quick overview of BTC, ETH, SOL, XRP with current price, 1m/5m momentum, and bias."""
    err = _check_rate_limit("snapshot")
    if err:
        return err
    data = await _api_get("/market-snapshot")
    return _fmt(data)


@mcp.tool()
async def get_performance() -> str:
    """DeepBlue bot trading performance: win rate, total P&L, recent trades.
    Verified track record from live Polymarket 5-min crypto trading."""
    err = _check_rate_limit("performance")
    if err:
        return err
    data = await _api_get("/performance")
    return _fmt(data)


@mcp.tool()
async def get_sentiment() -> str:
    """Crypto sentiment composite for BTC/ETH/SOL/XRP.
    Blends Fear & Greed Index (30%), whale/institutional exchange flow (20%),
    and 6 real-time technical indicators (50%).
    Per-coin scores from -1 (max bearish) to +1 (max bullish)."""
    err = _check_rate_limit("sentiment")
    if err:
        return err
    data = await _api_get("/sentiment/composite")
    return _fmt(data)


@mcp.tool()
async def get_market_intel() -> str:
    """BTC macro intelligence: funding rates, open interest trends,
    liquidation risk levels, and regime classification (trending/ranging/volatile)."""
    err = _check_rate_limit("market_intel")
    if err:
        return err
    data = await _api_get("/market-intel")
    return _fmt(data)


@mcp.tool()
async def get_polymarket_analytics() -> str:
    """Polymarket prediction market analytics: active positions, per-coin win rates,
    edge analysis, best/worst trading hours from DeepBlue's live trading bot."""
    err = _check_rate_limit("polymarket")
    if err:
        return err
    data = await _api_get("/polymarket")
    return _fmt(data)


@mcp.tool()
async def get_token_price(token: str) -> str:
    """Real-time token price. Supports: BTC, ETH, SOL, XRP, and other major tokens.
    Args:
        token: Token symbol (e.g. 'BTC', 'ETH', 'SOL')
    """
    err = _check_rate_limit("price")
    if err:
        return err
    data = await _api_get(f"/price/{token}")
    return _fmt(data)


@mcp.tool()
async def get_whale_moves() -> str:
    """On-chain whale movements and exchange flows.
    Tracks large BTC/ETH transfers, exchange inflows/outflows, and accumulation patterns."""
    err = _check_rate_limit("whales")
    if err:
        return err
    data = await _api_get("/whale-moves")
    return _fmt(data)


@mcp.tool()
async def get_market_mood() -> str:
    """Quick market mood check: fear/greed level, regime classification, per-asset directional bias.
    Lightweight alternative to full sentiment analysis."""
    err = _check_rate_limit("mood")
    if err:
        return err
    data = await _api_get("/market-mood")
    return _fmt(data)


@mcp.tool()
async def get_realtime_signal(coin: str = "btc") -> str:
    """Live directional signal for a single coin from Binance websocket feed.
    Args:
        coin: Coin to analyze — btc, eth, sol, or xrp (default: btc)
    Returns momentum, orderbook imbalance, aggressor ratio, and directional bias."""
    err = _check_rate_limit("realtime")
    if err:
        return err
    data = await _api_get(f"/btc/realtime-signal", params={"coin": coin})
    return _fmt(data)


# ── Entry point ─────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="DeepBlue MCP Server")
    parser.add_argument("--sse", action="store_true", help="Run SSE transport (default: stdio)")
    parser.add_argument("--port", type=int, default=4300, help="Port for SSE transport")
    parser.add_argument("--host", default="0.0.0.0", help="Host for SSE transport")
    args = parser.parse_args()

    if args.sse:
        mcp.run(transport="streamable-http", host=args.host, port=args.port)
    else:
        mcp.run(transport="stdio")