#!/usr/bin/env python3 """ Volatility Arbitrage: BTC 15-min binaries Kalshi KXBTC15M — IV vs 3-min rolling RV 6-exchange CCXT weighted average spot price No delta hedging — directional IV/RV signal only. """ import os, sys, time, math, base64, hashlib, hmac, asyncio, urllib.parse, json, subprocess, signal, argparse import concurrent.futures import requests from collections import deque from dataclasses import dataclass from typing import Optional, List, Dict from datetime import datetime, timezone from dotenv import load_dotenv from cryptography.hazmat.primitives import serialization, hashes from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.backends import default_backend import aiohttp from aiohttp import web import ccxt load_dotenv() # ── Telegram (non-blocking) ──────────────────────────────────────── _tg_pool = concurrent.futures.ThreadPoolExecutor(max_workers=2, thread_name_prefix="tg") class TelegramNotifier: def __init__(self, bot_token: str, chat_id: str): self.bot_token = bot_token self.chat_id = chat_id self.base_url = f"https://api.telegram.org/bot{bot_token}" def _do_send(self, text: str): try: requests.post(f"{self.base_url}/sendMessage", json={"chat_id": self.chat_id, "text": text, "parse_mode": "HTML"}, timeout=5) except Exception as e: print(f"⚠️ Telegram send failed: {e}") def send(self, text: str): _tg_pool.submit(self._do_send, text) # ── DB (optional) ─────────────────────────────────────────────────── try: from db import get_conn as _get_pg, create_all_tables, write_trade _pg = _get_pg(); create_all_tables(_pg) print("✓ PostgreSQL connected"); PG = True except Exception as e: print(f"⚠️ No PG: {e}"); _pg = None; PG = False def pg(): global _pg if not PG: return None try: _pg.cursor().execute("SELECT 1"); return _pg except Exception: try: _pg = _get_pg(); return _pg except Exception: return None # ── BTC Price Feed (6-exchange CCXT weighted average) ────────────── class BTCPriceFeed: def __init__(self): self.exchanges = { 'cryptocom': ccxt.cryptocom(), 'coinbase': ccxt.coinbase(), 'bitstamp': ccxt.bitstamp(), 'kraken': ccxt.kraken(), 'gemini': ccxt.gemini(), 'bullish': ccxt.bullish(), } print(f"✓ CCXT price feed: {', '.join(self.exchanges.keys())}") def fetch_prices(self) -> Dict[str, Optional[float]]: prices = {} for name, exchange in self.exchanges.items(): try: ticker = exchange.fetch_ticker('BTC/USD') prices[name] = ticker['last'] except: prices[name] = None return prices def get_average_price(self) -> Optional[float]: prices = self.fetch_prices() weighted_sum = 0 total_weight = 0 for name, price in prices.items(): if price is not None: weight = 2 if name in ['coinbase', 'cryptocom'] else 1 weighted_sum += price * weight total_weight += weight if total_weight == 0: return None return weighted_sum / total_weight # ── Binary Option Math (Black-76) ─────────────────────────────────── def N(x): return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0))) def n(x): return math.exp(-0.5*x*x) / math.sqrt(2.0*math.pi) def bin_price(F, K, T, sig): if T <= 0 or sig <= 0: return 1.0 if F > K else 0.0 sT = sig * math.sqrt(T) d1 = (math.log(F/K) + 0.5*sig**2*T) / sT d2 = d1 - sT return N(d2) def bin_delta(F, K, T, sig): if T <= 0 or sig <= 0: return 0.0 sT = sig * math.sqrt(T) d1 = (math.log(F/K) + 0.5*sig**2*T) / sT d2 = d1 - sT return n(d2) / (F * sig * math.sqrt(T)) def bin_gamma(F, K, T, sig): if T <= 0 or sig <= 0: return 0.0 sT = sig * math.sqrt(T) d1 = (math.log(F/K) + 0.5*sig**2*T) / sT d2 = d1 - sT return -n(d2) * d1 / (F**2 * sig**2 * T) def extract_iv(price, F, K, T): """Brent's method to find implied vol from binary option price.""" if price <= 0.01 or price >= 0.99 or T <= 0: return None tol = 0.0001 a, b = 0.01, 10.0 fa = bin_price(F, K, T, a) - price fb = bin_price(F, K, T, b) - price if fa * fb > 0: return None # no root in bracket if abs(fa) < abs(fb): a, b = b, a fa, fb = fb, fa c, fc = a, fa d = b - a mflag = True for _ in range(200): if abs(fb) < tol: return b if abs(b - a) < tol * 0.01: return b if fa != fc and fb != fc: # inverse quadratic interpolation s = (a*fb*fc/((fa-fb)*(fa-fc)) + b*fa*fc/((fb-fa)*(fb-fc)) + c*fa*fb/((fc-fa)*(fc-fb))) else: s = b - fb*(b-a)/(fb-fa) # conditions for bisection fallback if (not ((3*a+b)/4 < s < b or b < s < (3*a+b)/4) or (mflag and abs(s-b) >= abs(b-c)/2) or (not mflag and abs(s-b) >= abs(c-d)/2) or (mflag and abs(b-c) < tol) or (not mflag and abs(c-d) < tol)): s = (a+b)/2 mflag = True else: mflag = False fs = bin_price(F, K, T, s) - price d, c, fc = c, b, fb if fa * fs < 0: b, fb = s, fs else: a, fa = s, fs if abs(fa) < abs(fb): a, b = b, a fa, fb = fb, fa return b def realized_vol(prices, interval_sec): if len(prices) < 10: return None lr = [math.log(prices[i]/prices[i-1]) for i in range(1,len(prices)) if prices[i]>0 and prices[i-1]>0] if len(lr) < 5: return None mu = sum(lr)/len(lr) var = sum((r-mu)**2 for r in lr)/(len(lr)-1) return math.sqrt(var * 365.25*24*3600/interval_sec) # ── Strategy ──────────────────────────────────────────────────────── WARMUP_SAMPLES = 180 # 3 minutes at 1s sampling MARKET_DURATION = 900 # 15-min market = 900 seconds TRADE_WINDOW_START = 180 # start placing orders at 3 min in TRADE_WINDOW_END = 600 # stop placing orders at 10 min in @dataclass class Pos: side: Optional[str] = None contracts: int = 0 avg_price: float = 0.0 market_id: Optional[str] = None entry_iv: float = 0.0 entry_rv: float = 0.0 closing: bool = False class VolArb: def __init__(self, k_key, k_sec): # Kalshi self.k_key = k_key self.k_base = "https://api.elections.kalshi.com/trade-api/v2" self.k_pk = None if k_sec: kd = open(k_sec).read() if os.path.isfile(k_sec) else k_sec self.k_pk = serialization.load_pem_private_key( kd.encode() if isinstance(kd,str) else kd, password=None, backend=default_backend()) print("✓ Kalshi key loaded") # CCXT price feed self.price_feed = BTCPriceFeed() # State self.spot: Optional[float] = None self.spot_hist: deque = deque(maxlen=WARMUP_SAMPLES) self.sample_iv = 1 # sample every 1 second self.last_sample = 0.0 self.strike: Optional[float] = None self.market_id: Optional[str] = os.getenv("KALSHI_MARKET_ID") self.market_interval_minutes: Optional[int] = None self._parse_interval() self.yes_bid: Optional[float] = None self.yes_ask: Optional[float] = None self.mid: Optional[float] = None self.iv: Optional[float] = None self.rv: Optional[float] = None self.tte: Optional[float] = None self.delta = 0.0 self.gamma = 0.0 self.pos = Pos() self.edge_enter = 0.07 # 7% edge to enter self.edge_exit = 0.02 # 2% edge to close position self.contracts = 1 self.rv_price: Optional[float] = None # bin_price using RV self.iv_price: Optional[float] = None # bin_price using IV (= market mid) # Pending maker order self.pending_oid: Optional[str] = None self.pending_side: Optional[str] = None self.pending_price: Optional[float] = None self.pending_count: int = 0 # One trade per market self.traded_this_market: bool = False # Full orderbook for dashboard self.ob_yes: list = [] self.ob_no: list = [] self.queue_position: Optional[int] = None # Time series for dashboard charts (3 min rolling) self.iv_history: deque = deque(maxlen=WARMUP_SAMPLES) self.rv_history: deque = deque(maxlen=WARMUP_SAMPLES) self.ts_history: deque = deque(maxlen=WARMUP_SAMPLES) # Event log self.events: deque = deque(maxlen=100) # Websocket clients self.ws_clients: set = set() # Telegram self.telegram: Optional[TelegramNotifier] = None # Drawdown guard ($15 limit = 1500 cents) self.DRAWDOWN_LIMIT = 1500 self.balance: int = 0 self.peak_balance: int = 0 # RV filter: skip trades if RV outside range self.RV_MIN = 0.05 self.RV_MAX = 0.75 # Follow IV mode: trade WITH the market instead of fading it self.follow_iv = False self.running = False self.active = False self.paused = True self.auto_unpause = False # auto-unpause on market switch self.task: Optional[asyncio.Task] = None # ── Kalshi API ── def _k_sign(self, ts, method, path): if not self.k_pk: return "" msg = ts + method + "/trade-api/v2" + path.split('?')[0] sig = self.k_pk.sign(msg.encode(), padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.DIGEST_LENGTH), hashes.SHA256()) return base64.b64encode(sig).decode() async def _k(self, method, ep, data=None): url = f"{self.k_base}{ep}" ts = str(int(time.time()*1000)) h = {'KALSHI-ACCESS-KEY': self.k_key, 'KALSHI-ACCESS-SIGNATURE': self._k_sign(ts, method, ep), 'KALSHI-ACCESS-TIMESTAMP': ts, 'Content-Type': 'application/json'} try: to = aiohttp.ClientTimeout(total=5) async with aiohttp.ClientSession(timeout=to) as s: fn = {'GET':s.get,'POST':s.post,'DELETE':s.delete}[method] kw = {'headers':h} if method=='POST': kw['json']=data async with fn(url, **kw) as r: return await r.json() or {} except: return {} # ── Data ── async def fetch_spot(self): """Fetch BTC/USD from 6-exchange CCXT weighted average.""" loop = asyncio.get_event_loop() price = await loop.run_in_executor(None, self.price_feed.get_average_price) if price: self.spot = price return self.spot def _parse_interval(self): """Parse market interval from ticker, e.g. KXBTC15M → 15 minutes.""" if not self.market_id: return ticker = self.market_id.split('-')[0] if 'M' in ticker: try: m_idx = ticker.rfind('M') s = "" for i in range(m_idx - 1, -1, -1): if ticker[i].isdigit(): s = ticker[i] + s else: break if s: self.market_interval_minutes = int(s) print(f"✓ Interval: {self.market_interval_minutes} minutes") except: pass def _parse_market_time(self): """Parse expiry hour/minute from market ID, e.g. 26APR012000 → (20, 0).""" parts = self.market_id.split('-') if len(parts) < 2: return None tp = parts[1] try: return (int(tp[7:9]), int(tp[9:11])) except: return None def minutes_into_market(self) -> Optional[float]: """Minutes elapsed since market opened (same as rsibot).""" parts = self.market_id.split('-') if len(parts) < 2: return None tp = parts[1] try: month_map = {'JAN':1,'FEB':2,'MAR':3,'APR':4,'MAY':5,'JUN':6, 'JUL':7,'AUG':8,'SEP':9,'OCT':10,'NOV':11,'DEC':12} year = 2000 + int(tp[0:2]) month = month_map.get(tp[2:5].upper()) day = int(tp[5:7]) hour = int(tp[7:9]) minute = int(tp[9:11]) if not month or not self.market_interval_minutes: return None import datetime as dt_mod expiry = dt_mod.datetime(year, month, day, hour, minute) start = expiry - dt_mod.timedelta(minutes=self.market_interval_minutes) return (dt_mod.datetime.now() - start).total_seconds() / 60 except: return None async def check_market_expiry(self) -> bool: """Check if current market has expired, switch to next if so.""" if not self.market_interval_minutes: return False parts = self.market_id.split('-') if len(parts) < 2: return False tp = parts[1] try: import datetime as dt_mod month_map = {'JAN':1,'FEB':2,'MAR':3,'APR':4,'MAY':5,'JUN':6, 'JUL':7,'AUG':8,'SEP':9,'OCT':10,'NOV':11,'DEC':12} year = 2000 + int(tp[0:2]) month = month_map.get(tp[2:5].upper()) day = int(tp[5:7]) hour = int(tp[7:9]) minute = int(tp[9:11]) if not month: return False expiry = dt_mod.datetime(year, month, day, hour, minute) if dt_mod.datetime.now() >= expiry: return await self.switch_to_next_market() except Exception as e: print(f"⚠️ Error parsing expiry: {e}") return False async def switch_to_next_market(self) -> bool: """Compute next market ID and switch to it (same as rsibot).""" if not self.market_interval_minutes: return False parts = self.market_id.split('-') if len(parts) < 3: return False tp = parts[1] prefix = parts[0] try: import datetime as dt_mod month_map = {'JAN':1,'FEB':2,'MAR':3,'APR':4,'MAY':5,'JUN':6, 'JUL':7,'AUG':8,'SEP':9,'OCT':10,'NOV':11,'DEC':12} rev_month = {v: k for k, v in month_map.items()} year = 2000 + int(tp[0:2]) month = month_map.get(tp[2:5].upper()) day = int(tp[5:7]) hour = int(tp[7:9]) minute = int(tp[9:11]) if not month: return False nxt = (dt_mod.datetime(year, month, day, hour, minute) + dt_mod.timedelta(minutes=self.market_interval_minutes)) next_tp = (f"{nxt.year%100:02d}{rev_month[nxt.month]}" f"{nxt.day:02d}{nxt.hour:02d}{nxt.minute:02d}") next_sfx = f"{nxt.minute:02d}" next_id = f"{prefix}-{next_tp}-{next_sfx}" except Exception as e: print(f"⚠️ Error computing next market: {e}") return False old_id = self.market_id await self.cancel_pending() self.market_id = next_id self.pos = Pos() self.traded_this_market = False self.strike = None self.iv = None self.mid = None self.yes_bid = None self.yes_ask = None self.ob_yes = [] self.ob_no = [] self.paused = True # Fetch new strike, retry every 10s until available print(f"\n🔄 Market switched: {old_id} → {next_id}") for attempt in range(30): # up to 5 min of retries r = await self._k("GET", f"/markets/{self.market_id}") m = r.get("market", r) if m: floor_strike = m.get("floor_strike") if floor_strike is not None: self.strike = float(floor_strike) print(f" Strike: ${self.strike:,.2f}") break if attempt == 0: print(f" Waiting for strike...") await asyncio.sleep(10) if not self.strike: print(f" ⚠️ Could not fetch strike after retries") self.log_event("switch", f"{old_id} → {next_id}") if self.telegram: self.telegram.send(f"🔄 Market switched: {old_id[-12:]} → {next_id[-12:]}") await self.check_balance() if self.auto_unpause and not self.paused: pass # already unpaused (drawdown didn't trigger) elif self.auto_unpause: self.paused = False print(f" ▶️ Auto-resumed for next market") return True async def load_market(self): """Load market from KALSHI_MARKET_ID env var, fetch expiry info.""" if not self.market_id: print("❌ Set KALSHI_MARKET_ID in .env") return False # Parse strike from ticker: KXBTC15M-26APR010030-30 → strike after last dash # But BTC strikes are like 67500, 68000 etc. Parse from market API. r = await self._k("GET", f"/markets/{self.market_id}") m = r.get("market", r) if not m: print(f"❌ Market not found: {self.market_id}") return False # Get strike from floor_strike (same as rsi3.py) floor_strike = m.get("floor_strike") if floor_strike is not None: self.strike = float(floor_strike) print(f" Strike: ${self.strike:,.2f} (floor_strike)") else: print(f"⚠️ No floor_strike in market data for {self.market_id}") # Parse expiry/TTE from market ID (same as rsibot) elapsed_min = self.minutes_into_market() if elapsed_min is not None and self.market_interval_minutes: remaining_min = self.market_interval_minutes - elapsed_min self.tte = max(remaining_min * 60, 1) / (365.25*24*3600) else: self.tte = 1.0 / 8766.0 print(f" Market: {self.market_id}") if elapsed_min is not None: print(f" Elapsed: {elapsed_min:.1f} min ({elapsed_min*60:.0f}s)") print(f" TTE: {(self.market_interval_minutes or 15) - elapsed_min:.1f} min") return True def market_elapsed(self) -> float: """Seconds elapsed since market opened.""" m = self.minutes_into_market() return m * 60 if m is not None else 0 async def fetch_ob(self): if not self.market_id: return False r = await self._k("GET", f"/markets/{self.market_id}/orderbook") ob = r.get("orderbook_fp", {}) yb = ob.get("yes_dollars", []); nb = ob.get("no_dollars", []) self.ob_yes = sorted([[float(p), int(float(s))] for p, s in yb], key=lambda x: x[0], reverse=True) self.ob_no = sorted([[float(p), int(float(s))] for p, s in nb], key=lambda x: x[0], reverse=True) self.yes_bid = self.ob_yes[0][0] if self.ob_yes else None if self.ob_no: self.yes_ask = round(1.0 - self.ob_no[0][0], 4) else: self.yes_ask = None if self.yes_bid is not None and self.yes_ask is not None: self.mid = round((self.yes_bid + self.yes_ask)/2, 4) else: self.mid = self.yes_bid or self.yes_ask return self.mid is not None # ── Calcs ── def calc_iv(self): if self.mid and self.spot and self.strike and self.tte: self.iv = extract_iv(self.mid, self.spot, float(self.strike), self.tte) def calc_rv(self): if len(self.spot_hist) >= 10: self.rv = realized_vol(list(self.spot_hist), self.sample_iv) def calc_greeks(self): if self.spot and self.strike and self.tte and self.iv: self.delta = bin_delta(self.spot, float(self.strike), self.tte, self.iv) self.gamma = bin_gamma(self.spot, float(self.strike), self.tte, self.iv) def calc_edge(self): """Calculate fair price using RV vs market price using IV.""" self.iv_price = None self.rv_price = None if self.spot and self.strike and self.tte: if self.iv: self.iv_price = bin_price(self.spot, float(self.strike), self.tte, self.iv) if self.rv: self.rv_price = bin_price(self.spot, float(self.strike), self.tte, self.rv) def get_edge(self) -> Optional[float]: """Edge = rv_price - iv_price. Positive = YES underpriced, negative = NO underpriced.""" if self.rv_price is not None and self.iv_price is not None: return self.rv_price - self.iv_price return None # ── Trading ── async def kalshi_order(self, side, price, count): d = {"ticker": self.market_id, "side": side, "action": "buy", "count": count, "type": "limit", "client_order_id": f"vol-{side}-{int(time.time()*1000)}", f"{side}_price": int(round(price*100))} r = await self._k("POST", "/portfolio/orders", d) oid = r.get("order",{}).get("order_id") if r else None if oid: conn = pg() if conn: try: write_trade(conn, self.market_id, "placement", side, price, count, "volatility", order_id=oid) except: pass return oid async def cancel_pending(self): if self.pending_oid: await self._k("DELETE", f"/portfolio/orders/{self.pending_oid}") print(f"\n ✗ Cancelled pending {self.pending_side.upper()} order") self.log_event("cancel", f"Cancelled {self.pending_side.upper()} @ ${self.pending_price:.3f}") self.traded_this_market = False self.pending_oid = None self.pending_side = None self.pending_price = None self.pending_count = 0 async def long_vol(self): """Buy vol: place YES limit at the bid (maker).""" if not self.yes_bid or not self.market_id: print("❌ No bid"); return if self.pending_oid: print("⚠️ Already have pending order"); return price = self.yes_bid print(f"\n📈 LONG VOL: {self.contracts} YES @ ${price:.3f} (joining bid)") oid = await self.kalshi_order("yes", price, self.contracts) if oid: self.pending_oid = oid self.pending_side = "yes" self.pending_price = price self.pending_count = self.contracts print(f" ✓ Resting: {oid}") else: print(" ❌ Failed") async def short_vol(self): """Sell vol: place NO limit at the no-bid (maker).""" if not self.yes_ask or not self.market_id: print("❌ No ask"); return if self.pending_oid: print("⚠️ Already have pending order"); return no_price = round(1.0 - self.yes_ask, 4) print(f"\n📉 SHORT VOL: {self.contracts} NO @ ${no_price:.3f} (joining bid)") oid = await self.kalshi_order("no", no_price, self.contracts) if oid: self.pending_oid = oid self.pending_side = "no" self.pending_price = no_price self.pending_count = self.contracts print(f" ✓ Resting: {oid}") else: print(" ❌ Failed") async def check_pending_fill(self): if not self.pending_oid: self.queue_position = None return fills_task = self._k("GET", f"/portfolio/fills?ticker={self.market_id}&order_id={self.pending_oid}&limit=10") queue_task = self._k("GET", f"/portfolio/orders/queue_positions?market_tickers={self.market_id}") results = await asyncio.gather(fills_task, queue_task, return_exceptions=True) # Queue position (same pattern as auto1) self.queue_position = None if results[1] and not isinstance(results[1], Exception): queue_positions = results[1].get("queue_positions") if isinstance(results[1], dict) else None if queue_positions and isinstance(queue_positions, list): for qp in queue_positions: if qp.get("order_id") == self.pending_oid: self.queue_position = int(float( qp.get("queue_position_fp", qp.get("queue_position", "0")))) break # Fill detection via /portfolio/fills (same as auto1) if not results[0] or isinstance(results[0], Exception) or not isinstance(results[0], dict): return fills = results[0].get("fills") if not fills or not isinstance(fills, list): return filled_count = sum(int(float(f.get("count_fp", "0"))) for f in fills) if filled_count >= self.pending_count: side = self.pending_side price = self.pending_price count = self.pending_count is_close = self.pos.closing if is_close: print(f"\n✓ CLOSE FILLED: {count} {side.upper()} @ ${price:.3f}") self.log_event("kalshi_fill", f"CLOSE FILLED {count} {side.upper()} @ ${price:.3f}") if self.telegram: self.telegram.send( f"✅ CLOSE FILL: {count} {side.upper()} @ ${price:.3f}\n" f"{self._tg_vol_info()}") self.pos = Pos() else: print(f"\n✓ FILLED: {count} {side.upper()} @ ${price:.3f}") self.log_event("kalshi_fill", f"FILLED {count} {side.upper()} @ ${price:.3f}") if self.telegram: self.telegram.send( f"✅ FILL: {count} {side.upper()} @ ${price:.3f}\n" f"{self._tg_vol_info()}") self.pos.side = side self.pos.contracts += count self.pos.avg_price = price self.pos.market_id = self.market_id self.pos.entry_iv = self.iv or 0 self.pos.entry_rv = self.rv or 0 conn = pg() if conn: try: write_trade(conn, self.market_id, "close_fill" if is_close else "fill", side, price, count, "volatility", order_id=self.pending_oid) except: pass self.pending_oid = None self.pending_side = None self.pending_price = None self.pending_count = 0 async def follow_bid(self): if not self.pending_oid: return if self.pending_side == "yes": current_bid = self.yes_bid else: current_bid = round(1.0 - self.yes_ask, 4) if self.yes_ask else None if current_bid is None: return if abs(current_bid - self.pending_price) < 0.001: return old_price = self.pending_price amend_data = { "ticker": self.market_id, "side": self.pending_side, "action": "buy", "count": self.pending_count, f"{self.pending_side}_price": int(round(current_bid * 100))} r = await self._k("POST", f"/portfolio/orders/{self.pending_oid}/amend", amend_data) if r and r.get("order") and r["order"].get("status") != "canceled": self.pending_price = current_bid direction = "↑" if current_bid > old_price else "↓" print(f"\n{direction} {self.pending_side.upper()}: ${old_price:.3f} → ${current_bid:.3f}") else: self.pending_oid = None self.pending_side = None self.pending_price = None self.pending_count = 0 async def close_all(self): """Close binary position (maker on opposite side, with bid following).""" await self.cancel_pending() if self.pos.contracts == 0: print("⚠️ No position"); return if self.pos.side == "yes": no_bid = round(1.0 - (self.yes_ask or 0.99), 4) if self.yes_ask else 0.01 close_side, close_price = "no", no_bid else: close_side = "yes" close_price = self.yes_bid or 0.01 print(f"\n🔒 Closing {self.pos.contracts} {self.pos.side.upper()}" f" → {close_side.upper()} @ ${close_price:.3f} (maker)") oid = await self.kalshi_order(close_side, close_price, self.pos.contracts) if oid: self.pending_oid = oid self.pending_side = close_side self.pending_price = close_price self.pending_count = self.pos.contracts self.pos.closing = True print(f" ✓ Close order resting: {oid} (bid following active)") # ── Drawdown guard ── async def check_balance(self): result = await self._k("GET", "/portfolio/balance") if not result or 'balance' not in result: return self.balance = result['balance'] if self.balance > self.peak_balance: self.peak_balance = self.balance drawdown = self.peak_balance - self.balance if drawdown >= self.DRAWDOWN_LIMIT and not self.paused: self.paused = True msg = (f"🚨 DRAWDOWN GUARD: Balance ${self.balance/100:.2f} is " f"${drawdown/100:.2f} below peak ${self.peak_balance/100:.2f} " f"— auto-paused") print(f"\n{msg}") self.log_event("drawdown", msg) if self.telegram: self.telegram.send(msg) # ── Telegram helpers ── def _tg_vol_info(self) -> str: parts = [] if self.iv is not None: parts.append(f"IV: {self.iv*100:.1f}%") if self.rv is not None: parts.append(f"RV: {self.rv*100:.1f}%") if self.iv_price is not None: parts.append(f"IV_price: ${self.iv_price:.3f}") if self.rv_price is not None: parts.append(f"RV_price: ${self.rv_price:.3f}") if self.mid is not None: parts.append(f"Mid: ${self.mid:.3f}") return "\n".join(parts) # ── Event log ── def log_event(self, kind, msg): self.events.append({"ts": time.time(), "kind": kind, "msg": msg}) # ── Websocket ── def build_state(self) -> dict: spread = self.iv_rv_spread() elapsed = self.market_elapsed() in_window = TRADE_WINDOW_START <= elapsed <= TRADE_WINDOW_END return { "spot": self.spot, "strike": self.strike, "market_id": self.market_id, "tte_min": round(self.tte * 365.25 * 24 * 60, 1) if self.tte else None, "yes_bid": self.yes_bid, "yes_ask": self.yes_ask, "mid": self.mid, "iv": round(self.iv * 100, 2) if self.iv else None, "rv": round(self.rv * 100, 2) if self.rv else None, "spread": round(spread, 2) if spread is not None else None, "delta": round(self.delta, 8), "gamma": round(self.gamma, 10), "iv_price": round(self.iv_price, 4) if self.iv_price is not None else None, "rv_price": round(self.rv_price, 4) if self.rv_price is not None else None, "edge": round(self.get_edge(), 4) if self.get_edge() is not None else None, "ob_yes": self.ob_yes[:15], "ob_no": self.ob_no[:15], "pending": { "oid": self.pending_oid, "side": self.pending_side, "price": self.pending_price, "count": self.pending_count, "queue": self.queue_position, } if self.pending_oid else None, "position": { "side": self.pos.side, "contracts": self.pos.contracts, "entry_iv": round(self.pos.entry_iv * 100, 1), "entry_rv": round(self.pos.entry_rv * 100, 1), } if self.pos.contracts > 0 else None, "iv_history": list(self.iv_history), "rv_history": list(self.rv_history), "ts_history": list(self.ts_history), "samples": len(self.spot_hist), "max_samples": WARMUP_SAMPLES, "paused": self.paused, "active": self.active, "events": list(self.events)[-20:], "contracts_per_trade": self.contracts, "threshold": self.edge_enter, "elapsed": round(elapsed, 0), "in_window": in_window, } async def broadcast(self): if not self.ws_clients: return msg = json.dumps(self.build_state()) dead = set() for ws in self.ws_clients: try: await ws.send_str(msg) except: dead.add(ws) self.ws_clients -= dead async def ws_handler(self, request): ws = web.WebSocketResponse() await ws.prepare(request) self.ws_clients.add(ws) try: await ws.send_str(json.dumps(self.build_state())) async for msg in ws: pass finally: self.ws_clients.discard(ws) return ws # ── Loop ── async def refresh(self): await asyncio.gather(self.fetch_spot(), self.fetch_ob(), return_exceptions=True) now = time.time() if self.spot and now - self.last_sample >= self.sample_iv: self.spot_hist.append(self.spot); self.last_sample = now elapsed_min = self.minutes_into_market() if elapsed_min is not None and self.market_interval_minutes: remaining_min = self.market_interval_minutes - elapsed_min self.tte = max(remaining_min * 60, 1) / (365.25*24*3600) self.calc_iv(); self.calc_rv(); self.calc_greeks(); self.calc_edge() ts_now = now self.ts_history.append(ts_now) self.iv_history.append(round(self.iv * 100, 2) if self.iv else None) self.rv_history.append(round(self.rv * 100, 2) if self.rv else None) def status(self): sp = f"${self.spot:,.2f}" if self.spot else "?" k = f"${self.strike:,.2f}" if self.strike else "?" iv = f"{self.iv*100:.1f}%" if self.iv else "?" rv = f"{self.rv*100:.1f}%" if self.rv else "?" if self.iv and self.rv: spread = (self.iv-self.rv)*100 ss = f"{spread:+.1f}%" edge = self.get_edge() sig = "YES" if edge and edge>self.edge_enter else ("NO" if edge and edge<-self.edge_enter else "FLAT") else: ss="?"; sig="WAIT" bd = f"{self.yes_bid:.3f}" if self.yes_bid else "?" ak = f"{self.yes_ask:.3f}" if self.yes_ask else "?" tt = f"{self.tte*365.25*24*60:.0f}m" if self.tte else "?" elapsed = self.market_elapsed() ps = "FLAT" if self.pos.contracts > 0: ps = f"{self.pos.side.upper()}x{self.pos.contracts}" pnd = "" if self.pending_oid: pnd = f" PND:{self.pending_side.upper()}x{self.pending_count}@{self.pending_price:.3f}" pause = " PAUSED" if self.paused else "" window = f" W:{elapsed:.0f}s" if elapsed > 0 else "" print(f"\r[VOL] {sp} K:{k} {tt} | B:{bd}/A:{ak} | " f"IV:{iv} RV:{rv}({len(self.spot_hist)}) {ss} [{sig}] | " f"D:{self.delta:.6f} | {ps}{pnd}{window}{pause}", end="") sys.stdout.flush() def iv_rv_spread(self) -> Optional[float]: if self.iv is not None and self.rv is not None: return (self.iv - self.rv) * 100 return None async def loop(self): ls = time.time() while self.running and self.active: try: # Check if market expired → auto-switch await self.check_market_expiry() await self.refresh() elapsed = self.market_elapsed() in_window = TRADE_WINDOW_START <= elapsed <= TRADE_WINDOW_END if not self.paused: # Check fills await self.check_pending_fill() # Follow bid if order resting if self.pending_oid: await self.follow_bid() edge = self.get_edge() # Cancel resting orders if edge is null (can't compare) if edge is None and self.pending_oid: print(f"\n⚠️ IV/RV null — cancelling resting order") self.log_event("cancel", "IV/RV null — cancelled") await self.cancel_pending() if in_window: # Edge-based entry: # edge > 0 → RV says YES is worth more than IV prices it → buy YES # edge < 0 → RV says NO is worth more than IV prices it → buy NO # But direction also depends on spot vs strike: # edge > 0 + above strike → YES (status quo holds with more vol) # edge > 0 + below strike → YES (vol pushes through) # edge < 0 + above strike → NO (vol overpriced, stays but NO cheap) # edge < 0 + below strike → NO (status quo, stays below) above_strike = self.spot and self.strike and self.spot > self.strike if self.pos.contracts == 0 and not self.pending_oid and not self.traded_this_market and edge is not None: # RV filter: skip if outside 25-75% rv_ok = self.rv is not None and self.RV_MIN <= self.rv <= self.RV_MAX if not rv_ok and abs(edge) > self.edge_enter and self.rv is not None: rv_pct = self.rv * 100 msg = f"⏭ SKIP: RV {rv_pct:.1f}% outside {self.RV_MIN*100:.0f}-{self.RV_MAX*100:.0f}% filter | Edge {edge:+.3f}" print(f"\n{msg}") self.log_event("skip", msg) if self.telegram: self.telegram.send(f"{msg}\n{self._tg_vol_info()}") self.traded_this_market = True # skip counts as the 1 trade attempt elif rv_ok and abs(edge) > self.edge_enter: side_label = "above" if above_strike else "below" # follow_iv flips: trade WITH IV instead of fading it if self.follow_iv: go_yes = edge < -self.edge_enter else: go_yes = edge > self.edge_enter if go_yes: mode = "IV" if self.follow_iv else "RV" print(f"\n📈 Edge {edge:+.3f} ({edge*100:.1f}%), spot {side_label} strike — YES [{mode}]") self.log_event("signal", f"EDGE {edge:+.3f} {side_label} strike → YES [{mode}]") self.traded_this_market = True if self.telegram: self.telegram.send( f"📈 ORDER: YES x{self.contracts} @ ${self.yes_bid:.3f}\n" f"Edge: {edge:+.3f} ({edge*100:.1f}%) | {side_label} strike [{mode}]\n" f"{self._tg_vol_info()}") await self.long_vol() else: mode = "IV" if self.follow_iv else "RV" no_price = round(1.0 - (self.yes_ask or 0.99), 4) if self.yes_ask else 0.01 print(f"\n📉 Edge {edge:+.3f} ({edge*100:.1f}%), spot {side_label} strike — NO [{mode}]") self.log_event("signal", f"EDGE {edge:+.3f} {side_label} strike → NO [{mode}]") self.traded_this_market = True if self.telegram: self.telegram.send( f"📉 ORDER: NO x{self.contracts} @ ${no_price:.3f}\n" f"Edge: {edge:+.3f} ({edge*100:.1f}%) | {side_label} strike [{mode}]\n" f"{self._tg_vol_info()}") await self.short_vol() # Auto-cancel resting order if edge reverts below entry threshold if self.pending_oid and self.pos.contracts == 0 and edge is not None: if abs(edge) < self.edge_enter: print(f"\n↩ Edge reverted to {edge:+.3f} — cancelling") await self.cancel_pending() else: # Outside trading window: cancel any resting orders, pause if self.pending_oid: print(f"\n⏸ Outside window ({elapsed:.0f}s) — cancelling orders") self.log_event("window", f"Outside window ({elapsed:.0f}s) — cancelled") await self.cancel_pending() if elapsed > TRADE_WINDOW_END: print(f"\n⏸ Past trading window ({elapsed:.0f}s > {TRADE_WINDOW_END}s) — waiting for next market") self.log_event("window", f"Past window — paused") self.paused = True if time.time()-ls >= 2: self.status(); ls = time.time() await self.broadcast() await asyncio.sleep(0.67) except Exception as e: print(f"\n❌ Loop error: {e}") import traceback; traceback.print_exc() await asyncio.sleep(1) print(f"\n✓ Loop ended (running={self.running}, active={self.active})") if self.telegram: self.telegram.send(f"⚠️ Loop ended (running={self.running}, active={self.active})") async def warmup(self): """Collect 7 min of 1s spot samples for RV, load market early so IV streams too.""" print(f"\n⏳ Warming up — collecting {WARMUP_SAMPLES} spot samples (7 min)...") print(" Press R to pause/resume, Z to abort, V for status\n") # Fetch initial balance for drawdown tracking await self.check_balance() if self.balance > 0: print(f"💰 Balance: ${self.balance/100:.2f}") # Load market immediately so we can stream IV during warmup print("🔍 Loading market...") market_ok = await self.load_market() if not market_ok: print("⚠️ Market not loaded — will show RV only during warmup") while self.running and len(self.spot_hist) < WARMUP_SAMPLES: # Fetch spot + orderbook in parallel (if market loaded) if market_ok: await asyncio.gather(self.fetch_spot(), self.fetch_ob(), return_exceptions=True) else: await self.fetch_spot() if self.spot: now = time.time() if now - self.last_sample >= self.sample_iv: self.spot_hist.append(self.spot) self.last_sample = now n = len(self.spot_hist) if n >= 10: self.calc_rv() elapsed_min = self.minutes_into_market() if elapsed_min is not None and self.market_interval_minutes: remaining_min = self.market_interval_minutes - elapsed_min self.tte = max(remaining_min * 60, 1) / (365.25*24*3600) self.calc_iv() self.calc_greeks() self.calc_edge() # Append to chart history so IV+RV show during warmup self.ts_history.append(now) self.iv_history.append(round(self.iv * 100, 2) if self.iv else None) self.rv_history.append(round(self.rv * 100, 2) if self.rv else None) iv_str = f" | IV: {self.iv*100:.1f}%" if self.iv else "" rv_str = f" | RV: {self.rv*100:.1f}%" if self.rv else "" print(f"\r [{n}/{WARMUP_SAMPLES}] ${self.spot:,.2f}{rv_str}{iv_str}", end="") sys.stdout.flush() await self.broadcast() await asyncio.sleep(0.67) if not self.running: return self.calc_rv() print(f"\n\n✓ Warmup complete — {len(self.spot_hist)} samples") print(f" RV: {self.rv*100:.1f}%" if self.rv else " RV: computing...") if not market_ok: print("\n🔍 Loading market...") if not await self.load_market(): return print("\n🔍 Fetching orderbook...") if await self.fetch_ob(): if self.yes_bid: print(f" YES bid: ${self.yes_bid:.3f}") if self.yes_ask: print(f" YES ask: ${self.yes_ask:.3f}") self.calc_iv() if self.iv: print(f" IV: {self.iv*100:.1f}%") spread = self.iv_rv_spread() if spread is not None: print(f" IV-RV: {spread:+.1f}%") elapsed = self.market_elapsed() print(f"\n Market elapsed: {elapsed:.0f}s") print(f" Trading window: {TRADE_WINDOW_START}-{TRADE_WINDOW_END}s") self.active = True if self.auto_unpause: self.paused = False print(f"\n✓ Ready — AUTO-TRADING") else: self.paused = True print(f"\n✓ Ready — PAUSED. Press R to start trading.") print(f" Auto-trades when |edge| > {self.edge_enter*100:.1f}%") print(f" Orders only in {TRADE_WINDOW_START}-{TRADE_WINDOW_END}s window") print(f" RV filter: {self.RV_MIN*100:.0f}-{self.RV_MAX*100:.0f}%") if self.telegram: self.telegram.send( f"🚀 Volatility bot started\n" f"Market: {self.market_id}\n" f"Edge threshold: {self.edge_enter*100:.1f}%\n" f"RV filter: {self.RV_MIN*100:.0f}-{self.RV_MAX*100:.0f}%\n" f"Drawdown limit: ${self.DRAWDOWN_LIMIT/100:.0f}") async def main(): parser = argparse.ArgumentParser(description="Volatility Arb — BTC 15-min Binaries") parser.add_argument("--telegram", action="store_true", help="Headless mode: telegram notifications, no dashboard") parser.add_argument("--dashboard", action="store_true", help="Dashboard mode: websocket + frontend UI") parser.add_argument("--iv", action="store_true", help="Follow IV mode: trade WITH the market instead of fading it") parser.add_argument("--account", type=str, default=None, help="Account name (e.g. mom) — reads KALSHI_MOM_API_KEY, TELEGRAM_MOM_BOT_TOKEN etc.") args = parser.parse_args() # Load credentials based on --account flag if args.account: prefix = args.account.upper() kk = os.getenv(f"KALSHI_{prefix}_API_KEY") ks = os.getenv(f"KALSHI_{prefix}_API_SECRET") if not kk or not ks: print(f"❌ Missing KALSHI_{prefix}_API_KEY/KALSHI_{prefix}_API_SECRET"); sys.exit(1) else: kk = os.getenv("KALSHI_API_KEY") ks = os.getenv("KALSHI_API_SECRET") mid = os.getenv("KALSHI_MARKET_ID") if not kk or not ks: print("❌ Missing KALSHI_API_KEY/SECRET"); sys.exit(1) if not mid: print("❌ Missing KALSHI_MARKET_ID in .env"); sys.exit(1) a = VolArb(kk, ks) a.running = True # Telegram mode: headless, auto-unpause, no dashboard if args.telegram: if args.account: prefix = args.account.upper() tg_token = os.getenv(f"TELEGRAM_{prefix}_BOT_TOKEN") or os.getenv("TELEGRAM_BOT_TOKEN") tg_chat = os.getenv(f"TELEGRAM_{prefix}_CHAT_ID") or os.getenv("TELEGRAM_CHAT_ID") else: tg_token = os.getenv("TELEGRAM_BOT_TOKEN") tg_chat = os.getenv("TELEGRAM_CHAT_ID") if not tg_token or not tg_chat: print("❌ Missing TELEGRAM_BOT_TOKEN/TELEGRAM_CHAT_ID"); sys.exit(1) a.telegram = TelegramNotifier(tg_token, tg_chat) a.auto_unpause = True acct = f" ({args.account})" if args.account else "" print(f"✓ Telegram mode{acct} — headless, auto-unpause") if args.iv: a.follow_iv = True print("✓ Follow IV mode — trading WITH the market") # Dashboard mode: websocket + frontend runner = None frontend_proc = None if args.dashboard: app = web.Application() app.router.add_get("/ws", a.ws_handler) runner = web.AppRunner(app) await runner.setup() ws_port = int(os.getenv("WS_PORT", "8765")) site = web.TCPSite(runner, "0.0.0.0", ws_port) await site.start() print(f"✓ Dashboard WS on ws://localhost:{ws_port}/ws") frontend_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "frontend") if os.path.isdir(frontend_dir) and os.path.isfile(os.path.join(frontend_dir, "package.json")): frontend_proc = subprocess.Popen( ["npm", "run", "dev"], cwd=frontend_dir, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, preexec_fn=os.setsid) print(f"✓ Frontend dev server started (http://localhost:5174)") print("\n" + "="*70) print("VOLATILITY ARB — BTC 15-min Binaries") print("CCXT 6-exchange avg spot vs Kalshi IV") print("="*70) print(f"\n KALSHI_MARKET_ID = {mid}") print(f" Trading window = {TRADE_WINDOW_START}-{TRADE_WINDOW_END}s into market") print(f" RV window = {WARMUP_SAMPLES}s (3 min rolling)") print(f" RV filter = 25-75%") print(f" Drawdown limit = $15") mode = "telegram (headless)" if args.telegram else "dashboard" if args.dashboard else "interactive" if args.iv: mode += " + follow-IV" print(f" Mode = {mode}\n") if args.telegram: # Headless: warmup then loop forever await a.warmup() while a.running: if a.task and not a.task.done(): await asyncio.sleep(5) else: # Loop task ended — restart it print("\n⚠️ Loop task ended — restarting") if a.telegram: a.telegram.send("⚠️ Loop task ended — restarting") a.active = True a.task = asyncio.create_task(a.loop()) else: # Interactive mode with commands print("Controls:") print(" [G] Start (3-min warmup → auto-trade)") print(" [V] Vol analysis detail") print(" [N] Cancel pending maker order") print(" [+] More contracts/trade [-] Fewer") print(" [R] Pause/resume") print(" [Z] Close all + stop") print(" [Q] Quit") print("="*70) lp = asyncio.get_event_loop() try: while a.running: c = await lp.run_in_executor(None, lambda: input("Command: ").strip().upper()) if c == "G": await a.warmup() elif c == "V": iv = f"{a.iv*100:.1f}%" if a.iv else "N/A" rv = f"{a.rv*100:.1f}%" if a.rv else "N/A" sp = (a.iv - a.rv) * 100 if a.iv and a.rv else None print(f"\n{'='*50}") if a.spot: print(f" Spot: ${a.spot:,.2f} (6-exchange avg)") if a.strike: print(f" Strike: ${a.strike:,}") if a.mid: print(f" Mid: ${a.mid:.4f}") if a.tte: print(f" TTE: {a.tte*365.25*24*60:.1f} min") elapsed = a.market_elapsed() print(f" Elapsed: {elapsed:.0f}s (window: {TRADE_WINDOW_START}-{TRADE_WINDOW_END}s)") print(f" IV: {iv} (annualized)") print(f" RV: {rv} ({len(a.spot_hist)} samples)") if sp is not None: print(f" IV-RV: {sp:+.1f}%") edge = a.get_edge() if edge is not None: print(f" Edge: {edge:+.3f} ({edge*100:+.1f}%)") if edge > a.edge_enter: print(" Signal: YES") elif edge < -a.edge_enter: print(" Signal: NO") else: print(f" Signal: FLAT (±{a.edge_enter*100:.1f}%)") print(f" Delta: {a.delta:.8f} /contract") print(f" Gamma: {a.gamma:.10f} /contract") print(f" Size: {a.contracts}/trade") if a.pos.contracts > 0: print(f" Pos: {a.pos.side.upper()} x{a.pos.contracts}") print(f" Entry: IV={a.pos.entry_iv*100:.1f}% RV={a.pos.entry_rv*100:.1f}%") if a.pending_oid: print(f" Pending: {a.pending_side.upper()} x{a.pending_count} @ ${a.pending_price:.3f}") print(f"{'='*50}") elif c == "N": await a.cancel_pending() elif c == "+": a.contracts = min(a.contracts + 1, 200) print(f" {a.contracts}/trade") elif c == "-": a.contracts = max(a.contracts - 1, 1) print(f" {a.contracts}/trade") elif c == "R": a.paused = not a.paused if not a.paused: a.peak_balance = a.balance # reset drawdown on resume print(f"\n {'Paused' if a.paused else 'Resumed'}") elif c == "Z": await a.cancel_pending() await a.close_all() a.active = False elif c == "Q": if a.pos.contracts > 0: await a.cancel_pending() await a.close_all() a.running = False a.active = False else: print("❌ Invalid") finally: if frontend_proc: os.killpg(os.getpgid(frontend_proc.pid), signal.SIGTERM) frontend_proc.wait() if runner: await runner.cleanup() sys.exit(0) if __name__ == "__main__": asyncio.run(main())