Upload builderbrain/polymarket_client.py

builderbrain/polymarket_client.py

@@ -0,0 +1,350 @@
+"""
+Polymarket API Client + Builder Code Router
+===========================================
+
+Handles:
+- Live market data ingestion (prices, liquidity, orderbook)
+- Builder code registration and routing
+- Order construction and execution
+- Paper trading mode for hackathon demo
+
+Builder codes earn fees on every trade routed through them.
+This is the monetization layer for the intelligence agent.
+"""
+
+import requests
+import json
+from dataclasses import dataclass, asdict
+from typing import List, Dict, Optional, Any
+from datetime import datetime
+import hashlib
+
+
+@dataclass
+class PolymarketMarket:
+    """Simplified Polymarket market representation."""
+    market_id: str
+    slug: str
+    question: str
+    description: str
+    active: bool
+    closed: bool
+    best_yes_price: float  # 0-1
+    best_no_price: float
+    volume_24h: float
+    liquidity: float
+    resolution_source: str
+    end_date: str
+    category: str
+    
+    @property
+    def implied_yes_prob(self) -> float:
+        return self.best_yes_price
+    
+    @property
+    def spread(self) -> float:
+        return abs(self.best_yes_price + self.best_no_price - 1.0)
+
+
+@dataclass
+class OrderIntent:
+    """An order ready for execution."""
+    market_id: str
+    side: str  # 'BUY' or 'SELL'
+    outcome: str  # 'YES' or 'NO'
+    size: float  # USD notional
+    price: float  # limit price 0-1
+    builder_code: Optional[str] = None
+    client_order_id: Optional[str] = None
+
+
+@dataclass
+class BuilderCode:
+    """A registered builder code for fee sharing."""
+    code: str
+    name: str
+    description: str
+    fee_share_bps: int  # basis points earned on routed volume
+    total_volume_usd: float = 0.0
+    total_fees_earned_usd: float = 0.0
+    markets: List[str] = None
+    
+    def __post_init__(self):
+        if self.markets is None:
+            self.markets = []
+
+
+class PolymarketClient:
+    """
+    Polymarket API client with builder code support.
+    
+    Uses Gamma API for market data.
+    Supports paper trading for hackathon demos.
+    """
+    
+    GAMMA_API = "https://gamma-api.polymarket.com"
+    CLOB_API = "https://clob.polymarket.com"
+    
+    def __init__(self, api_key: Optional[str] = None, paper_trade: bool = True):
+        self.api_key = api_key
+        self.paper_trade = paper_trade
+        self.session = requests.Session()
+        self.builder_codes: Dict[str, BuilderCode] = {}
+        self.paper_portfolio: Dict[str, Any] = {}
+        self.trade_history: List[Dict] = []
+    
+    # ────────────────────────────── Market Data ──────────────────────────────
+    
+    def fetch_markets(
+        self,
+        active: bool = True,
+        limit: int = 100,
+        category: Optional[str] = None,
+    ) -> List[PolymarketMarket]:
+        """Fetch live markets from Polymarket Gamma API."""
+        url = f"{self.GAMMA_API}/markets"
+        params = {
+            "active": str(active).lower(),
+            "limit": limit,
+            "sort": "volume24hr",
+            "order": "desc",
+        }
+        if category:
+            params["category"] = category
+        
+        try:
+            resp = self.session.get(url, params=params, timeout=15)
+            resp.raise_for_status()
+            data = resp.json()
+            
+            markets = []
+            for m in data.get("markets", data if isinstance(data, list) else []):
+                # Handle different API response shapes
+                if isinstance(m, dict):
+                    markets.append(self._parse_market(m))
+            
+            return markets
+        except Exception as e:
+            print(f"[PolymarketClient] fetch error: {e}")
+            return []
+    
+    def _parse_market(self, raw: Dict) -> PolymarketMarket:
+        """Parse raw API response into PolymarketMarket."""
+        # Handle nested structure
+        market_data = raw.get("market", raw)
+        
+        # Prices can be in different fields depending on API version
+        yes_price = market_data.get("bestYesPrice", market_data.get("yesPrice", 0.5))
+        no_price = market_data.get("bestNoPrice", market_data.get("noPrice", 0.5))
+        
+        # Normalize prices to 0-1
+        if yes_price > 1:
+            yes_price /= 100
+        if no_price > 1:
+            no_price /= 100
+        
+        return PolymarketMarket(
+            market_id=market_data.get("id", market_data.get("slug", "unknown")),
+            slug=market_data.get("slug", ""),
+            question=market_data.get("question", ""),
+            description=market_data.get("description", "")[:200],
+            active=market_data.get("active", True),
+            closed=market_data.get("closed", False),
+            best_yes_price=float(yes_price),
+            best_no_price=float(no_price),
+            volume_24h=float(market_data.get("volume24hr", market_data.get("volume24h", 0))),
+            liquidity=float(market_data.get("liquidity", market_data.get("volume", 0))),
+            resolution_source=market_data.get("resolutionSource", ""),
+            end_date=market_data.get("endDate", ""),
+            category=market_data.get("category", "general"),
+        )
+    
+    def fetch_orderbook(self, market_id: str) -> Optional[Dict]:
+        """Fetch L2 orderbook for a market."""
+        url = f"{self.CLOB_API}/book/{market_id}"
+        try:
+            resp = self.session.get(url, timeout=10)
+            resp.raise_for_status()
+            return resp.json()
+        except Exception as e:
+            print(f"[PolymarketClient] orderbook error: {e}")
+            return None
+    
+    # ────────────────────────────── Builder Codes ──────────────────────────────
+    
+    def register_builder_code(
+        self,
+        code: str,
+        name: str,
+        description: str,
+        fee_share_bps: int = 10,
+    ) -> BuilderCode:
+        """Register a builder code for fee sharing."""
+        bc = BuilderCode(
+            code=code,
+            name=name,
+            description=description,
+            fee_share_bps=fee_share_bps,
+        )
+        self.builder_codes[code] = bc
+        return bc
+    
+    def route_order(
+        self,
+        order: OrderIntent,
+        builder_code: Optional[str] = None,
+    ) -> Dict:
+        """
+        Route an order with builder code attribution.
+        
+        In paper trade mode, simulates execution.
+        In live mode, would submit to Polymarket CLOB.
+        """
+        # Attach builder code if provided
+        if builder_code and builder_code in self.builder_codes:
+            bc = self.builder_codes[builder_code]
+            order.builder_code = builder_code
+            bc.total_volume_usd += order.size
+            bc.total_fees_earned_usd += order.size * bc.fee_share_bps / 10000
+            if order.market_id not in bc.markets:
+                bc.markets.append(order.market_id)
+        
+        if self.paper_trade:
+            return self._paper_execute(order)
+        else:
+            return self._live_execute(order)
+    
+    def _paper_execute(self, order: OrderIntent) -> Dict:
+        """Simulate order execution for hackathon demo."""
+        execution = {
+            "status": "FILLED",
+            "order_id": f"paper_{hashlib.sha256(json.dumps(asdict(order)).encode()).hexdigest()[:16]}",
+            "market_id": order.market_id,
+            "side": order.side,
+            "outcome": order.outcome,
+            "size_usd": order.size,
+            "price": order.price,
+            "filled_at": datetime.utcnow().isoformat(),
+            "builder_code": order.builder_code,
+            "paper_trade": True,
+            "fees_paid_usd": order.size * 0.002,  # 20 bps
+        }
+        
+        # Update paper portfolio
+        key = f"{order.market_id}_{order.outcome}"
+        if key not in self.paper_portfolio:
+            self.paper_portfolio[key] = {"size": 0, "avg_price": 0}
+        
+        pos = self.paper_portfolio[key]
+        new_size = pos["size"] + order.size
+        pos["avg_price"] = (pos["size"] * pos["avg_price"] + order.size * order.price) / new_size
+        pos["size"] = new_size
+        
+        self.trade_history.append(execution)
+        return execution
+    
+    def _live_execute(self, order: OrderIntent) -> Dict:
+        """Placeholder for live execution via CLOB API."""
+        # TODO: Implement Polymarket CLOB signing + submission
+        # Requires API key + wallet integration
+        return {
+            "status": "PENDING",
+            "order": asdict(order),
+            "note": "Live execution requires CLOB API key + wallet",
+        }
+    
+    # ────────────────────────────── Portfolio ──────────────────────────────
+    
+    def get_paper_portfolio(self) -> Dict:
+        """Get current paper trading portfolio."""
+        return {
+            "positions": self.paper_portfolio,
+            "trade_count": len(self.trade_history),
+            "total_volume": sum(t["size_usd"] for t in self.trade_history),
+            "total_fees": sum(t.get("fees_paid_usd", 0) for t in self.trade_history),
+            "builder_codes": {
+                code: {
+                    "volume": bc.total_volume_usd,
+                    "fees_earned": bc.total_fees_earned_usd,
+                    "markets": bc.markets,
+                }
+                for code, bc in self.builder_codes.items()
+            },
+        }
+    
+    def get_trade_history(self, limit: int = 50) -> List[Dict]:
+        """Get recent trade history."""
+        return self.trade_history[-limit:]
+
+
+class BuilderCodeRouter:
+    """
+    Intelligent builder code selection and routing.
+    
+    Routes orders to the most appropriate builder code based on:
+    - Market category alignment
+    - Fee share optimization
+    - Volume tracking for leaderboard
+    """
+    
+    def __init__(self, client: PolymarketClient):
+        self.client = client
+        self.category_map: Dict[str, str] = {}
+    
+    def register_category_mapping(self, category: str, builder_code: str):
+        """Map a market category to a builder code."""
+        self.category_map[category] = builder_code
+    
+    def select_builder_code(
+        self,
+        market: PolymarketMarket,
+        strategy: str = "max_fee_share",
+    ) -> Optional[str]:
+        """Select best builder code for a market."""
+        if strategy == "category_match":
+            return self.category_map.get(market.category)
+        
+        elif strategy == "max_fee_share":
+            # Route to builder code with highest fee share
+            # (incentivizes registering competitive codes)
+            best = None
+            best_bps = 0
+            for code, bc in self.client.builder_codes.items():
+                if bc.fee_share_bps > best_bps:
+                    best = code
+                    best_bps = bc.fee_share_bps
+            return best
+        
+        elif strategy == "volume_leader":
+            # Route to builder with most volume (social proof)
+            best = None
+            best_vol = 0
+            for code, bc in self.client.builder_codes.items():
+                if bc.total_volume_usd > best_vol:
+                    best = code
+                    best_vol = bc.total_volume_usd
+            return best
+        
+        return None
+    
+    def route_with_intent(
+        self,
+        market: PolymarketMarket,
+        side: str,
+        outcome: str,
+        size_usd: float,
+        price: float,
+    ) -> Dict:
+        """Full routing pipeline: select builder code + execute."""
+        builder_code = self.select_builder_code(market)
+        
+        order = OrderIntent(
+            market_id=market.market_id,
+            side=side,
+            outcome=outcome,
+            size=size_usd,
+            price=price,
+            builder_code=builder_code,
+        )
+        
+        return self.client.route_order(order, builder_code)