Benjamin/Kalshi-Bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

changes to bot.py certificate store

Browse Source
This commit is contained in:
Benjamin Adovasio
2026-02-09 19:56:34 -05:00
parent 3083ff471d
commit dd292741c1
4 changed files with 110 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

249
bot.py
View File

@@ -1,11 +1,10 @@
from __future__ import annotations
import datetime as dt
from http import client
import json
import time
import uuid
from typing import Dict, List, Optional, Tuple
from typing import Dict, Tuple
import yaml
@@ -16,14 +15,43 @@ from spot_feed import SpotFeed
from storage import Storage
# ---------- helpers ----------
def _parse_iso_z(s: str) -> dt.datetime:
# Example: "2023-11-07T05:31:56Z"
"""Parse ISO timestamps ending in Z."""
return dt.datetime.fromisoformat(s.replace("Z", "+00:00"))
def discover_open_crypto_15m_markets(client: KalshiClient, symbols: list[str]) -> dict[str, dict]:
def dollars_to_cents_price(p: float) -> int:
"""Convert $0.00–$1.00 price to 1–99 cents."""
return max(1, min(99, int(round(p * 100))))
def extract_strike_and_rule(market: dict) -> Tuple[float, bool]:
"""
Discover open 15-minute crypto markets by scanning open markets directly.
Returns mapping: symbol -> market_dict (nearest close)
Returns (strike, resolves_yes_if_spot_ge_strike).
"""
strike_type = market.get("strike_type")
if strike_type == "greater":
return float(market["floor_strike"]), True
if strike_type == "less":
return float(market["cap_strike"]), False
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"Unable to determine strike from market: {json.dumps(market)[:400]}")
def discover_open_crypto_15m_markets(
client: KalshiClient,
symbols: list[str],
) -> dict[str, dict]:
"""
Discover nearest-closing open 15-minute crypto markets by scanning open markets directly.
"""
data = client.get_markets(series_ticker=None, status="open", limit=500)
markets = data.get("markets", [])
@@ -33,6 +61,7 @@ def discover_open_crypto_15m_markets(client: KalshiClient, symbols: list[str]) -
for m in markets:
title = (m.get("title") or "").upper()
if "UP OR DOWN" not in title:
continue
if "15" not in title:
@@ -53,99 +82,54 @@ def discover_open_crypto_15m_markets(client: KalshiClient, symbols: list[str]) -
return per_symbol
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))
# ---------- main bot ----------
def main() -> None:
with open("config.yaml", "r") as f:
cfg = yaml.safe_load(f)
mode = cfg["mode"]
mode = cfg["mode"] # "paper" or "live"
symbols = cfg["symbols"]
client = KalshiClient.from_env()
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"]))
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"])
risk = RiskManager(
cfg["daily_loss_limit"],
cfg["max_consecutive_losses"],
)
last_traded_market: Dict[str, str] = {} # symbol -> market_ticker
last_traded_market: Dict[str, str] = {}
print(f"[init] mode={mode} series_map={series_map}")
print(f"[init] mode={mode} symbols={symbols}")
while True:
if risk.trading_halted():
print("[risk] Trading halted by risk manager. Sleeping 60s.")
print("[risk] Trading halted — sleeping 60s")
time.sleep(60)
continue
# refresh spot history
# update spot feed
try:
spot.update()
except Exception as e:
print(f"[spot] Error fetching spot: {e}")
print(f"[spot] update failed: {e}")
time.sleep(5)
continue
now = dt.datetime.now(dt.timezone.utc)
symbols = cfg["symbols"]
markets_by_symbol = discover_open_crypto_15m_markets(client, symbols)
for sym, m in markets_by_symbol.items():
try:
market_ticker = m["ticker"]
close_time = _parse_iso_z(m["close_time"])
# Only fire in a tight window starting at (close_time - lead_seconds)
# Trade window: T−6 minutes for a short window
lead = dt.timedelta(seconds=int(cfg["lead_seconds"]))
window = dt.timedelta(seconds=int(cfg["trade_window_seconds"]))
start = close_time - lead
@@ -154,61 +138,37 @@ def main() -> None:
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
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
)
if not (cfg["min_market_prob"] <= market_prob <= cfg["max_market_prob"]):
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
jump = abs(
spot.returns_over_window(sym, int(cfg["jump_lookback_seconds"]))
)
if jump > cfg["max_abs_jump"]:
continue
# Fair probability
spot_px = spot.latest(sym)
sigma = spot.realized_vol(sym)
time_remaining = max(1.0, (close_time - now).total_seconds())
# IMPORTANT: settlement is the AVERAGE of the final 60 seconds
time_remaining = max(
60.0,
(close_time - now).total_seconds(),
)
fair = fair_prob_threshold(
spot=spot_px,
@@ -219,54 +179,55 @@ def main() -> None:
)
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
)
if edge < cfg["min_edge"]:
continue
# Determine limit price (cents)
improve = int(cfg.get("limit_price_improve_cents", 0))
yes_ask_cents = dollars_to_cents_price(yes_ask)
improve = int(cfg.get("limit_price_improve_cents", 0))
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)
max_cost_cents = int(round(cfg["max_cost_dollars"] * 100))
count = 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"
symbol=sym,
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]}"
client_order_id = f"{sym}-{uuid.uuid4().hex[:10]}"
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}")
print(
f"[PAPER] {sym} {market_ticker} "
f"count={count} limit={limit_cents}c "
f"edge={edge:.3f}"
)
storage.log_order(
market_ticker,
order_id=None,
mode="paper",
status="simulated",
details=f"count={count} limit={limit_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}")
print(
f"[LIVE] {sym} {market_ticker} "
f"count={count} limit={limit_cents}c "
f"edge={edge:.3f}"
)
resp = client.create_order(
ticker=market_ticker,
side="yes",
@@ -278,14 +239,18 @@ def main() -> None:
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])
storage.log_order(
market_ticker,
order_id=order.get("order_id"),
mode="live",
status=order.get("status", "unknown"),
details=json.dumps(order)[:1500],
)
last_traded_market[sym] = market_ticker
except Exception as e:
print(f"[loop] Error for {sym}: {e}")
print(f"[loop] error for {sym}: {e}")
time.sleep(2)

4
spot_feed.py
View File

@@ -4,6 +4,7 @@ import time
from collections import deque
from dataclasses import dataclass
from typing import Deque, Dict, Tuple
import certifi
import requests
@@ -24,10 +25,9 @@ class SpotFeed:
self.history: Dict[str, Deque[SpotPoint]] = {sym: deque() for sym in urls.keys()}
def _fetch(self, url: str) -> float:
r = requests.get(url, timeout=10)
r = requests.get(url, timeout=10, verify=certifi.where())
r.raise_for_status()
data = r.json()
# Coinbase shape: {"data": {"amount": "70428.82", "currency": "USD"}}
return float(data["data"]["amount"])
def update(self) -> Dict[str, float]:

BIN
storage.sqlite-shm Normal file
View File

Binary file not shown.

0
storage.sqlite-wal Normal file
View File