hk-weather-mkt/markets/polymarket_client.py

"""Polymarket API client for market data and trading.

Uses the Gamma Markets API for market discovery and CLOB for order execution.
"""

import json
import time
import hashlib
from datetime import datetime
from typing import Optional, Dict, List
from urllib.parse import urlencode

import requests

from config import (
    POLYMARKET_GAMMA_API,
    POLYMARKET_CLOB_API,
    MIN_LIQUIDITY_USDC,
)


class PolymarketClient:
    """Client for Polymarket prediction markets."""

    def __init__(self):
        self.gamma_url = POLYMARKET_GAMMA_API
        self.clob_url = POLYMARKET_CLOB_API
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "HK-Weather-Market/1.0",
            "Accept": "application/json",
        })

    def search_markets(
        self,
        query: str = "hong kong weather",
        active: bool = True,
        limit: int = 20,
    ) -> List[Dict]:
        """Search for weather-related markets on Polymarket."""
        params = {
            "query": query,
            "active": str(active).lower(),
            "closed": "false",
            "limit": limit,
            "archived": "false",
            "order": "liquidity",
        }

        try:
            url = f"{self.gamma_url}/markets?{urlencode(params)}"
            resp = self.session.get(url, timeout=15)
            resp.raise_for_status()
            markets = resp.json()

            return [
                {
                    "id": m.get("id"),
                    "question": m.get("question"),
                    "condition_id": m.get("conditionId"),
                    "slug": m.get("slug"),
                    "volume": float(m.get("volume", 0)),
                    "liquidity": float(m.get("liquidity", 0)),
                    "volume_24hr": float(m.get("volume24hr", 0)),
                    "end_date": m.get("endDateIso"),
                    "start_date": m.get("startDateIso"),
                    "active": m.get("active"),
                    "closed": m.get("closed"),
                    "outcome_prices": json.loads(m.get("outcomePrices", "[]")),
                    "outcomes": json.loads(m.get("outcomes", "[]")),
                    "description": m.get("description", ""),
                    "category": m.get("category", ""),
                    "tags": m.get("tags", []),
                }
                for m in markets
            ]

        except Exception as e:
            print(f"Polymarket search error: {e}")
            return []

    def get_market(self, condition_id: str) -> Optional[Dict]:
        """Get a single market by condition ID."""
        try:
            url = f"{self.gamma_url}/markets/{condition_id}"
            resp = self.session.get(url, timeout=15)
            resp.raise_for_status()
            m = resp.json()

            return {
                "id": m.get("id"),
                "question": m.get("question"),
                "condition_id": m.get("conditionId"),
                "slug": m.get("slug"),
                "volume": float(m.get("volume", 0)),
                "liquidity": float(m.get("liquidity", 0)),
                "end_date": m.get("endDateIso"),
                "active": m.get("active"),
                "closed": m.get("closed"),
                "outcome_prices": json.loads(m.get("outcomePrices", "[]")),
                "outcomes": json.loads(m.get("outcomes", "[]")),
                "description": m.get("description", ""),
            }

        except Exception as e:
            print(f"Polymarket get_market error: {e}")
            return None

    def get_market_price(self, token_id: str) -> Optional[float]:
        """Get current price for a specific outcome token."""
        try:
            url = f"{self.clob_url}/price?token_id={token_id}&side=buy"
            resp = self.session.get(url, timeout=10)
            if resp.status_code == 200:
                data = resp.json()
                return float(data.get("price", 0))
        except Exception as e:
            print(f"Polymarket price error: {e}")
        return None

    def get_order_book(self, token_id: str) -> Dict:
        """Get order book for a token."""
        try:
            url = f"{self.clob_url}/book?token_id={token_id}"
            resp = self.session.get(url, timeout=10)
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            print(f"Polymarket orderbook error: {e}")
            return {}

    def find_relevant_weather_markets(self) -> List[Dict]:
        """Find all weather-related markets relevant to Hong Kong."""
        queries = [
            "hong kong weather",
            "hong kong temperature",
            "hong kong typhoon",
            "hong kong rain",
            "asia typhoon",
            "south china sea",
        ]

        all_markets = []
        seen_ids = set()

        for query in queries:
            markets = self.search_markets(query=query)
            for m in markets:
                if m["id"] not in seen_ids and m.get("active") and not m.get("closed"):
                    seen_ids.add(m["id"])
                    all_markets.append(m)

        all_markets.sort(key=lambda m: m.get("liquidity", 0), reverse=True)

        return all_markets

    def get_market_implied_probability(
        self, condition_id: str, outcome_index: int = 0
    ) -> Optional[float]:
        """Get market-implied probability for a specific outcome.

        Uses midpoint of best bid/ask when available, otherwise last price.
        """
        market = self.get_market(condition_id)
        if not market or "outcome_prices" not in market:
            return None

        if outcome_index < len(market["outcome_prices"]):
            return float(market["outcome_prices"][outcome_index])

        return None

    def get_clob_token_id(self, condition_id: str, outcome_index: int = 0) -> Optional[str]:
        """Get CLOB token ID for a market outcome.

        Token IDs are derived deterministically from condition ID + outcome index.
        """
        try:
            url = f"{self.clob_url}/markets/{condition_id}"
            resp = self.session.get(url, timeout=10)
            resp.raise_for_status()
            data = resp.json()

            tokens = data.get("tokens", [])
            if outcome_index < len(tokens):
                return tokens[outcome_index].get("token_id")
        except Exception as e:
            print(f"CLOB token ID error: {e}")

        return None

    def create_market(
        self,
        question: str,
        outcomes: List[str],
        end_date: str,
        description: str = "",
    ) -> Optional[Dict]:
        """
        Create a new market on Polymarket.
        NOTE: Requires whitelisted API key and Polygonscan approval.
        Markets go through curation before going live.
        """
        print("Market creation requires curation approval from Polymarket.")
        print(f"Would create: {question}")
        print(f"Outcomes: {outcomes}")
        print(f"End date: {end_date}")
        return {
            "status": "proposed",
            "question": question,
            "outcomes": outcomes,
            "end_date": end_date,
            "note": "Submit via Polymarket UI or contact partnerships@polymarket.com",
        }