""" Live cross-platform scan: pull open markets from Polymarket (Gamma + CLOB) and Kalshi (trade-api v2), pair up markets that ask the same question, and run the pairs through CrossPlatformArbitrage. """ import asyncio import json import logging import math import re import ssl import unicodedata from collections import defaultdict from dataclasses import dataclass, field from datetime import date, datetime, timedelta, timezone from typing import Optional from zoneinfo import ZoneInfo import aiohttp import certifi from src.strategies.arbitrage import ArbOpportunity, CrossPlatformArbitrage from src.strategies.market_pairs import CURATED_PAIRS logger = logging.getLogger(__name__) GAMMA_MARKETS_URL = "https://gamma-api.polymarket.com/markets" CLOB_BOOK_URL = "https://clob.polymarket.com/book" KALSHI_EVENTS_URL = "https://api.elections.kalshi.com/trade-api/v2/events" KALSHI_MARKETS_URL = "https://api.elections.kalshi.com/trade-api/v2/markets" TIMEOUT = aiohttp.ClientTimeout(total=15) # Some Python installs (e.g. python.org macOS builds) ship without a CA bundle. SSL_CONTEXT = ssl.create_default_context(cafile=certifi.where()) STOPWORDS = { "will", "the", "a", "an", "of", "in", "on", "to", "be", "by", "is", "for", "at", "and", "or", "who", "what", "which", "how", "does", "do", "vs", "v", } # Words that change what a contract pays out on. Both sides must use the same # set: "win the nomination" vs "run for the nomination", "at least 4" vs "4", # or "finish second" vs "win the first round" are different bets. OUTCOME_TERMS = { "win", "wins", "won", "run", "lose", "finish", "place", "first", "second", "third", "perform", "meet", "receive", "become", "resign", "leave", "out", "next", "nominee", "nomination", "election", "primary", "caucus", "runoff", "round", "least", "exactly", "more", "less", "fewer", "above", "below", "over", "under", "before", "after", "through", "between", } # Polymarket sports series (year suffix stripped) -> Kalshi game-winner series. # Only leagues where a game can't end in a draw, so "Team A wins" on one venue # is exactly "Team B doesn't win" on the other. GAME_LEAGUES = { "mlb": "KXMLBGAME", "nfl": "KXNFLGAME", "cfb": "KXNCAAFGAME", "nhl": "KXNHLGAME", "nba": "KXNBAGAME", "wnba": "KXWNBAGAME", "kbo": "KXKBOGAME", "counter-strike": "KXCS2GAME", "league-of-legends": "KXLOLGAME", "valorant": "KXVALORANTGAME", "dota-2": "KXDOTA2GAME", "rainbow-six-siege": "KXR6GAME", } KALSHI_GAME_SERIES = set(GAME_LEAGUES.values()) # Polymarket names NFL/NHL teams by nickname only; Kalshi uses the city # ("Los Angeles R", "New York I"). Expand nicknames so teams_match() can pair them. TEAM_ALIASES = { "mlb": {"Athletics": "A's", "Chicago White Sox": "Chicago WS"}, "nfl": { "Cardinals": "Arizona Cardinals", "Falcons": "Atlanta Falcons", "Ravens": "Baltimore Ravens", "Bills": "Buffalo Bills", "Panthers": "Carolina Panthers", "Bears": "Chicago Bears", "Bengals": "Cincinnati Bengals", "Browns": "Cleveland Browns", "Cowboys": "Dallas Cowboys", "Broncos": "Denver Broncos", "Lions": "Detroit Lions", "Packers": "Green Bay Packers", "Texans": "Houston Texans", "Colts": "Indianapolis Colts", "Jaguars": "Jacksonville Jaguars", "Chiefs": "Kansas City Chiefs", "Raiders": "Las Vegas Raiders", "Chargers": "Los Angeles Chargers", "Rams": "Los Angeles Rams", "Dolphins": "Miami Dolphins", "Vikings": "Minnesota Vikings", "Patriots": "New England Patriots", "Saints": "New Orleans Saints", "Giants": "New York Giants", "Jets": "New York Jets", "Eagles": "Philadelphia Eagles", "Steelers": "Pittsburgh Steelers", "49ers": "San Francisco 49ers", "Seahawks": "Seattle Seahawks", "Buccaneers": "Tampa Bay Buccaneers", "Titans": "Tennessee Titans", "Commanders": "Washington Commanders", }, "nhl": { "Ducks": "Anaheim Ducks", "Bruins": "Boston Bruins", "Sabres": "Buffalo Sabres", "Flames": "Calgary Flames", "Hurricanes": "Carolina Hurricanes", "Blackhawks": "Chicago Blackhawks", "Avalanche": "Colorado Avalanche", "Blue Jackets": "Columbus Blue Jackets", "Stars": "Dallas Stars", "Red Wings": "Detroit Red Wings", "Oilers": "Edmonton Oilers", "Panthers": "Florida Panthers", "Kings": "Los Angeles Kings", "Wild": "Minnesota Wild", "Canadiens": "Montreal Canadiens", "Predators": "Nashville Predators", "Devils": "New Jersey Devils", "Islanders": "New York Islanders", "Rangers": "New York Rangers", "Senators": "Ottawa Senators", "Flyers": "Philadelphia Flyers", "Penguins": "Pittsburgh Penguins", "Sharks": "San Jose Sharks", "Kraken": "Seattle Kraken", "Blues": "St. Louis Blues", "Lightning": "Tampa Bay Lightning", "Maple Leafs": "Toronto Maple Leafs", "Mammoth": "Utah Mammoth", "Utah Hockey Club": "Utah Hockey Club", "Canucks": "Vancouver Canucks", "Golden Knights": "Vegas Golden Knights", "Capitals": "Washington Capitals", "Jets": "Winnipeg Jets", }, } EASTERN = ZoneInfo("America/New_York") @dataclass class MarketMatch: pm_id: str kalshi_id: str pm_name: str kalshi_name: str score: float verified: bool = False # hand-verified pair from CURATED_PAIRS kind: str = "event" # "event" (question match) or "game" (sports/esports winner) starts_at: Optional[datetime] = None # game start time, for "game" matches @dataclass class ScanResult: opportunities: list[ArbOpportunity] = field(default_factory=list) matches: list[MarketMatch] = field(default_factory=list) polymarket_count: int = 0 kalshi_count: int = 0 errors: list[str] = field(default_factory=list) # --------------------------------------------------------------------------- # Parsing # --------------------------------------------------------------------------- def _json_list(value) -> list: """Gamma returns list fields (outcomes, outcomePrices, clobTokenIds) as JSON strings.""" if isinstance(value, list): return value if isinstance(value, str): try: parsed = json.loads(value) return parsed if isinstance(parsed, list) else [] except json.JSONDecodeError: return [] return [] def parse_polymarket_markets(raw: list[dict]) -> dict[str, dict]: """ Keep binary Yes/No order-book markets only, so "YES" means the same thing on both venues. Prices here are indicative; real depth comes from the CLOB book. """ markets = {} for m in raw: if not m.get("enableOrderBook"): continue outcomes = [str(o).strip().lower() for o in _json_list(m.get("outcomes"))] token_ids = _json_list(m.get("clobTokenIds")) if outcomes != ["yes", "no"] or len(token_ids) != 2: continue try: bid = float(m.get("bestBid") or 0) ask = float(m.get("bestAsk") or 0) except (TypeError, ValueError): continue if not (0 < bid < 1 or 0 < ask < 1): continue markets[m["conditionId"]] = { "name": m.get("question", ""), "yes_token": str(token_ids[0]), "bid": bid, "ask": ask, } return markets def _kalshi_price(m: dict, field_name: str) -> float: """Kalshi v2 now quotes `_dollars` strings; fall back to legacy cents.""" dollars = m.get(f"{field_name}_dollars") if dollars is not None: return float(dollars) cents = m.get(field_name) return float(cents) / 100 if cents is not None else 0.0 def _kalshi_size(m: dict, field_name: str) -> float: value = m.get(f"{field_name}_fp", m.get(field_name)) return float(value) if value is not None else 0.0 def _parse_time(value) -> Optional[datetime]: if not value: return None try: dt = datetime.fromisoformat(str(value).replace("Z", "+00:00").replace(" ", "T")) except ValueError: return None return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) def parse_kalshi_markets(raw: list[dict]) -> dict[str, dict]: markets = {} for m in raw: try: bid = _kalshi_price(m, "yes_bid") ask = _kalshi_price(m, "yes_ask") bid_size = _kalshi_size(m, "yes_bid_size") ask_size = _kalshi_size(m, "yes_ask_size") except (TypeError, ValueError): continue has_bid = 0 < bid < 1 and bid_size > 0 has_ask = 0 < ask < 1 and ask_size > 0 if not (has_bid or has_ask): continue title = m.get("title", "") sub = m.get("yes_sub_title") or "" name = title if not sub or sub.lower() in title.lower() else f"{title} {sub}" markets[m["ticker"]] = { "name": name, "event_ticker": m.get("event_ticker", ""), "team": sub, "resolves_at": _parse_time(m.get("expected_expiration_time") or m.get("close_time")), "bid": bid if has_bid else 0.0, "bid_size": bid_size if has_bid else 0.0, "ask": ask if has_ask else 0.0, "ask_size": ask_size if has_ask else 0.0, } return markets def parse_clob_book(book: dict) -> Optional[dict]: """Best YES bid/ask and their sizes from a Polymarket CLOB book.""" try: bids = [(float(b["price"]), float(b["size"])) for b in book.get("bids", [])] asks = [(float(a["price"]), float(a["size"])) for a in book.get("asks", [])] except (KeyError, TypeError, ValueError): return None best_bid = max(bids, default=(0.0, 0.0)) best_ask = min(asks, default=(0.0, 0.0)) if best_bid[0] <= 0 and best_ask[0] <= 0: return None return {"bid": best_bid[0], "bid_size": best_bid[1], "ask": best_ask[0], "ask_size": best_ask[1]} # --------------------------------------------------------------------------- # Matching # --------------------------------------------------------------------------- def _tokens(text: str) -> set[str]: text = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode().lower() text = re.sub(r"\b([a-z])\.(?=[a-z]\b)", r"\1", text) # "u.s." -> "us." words = re.findall(r"[a-z0-9]+(?:\.[0-9]+)?", text) return {w for w in words if w not in STOPWORDS} def _numbers(tokens: set[str]) -> set[str]: return {t for t in tokens if any(c.isdigit() for c in t)} def match_markets(poly: dict[str, dict], kalshi: dict[str, dict], min_score: float = 0.75, max_token_df: int = 200) -> list[MarketMatch]: """ One-to-one matching on IDF-weighted Jaccard similarity, so rare tokens (names, places) outweigh template words like "win the presidential election". Any numbers (years, thresholds, dates) must match exactly, since "above 4%" vs "above 5%" or "by 2026" vs "by 2027" are different contracts, and so must OUTCOME_TERMS. """ pm_tokens = {p: _tokens(m["name"]) for p, m in poly.items()} k_tokens = {k: _tokens(m["name"]) for k, m in kalshi.items()} df = defaultdict(int) for toks in list(pm_tokens.values()) + list(k_tokens.values()): for t in toks: df[t] += 1 n_docs = len(pm_tokens) + len(k_tokens) idf = {t: math.log(n_docs / c) for t, c in df.items()} index = defaultdict(set) for k, toks in k_tokens.items(): for t in toks: index[t].add(k) def weight(toks: set[str]) -> float: return sum(idf[t] for t in toks) candidates = [] for pm_id, pm_toks in pm_tokens.items(): if not pm_toks: continue pm_nums = _numbers(pm_toks) pm_outcome = pm_toks & OUTCOME_TERMS shared = defaultdict(int) for t in pm_toks: posting = index.get(t) if posting and len(posting) <= max_token_df: for k in posting: shared[k] += 1 for k in shared: k_toks = k_tokens[k] if _numbers(k_toks) != pm_nums or (k_toks & OUTCOME_TERMS) != pm_outcome: continue union = weight(pm_toks | k_toks) score = weight(pm_toks & k_toks) / union if union else 0.0 if score >= min_score: candidates.append((score, pm_id, k)) matches = [] used_pm, used_k = set(), set() for score, pm_id, k in sorted(candidates, reverse=True): if pm_id in used_pm or k in used_k: continue used_pm.add(pm_id) used_k.add(k) matches.append(MarketMatch(pm_id, k, poly[pm_id]["name"], kalshi[k]["name"], score)) return matches def curated_matches(poly: dict[str, dict], kalshi: dict[str, dict]) -> list[MarketMatch]: """Hand-verified pairs that currently have quotes on both venues.""" return [ MarketMatch(pm_id, ticker, poly[pm_id]["name"], kalshi[ticker]["name"], 1.0, verified=True) for _, _, pm_id, ticker in CURATED_PAIRS if pm_id in poly and ticker in kalshi ] def _league(m: dict) -> str: events = m.get("events") or [{}] return re.sub(r"-\d{4}$", "", events[0].get("seriesSlug") or "") def parse_polymarket_games(raw: list[dict]) -> dict[str, dict]: """ Two-team moneyline markets. The first outcome's token is treated as YES (" wins"); the second outcome is its NO side. """ games = {} for m in raw: league = _league(m) if league not in GAME_LEAGUES or not m.get("enableOrderBook"): continue outcomes = [str(t) for t in _json_list(m.get("outcomes"))] aliases = TEAM_ALIASES.get(league, {}) teams = [aliases.get(t, t) for t in outcomes] # for matching; display keeps the originals token_ids = _json_list(m.get("clobTokenIds")) start = _parse_time(m.get("gameStartTime")) if len(teams) != 2 or len(token_ids) != 2 or start is None: continue local_start = start.astimezone(EASTERN) games[m["conditionId"]] = { # Teams often play on back-to-back days, so the date is part of the name. "name": f"{outcomes[0]} to beat {outcomes[1]} ({local_start:%a %b} {local_start.day})", "yes_token": str(token_ids[0]), "league": league, "teams": teams, "start": start, "date": local_start.date(), } return games def _team_tokens(name: str) -> list[str]: name = unicodedata.normalize("NFKD", name).encode("ascii", "ignore").decode().lower() return re.findall(r"[a-z0-9]+", name) def _prefix_run(short: list[str], long: list[str]) -> bool: """Each short token is a prefix of the matching long token, in a contiguous run.""" if not short or len(short) > len(long): return False return any( all(long[i + j].startswith(t) for j, t in enumerate(short)) for i in range(len(long) - len(short) + 1) ) def teams_match(kalshi_team: str, pm_team: str) -> bool: """"New York Y" ~ "New York Yankees", "Montana St." ~ "Montana State", "Liquid" ~ "Team Liquid".""" k, p = _team_tokens(kalshi_team), _team_tokens(pm_team) return _prefix_run(k, p) or _prefix_run(p, k) def _kalshi_game_date(event_ticker: str) -> Optional[date]: # KXMLBGAME-26SEP271520BALNYY -> 2026-09-27 parts = event_ticker.split("-") if len(parts) < 2: return None try: return datetime.strptime(parts[1][:7], "%y%b%d").date() except ValueError: return None def match_games(pm_games: dict[str, dict], kalshi: dict[str, dict]) -> list[MarketMatch]: """ Pair a Polymarket moneyline with a Kalshi game event on league + date + both teams. A pairing is kept only when it is unambiguous in both directions. """ events = defaultdict(list) # (series, date) -> [(event_ticker, [(ticker, team), ...])] by_event = defaultdict(list) for ticker, km in kalshi.items(): series = km.get("event_ticker", "").split("-")[0] if series in KALSHI_GAME_SERIES and km.get("team"): by_event[km["event_ticker"]].append((ticker, km["team"])) for event_ticker, markets in by_event.items(): game_date = _kalshi_game_date(event_ticker) if len(markets) == 2 and game_date: events[(event_ticker.split("-")[0], game_date)].append((event_ticker, markets)) def find(g: dict, day: date) -> list[tuple[str, str]]: found = [] team0, team1 = g["teams"] for event_ticker, markets in events.get((GAME_LEAGUES[g["league"]], day), []): (t_a, team_a), (t_b, team_b) = markets if teams_match(team_a, team0) and teams_match(team_b, team1) \ and not (teams_match(team_a, team1) or teams_match(team_b, team0)): found.append((event_ticker, t_a)) elif teams_match(team_b, team0) and teams_match(team_a, team1) \ and not (teams_match(team_b, team1) or teams_match(team_a, team0)): found.append((event_ticker, t_b)) return found candidates = {} # pm_id -> [(event_ticker, kalshi ticker for team 0)] for pm_id, g in pm_games.items(): # Same-date first: teams in a series play on back-to-back days, so the # +/- 1 day fallback (for timezone edge cases) is only used if needed. cands = find(g, g["date"]) if not cands: cands = find(g, g["date"] - timedelta(days=1)) + find(g, g["date"] + timedelta(days=1)) if cands: candidates[pm_id] = cands event_claims = defaultdict(int) for cands in candidates.values(): for event_ticker, _ in cands: event_claims[event_ticker] += 1 matches = [] for pm_id, cands in candidates.items(): if len(cands) != 1 or event_claims[cands[0][0]] != 1: continue ticker = cands[0][1] matches.append(MarketMatch(pm_id, ticker, pm_games[pm_id]["name"], kalshi[ticker]["name"], 1.0, kind="game", starts_at=pm_games[pm_id]["start"])) return matches # --------------------------------------------------------------------------- # Fetching # --------------------------------------------------------------------------- async def fetch_polymarket(session: aiohttp.ClientSession, pages: int = 10, page_size: int = 100) -> list[dict]: # Gamma caps `limit` at 100 per request. raw = [] for page in range(pages): params = { "active": "true", "closed": "false", "limit": page_size, "offset": page * page_size, "order": "volume24hr", "ascending": "false", } async with session.get(GAMMA_MARKETS_URL, params=params, timeout=TIMEOUT) as resp: resp.raise_for_status() batch = await resp.json() raw.extend(batch) if len(batch) < page_size: break return raw async def fetch_polymarket_by_ids(session: aiohttp.ClientSession, condition_ids: list[str], batch_size: int = 30) -> list[dict]: """Fetch specific markets by conditionId, batched to keep request URLs short.""" raw = [] for i in range(0, len(condition_ids), batch_size): params = [("condition_ids", cid) for cid in condition_ids[i:i + batch_size]] async with session.get(GAMMA_MARKETS_URL, params=params, timeout=TIMEOUT) as resp: resp.raise_for_status() raw.extend(await resp.json()) return raw async def fetch_kalshi_games(session: aiohttp.ClientSession, page_size: int = 1000) -> list[dict]: """Open game-winner markets per series; the events listing doesn't surface all of them.""" async def one_series(series: str) -> list[dict]: out, cursor = [], None while True: params = {"series_ticker": series, "status": "open", "limit": page_size} if cursor: params["cursor"] = cursor async with session.get(KALSHI_MARKETS_URL, params=params, timeout=TIMEOUT) as resp: resp.raise_for_status() data = await resp.json() out.extend(data.get("markets", [])) cursor = data.get("cursor") if not cursor or not data.get("markets"): return out pages = await asyncio.gather(*(one_series(s) for s in sorted(KALSHI_GAME_SERIES))) return [m for page in pages for m in page] async def fetch_polymarket_games(session: aiohttp.ClientSession, days: int = 16, page_size: int = 100, max_offset: int = 2000) -> list[dict]: """ Moneyline markets ending in the next `days` days (Polymarket sets a game's end date about a week after kickoff). Queried one day at a time because Gamma refuses offsets past ~2,100 and soccer alone fills that. """ now = datetime.now(timezone.utc) sem = asyncio.Semaphore(8) async def one_day(d: int) -> list[dict]: lo, hi = now + timedelta(days=d), now + timedelta(days=d + 1) out = [] async with sem: for offset in range(0, max_offset, page_size): params = { "active": "true", "closed": "false", "limit": page_size, "offset": offset, "sports_market_types": "moneyline", "end_date_min": lo.isoformat(), "end_date_max": hi.isoformat(), } async with session.get(GAMMA_MARKETS_URL, params=params, timeout=TIMEOUT) as resp: resp.raise_for_status() batch = await resp.json() if not isinstance(batch, list): break out.extend(batch) if len(batch) < page_size: break return out pages = await asyncio.gather(*(one_day(d) for d in range(days))) return [m for page in pages for m in page] async def fetch_kalshi(session: aiohttp.ClientSession, max_pages: int = 40, page_size: int = 200) -> list[dict]: """ Walk open events with nested markets. The flat /markets listing is dominated by thousands of sports player props, so this covers every category instead. """ raw, cursor = [], None for _ in range(max_pages): params = {"status": "open", "limit": page_size, "with_nested_markets": "true"} if cursor: params["cursor"] = cursor async with session.get(KALSHI_EVENTS_URL, params=params, timeout=TIMEOUT) as resp: resp.raise_for_status() data = await resp.json() for event in data.get("events", []): raw.extend(m for m in event.get("markets") or [] if m.get("status") in (None, "active", "open")) cursor = data.get("cursor") if not cursor: break return raw async def fetch_clob_book(session: aiohttp.ClientSession, token_id: str) -> Optional[dict]: try: async with session.get(CLOB_BOOK_URL, params={"token_id": token_id}, timeout=TIMEOUT) as resp: if resp.status != 200: return None return parse_clob_book(await resp.json()) except (aiohttp.ClientError, asyncio.TimeoutError): return None async def run_scan(min_profit_threshold: float = 0.005, min_match_score: float = 0.75) -> ScanResult: result = ScanResult() async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=SSL_CONTEXT)) as session: pm_raw, curated_raw, games_raw, k_raw, k_games_raw = await asyncio.gather( fetch_polymarket(session), fetch_polymarket_by_ids(session, [pm_id for _, _, pm_id, _ in CURATED_PAIRS]), fetch_polymarket_games(session), fetch_kalshi(session), fetch_kalshi_games(session), return_exceptions=True, ) if isinstance(pm_raw, BaseException): result.errors.append(f"Polymarket API: {pm_raw}") pm_raw = [] if isinstance(curated_raw, BaseException): result.errors.append(f"Polymarket API (curated pairs): {curated_raw}") curated_raw = [] if isinstance(games_raw, BaseException): result.errors.append(f"Polymarket API (games): {games_raw}") games_raw = [] if isinstance(k_raw, BaseException): result.errors.append(f"Kalshi API: {k_raw}") k_raw = [] if isinstance(k_games_raw, BaseException): result.errors.append(f"Kalshi API (games): {k_games_raw}") k_games_raw = [] poly = parse_polymarket_markets(pm_raw + curated_raw) pm_games = parse_polymarket_games(games_raw) kalshi = parse_kalshi_markets(k_raw + k_games_raw) result.polymarket_count = len(poly) + len(pm_games) result.kalshi_count = len(kalshi) matches = curated_matches(poly, kalshi) + match_games(pm_games, kalshi) used_pm = {m.pm_id for m in matches} | set(pm_games) used_k = {m.kalshi_id for m in matches} matches += match_markets( {k: v for k, v in poly.items() if k not in used_pm}, {k: v for k, v in kalshi.items() if k not in used_k}, min_score=min_match_score, ) tokens = {**{k: v["yes_token"] for k, v in poly.items()}, **{k: v["yes_token"] for k, v in pm_games.items()}} # Gamma only gives top-of-book prices; pull real depth for matched markets. books = await asyncio.gather(*(fetch_clob_book(session, tokens[m.pm_id]) for m in matches)) engine = CrossPlatformArbitrage(min_profit_threshold=min_profit_threshold) for match, book in zip(matches, books): if book is None: continue result.matches.append(match) km = kalshi[match.kalshi_id] engine.add_mapping(match.pm_id, match.kalshi_id, name=match.pm_name) engine.update_state("polymarket", match.pm_id, book["bid"], book["bid_size"], book["ask"], book["ask_size"]) engine.update_state("kalshi", match.kalshi_id, km["bid"], km["bid_size"], km["ask"], km["ask_size"]) result.opportunities = engine.scan_opportunities() starts = {m.pm_id: m.starts_at for m in result.matches} now = datetime.now(timezone.utc) for opp in result.opportunities: opp.resolves_at = kalshi[opp.market_id_kalshi]["resolves_at"] start = starts.get(opp.market_id_pm) # Prices move by the second once a game is underway, so a gap between two # snapshots taken moments apart is usually stale quotes, not an arb. opp.in_progress = bool(start and start <= now) return result