""" starter_4_cross_platform_arb.py — Cross-platform arbitrage (Polymarket vs Kalshi). Detects price differentials between equivalent contracts on Polymarket and Kalshi. When |p_polymarket - p_kalshi| > threshold, buys the cheap side and sells (or hedges with NO) on the expensive side. NOTE: This is structurally HARDER than single-venue arb because: - Settlement criteria differ subtly between venues. A "Trump wins NH" market can resolve differently. Manually map and verify each pair. - Withdrawal/transfer between Polygon (USDC.e) and Kalshi (USD bank) is slow and not zero-cost. Capital must be pre-positioned on both sides. - Kalshi has 1-2% fees too, plus ACH delays. Calculate net carefully. - Kalshi's API requires KYC/identity. This template uses a stub for Kalshi. Customize: PAIRS (mapping table), MIN_EDGE_USD, MAX_TRADE_USD. Run: python main.py Defaults to DRY_RUN=true. Kalshi side defaults to mock/stub. """ from __future__ import annotations import asyncio import os import signal import time from dataclasses import dataclass from typing import Optional from dotenv import load_dotenv from common import ( JsonlLogger, OrderIntent, RiskLayer, cancel_all, close_http, fetch_market_by_slug, get_http, get_market_book, healthcheck, make_client, parse_token_ids, place_fok_with_retry, ) # --------------------------------------------------------------------------- # Pair mapping — YOU MUST CURATE THIS BY HAND. # Each entry: a Polymarket slug paired with the Kalshi event/series ticker. # Verify settlement criteria match BEFORE adding here. # --------------------------------------------------------------------------- @dataclass class Pair: label: str polymarket_slug: str kalshi_ticker: str # Kalshi market ticker (e.g. "KXPRES-28-DJT") polymarket_outcome: str # "Yes" or "No" — which side maps to Kalshi YES PAIRS: list[Pair] = [ # EDIT ME — these are placeholders. Verify on each platform before using. # Pair(label="2028 Trump", polymarket_slug="will-trump-win-2028", kalshi_ticker="KXPRES-28-DJT", polymarket_outcome="Yes"), ] MIN_EDGE_USD = float(os.getenv("MIN_EDGE_USD", "0.02")) # higher than single-venue: more friction MAX_TRADE_USD = float(os.getenv("MAX_TRADE_USD", "50")) SCAN_INTERVAL_SEC = float(os.getenv("SCAN_INTERVAL_SEC", "15")) DRY_RUN = os.getenv("DRY_RUN", "true").lower() != "false" USE_REAL_KALSHI = os.getenv("USE_REAL_KALSHI", "false").lower() == "true" # Fees (round-trip, both sides). Calibrate to your account tier. POLY_FEE_BPS = 200 # 2% KALSHI_FEE_BPS = 100 # 1% TOTAL_FEE = (POLY_FEE_BPS + KALSHI_FEE_BPS) / 10_000 # --------------------------------------------------------------------------- # Kalshi adapter — STUB. Replace with real client (kalshi-python) when ready. # --------------------------------------------------------------------------- class KalshiClient: """Minimal Kalshi adapter. Use the official kalshi-python SDK in prod.""" def __init__(self, use_real: bool = False): self.use_real = use_real self._http_started = False async def get_best_quote(self, ticker: str) -> tuple[Optional[float], Optional[float]]: """Returns (best_yes_ask, best_no_ask) in dollars.""" if not self.use_real: # Stubbed — returns mid-market dummy values for paper testing. return (0.50, 0.50) # Real implementation: GET https://api.elections.kalshi.com/trade-api/v2/markets/{ticker}/orderbook http = await get_http() url = f"https://api.elections.kalshi.com/trade-api/v2/markets/{ticker}/orderbook" try: r = await http.get(url, headers={"Accept": "application/json"}) r.raise_for_status() data = r.json() ob = data.get("orderbook", {}) yes_asks = ob.get("yes") or [] no_asks = ob.get("no") or [] yes = (yes_asks[0][0] / 100.0) if yes_asks else None # Kalshi uses cents no = (no_asks[0][0] / 100.0) if no_asks else None return yes, no except Exception: return None, None async def place_order(self, ticker: str, side: str, count: int, price_cents: int) -> dict: """side: 'yes' or 'no'. price_cents: 1..99.""" if not self.use_real or DRY_RUN: return {"status": "stubbed", "ticker": ticker, "side": side, "count": count} # Real impl needs auth headers, signed payload — see Kalshi API docs. raise NotImplementedError("Implement Kalshi place_order with kalshi-python SDK") # --------------------------------------------------------------------------- # Strategy # --------------------------------------------------------------------------- async def evaluate_pair(poly_client, kalshi: KalshiClient, pair: Pair) -> Optional[dict]: # Polymarket side pm_market = await fetch_market_by_slug(pair.polymarket_slug) if not pm_market: return None yes_id, no_id = parse_token_ids(pm_market) target_token = yes_id if pair.polymarket_outcome.lower() == "yes" else no_id pm_book = await asyncio.to_thread(get_market_book, poly_client, target_token) pm_asks = pm_book.get("asks") or [] pm_bids = pm_book.get("bids") or [] if not pm_asks or not pm_bids: return None pm_ask = float(pm_asks[0]["price"]) pm_bid = float(pm_bids[0]["price"]) # Kalshi side k_yes_ask, k_no_ask = await kalshi.get_best_quote(pair.kalshi_ticker) if k_yes_ask is None or k_no_ask is None: return None # Best bid on Kalshi YES = 1 - best ask on NO (binary identity) k_yes_bid = 1.0 - k_no_ask # Two directions: # (a) Buy Polymarket YES, sell Kalshi YES: edge = k_yes_bid - pm_ask # (b) Buy Kalshi YES, sell Polymarket YES: edge = pm_bid - k_yes_ask edge_a = k_yes_bid - pm_ask - TOTAL_FEE edge_b = pm_bid - k_yes_ask - TOTAL_FEE if edge_a > MIN_EDGE_USD and edge_a >= edge_b: return { "direction": "buy_poly_sell_kalshi", "pair": pair.label, "pm_ask": pm_ask, "k_yes_bid": k_yes_bid, "edge": edge_a, "token_id": target_token, } if edge_b > MIN_EDGE_USD: return { "direction": "buy_kalshi_sell_poly", "pair": pair.label, "pm_bid": pm_bid, "k_yes_ask": k_yes_ask, "edge": edge_b, "token_id": target_token, } return None async def main(): load_dotenv() log = JsonlLogger("logs/cross_arb_v1") print("=" * 60) print("Cross-Platform Arbitrage (Polymarket ⟷ Kalshi)") print("=" * 60) print(f"Mode: {'DRY-RUN' if DRY_RUN else '🔴 LIVE'}") print(f"Kalshi: {'REAL' if USE_REAL_KALSHI else 'STUB (paper)'}") print(f"Pairs loaded: {len(PAIRS)}") print(f"Min edge: ${MIN_EDGE_USD} Max trade: ${MAX_TRADE_USD}") print("=" * 60) if not PAIRS: print("⚠️ No pairs configured. Edit PAIRS list and re-run.") return client = make_client() healthcheck(client) kalshi = KalshiClient(use_real=USE_REAL_KALSHI) risk = RiskLayer.from_env(strategy_id="cross_arb_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()) while not stop.is_set(): t0 = time.time() for pair in PAIRS: try: opp = await evaluate_pair(client, kalshi, pair) except Exception as e: log.log("pair_error", pair=pair.label, error=str(e)) continue if not opp: continue print(f"\n🎯 {opp['pair']}: {opp['direction']} edge=${opp['edge']:.4f}") log.log("cross_opportunity", **opp) if DRY_RUN: continue size_usd = min(MAX_TRADE_USD, 50.0) if "buy_poly" in opp["direction"]: price = opp["pm_ask"] size = size_usd / price intent = OrderIntent( token_id=opp["token_id"], side="BUY", size=size, price=price, strategy_id="cross_arb_v1", market_slug=pair.polymarket_slug, ) if not risk.allow(intent): continue # Hedge first: open Kalshi short YES (i.e., buy NO) k_count = int(size) k_price_cents = int(round((1 - opp["k_yes_bid"]) * 100)) await kalshi.place_order(pair.kalshi_ticker, "no", k_count, k_price_cents) # Then fill Polymarket leg r = await asyncio.to_thread( place_fok_with_retry, client, opp["token_id"], "BUY", size, price ) risk.record_fill(intent, r.filled_size) log.log("cross_executed", direction=opp["direction"], poly=r.__dict__, kalshi={"ticker": pair.kalshi_ticker, "side": "no", "count": k_count}) elapsed = time.time() - t0 try: await asyncio.wait_for(stop.wait(), timeout=max(0, SCAN_INTERVAL_SEC - elapsed)) except asyncio.TimeoutError: pass if not DRY_RUN: try: cancel_all(client) except Exception: pass await close_http() print("Done.") if __name__ == "__main__": asyncio.run(main())