diff --git a/bot.py b/bot.py new file mode 100644 index 0000000..e36850f --- /dev/null +++ b/bot.py @@ -0,0 +1,283 @@ +from __future__ import annotations + +import datetime as dt +import json +import time +import uuid +from typing import Dict, List, Optional, Tuple + +import yaml + +from fair_prob import fair_prob_threshold +from kalshi_client import KalshiClient +from risk import RiskManager +from spot_feed import SpotFeed +from storage import Storage + + +def _parse_iso_z(s: str) -> dt.datetime: + # Example: "2023-11-07T05:31:56Z" + return dt.datetime.fromisoformat(s.replace("Z", "+00:00")) + + +def discover_crypto_15m_series(client: KalshiClient, category: str, title_contains: str, symbols: List[str]) -> Dict[str, str]: + """ + Returns mapping symbol -> series_ticker by scanning /series?category=crypto. + """ + data = client.get_series_list(category=category) + series = data.get("series", []) + out: Dict[str, str] = {} + for s in series: + title = (s.get("title") or "").upper() + if title_contains.upper() not in title: + continue + for sym in symbols: + if sym.upper() in title and sym.upper() not in out: + out[sym.upper()] = s["ticker"] + return out + + +def choose_current_market(client: KalshiClient, series_ticker: str) -> Optional[dict]: + """ + From open markets in a series, pick the one with the nearest close_time in the future. + """ + data = client.get_markets(series_ticker=series_ticker, status="open", limit=200) + markets = data.get("markets", []) + now = dt.datetime.now(dt.timezone.utc) + + best = None + best_close = None + for m in markets: + close_time = _parse_iso_z(m["close_time"]) + if close_time <= now: + continue + if best_close is None or close_time < best_close: + best = m + best_close = close_time + return best + + +def extract_strike_and_rule(market: dict) -> Tuple[float, bool]: + """ + Returns (strike, resolves_yes_if_spot_ge_strike). + For many threshold markets, Kalshi provides strike_type + floor_strike/cap_strike. + """ + strike_type = market.get("strike_type") # e.g., "greater" or "less" + if strike_type == "greater": + strike = float(market.get("floor_strike")) + return strike, True # YES if spot >= strike + if strike_type == "less": + strike = float(market.get("cap_strike")) + return strike, False # YES if spot <= strike + + # Fallback: if missing, try floor_strike then cap_strike + if market.get("floor_strike") is not None: + return float(market["floor_strike"]), True + if market.get("cap_strike") is not None: + return float(market["cap_strike"]), False + + raise RuntimeError(f"Could not determine strike from market fields: {json.dumps(market)[:500]}") + + +def dollars_to_cents_price(p: float) -> int: + # p is dollars 0.00..1.00; convert to cents 1..99 + c = int(round(p * 100)) + return max(1, min(99, c)) + + +def main() -> None: + with open("config.yaml", "r") as f: + cfg = yaml.safe_load(f) + + mode = cfg["mode"] + client = KalshiClient.from_env() + + series_map = discover_crypto_15m_series( + client, + category=cfg["category"], + title_contains=cfg["series_title_contains"], + symbols=cfg["symbols"], + ) + if len(series_map) == 0: + raise RuntimeError("Could not discover any matching 15-min crypto series. Check title/category filters.") + + spot = SpotFeed(urls=cfg["coinbase"], lookback_seconds=int(cfg["vol_lookback_seconds"])) + storage = Storage("storage.sqlite") + risk = RiskManager(cfg["daily_loss_limit"], cfg["max_consecutive_losses"]) + + last_traded_market: Dict[str, str] = {} # symbol -> market_ticker + + print(f"[init] mode={mode} series_map={series_map}") + + while True: + if risk.trading_halted(): + print("[risk] Trading halted by risk manager. Sleeping 60s.") + time.sleep(60) + continue + + # refresh spot history + try: + spot.update() + except Exception as e: + print(f"[spot] Error fetching spot: {e}") + time.sleep(5) + continue + + now = dt.datetime.now(dt.timezone.utc) + + for sym, series_ticker in series_map.items(): + try: + m = choose_current_market(client, series_ticker) + if not m: + continue + + market_ticker = m["ticker"] + close_time = _parse_iso_z(m["close_time"]) + + # Only fire in a tight window starting at (close_time - lead_seconds) + lead = dt.timedelta(seconds=int(cfg["lead_seconds"])) + window = dt.timedelta(seconds=int(cfg["trade_window_seconds"])) + start = close_time - lead + end = start + window + + if not (start <= now <= end): + continue + + # prevent duplicate trade on same market + if last_traded_market.get(sym) == market_ticker: + continue + + # Pull full market details (strike fields more reliable) + market_full = client.get_market(market_ticker)["market"] + strike, yes_if_ge = extract_strike_and_rule(market_full) + + # Market best prices (dollars strings included) + yes_bid = float(market_full.get("yes_bid_dollars") or 0.0) + yes_ask = float(market_full.get("yes_ask_dollars") or 1.0) + spread = yes_ask - yes_bid + + market_prob = yes_ask # to buy YES, you pay the ask + + # Guardrails: spread + prob bounds + if spread > float(cfg["max_spread_dollars"]): + reason = f"skip: spread {spread:.4f} > max" + storage.log_decision( + ts=time.time(), + symbol=sym, series_ticker=series_ticker, market_ticker=market_ticker, + close_time=m["close_time"], strike=strike, side="yes", + market_prob=market_prob, fair_prob=0.0, edge=0.0, + spread=spread, jump=0.0, reason=reason + ) + continue + + if not (float(cfg["min_market_prob"]) <= market_prob <= float(cfg["max_market_prob"])): + reason = f"skip: market_prob {market_prob:.4f} outside bounds" + storage.log_decision( + ts=time.time(), + symbol=sym, series_ticker=series_ticker, market_ticker=market_ticker, + close_time=m["close_time"], strike=strike, side="yes", + market_prob=market_prob, fair_prob=0.0, edge=0.0, + spread=spread, jump=0.0, reason=reason + ) + continue + + # Jump filter (tail risk) + jump = abs(spot.returns_over_window(sym, int(cfg["jump_lookback_seconds"]))) + if jump > float(cfg["max_abs_jump"]): + reason = f"skip: jump {jump:.5f} > max" + storage.log_decision( + ts=time.time(), + symbol=sym, series_ticker=series_ticker, market_ticker=market_ticker, + close_time=m["close_time"], strike=strike, side="yes", + market_prob=market_prob, fair_prob=0.0, edge=0.0, + spread=spread, jump=jump, reason=reason + ) + continue + + # Fair probability + spot_px = spot.latest(sym) + sigma = spot.realized_vol(sym) + time_remaining = max(1.0, (close_time - now).total_seconds()) + + fair = fair_prob_threshold( + spot=spot_px, + strike=strike, + sigma_per_second=sigma, + time_remaining_seconds=time_remaining, + resolves_yes_if_spot_ge_strike=yes_if_ge, + ) + + edge = fair - market_prob + if edge < float(cfg["min_edge"]): + reason = f"skip: edge {edge:.4f} < min" + storage.log_decision( + ts=time.time(), + symbol=sym, series_ticker=series_ticker, market_ticker=market_ticker, + close_time=m["close_time"], strike=strike, side="yes", + market_prob=market_prob, fair_prob=fair, edge=edge, + spread=spread, jump=jump, reason=reason + ) + continue + + # Determine limit price (cents) + improve = int(cfg.get("limit_price_improve_cents", 0)) + yes_ask_cents = dollars_to_cents_price(yes_ask) + limit_cents = max(1, yes_ask_cents - improve) + + # Size by max_cost + max_cost_cents = int(round(float(cfg["max_cost_dollars"]) * 100)) + # worst-case cost ≈ count * price_cents + count = max(0, max_cost_cents // limit_cents) + if count <= 0: + reason = "skip: count computed as 0" + storage.log_decision( + ts=time.time(), + symbol=sym, series_ticker=series_ticker, market_ticker=market_ticker, + close_time=m["close_time"], strike=strike, side="yes", + market_prob=market_prob, fair_prob=fair, edge=edge, + spread=spread, jump=jump, reason=reason + ) + continue + + # Log decision + storage.log_decision( + ts=time.time(), + symbol=sym, series_ticker=series_ticker, market_ticker=market_ticker, + close_time=m["close_time"], strike=strike, side="yes", + market_prob=market_prob, fair_prob=fair, edge=edge, + spread=spread, jump=jump, reason="trade" + ) + + client_order_id = f"{sym}-{uuid.uuid4().hex[:12]}" + + if mode == "paper": + print(f"[PAPER] {sym} trade {market_ticker} count={count} limit={limit_cents}c max_cost={max_cost_cents}c edge={edge:.3f}") + storage.log_order(market_ticker, order_id=None, mode="paper", status="simulated", + details=f"count={count} yes_price={limit_cents} buy_max_cost={max_cost_cents} edge={edge:.4f}") + else: + print(f"[LIVE] {sym} placing order {market_ticker} count={count} limit={limit_cents}c max_cost={max_cost_cents}c edge={edge:.3f}") + resp = client.create_order( + ticker=market_ticker, + side="yes", + action="buy", + count=count, + yes_price_cents=limit_cents, + buy_max_cost_cents=max_cost_cents, + time_in_force="fill_or_kill", + client_order_id=client_order_id, + ) + order = resp.get("order", {}) + order_id = order.get("order_id") + status = order.get("status", "unknown") + storage.log_order(market_ticker, order_id=order_id, mode="live", status=status, details=json.dumps(order)[:2000]) + + last_traded_market[sym] = market_ticker + + except Exception as e: + print(f"[loop] Error for {sym}: {e}") + + time.sleep(2) + + +if __name__ == "__main__": + main()