Create polymarket.py

app/polymarket.py

@@ -0,0 +1,611 @@
+# =============================================================================
+# app/polymarket.py
+# Polymarket Analysis Tool — Single File Module
+# Part of: Universal MCP Hub (Sandboxed) - based on PyFundaments Architecture
+# Copyright 2026 - Volkan Kücükbudak
+# Apache License V. 2 + ESOL 1.1
+# Repo: https://github.com/VolkanSah/Universal-MCP-Hub-sandboxed
+# =============================================================================
+# ARCHITECTURE NOTE:
+#   This file lives exclusively in app/ and is ONLY registered by app/mcp.py.
+#   NO direct access to fundaments/*, .env, or Guardian (main.py).
+#   All config comes from app/.pyfun via app/config.py.
+#
+# LAZY INIT PRINCIPLE:
+#   initialize() is called on first tool use — NOT during startup.
+#   This keeps startup fast and avoids crashes if Gamma API is unreachable.
+#
+# SANDBOX RULES:
+#   - SQLite cache is app/* internal state — NOT postgresql.py (Guardian-only)
+#   - LLM calls are optional — gracefully skipped if no provider key is set
+#   - No global state leaks outside this file
+#   - Read-Only access to Polymarket — no transactions, no auth needed
+# =============================================================================
+
+import asyncio
+import aiosqlite
+import aiohttp
+import logging
+import json
+import os
+from datetime import datetime, timezone
+from typing import Optional
+
+logger = logging.getLogger("polymarket")
+
+# =============================================================================
+# Constants
+# =============================================================================
+
+GAMMA_API       = "https://gamma-api.polymarket.com"
+CACHE_DB        = "app/polymarket_cache.db"
+FETCH_INTERVAL  = 300       # seconds — 5 min between API pulls
+MARKET_LIMIT    = 100       # max markets per fetch (rate limit friendly)
+PRICE_DECIMALS  = 2         # rounding for probability display
+
+# Gamma API uses its own tag/category system.
+# We map their tags to our simplified categories for filtering.
+CATEGORY_MAP = {
+    "politics":   ["politics", "elections", "government", "trump", "us-politics"],
+    "crypto":     ["crypto", "bitcoin", "ethereum", "defi", "web3"],
+    "economics":  ["economics", "inflation", "fed", "interest-rates", "finance"],
+    "energy":     ["energy", "oil", "gas", "renewables", "climate"],
+    "tech":       ["technology", "ai", "spacex", "elon-musk", "science"],
+    "sports":     ["sports", "football", "soccer", "nba", "nfl", "esports"],
+    "world":      ["world", "geopolitics", "war", "nato", "china", "russia"],
+}
+
+# =============================================================================
+# Internal State — lazy init guard
+# =============================================================================
+
+_initialized    = False
+_scheduler_task = None   # asyncio background task handle
+
+# =============================================================================
+# SECTION 1 — Init (Lazy)
+# =============================================================================
+
+async def initialize() -> None:
+    """
+    Lazy initializer — called on first tool use.
+    Sets up SQLite cache and starts background scheduler.
+    Idempotent — safe to call multiple times.
+    """
+    global _initialized, _scheduler_task
+
+    if _initialized:
+        return
+
+    logger.info("Polymarket module initializing (lazy)...")
+
+    await _init_cache()
+
+    # Start background scheduler as asyncio task
+    if _scheduler_task is None or _scheduler_task.done():
+        _scheduler_task = asyncio.create_task(_scheduler_loop())
+        logger.info(f"Scheduler started — fetching every {FETCH_INTERVAL}s.")
+
+    _initialized = True
+    logger.info("Polymarket module ready.")
+
+
+# =============================================================================
+# SECTION 2 — SQLite Cache (app/* internal, NOT postgresql.py!)
+# =============================================================================
+
+async def _init_cache() -> None:
+    """Create SQLite tables if they don't exist."""
+    async with aiosqlite.connect(CACHE_DB) as db:
+        await db.execute("""
+            CREATE TABLE IF NOT EXISTS markets (
+                id          TEXT PRIMARY KEY,
+                slug        TEXT,
+                question    TEXT,
+                category    TEXT,
+                probability REAL,
+                volume      REAL,
+                liquidity   REAL,
+                end_date    TEXT,
+                active      INTEGER,
+                data        TEXT,
+                fetched_at  TEXT
+            )
+        """)
+        await db.execute("""
+            CREATE INDEX IF NOT EXISTS idx_category ON markets(category);
+        """)
+        await db.execute("""
+            CREATE INDEX IF NOT EXISTS idx_active ON markets(active);
+        """)
+        await db.commit()
+    logger.info("Cache initialized.")
+
+
+async def _store_markets(markets: list) -> None:
+    """Upsert markets into SQLite cache."""
+    if not markets:
+        return
+
+    now = datetime.now(timezone.utc).isoformat()
+
+    async with aiosqlite.connect(CACHE_DB) as db:
+        for m in markets:
+            category = _categorize_market(m)
+            prob     = _extract_probability(m)
+
+            await db.execute("""
+                INSERT OR REPLACE INTO markets
+                    (id, slug, question, category, probability, volume, liquidity, end_date, active, data, fetched_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """, (
+                str(m.get("id", "")),
+                m.get("slug", ""),
+                m.get("question", ""),
+                category,
+                prob,
+                float(m.get("volume", 0) or 0),
+                float(m.get("liquidity", 0) or 0),
+                m.get("endDate", ""),
+                1 if m.get("active", False) else 0,
+                json.dumps(m),
+                now,
+            ))
+        await db.commit()
+
+    logger.info(f"Cached {len(markets)} markets.")
+
+
+async def _get_cached_markets(
+    category: Optional[str] = None,
+    active_only: bool = True,
+    limit: int = 50
+) -> list:
+    """Retrieve markets from SQLite cache with optional filters."""
+    async with aiosqlite.connect(CACHE_DB) as db:
+        db.row_factory = aiosqlite.Row
+
+        conditions = []
+        params     = []
+
+        if active_only:
+            conditions.append("active = 1")
+
+        if category and category.lower() in CATEGORY_MAP:
+            conditions.append("category = ?")
+            params.append(category.lower())
+
+        where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
+
+        params.append(limit)
+        cursor = await db.execute(
+            f"SELECT * FROM markets {where} ORDER BY volume DESC LIMIT ?",
+            params
+        )
+        rows = await cursor.fetchall()
+        return [dict(r) for r in rows]
+
+
+async def _get_market_by_id(market_id: str) -> Optional[dict]:
+    """Retrieve a single market by ID from cache."""
+    async with aiosqlite.connect(CACHE_DB) as db:
+        db.row_factory = aiosqlite.Row
+        cursor = await db.execute(
+            "SELECT * FROM markets WHERE id = ?", (market_id,)
+        )
+        row = await cursor.fetchone()
+        return dict(row) if row else None
+
+
+async def _get_cache_stats() -> dict:
+    """Return basic cache statistics."""
+    async with aiosqlite.connect(CACHE_DB) as db:
+        cursor = await db.execute("SELECT COUNT(*) FROM markets WHERE active = 1")
+        active = (await cursor.fetchone())[0]
+        cursor = await db.execute("SELECT MAX(fetched_at) FROM markets")
+        last   = (await cursor.fetchone())[0]
+    return {"active_markets": active, "last_fetch": last}
+
+
+# =============================================================================
+# SECTION 3 — Fetcher (Gamma API — Read Only, no auth)
+# =============================================================================
+
+async def _fetch_from_api(limit: int = MARKET_LIMIT) -> list:
+    """
+    Fetch active markets from Polymarket Gamma API.
+    Read-Only — no auth, no transactions.
+    """
+    params = {
+        "active":   "true",
+        "archived": "false",
+        "closed":   "false",
+        "limit":    limit,
+        "order":    "volume",
+        "ascending":"false",
+    }
+
+    async with aiohttp.ClientSession() as session:
+        async with session.get(
+            f"{GAMMA_API}/markets",
+            params=params,
+            timeout=aiohttp.ClientTimeout(total=30)
+        ) as resp:
+            resp.raise_for_status()
+            data = await resp.json()
+            logger.info(f"Fetched {len(data)} markets from Gamma API.")
+            return data
+
+
+async def _scheduler_loop() -> None:
+    """
+    Background task — polls Gamma API every FETCH_INTERVAL seconds.
+    Runs indefinitely inside asyncio event loop.
+    """
+    logger.info("Scheduler loop started.")
+
+    while True:
+        try:
+            markets = await _fetch_from_api()
+            await _store_markets(markets)
+        except aiohttp.ClientError as e:
+            logger.warning(f"Gamma API fetch failed (network): {e}")
+        except Exception as e:
+            logger.error(f"Scheduler error: {e}")
+
+        await asyncio.sleep(FETCH_INTERVAL)
+
+
+# =============================================================================
+# SECTION 4 — Filter & Categorization
+# =============================================================================
+
+def _categorize_market(market: dict) -> str:
+    """
+    Map a Polymarket market to our simplified category system.
+    Uses tags array from Gamma API response.
+    Falls back to 'other' if no match found.
+    """
+    tags = []
+
+    # Gamma API returns tags as list of dicts or strings
+    for t in market.get("tags", []):
+        if isinstance(t, dict):
+            tags.append(t.get("slug", "").lower())
+            tags.append(t.get("label", "").lower())
+        elif isinstance(t, str):
+            tags.append(t.lower())
+
+    # Also check question text for keywords
+    question = market.get("question", "").lower()
+
+    for category, keywords in CATEGORY_MAP.items():
+        for kw in keywords:
+            if kw in tags or kw in question:
+                return category
+
+    return "other"
+
+
+def _extract_probability(market: dict) -> float:
+    """
+    Extract YES probability from market data.
+    Polymarket stores outcome prices as probability (0.0 - 1.0).
+    Returns probability as percentage (0-100).
+    """
+    try:
+        # outcomePrices is a JSON string like '["0.73", "0.27"]'
+        prices = market.get("outcomePrices")
+        if isinstance(prices, str):
+            prices = json.loads(prices)
+        if prices and len(prices) > 0:
+            return round(float(prices[0]) * 100, PRICE_DECIMALS)
+    except (ValueError, TypeError, json.JSONDecodeError):
+        pass
+    return 0.0
+
+
+def _format_market_simple(market: dict) -> dict:
+    """
+    Format a market for human-readable output.
+    Used by all public tools for consistent output.
+    """
+    prob = market.get("probability", 0.0)
+
+    # Simple plain-language probability label
+    if prob >= 80:
+        sentiment = "sehr wahrscheinlich"
+    elif prob >= 60:
+        sentiment = "wahrscheinlich"
+    elif prob >= 40:
+        sentiment = "ungewiss"
+    elif prob >= 20:
+        sentiment = "unwahrscheinlich"
+    else:
+        sentiment = "sehr unwahrscheinlich"
+
+    return {
+        "id":          market.get("id"),
+        "question":    market.get("question"),
+        "category":    market.get("category"),
+        "probability": f"{prob}%",
+        "sentiment":   sentiment,
+        "volume_usd":  f"${market.get('volume', 0):,.0f}",
+        "liquidity":   f"${market.get('liquidity', 0):,.0f}",
+        "end_date":    market.get("end_date", ""),
+        "slug":        market.get("slug", ""),
+        "url":         f"https://polymarket.com/event/{market.get('slug', '')}",
+    }
+
+
+# =============================================================================
+# SECTION 5 — LLM Adapter (Optional — graceful fallback if no key)
+# =============================================================================
+
+async def _llm_analyze(prompt: str) -> Optional[str]:
+    """
+    Send prompt to available LLM provider.
+    Checks for API keys in order: Anthropic → OpenRouter → HuggingFace.
+    Returns None if no provider is available — caller handles fallback.
+    """
+    # --- Anthropic Claude ---
+    anthropic_key = os.getenv("ANTHROPIC_API_KEY")
+    if anthropic_key:
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    "https://api.anthropic.com/v1/messages",
+                    headers={
+                        "x-api-key":           anthropic_key,
+                        "anthropic-version":   "2023-06-01",
+                        "content-type":        "application/json",
+                    },
+                    json={
+                        "model":      "claude-haiku-4-5-20251001",
+                        "max_tokens": 512,
+                        "messages":   [{"role": "user", "content": prompt}],
+                    },
+                    timeout=aiohttp.ClientTimeout(total=30)
+                ) as resp:
+                    resp.raise_for_status()
+                    data = await resp.json()
+                    return data["content"][0]["text"]
+        except Exception as e:
+            logger.warning(f"Anthropic LLM call failed: {e}")
+
+    # --- OpenRouter fallback ---
+    openrouter_key = os.getenv("OPENROUTER_API_KEY")
+    if openrouter_key:
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    "https://openrouter.ai/api/v1/chat/completions",
+                    headers={
+                        "Authorization": f"Bearer {openrouter_key}",
+                        "content-type":  "application/json",
+                    },
+                    json={
+                        "model":      "mistralai/mistral-7b-instruct",
+                        "max_tokens": 512,
+                        "messages":   [{"role": "user", "content": prompt}],
+                    },
+                    timeout=aiohttp.ClientTimeout(total=30)
+                ) as resp:
+                    resp.raise_for_status()
+                    data = await resp.json()
+                    return data["choices"][0]["message"]["content"]
+        except Exception as e:
+            logger.warning(f"OpenRouter LLM call failed: {e}")
+
+    # --- HuggingFace fallback ---
+    hf_key = os.getenv("HF_API_KEY")
+    if hf_key:
+        try:
+            model = "mistralai/Mistral-7B-Instruct-v0.3"
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"https://api-inference.huggingface.co/models/{model}/v1/chat/completions",
+                    headers={
+                        "Authorization": f"Bearer {hf_key}",
+                        "content-type":  "application/json",
+                    },
+                    json={
+                        "model":      model,
+                        "max_tokens": 512,
+                        "messages":   [{"role": "user", "content": prompt}],
+                    },
+                    timeout=aiohttp.ClientTimeout(total=60)
+                ) as resp:
+                    resp.raise_for_status()
+                    data = await resp.json()
+                    return data["choices"][0]["message"]["content"]
+        except Exception as e:
+            logger.warning(f"HuggingFace LLM call failed: {e}")
+
+    logger.info("No LLM provider available — returning None.")
+    return None
+
+
+# =============================================================================
+# SECTION 6 — Public Tools (registered by mcp.py)
+# =============================================================================
+
+async def get_markets(
+    category: Optional[str] = None,
+    limit: int = 20
+) -> list:
+    """
+    MCP Tool: Get active prediction markets from cache.
+
+    Args:
+        category: Filter by category. Options: politics, crypto, economics,
+                  energy, tech, sports, world, other. None = all categories.
+        limit:    Max number of markets to return (default 20, max 100).
+
+    Returns:
+        List of formatted market dicts with human-readable probability.
+    """
+    await initialize()
+
+    limit   = min(limit, 100)
+    markets = await _get_cached_markets(category=category, limit=limit)
+
+    if not markets:
+        return [{"info": "No markets in cache yet. Try again in 30 seconds."}]
+
+    return [_format_market_simple(m) for m in markets]
+
+
+async def trending_markets(limit: int = 10) -> list:
+    """
+    MCP Tool: Get top trending markets by trading volume.
+
+    Args:
+        limit: Number of trending markets to return (default 10).
+
+    Returns:
+        List of top markets sorted by volume descending.
+    """
+    await initialize()
+
+    markets = await _get_cached_markets(active_only=True, limit=limit)
+
+    if not markets:
+        return [{"info": "No markets in cache yet. Try again in 30 seconds."}]
+
+    return [_format_market_simple(m) for m in markets]
+
+
+async def analyze_market(market_id: str) -> dict:
+    """
+    MCP Tool: Get LLM analysis of a single prediction market.
+    Falls back to structured data summary if no LLM key is configured.
+
+    Args:
+        market_id: Polymarket market ID from get_markets() results.
+
+    Returns:
+        Dict with market data + LLM analysis (or structured fallback).
+    """
+    await initialize()
+
+    market = await _get_market_by_id(market_id)
+
+    if not market:
+        return {"error": f"Market '{market_id}' not found in cache."}
+
+    formatted = _format_market_simple(market)
+
+    # Build LLM prompt
+    prompt = (
+        f"Analysiere diesen Prediction Market kurz und präzise:\n\n"
+        f"Frage: {market.get('question')}\n"
+        f"Wahrscheinlichkeit YES: {formatted['probability']}\n"
+        f"Handelsvolumen: {formatted['volume_usd']}\n"
+        f"Kategorie: {market.get('category')}\n"
+        f"Läuft bis: {market.get('end_date', 'unbekannt')}\n\n"
+        f"Erkläre in 2-3 Sätzen was der Markt aussagt und was das für "
+        f"Alltagsentscheidungen bedeuten könnte. Keine Finanzberatung."
+    )
+
+    analysis = await _llm_analyze(prompt)
+
+    if analysis:
+        formatted["analysis"] = analysis
+    else:
+        # Structured fallback — no LLM needed
+        prob = market.get("probability", 0)
+        formatted["analysis"] = (
+            f"Der Markt bewertet '{market.get('question')}' mit "
+            f"{prob}% Wahrscheinlichkeit. "
+            f"Für eine KI-Analyse bitte LLM Provider API Key konfigurieren."
+        )
+
+    return formatted
+
+
+async def summary_report(category: Optional[str] = None) -> dict:
+    """
+    MCP Tool: Generate a summary report for a category or all markets.
+    Uses LLM if available, falls back to structured statistics.
+
+    Args:
+        category: Category to summarize. None = all active markets.
+
+    Returns:
+        Dict with statistics and optional LLM narrative summary.
+    """
+    await initialize()
+
+    markets = await _get_cached_markets(category=category, limit=50)
+
+    if not markets:
+        return {"error": "No markets in cache yet. Try again in 30 seconds."}
+
+    # --- Build statistics (always available, no LLM needed) ---
+    probs      = [m["probability"] for m in markets if m.get("probability")]
+    avg_prob   = round(sum(probs) / len(probs), 1) if probs else 0
+    total_vol  = sum(m.get("volume", 0) for m in markets)
+
+    # Top 3 by volume
+    top3 = [_format_market_simple(m) for m in markets[:3]]
+
+    stats = {
+        "category":       category or "all",
+        "market_count":   len(markets),
+        "avg_probability":f"{avg_prob}%",
+        "total_volume":   f"${total_vol:,.0f}",
+        "top_markets":    top3,
+        "generated_at":   datetime.now(timezone.utc).isoformat(),
+    }
+
+    # --- LLM narrative (optional) ---
+    market_list = "\n".join([
+        f"- {m.get('question')} ({m.get('probability', 0)}% YES, Vol: ${m.get('volume', 0):,.0f})"
+        for m in markets[:10]
+    ])
+
+    prompt = (
+        f"Erstelle eine kurze Zusammenfassung (3-4 Sätze) der aktuellen "
+        f"Prediction Market Lage{' für ' + category if category else ''}:\n\n"
+        f"{market_list}\n\n"
+        f"Was sind die auffälligsten Trends? Sachlich, keine Finanzberatung."
+    )
+
+    narrative = await _llm_analyze(prompt)
+    if narrative:
+        stats["narrative"] = narrative
+
+    return stats
+
+
+async def get_cache_info() -> dict:
+    """
+    MCP Tool: Get cache status and available categories.
+    Useful for debugging and monitoring.
+
+    Returns:
+        Dict with cache stats and available category list.
+    """
+    await initialize()
+
+    cache_stats = await _get_cache_stats()
+
+    return {
+        **cache_stats,
+        "fetch_interval_seconds": FETCH_INTERVAL,
+        "available_categories":   list(CATEGORY_MAP.keys()) + ["other"],
+        "gamma_api":              GAMMA_API,
+        "llm_available":          any([
+            os.getenv("ANTHROPIC_API_KEY"),
+            os.getenv("OPENROUTER_API_KEY"),
+            os.getenv("HF_API_KEY"),
+        ]),
+    }
+
+
+# =============================================================================
+# Direct execution guard
+# =============================================================================
+if __name__ == "__main__":
+    print("WARNING: Run via main.py → app.py → mcp.py, not directly.")