MCP for Trading and Prediction Market Agents

// MCP config: multiple data sources for a trading agent
{
  "mcpServers": {
    "prediction-markets": {
      "command": "npx",
      "args": ["predmcp-server"],
      "env": {"POLYMARKET_API_KEY": "..."}
    },
    "search-signals": {
      "command": "npx",
      "args": ["@scavio/mcp-server"],
      "env": {"SCAVIO_API_KEY": "..."}
    },
    "onchain-data": {
      "command": "npx",
      "args": ["defi-mcp-server"],
      "env": {"RPC_URL": "..."}
    }
  }
}

import requests, os
from datetime import datetime

def news_market_correlation(event: str) -> dict:
    """Compare news coverage volume with prediction market odds."""
    headers = {"x-api-key": os.environ["SCAVIO_API_KEY"]}

    # Get current news coverage for the event
    resp = requests.post(
        "https://api.scavio.dev/api/v1/search",
        headers=headers,
        json={
            "query": event,
            "platform": "google",
            "num_results": 20,
        },
        timeout=10,
    )
    results = resp.json()
    organic = results.get("organic_results", [])
    news_results = [r for r in organic if "news" in r.get("link", "").lower()
                    or r.get("date", "")]

    # Check AI Overview for consensus narrative
    ai_overview = results.get("ai_overview", {})
    consensus = ai_overview.get("text", "")

    return {
        "event": event,
        "news_coverage_count": len(news_results),
        "total_serp_mentions": len(organic),
        "consensus_narrative": consensus[:200],
        "top_headlines": [r["title"] for r in news_results[:5]],
        "checked_at": datetime.now().isoformat(),
    }

# Check news coverage for prediction market events
signal = news_market_correlation("2026 federal reserve rate decision")
# If news coverage spikes but odds haven't moved -> potential opportunity

def social_sentiment_scan(topics: list) -> dict:
    """Scan Reddit and social for sentiment shifts before market moves."""
    headers = {"x-api-key": os.environ["SCAVIO_API_KEY"]}
    signals = {}

    for topic in topics:
        # Reddit discussion volume
        resp = requests.post(
            "https://api.scavio.dev/api/v1/search",
            headers=headers,
            json={
                "query": f"reddit {topic}",
                "platform": "google",
                "num_results": 10,
            },
            timeout=10,
        )
        reddit_results = resp.json().get("organic_results", [])

        # Look for sentiment in snippets
        bullish_words = ["bullish", "moon", "buy", "undervalued", "opportunity"]
        bearish_words = ["bearish", "crash", "sell", "overvalued", "bubble"]

        bull_count = sum(
            1 for r in reddit_results
            if any(w in r.get("snippet", "").lower() for w in bullish_words)
        )
        bear_count = sum(
            1 for r in reddit_results
            if any(w in r.get("snippet", "").lower() for w in bearish_words)
        )

        signals[topic] = {
            "reddit_mentions": len(reddit_results),
            "bullish_signals": bull_count,
            "bearish_signals": bear_count,
            "sentiment_ratio": bull_count / max(bear_count, 1),
        }

    return signals

# Scan sentiment for prediction market categories
sentiment = social_sentiment_scan([
    "bitcoin 2026",
    "AI regulation",
    "fed rate cut",
    "election polls",
])

class TradingSignalPipeline:
    """Combine prediction market data with news and social signals."""

    def __init__(self):
        self.search_key = os.environ["SCAVIO_API_KEY"]

    def evaluate_opportunity(self, market_event: str, current_odds: float) -> dict:
        # Layer 1: News coverage volume and direction
        news = news_market_correlation(market_event)

        # Layer 2: Social sentiment
        sentiment = social_sentiment_scan([market_event])

        # Layer 3: Information edge detection
        edge = self.detect_edge(news, sentiment, current_odds)

        return {
            "event": market_event,
            "current_odds": current_odds,
            "news_volume": news["news_coverage_count"],
            "social_sentiment": sentiment.get(market_event, {}),
            "edge_signal": edge,
        }

    def detect_edge(self, news: dict, sentiment: dict, odds: float) -> str:
        """Detect when information flow suggests odds are mispriced."""
        news_volume = news["news_coverage_count"]
        # High news volume + low odds shift = potential edge
        if news_volume > 10 and odds < 0.3:
            return "HIGH_NEWS_LOW_ODDS: possible underpriced event"
        if news_volume < 3 and odds > 0.7:
            return "LOW_NEWS_HIGH_ODDS: possible overpriced event"
        return "NO_CLEAR_EDGE"