import subprocess import json import os from dotenv import load_dotenv from loguru import logger load_dotenv() _DEMO_MODE = False def is_demo_mode() -> bool: return _DEMO_MODE DEMO_MARKETS = [ { "slug": "demo-nyc-high-above-87f", "title": "Will the high temperature in New York City exceed 87°F?", "yes_price": 0.30, "no_price": 0.70, "volume": 15200, "city": "New York", }, { "slug": "demo-miami-high-above-95f", "title": "Will the high temperature in Miami exceed 95°F?", "yes_price": 0.62, "no_price": 0.38, "volume": 8400, "city": "Miami", }, { "slug": "demo-london-high-above-75f", "title": "Will the high temperature in London exceed 75°F?", "yes_price": 0.45, "no_price": 0.55, "volume": 6100, "city": "London", }, { "slug": "demo-tokyo-high-above-90f", "title": "Will the high temperature in Tokyo exceed 90°F?", "yes_price": 0.71, "no_price": 0.29, "volume": 9800, "city": "Tokyo", }, { "slug": "demo-la-high-above-85f", "title": "Will the high temperature in Los Angeles exceed 85°F?", "yes_price": 0.55, "no_price": 0.45, "volume": 11300, "city": "Los Angeles", }, ] def _run_pm(args: list[str]) -> dict | list | None: account = os.getenv("PM_TRADER_ACCOUNT", "weather_agent") data_dir = os.getenv("PM_TRADER_DATA_DIR", "./results/paper_trades") cmd = ["pm-trader", "--account", account, "--data-dir", data_dir, *args] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if result.returncode != 0: logger.warning(f"pm-trader error: {result.stderr.strip()}") return None return json.loads(result.stdout) except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError) as e: logger.error(f"pm-trader call failed: {e}") return None _GAMMA_FAILED = False def _fetch_gamma_api(endpoint: str, params: dict | None = None) -> dict | list | None: global _GAMMA_FAILED if _GAMMA_FAILED: return None import requests as req try: resp = req.get( f"https://gamma-api.polymarket.com/{endpoint}", params=params, timeout=(5, 10), ) resp.raise_for_status() return resp.json() except Exception as e: logger.debug(f"Gamma API call failed: {e}") _GAMMA_FAILED = True return None def _set_demo_mode(): global _DEMO_MODE _DEMO_MODE = True logger.warning("Polymarket API unavailable — using demo market data") def _is_weather_market(market: dict) -> bool: import re title = str(market.get("title") or market.get("question") or "") slug = str(market.get("slug") or "") text = (title + " " + slug).lower() city_patterns = [r"\bnew york\b", r"\bnyc\b", r"\blondon\b", r"\bmiami\b", r"\btokyo\b", r"\blos angeles\b", r"\bla\b"] weather_patterns = [r"\btemperature\b", r"\bweather\b", r"°f", r"\bdegree", r"\bhigh\b", r"\blow\b", r"\bforecast\b", r"\bcelsius\b", r"\bfahrenheit\b", r"\btemp\b", r"\brain\b", r"\bsnow\b", r"\bwind\b", r"\bhumidity\b"] has_city = any(re.search(p, text) for p in city_patterns) has_weather = any(re.search(p, text) for p in weather_patterns) return has_city or has_weather def search_weather_markets() -> list[dict]: global _DEMO_MODE _DEMO_MODE = False try: gamma_result = _fetch_gamma_api("markets", { "active": "true", "closed": "false", "limit": 500, }) if gamma_result and isinstance(gamma_result, list): found = [] seen_slugs = set() for market in gamma_result: if not _is_weather_market(market): continue slug = market.get("slug", "") if slug not in seen_slugs: seen_slugs.add(slug) prices = _parse_json_field(market.get("outcomePrices")) found.append({ "slug": slug, "title": market.get("question", slug), "yes_price": float(prices[0]) if prices and len(prices) > 0 else None, "no_price": float(prices[1]) if prices and len(prices) > 1 else None, "volume": market.get("volume24hr", 0), }) if found: logger.info(f"Found {len(found)} weather-related markets on Polymarket") return found except Exception as e: logger.warning(f"Gamma API search failed: {e}") _set_demo_mode() logger.info(f"Using {len(DEMO_MARKETS)} demo markets as fallback") return list(DEMO_MARKETS) def get_market_odds(market_slug: str) -> dict | None: global _DEMO_MODE if not market_slug: return None if _DEMO_MODE: for m in DEMO_MARKETS: if m["slug"] == market_slug: return { "slug": market_slug, "yes_price": m["yes_price"], "no_price": m["no_price"], "spread": None, "volume": m["volume"], } return None try: gamma_result = _fetch_gamma_api("markets", {"slug": market_slug}) if gamma_result and isinstance(gamma_result, list) and len(gamma_result) > 0: m = gamma_result[0] prices = _parse_json_field(m.get("outcomePrices")) _DEMO_MODE = False return { "slug": market_slug, "yes_price": float(prices[0]) if prices and len(prices) > 0 else None, "no_price": float(prices[1]) if prices and len(prices) > 1 else None, "spread": None, "volume": m.get("volume24hr", 0), } except Exception as e: logger.warning(f"Gamma API odds fetch failed for {market_slug}: {e}") for m in DEMO_MARKETS: if m["slug"] == market_slug: logger.info(f"Using demo odds for {market_slug}") _set_demo_mode() return { "slug": market_slug, "yes_price": m["yes_price"], "no_price": m["no_price"], "spread": None, "volume": m["volume"], } return None def get_portfolio() -> dict | None: result = _run_pm(["portfolio"]) if result and isinstance(result, dict) and result.get("ok"): return result.get("data") return None def get_stats() -> dict | None: result = _run_pm(["stats"]) if result and isinstance(result, dict) and result.get("ok"): return result.get("data") return None def get_trade_history(limit: int = 50) -> list | None: result = _run_pm(["history", "--limit", str(limit)]) if result and isinstance(result, dict) and result.get("ok"): data = result.get("data", []) return data if isinstance(data, list) else [] return None def init_account(balance: float = 10000.0) -> bool: account = os.getenv("PM_TRADER_ACCOUNT", "weather_agent") data_dir = os.getenv("PM_TRADER_DATA_DIR", "./results/paper_trades") os.makedirs(data_dir, exist_ok=True) cmd = ["pm-trader", "--account", account, "--data-dir", data_dir, "init", "--balance", str(balance)] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=15) success = result.returncode == 0 if success: logger.success(f"Paper account '{account}' initialized with ${balance:,.0f}") else: if "already exists" in result.stderr.lower(): logger.info(f"Account '{account}' already exists, skipping init") return True logger.warning(f"Account init issue: {result.stderr.strip()}") return success except Exception as e: logger.error(f"Account init failed: {e}") return False def _parse_json_field(value) -> list: if isinstance(value, list): return value if isinstance(value, str): try: return json.loads(value) except (json.JSONDecodeError, TypeError): pass return [] def summarize_markets(markets: list[dict]) -> str: if not markets: return "No weather markets found on Polymarket currently." lines = ["=== Active Weather Markets ==="] for m in markets[:10]: title = m.get("title") or m.get("question") or m.get("slug", "Unknown") slug = m.get("slug", "?") volume = m.get("volume") or m.get("volume24hr") or 0 yes_price = m.get("yes_price") if yes_price is None: prices = m.get("outcome_prices") or _parse_json_field(m.get("outcomePrices")) yes_price = f"{float(prices[0]):.3f}" if prices else "?" lines.append(f" [{slug}] {title}") lines.append(f" YES: {yes_price} | Volume: ${volume:,.0f}") return "\n".join(lines)