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

removed old files to recode

Browse Source
This commit is contained in:
Benjamin Adovasio
2026-02-10 12:00:42 -05:00
parent e2d74bac46
commit 681231ce04
9 changed files with 3 additions and 683 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
.DS_Store vendored Normal file
View File

Binary file not shown.

259
bot.py
View File

@@ -1,259 +0,0 @@
from __future__ import annotations
import datetime as dt
import json
import time
import uuid
from typing import Dict, 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
# ---------- helpers ----------
def _parse_iso_z(s: str) -> dt.datetime:
"""Parse ISO timestamps ending in Z."""
return dt.datetime.fromisoformat(s.replace("Z", "+00:00"))
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]:
"""
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", [])
now = dt.datetime.now(dt.timezone.utc)
per_symbol: dict[str, dict] = {}
for m in markets:
title = (m.get("title") or "").upper()
if "UP OR DOWN" not in title:
continue
if "15" not in title:
continue
for sym in symbols:
if sym.upper() not in title:
continue
close_time = _parse_iso_z(m["close_time"])
if close_time <= now:
continue
prev = per_symbol.get(sym)
if prev is None or close_time < _parse_iso_z(prev["close_time"]):
per_symbol[sym] = m
return per_symbol
# ---------- main bot ----------
def main() -> None:
with open("config.yaml", "r") as f:
cfg = yaml.safe_load(f)
mode = cfg["mode"] # "paper" or "live"
symbols = cfg["symbols"]
client = KalshiClient.from_env()
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] = {}
print(f"[init] mode={mode} symbols={symbols}")
while True:
if risk.trading_halted():
print("[risk] Trading halted — sleeping 60s")
time.sleep(60)
continue
# update spot feed
try:
spot.update()
except Exception as e:
print(f"[spot] update failed: {e}")
time.sleep(5)
continue
now = dt.datetime.now(dt.timezone.utc)
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"])
# 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
end = start + window
if not (start <= now <= end):
continue
if last_traded_market.get(sym) == market_ticker:
continue
market_full = client.get_market(market_ticker)["market"]
strike, yes_if_ge = extract_strike_and_rule(market_full)
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
if spread > float(cfg["max_spread_dollars"]):
continue
if not (cfg["min_market_prob"] <= market_prob <= cfg["max_market_prob"]):
continue
jump = abs(
spot.returns_over_window(sym, int(cfg["jump_lookback_seconds"]))
)
if jump > cfg["max_abs_jump"]:
continue
spot_px = spot.latest(sym)
sigma = spot.realized_vol(sym)
# 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,
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 < cfg["min_edge"]:
continue
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)
max_cost_cents = int(round(cfg["max_cost_dollars"] * 100))
count = max_cost_cents // limit_cents
if count <= 0:
continue
storage.log_decision(
ts=time.time(),
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[:10]}"
if mode == "paper":
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} {market_ticker} "
f"count={count} limit={limit_cents}c "
f"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", {})
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"[heartbeat] {dt.datetime.utcnow().isoformat()}Z")
time.sleep(2)
if __name__ == "__main__":
main()

39
config.yaml
View File

@@ -1,39 +0,0 @@
mode: paper # paper or live
symbols: ["BTC", "ETH", "SOL"]
series_title_contains: "Up or Down - 15 minutes"
category: "crypto"
# When to fire relative to market close time
trade_window_seconds: 20 # only trade within this window starting at (close_time - lead_seconds)
lead_seconds: 360 # 6 minutes
# Guardrails
min_market_prob: 0.80
max_market_prob: 0.97
min_edge: 0.05 # fair_prob - market_prob
max_spread_dollars: 0.02
# Volatility / tail-risk controls (spot feed)
vol_lookback_seconds: 900 # 15 minutes of spot history
jump_lookback_seconds: 120 # 2 minutes
max_abs_jump: 0.0015 # 0.15% over jump_lookback_seconds
# Position sizing
max_cost_dollars: 2.00 # cap per trade (uses buy_max_cost)
# Risk limits
daily_loss_limit: 10.0
max_consecutive_losses: 3
# Pricing
limit_price_improve_cents: 0 # 0 = use current yes_ask; >0 = bid cheaper by N cents
# Coinbase spot feed endpoints (simple + free)
coinbase:
BTC: "https://api.coinbase.com/v2/prices/BTC-USD/spot"
ETH: "https://api.coinbase.com/v2/prices/ETH-USD/spot"
SOL: "https://api.coinbase.com/v2/prices/SOL-USD/spot"
#testing

38
fair_prob.py
View File

@@ -1,38 +0,0 @@
from __future__ import annotations
import math
def _norm_cdf(z: float) -> float:
# Standard normal CDF via erf (no scipy dependency)
return 0.5 * (1.0 + math.erf(z / math.sqrt(2.0)))
def fair_prob_threshold(
*,
spot: float,
strike: float,
sigma_per_second: float,
time_remaining_seconds: float,
resolves_yes_if_spot_ge_strike: bool,
) -> float:
"""
Conservative approximation:
spot(t) ~ Normal(spot, spot*sigma*sqrt(t))
and compute P(spot_T >= strike) or P(spot_T <= strike).
"""
if spot <= 0 or strike <= 0:
return 0.5
t = max(1.0, float(time_remaining_seconds))
sigma = max(1e-9, float(sigma_per_second))
stdev = spot * sigma * math.sqrt(t)
if stdev <= 0:
return 0.5
z = (spot - strike) / stdev
p_ge = _norm_cdf(z) # P(spot_T >= strike)
fair = p_ge if resolves_yes_if_spot_ge_strike else (1.0 - p_ge)
# clip away from certainty (tail risk)
return max(0.03, min(0.97, fair))

148
kalshi_client.py
View File

@@ -1,148 +0,0 @@
from __future__ import annotations
import base64
import dataclasses
import datetime as dt
import os
import time
from typing import Any, Dict, Optional
import requests
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa
@dataclasses.dataclass(frozen=True)
class KalshiConfig:
env: str # "demo" or "prod"
api_key_id: str
private_key_path: str
@property
def base_url(self) -> str:
# Docs show demo at demo-api.kalshi.co and public market data at api.elections.kalshi.com.
# We’ll use:
# - demo trading: https://demo-api.kalshi.co
# - prod/public: https://api.elections.kalshi.com
if self.env.lower() == "demo":
return "https://demo-api.kalshi.co"
return "https://api.elections.kalshi.com"
def _load_private_key_from_file(file_path: str) -> rsa.RSAPrivateKey:
with open(file_path, "rb") as key_file:
private_key = serialization.load_pem_private_key(
key_file.read(),
password=None,
backend=default_backend(),
)
if not isinstance(private_key, rsa.RSAPrivateKey):
raise ValueError("Private key is not an RSA private key")
return private_key
def _sign_pss_text(private_key: rsa.RSAPrivateKey, text: str) -> str:
message = text.encode("utf-8")
signature = private_key.sign(
message,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.DIGEST_LENGTH,
),
hashes.SHA256(),
)
return base64.b64encode(signature).decode("utf-8")
class KalshiClient:
"""
Minimal REST client using Kalshi's signed headers:
KALSHI-ACCESS-KEY, KALSHI-ACCESS-TIMESTAMP (ms), KALSHI-ACCESS-SIGNATURE
Signature = RSA-PSS(SHA256) over: timestamp + METHOD + path_without_query
"""
def __init__(self, cfg: KalshiConfig, session: Optional[requests.Session] = None):
self.cfg = cfg
self.session = session or requests.Session()
self._private_key = _load_private_key_from_file(cfg.private_key_path)
@staticmethod
def from_env() -> "KalshiClient":
env = os.getenv("KALSHI_ENV", "demo")
api_key_id = os.environ["KALSHI_API_KEY_ID"]
key_path = os.environ["KALSHI_PRIVATE_KEY_PATH"]
return KalshiClient(KalshiConfig(env=env, api_key_id=api_key_id, private_key_path=key_path))
def _auth_headers(self, method: str, path: str) -> Dict[str, str]:
ts = str(int(time.time() * 1000))
path_wo_query = path.split("?")[0]
msg = f"{ts}{method.upper()}{path_wo_query}"
sig = _sign_pss_text(self._private_key, msg)
return {
"KALSHI-ACCESS-KEY": self.cfg.api_key_id,
"KALSHI-ACCESS-TIMESTAMP": ts,
"KALSHI-ACCESS-SIGNATURE": sig,
}
def _request(self, method: str, path: str, *, params: Optional[dict] = None, json: Optional[dict] = None, auth: bool = False) -> Dict[str, Any]:
url = self.cfg.base_url + path
headers: Dict[str, str] = {"Content-Type": "application/json"}
if auth:
headers.update(self._auth_headers(method, path))
resp = self.session.request(method=method, url=url, params=params, json=json, headers=headers, timeout=15)
if resp.status_code >= 400:
raise RuntimeError(f"Kalshi API error {resp.status_code}: {resp.text}")
if resp.status_code == 204:
return {}
return resp.json()
# ---- Public market-data endpoints (no auth required in docs) ----
def get_series_list(self, *, category: str) -> Dict[str, Any]:
return self._request("GET", "/trade-api/v2/series", params={"category": category}, auth=False)
def get_markets(self, *, series_ticker: str, status: str = "open", limit: int = 200) -> Dict[str, Any]:
params = {"series_ticker": series_ticker, "status": status, "limit": limit}
return self._request("GET", "/trade-api/v2/markets", params=params, auth=False)
def get_market(self, ticker: str) -> Dict[str, Any]:
return self._request("GET", f"/trade-api/v2/markets/{ticker}", auth=False)
# ---- Trading endpoints (auth required) ----
def create_order(
self,
*,
ticker: str,
side: str,
action: str,
count: int,
yes_price_cents: Optional[int] = None,
no_price_cents: Optional[int] = None,
buy_max_cost_cents: Optional[int] = None,
time_in_force: str = "fill_or_kill",
client_order_id: Optional[str] = None,
) -> Dict[str, Any]:
payload: Dict[str, Any] = {
"ticker": ticker,
"side": side, # "yes" or "no"
"action": action, # "buy" or "sell"
"count": int(count),
"type": "limit",
"time_in_force": time_in_force,
}
if client_order_id:
payload["client_order_id"] = client_order_id
if yes_price_cents is not None:
payload["yes_price"] = int(yes_price_cents)
if no_price_cents is not None:
payload["no_price"] = int(no_price_cents)
if buy_max_cost_cents is not None:
payload["buy_max_cost"] = int(buy_max_cost_cents)
return self._request("POST", "/trade-api/v2/portfolio/orders", json=payload, auth=True)
def get_order(self, order_id: str) -> Dict[str, Any]:
return self._request("GET", f"/trade-api/v2/portfolio/orders/{order_id}", auth=True)
def cancel_order(self, order_id: str) -> Dict[str, Any]:
return self._request("DELETE", f"/trade-api/v2/portfolio/orders/{order_id}", auth=True)

5
requirements.txt
View File

@@ -1,4 +1,5 @@
cryptography>=42.0.0
requests>=2.31.0 requests>=2.31.0
PyYAML>=6.0.1 PyYAML>=6.0.1
s cryptography>=41.0.0
sds

35
risk.py
View File

@@ -1,35 +0,0 @@
from __future__ import annotations
import datetime as dt
class RiskManager:
def __init__(self, daily_loss_limit: float, max_consecutive_losses: int):
self.daily_loss_limit = float(daily_loss_limit)
self.max_consecutive_losses = int(max_consecutive_losses)
self._day = dt.date.today()
self._daily_pnl = 0.0
self._consec_losses = 0
def _roll_day(self) -> None:
today = dt.date.today()
if today != self._day:
self._day = today
self._daily_pnl = 0.0
self._consec_losses = 0
def trading_halted(self) -> bool:
self._roll_day()
if self._daily_pnl <= -self.daily_loss_limit:
return True
if self._consec_losses >= self.max_consecutive_losses:
return True
return False
def record_trade_result(self, pnl: float) -> None:
self._roll_day()
self._daily_pnl += float(pnl)
if pnl < 0:
self._consec_losses += 1
else:
self._consec_losses = 0

98
spot_feed.py
View File

@@ -1,98 +0,0 @@
from __future__ import annotations
import time
from collections import deque
from dataclasses import dataclass
from typing import Deque, Dict, Tuple
import certifi
import requests
@dataclass
class SpotPoint:
ts: float
price: float
class SpotFeed:
"""
Simple polling spot feed (Coinbase spot). Keeps rolling history for volatility + jump checks.
"""
def __init__(self, urls: Dict[str, str], lookback_seconds: int):
self.urls = urls
self.lookback_seconds = lookback_seconds
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, verify=certifi.where())
r.raise_for_status()
data = r.json()
return float(data["data"]["amount"])
def update(self) -> Dict[str, float]:
now = time.time()
out: Dict[str, float] = {}
for sym, url in self.urls.items():
px = self._fetch(url)
out[sym] = px
dq = self.history[sym]
dq.append(SpotPoint(ts=now, price=px))
# trim
cutoff = now - self.lookback_seconds
while dq and dq[0].ts < cutoff:
dq.popleft()
return out
def latest(self, sym: str) -> float:
dq = self.history[sym]
if not dq:
raise RuntimeError(f"No spot data yet for {sym}")
return dq[-1].price
def returns_over_window(self, sym: str, window_seconds: int) -> float:
dq = self.history[sym]
if len(dq) < 2:
return 0.0
now = dq[-1].ts
cutoff = now - window_seconds
# find earliest point >= cutoff
base = dq[0]
for p in dq:
if p.ts >= cutoff:
base = p
break
if base.price <= 0:
return 0.0
return (dq[-1].price / base.price) - 1.0
def realized_vol(self, sym: str) -> float:
"""
Very conservative realized vol estimate from simple returns in the stored history.
Returns a per-second sigma (not annualized).
"""
dq = self.history[sym]
if len(dq) < 3:
return 0.0
rets = []
for i in range(1, len(dq)):
p0 = dq[i - 1].price
p1 = dq[i].price
if p0 > 0:
rets.append((p1 / p0) - 1.0)
if len(rets) < 2:
return 0.0
# compute stddev of returns per sample
mean = sum(rets) / len(rets)
var = sum((r - mean) ** 2 for r in rets) / (len(rets) - 1)
# Estimate average sampling interval
dt_avg = (dq[-1].ts - dq[0].ts) / max(1, (len(dq) - 1))
if dt_avg <= 0:
return 0.0
# Convert return std per sample to per-second sigma
std_per_sample = var ** 0.5
sigma_per_second = std_per_sample / (dt_avg ** 0.5)
return sigma_per_second

64
storage.py
View File

@@ -1,64 +0,0 @@
from __future__ import annotations
import sqlite3
import time
from typing import Any, Dict, Optional
class Storage:
def __init__(self, path: str = "storage.sqlite"):
self.conn = sqlite3.connect(path)
self.conn.execute("PRAGMA journal_mode=WAL;")
self._init()
def _init(self) -> None:
self.conn.execute(
"""
CREATE TABLE IF NOT EXISTS decisions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts REAL NOT NULL,
symbol TEXT NOT NULL,
series_ticker TEXT NOT NULL,
market_ticker TEXT NOT NULL,
close_time TEXT NOT NULL,
strike REAL NOT NULL,
side TEXT NOT NULL,
market_prob REAL NOT NULL,
fair_prob REAL NOT NULL,
edge REAL NOT NULL,
spread REAL NOT NULL,
jump REAL NOT NULL,
reason TEXT NOT NULL
)
"""
)
self.conn.execute(
"""
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts REAL NOT NULL,
market_ticker TEXT NOT NULL,
order_id TEXT,
mode TEXT NOT NULL,
status TEXT NOT NULL,
details TEXT
)
"""
)
self.conn.commit()
def log_decision(self, **row: Any) -> None:
cols = ",".join(row.keys())
qs = ",".join(["?"] * len(row))
self.conn.execute(f"INSERT INTO decisions ({cols}) VALUES ({qs})", list(row.values()))
self.conn.commit()
def log_order(self, market_ticker: str, order_id: Optional[str], mode: str, status: str, details: str = "") -> None:
self.conn.execute(
"INSERT INTO orders (ts, market_ticker, order_id, mode, status, details) VALUES (?, ?, ?, ?, ?, ?)",
(time.time(), market_ticker, order_id, mode, status, details),
)
self.conn.commit()
def close(self) -> None:
self.conn.close()