""" starter_5_endgame_sweep.py — End-game sweep bot. When a market is "de-facto" resolved (e.g. winner declared, but UMA hasn't settled yet), the YES token often trades at 0.95-0.99. Holding to settlement yields 1-5% in days/hours. Stack many of these and you've got a tidy carry strategy with negligible directional risk — IF you read the resolution criteria correctly. Risk: UMA disputes. About 0.1-0.5% of markets get an unexpected resolution through dispute. You don't want to hold $10k notional on one bet. Customize: ENTRY_PRICE_MIN/MAX, MAX_PER_MARKET_USD, MAX_DAYS_TO_RESOLUTION. Run: python main.py Defaults to DRY_RUN=true. """ from __future__ import annotations import asyncio import os import signal import time from datetime import datetime, timezone from dotenv import load_dotenv from common import ( JsonlLogger, OrderIntent, RiskLayer, book_depth_at_price, cancel_all, close_http, fetch_active_markets, get_market_book, healthcheck, make_client, parse_token_ids, place_fok_with_retry, ) # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- # Sweet spot: high enough that resolution is essentially priced in, low enough # to leave a few cents of carry. ENTRY_PRICE_MIN = float(os.getenv("ENTRY_PRICE_MIN", "0.95")) ENTRY_PRICE_MAX = float(os.getenv("ENTRY_PRICE_MAX", "0.99")) # Don't put more than this on any single market. UMA dispute = 100% loss. MAX_PER_MARKET_USD = float(os.getenv("MAX_PER_MARKET_USD", "50")) # Skip markets that are too far out — capital lockup hurts annualized return. MAX_DAYS_TO_RESOLUTION = float(os.getenv("MAX_DAYS_TO_RESOLUTION", "30")) # Annualized minimum carry (filters out 1¢ of edge over 60 days). MIN_APR = float(os.getenv("MIN_APR", "0.30")) # 30% # Min depth at our limit price. MIN_DEPTH_USD = float(os.getenv("MIN_DEPTH_USD", "10")) SCAN_INTERVAL_SEC = float(os.getenv("SCAN_INTERVAL_SEC", "30")) DRY_RUN = os.getenv("DRY_RUN", "true").lower() != "false" def days_to_resolution(market: dict) -> float: end = market.get("endDate") or market.get("end_date") or market.get("end_date_iso") if not end: return 999.0 try: if isinstance(end, str): end_dt = datetime.fromisoformat(end.replace("Z", "+00:00")) else: end_dt = datetime.fromtimestamp(float(end), tz=timezone.utc) delta = end_dt - datetime.now(timezone.utc) return max(0.0, delta.total_seconds() / 86400.0) except Exception: return 999.0 def implied_apr(price: float, days_left: float) -> float: if days_left <= 0 or price <= 0: return 0.0 # Assuming YES resolves to $1: gross_return = (1 - price) / price # Annualize linearly (close enough for short horizons). gross = (1 - price) / price return gross * (365.0 / days_left) # --------------------------------------------------------------------------- # Strategy # --------------------------------------------------------------------------- async def scan_once(client, risk: RiskLayer, log: JsonlLogger) -> int: markets = await fetch_active_markets(closed=False, limit=300) found = 0 for m in markets: days = days_to_resolution(m) if days > MAX_DAYS_TO_RESOLUTION: continue try: yes_id, no_id = parse_token_ids(m) except Exception: continue # Check both sides — sometimes NO is the high-confidence side. for side_label, token_id in [("YES", yes_id), ("NO", no_id)]: book = await asyncio.to_thread(get_market_book, client, token_id) asks = book.get("asks") or [] if not asks: continue ask = float(asks[0]["price"]) if not (ENTRY_PRICE_MIN <= ask <= ENTRY_PRICE_MAX): continue apr = implied_apr(ask, days) if apr < MIN_APR: continue depth = book_depth_at_price(book, "buy", ask) depth_usd = depth * ask if depth_usd < MIN_DEPTH_USD: continue size_usd = min(MAX_PER_MARKET_USD, depth_usd) size = size_usd / ask slug = m.get("slug") or m.get("question", "")[:40] print(f"\n💎 ENDGAME: {slug} ({side_label})") print(f" price={ask:.4f} days={days:.1f} APR={apr:.1%} size=${size_usd:.2f}") log.log("endgame_opportunity", market=slug, side=side_label, price=ask, days_left=days, apr=apr, size_usd=size_usd) found += 1 if DRY_RUN: continue intent = OrderIntent( token_id=token_id, side="BUY", size=size, price=ask, strategy_id="endgame_v1", market_slug=slug, ) if not risk.allow(intent): log.log("endgame_skipped", reason="risk_layer", market=slug) continue try: r = await asyncio.to_thread( place_fok_with_retry, client, token_id, "BUY", size, ask ) risk.record_fill(intent, r.filled_size) log.log("endgame_executed", market=slug, **r.__dict__) print(f" ✅ filled {r.filled_size:.2f} @ ~{ask:.4f}") except Exception as e: log.log("endgame_error", market=slug, error=str(e)) print(f" ❌ {e}") return found async def main(): load_dotenv() log = JsonlLogger("logs/endgame_v1") print("=" * 60) print("Polymarket End-Game Sweep") print("=" * 60) print(f"Mode: {'DRY-RUN' if DRY_RUN else '🔴 LIVE'}") print(f"Entry range: {ENTRY_PRICE_MIN}–{ENTRY_PRICE_MAX}") print(f"Max days: {MAX_DAYS_TO_RESOLUTION}") print(f"Min APR: {MIN_APR:.0%}") print(f"Max per market: ${MAX_PER_MARKET_USD}") print("=" * 60) client = make_client() healthcheck(client) risk = RiskLayer.from_env(strategy_id="endgame_v1") stop = asyncio.Event() for sig in (signal.SIGINT, signal.SIGTERM): try: asyncio.get_running_loop().add_signal_handler(sig, stop.set) except NotImplementedError: signal.signal(sig, lambda *_: stop.set()) cycles = 0 while not stop.is_set(): t0 = time.time() try: n = await scan_once(client, risk, log) cycles += 1 print(f"[cycle {cycles}] scanned in {time.time() - t0:.1f}s — {n} entries") except Exception as e: log.log("scan_error", error=str(e)) print(f"⚠️ {e}") try: await asyncio.wait_for(stop.wait(), timeout=SCAN_INTERVAL_SEC) except asyncio.TimeoutError: pass await close_http() print("Done.") if __name__ == "__main__": asyncio.run(main())