added README.md

.env.example

@@ -0,0 +1,8 @@
+# Kalshi environment: demo or prod
+export KALSHI_ENV="demo"
+
+# Your API Key ID (UUID shown when you create the key)
+export KALSHI_API_KEY_ID="REPLACE_ME"
+
+# Path to the downloaded private key (.key) from Kalshi
+export KALSHI_PRIVATE_KEY_PATH="/absolute/path/to/kalshi-private.key"

README.md

@@ -0,0 +1,20 @@
+# Kalshi 15m Crypto Probability Bot
+
+Strategy:
+- Every loop, discover the nearest-close open 15-minute "Up or Down" market for BTC/ETH/SOL
+- At T - 6 minutes (configurable), if YES ask or NO ask is between 70% and 95%, place a bet
+- Stake tiers:
+  - 70–80% => $1
+  - 80–90% => $2
+  - 90–95% => $2 (change in config.yaml)
+
+## Setup (Mac / Linux)
+
+```bash
+python3 -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+cp .env.example .env
+# edit .env with your key id + private key path
+source .env
+python bot.py

bot.py

@@ -0,0 +1,327 @@
+from __future__ import annotations
+
+import datetime as dt
+import json
+import time
+import uuid
+from dataclasses import dataclass
+from typing import Dict, List, Optional, Tuple
+
+import yaml
+
+from kalshi_client import KalshiClient, KalshiConfig
+from storage import Storage, DecisionRow
+
+
+def _parse_iso_z(s: str) -> dt.datetime:
+    # example: "2026-02-10T15:29:16Z"
+    return dt.datetime.fromisoformat(s.replace("Z", "+00:00"))
+
+
+def _market_prob_from_asks(m: dict) -> Tuple[Optional[float], Optional[float]]:
+    """
+    Return (yes_ask_prob, no_ask_prob) in 0..1 using whatever fields are present.
+    Kalshi market objects often include yes_ask/yes_bid (cents) and/or yes_ask_dollars (float).
+    """
+    def read_prob(prefix: str) -> Optional[float]:
+        cents_key = f"{prefix}_ask"
+        dollars_key = f"{prefix}_ask_dollars"
+        if m.get(dollars_key) is not None:
+            return float(m[dollars_key])
+        if m.get(cents_key) is not None:
+            return float(m[cents_key]) / 100.0
+        return None
+
+    yes_p = read_prob("yes")
+    no_p = read_prob("no")
+    return yes_p, no_p
+
+
+def _yes_bid_ask_spread(m: dict) -> Tuple[float, float, float]:
+    def read(prefix: str, side: str) -> Optional[float]:
+        cents_key = f"{prefix}_{side}"
+        dollars_key = f"{prefix}_{side}_dollars"
+        if m.get(dollars_key) is not None:
+            return float(m[dollars_key])
+        if m.get(cents_key) is not None:
+            return float(m[cents_key]) / 100.0
+        return None
+
+    yes_bid = read("yes", "bid") or 0.0
+    yes_ask = read("yes", "ask") or 1.0
+    return yes_bid, yes_ask, max(0.0, yes_ask - yes_bid)
+
+
+def _stake_for_prob(prob: float, tiers: List[dict]) -> Optional[float]:
+    for t in tiers:
+        if float(t["min"]) <= prob < float(t["max"]):
+            return float(t["stake_dollars"])
+    # allow exact upper bound match (e.g., prob==0.95)
+    for t in tiers:
+        if abs(prob - float(t["max"])) < 1e-12:
+            return float(t["stake_dollars"])
+    return None
+
+
+def _dollars_to_cents_price(p: float) -> int:
+    c = int(round(p * 100))
+    return max(1, min(99, c))
+
+
+def discover_open_crypto_15m_markets(client: KalshiClient, symbols: list[str]) -> dict[str, dict]:
+    """
+    Scan open markets and pick the nearest-to-close market for each symbol.
+    Uses title heuristics so you don't need series discovery (more robust).
+    """
+    data = client.list_open_markets(limit=1000)
+    markets = data.get("markets") or []
+
+    now = dt.datetime.now(dt.timezone.utc)
+    per_symbol: dict[str, dict] = {}
+
+    for m in markets:
+        title = (m.get("title") or "").upper()
+
+        # Heuristics: 15-min, Up/Down, and the symbol present
+        if "UP OR DOWN" not in title:
+            continue
+        if "15" not in title:
+            continue
+
+        close_time_s = m.get("close_time")
+        if not close_time_s:
+            continue
+        close_time = _parse_iso_z(close_time_s)
+        if close_time <= now:
+            continue
+
+        for sym in symbols:
+            if sym.upper() not in title:
+                continue
+
+            prev = per_symbol.get(sym)
+            if prev is None or _parse_iso_z(prev["close_time"]) > close_time:
+                per_symbol[sym] = m
+
+    return per_symbol
+
+
+def main() -> None:
+    cfg = yaml.safe_load(open("config.yaml", "r"))
+
+    mode = cfg["mode"].lower()
+    symbols = cfg["symbols"]
+    lead_seconds = int(cfg["lead_seconds"])
+    trade_window_seconds = int(cfg["trade_window_seconds"])
+    min_prob = float(cfg["min_prob"])
+    max_prob = float(cfg["max_prob"])
+    tiers = list(cfg["tiers"])
+    max_spread = float(cfg["max_spread_dollars"])
+    improve_cents = int(cfg.get("improve_cents", 0))
+    tif = str(cfg.get("time_in_force", "fill_or_kill"))
+    poll_seconds = float(cfg.get("poll_seconds", 2))
+    storage = Storage(cfg.get("sqlite_path", "storage.sqlite"))
+
+    client = KalshiClient(KalshiConfig(env=str((__import__("os").getenv("KALSHI_ENV") or "demo"))))
+
+    print(f"[init] mode={mode} symbols={symbols} lead_seconds={lead_seconds} window={trade_window_seconds}s")
+
+    last_traded_market: Dict[str, str] = {}  # sym -> market_ticker
+
+    while True:
+        now = dt.datetime.now(dt.timezone.utc)
+        print(f"[heartbeat] {now.isoformat().replace('+00:00','Z')}")
+
+        try:
+            markets_by_symbol = discover_open_crypto_15m_markets(client, symbols)
+        except Exception as e:
+            print(f"[kalshi] market discovery failed: {e}")
+            time.sleep(poll_seconds)
+            continue
+
+        for sym, m in markets_by_symbol.items():
+            try:
+                market_ticker = m["ticker"]
+                close_time = _parse_iso_z(m["close_time"])
+
+                # Fire only in a narrow window at T - lead_seconds
+                start = close_time - dt.timedelta(seconds=lead_seconds)
+                end = start + dt.timedelta(seconds=trade_window_seconds)
+                if not (start <= now <= end):
+                    continue
+
+                # Dedup per symbol per market
+                if last_traded_market.get(sym) == market_ticker:
+                    continue
+
+                # Pull full market object (more reliable fields)
+                market_full = client.get_market(market_ticker).get("market") or m
+
+                yes_p, no_p = _market_prob_from_asks(market_full)
+                if yes_p is None or no_p is None:
+                    storage.log_decision(
+                        DecisionRow(
+                            ts=time.time(),
+                            symbol=sym,
+                            market_ticker=market_ticker,
+                            close_time=market_full.get("close_time", ""),
+                            side="",
+                            prob=0.0,
+                            yes_bid=0.0,
+                            yes_ask=0.0,
+                            spread=0.0,
+                            stake_dollars=0.0,
+                            count=0,
+                            limit_cents=0,
+                            reason="skip: missing yes/no ask fields",
+                        )
+                    )
+                    continue
+
+                yes_bid, yes_ask, spread = _yes_bid_ask_spread(market_full)
+                if spread > max_spread:
+                    storage.log_decision(
+                        DecisionRow(
+                            ts=time.time(), symbol=sym, market_ticker=market_ticker,
+                            close_time=market_full.get("close_time",""),
+                            side="",
+                            prob=0.0,
+                            yes_bid=yes_bid, yes_ask=yes_ask, spread=spread,
+                            stake_dollars=0.0, count=0, limit_cents=0,
+                            reason=f"skip: spread {spread:.4f} > {max_spread:.4f}",
+                        )
+                    )
+                    continue
+
+                # Choose side(s) that qualify: any of the 6 outcomes (YES/NO for each symbol)
+                candidates: List[Tuple[str, float]] = []
+                if min_prob <= yes_p <= max_prob:
+                    candidates.append(("yes", yes_p))
+                if min_prob <= no_p <= max_prob:
+                    candidates.append(("no", no_p))
+
+                if not candidates:
+                    storage.log_decision(
+                        DecisionRow(
+                            ts=time.time(), symbol=sym, market_ticker=market_ticker,
+                            close_time=market_full.get("close_time",""),
+                            side="",
+                            prob=max(yes_p, no_p),
+                            yes_bid=yes_bid, yes_ask=yes_ask, spread=spread,
+                            stake_dollars=0.0, count=0, limit_cents=0,
+                            reason=f"skip: prob not in [{min_prob:.2f},{max_prob:.2f}] (yes={yes_p:.3f} no={no_p:.3f})",
+                        )
+                    )
+                    continue
+
+                # If both somehow qualify (rare/unexpected), prefer the higher probability side
+                side, prob = sorted(candidates, key=lambda x: x[1], reverse=True)[0]
+
+                stake = _stake_for_prob(prob, tiers)
+                if stake is None:
+                    storage.log_decision(
+                        DecisionRow(
+                            ts=time.time(), symbol=sym, market_ticker=market_ticker,
+                            close_time=market_full.get("close_time",""),
+                            side=side,
+                            prob=prob,
+                            yes_bid=yes_bid, yes_ask=yes_ask, spread=spread,
+                            stake_dollars=0.0, count=0, limit_cents=0,
+                            reason="skip: no tier matched prob",
+                        )
+                    )
+                    continue
+
+                # Determine limit price (in cents) using ask for that side
+                if side == "yes":
+                    ask = float(market_full.get("yes_ask_dollars") or (market_full.get("yes_ask", 99) / 100.0))
+                else:
+                    ask = float(market_full.get("no_ask_dollars") or (market_full.get("no_ask", 99) / 100.0))
+
+                ask_cents = _dollars_to_cents_price(ask)
+                limit_cents = max(1, ask_cents - improve_cents)
+
+                # Size contracts to spend up to stake_dollars
+                max_cost_cents = int(round(stake * 100))
+                count = max_cost_cents // limit_cents
+                if count <= 0:
+                    storage.log_decision(
+                        DecisionRow(
+                            ts=time.time(), symbol=sym, market_ticker=market_ticker,
+                            close_time=market_full.get("close_time",""),
+                            side=side, prob=prob,
+                            yes_bid=yes_bid, yes_ask=yes_ask, spread=spread,
+                            stake_dollars=stake, count=0, limit_cents=limit_cents,
+                            reason="skip: count computed as 0",
+                        )
+                    )
+                    continue
+
+                storage.log_decision(
+                    DecisionRow(
+                        ts=time.time(), symbol=sym, market_ticker=market_ticker,
+                        close_time=market_full.get("close_time",""),
+                        side=side, prob=prob,
+                        yes_bid=yes_bid, yes_ask=yes_ask, spread=spread,
+                        stake_dollars=stake, count=count, limit_cents=limit_cents,
+                        reason="trade",
+                    )
+                )
+
+                client_order_id = f"{sym}-{side}-{uuid.uuid4().hex[:12]}"
+
+                if mode == "paper":
+                    print(
+                        f"[PAPER] {sym} {side.upper()} prob={prob:.3f} "
+                        f"{market_ticker} count={count} limit={limit_cents}c stake=${stake:.2f}"
+                    )
+                    storage.log_order(
+                        market_ticker=market_ticker,
+                        mode="paper",
+                        client_order_id=client_order_id,
+                        order_id=None,
+                        status="simulated",
+                        details=json.dumps(
+                            {"symbol": sym, "side": side, "prob": prob, "count": count, "limit_cents": limit_cents, "stake": stake}
+                        ),
+                    )
+                else:
+                    order = {
+                        "ticker": market_ticker,
+                        "action": "buy",
+                        "side": side,
+                        "count": int(count),
+                        "type": "limit",
+                        "client_order_id": client_order_id,
+                        "time_in_force": tif,
+                    }
+                    if side == "yes":
+                        order["yes_price"] = int(limit_cents)
+                    else:
+                        order["no_price"] = int(limit_cents)
+
+                    print(
+                        f"[LIVE] {sym} {side.upper()} prob={prob:.3f} "
+                        f"{market_ticker} count={count} limit={limit_cents}c stake=${stake:.2f}"
+                    )
+                    resp = client.create_order(order)
+                    o = resp.get("order", {})
+                    storage.log_order(
+                        market_ticker=market_ticker,
+                        mode="live",
+                        client_order_id=client_order_id,
+                        order_id=o.get("order_id"),
+                        status=o.get("status", "unknown"),
+                        details=json.dumps(o)[:2000],
+                    )
+
+                last_traded_market[sym] = market_ticker
+
+            except Exception as e:
+                print(f"[loop] error sym={sym}: {e}")
+
+        time.sleep(poll_seconds)
+
+
+if __name__ == "__main__":
+    main()

config.yaml

@@ -0,0 +1,32 @@
+# paper = log-only (no orders placed)
+# live  = real orders placed
+mode: paper
+
+# Crypto symbols to trade
+symbols: ["BTC", "ETH", "SOL"]
+
+# Timing
+lead_seconds: 360            # 6 minutes before close
+trade_window_seconds: 25     # only attempt inside this window (prevents spam)
+
+# Probability filter (derived from ask price)
+min_prob: 0.70
+max_prob: 0.95
+
+# Bet sizing tiers based on implied probability
+# stake_dollars is total max cost you want to spend on that bet
+tiers:
+  - { min: 0.70, max: 0.80, stake_dollars: 1.00 }
+  - { min: 0.80, max: 0.90, stake_dollars: 2.00 }
+  - { min: 0.90, max: 0.95, stake_dollars: 2.00 }
+
+# Microstructure guardrails (optional but recommended)
+max_spread_dollars: 0.08   # skip if yes_ask - yes_bid > this (wide spreads = junk/illiquid)
+
+# Order placement behavior
+time_in_force: "fill_or_kill"   # avoids hanging orders
+improve_cents: 0                # pay ask by default; set 1 to try to improve by 1c
+
+# Logging / persistence
+sqlite_path: "storage.sqlite"
+poll_seconds: 2

kalshi_client.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+import base64
+import datetime as dt
+import os
+from dataclasses import dataclass
+from typing import Any, Dict, Optional
+from urllib.parse import urljoin
+
+import requests
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import padding
+
+
+@dataclass(frozen=True)
+class KalshiConfig:
+    env: str  # "demo" or "prod"
+
+
+class KalshiClient:
+    """
+    Minimal Kalshi REST client (public market data + authenticated trading).
+    Auth scheme: KALSHI-ACCESS-KEY, KALSHI-ACCESS-TIMESTAMP (ms), KALSHI-ACCESS-SIGNATURE
+    where signature is RSA-PSS-SHA256 of: timestamp + METHOD + path_without_query. :contentReference[oaicite:1]{index=1}
+    """
+
+    def __init__(self, cfg: KalshiConfig, session: Optional[requests.Session] = None):
+        self.cfg = cfg
+        self.session = session or requests.Session()
+
+        # Base URL differs for demo vs prod per Kalshi quick-start docs. :contentReference[oaicite:2]{index=2}
+        if cfg.env.lower() == "demo":
+            self.base_url = "https://demo-api.kalshi.co"
+        elif cfg.env.lower() in ("prod", "production"):
+            self.base_url = "https://api.kalshi.com"
+        else:
+            raise ValueError("KALSHI_ENV must be 'demo' or 'prod'")
+
+        # Auth fields are optional in paper mode; only required when placing orders
+        self.api_key_id = os.getenv("KALSHI_API_KEY_ID")
+        self.private_key_path = os.getenv("KALSHI_PRIVATE_KEY_PATH")
+        self._private_key = None
+
+    def _load_private_key(self) -> None:
+        if self._private_key is not None:
+            return
+        if not self.private_key_path:
+            raise RuntimeError("Missing KALSHI_PRIVATE_KEY_PATH (did you source .env?)")
+        with open(self.private_key_path, "rb") as f:
+            self._private_key = serialization.load_pem_private_key(f.read(), password=None)
+
+    def _timestamp_ms(self) -> str:
+        return str(int(dt.datetime.now(dt.timezone.utc).timestamp() * 1000))
+
+    def _sign(self, timestamp_ms: str, method: str, path: str) -> str:
+        """
+        Sign timestamp + METHOD + path_without_query using RSA-PSS(SHA256), return base64 signature.
+        :contentReference[oaicite:3]{index=3}
+        """
+        self._load_private_key()
+        assert self._private_key is not None
+
+        path_wo_query = path.split("?", 1)[0]
+        msg = f"{timestamp_ms}{method.upper()}{path_wo_query}".encode("utf-8")
+
+        sig = self._private_key.sign(
+            msg,
+            padding.PSS(
+                mgf=padding.MGF1(hashes.SHA256()),
+                salt_length=padding.PSS.DIGEST_LENGTH,
+            ),
+            hashes.SHA256(),
+        )
+        return base64.b64encode(sig).decode("utf-8")
+
+    def _auth_headers(self, method: str, path: str) -> Dict[str, str]:
+        if not self.api_key_id:
+            raise RuntimeError("Missing KALSHI_API_KEY_ID (did you source .env?)")
+        ts = self._timestamp_ms()
+        sig = self._sign(ts, method, path)
+        return {
+            "KALSHI-ACCESS-KEY": self.api_key_id,
+            "KALSHI-ACCESS-TIMESTAMP": ts,
+            "KALSHI-ACCESS-SIGNATURE": sig,
+        }
+
+    def get_public(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        url = urljoin(self.base_url, path)
+        r = self.session.get(url, params=params, timeout=20)
+        r.raise_for_status()
+        return r.json()
+
+    def get_authed(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        url = urljoin(self.base_url, path)
+        headers = self._auth_headers("GET", path)
+        r = self.session.get(url, headers=headers, params=params, timeout=20)
+        r.raise_for_status()
+        return r.json()
+
+    def post_authed(self, path: str, json_body: Dict[str, Any]) -> Dict[str, Any]:
+        url = urljoin(self.base_url, path)
+        headers = self._auth_headers("POST", path)
+        headers["Content-Type"] = "application/json"
+        r = self.session.post(url, headers=headers, json=json_body, timeout=20)
+        r.raise_for_status()
+        return r.json()
+
+    # Convenience wrappers
+    def list_open_markets(self, limit: int = 1000) -> Dict[str, Any]:
+        # public market data endpoint per quick-start :contentReference[oaicite:4]{index=4}
+        return self.get_public("/trade-api/v2/markets", params={"status": "open", "limit": str(limit)})
+
+    def get_market(self, ticker: str) -> Dict[str, Any]:
+        return self.get_public(f"/trade-api/v2/markets/{ticker}")
+
+    def create_order(self, order_data: Dict[str, Any]) -> Dict[str, Any]:
+        # POST /trade-api/v2/portfolio/orders per quick-start :contentReference[oaicite:5]{index=5}
+        return self.post_authed("/trade-api/v2/portfolio/orders", order_data)

storage.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+import sqlite3
+import time
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class DecisionRow:
+    ts: float
+    symbol: str
+    market_ticker: str
+    close_time: str
+    side: str
+    prob: float
+    yes_bid: float
+    yes_ask: float
+    spread: float
+    stake_dollars: float
+    count: int
+    limit_cents: int
+    reason: str
+
+
+class Storage:
+    def __init__(self, path: str):
+        self.path = path
+        self._init()
+
+    def _init(self) -> None:
+        con = sqlite3.connect(self.path)
+        cur = con.cursor()
+        cur.execute(
+            """
+            CREATE TABLE IF NOT EXISTS decisions (
+                ts REAL,
+                symbol TEXT,
+                market_ticker TEXT,
+                close_time TEXT,
+                side TEXT,
+                prob REAL,
+                yes_bid REAL,
+                yes_ask REAL,
+                spread REAL,
+                stake_dollars REAL,
+                count INTEGER,
+                limit_cents INTEGER,
+                reason TEXT
+            )
+            """
+        )
+        cur.execute(
+            """
+            CREATE TABLE IF NOT EXISTS orders (
+                ts REAL,
+                market_ticker TEXT,
+                mode TEXT,
+                client_order_id TEXT,
+                order_id TEXT,
+                status TEXT,
+                details TEXT
+            )
+            """
+        )
+        con.commit()
+        con.close()
+
+    def log_decision(self, row: DecisionRow) -> None:
+        con = sqlite3.connect(self.path)
+        cur = con.cursor()
+        cur.execute(
+            """
+            INSERT INTO decisions VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)
+            """,
+            (
+                row.ts,
+                row.symbol,
+                row.market_ticker,
+                row.close_time,
+                row.side,
+                row.prob,
+                row.yes_bid,
+                row.yes_ask,
+                row.spread,
+                row.stake_dollars,
+                row.count,
+                row.limit_cents,
+                row.reason,
+            ),
+        )
+        con.commit()
+        con.close()
+
+    def log_order(
+        self,
+        market_ticker: str,
+        mode: str,
+        client_order_id: str,
+        order_id: Optional[str],
+        status: str,
+        details: str,
+    ) -> None:
+        con = sqlite3.connect(self.path)
+        cur = con.cursor()
+        cur.execute(
+            "INSERT INTO orders VALUES (?,?,?,?,?,?,?)",
+            (time.time(), market_ticker, mode, client_order_id, order_id, status, details),
+        )
+        con.commit()
+        con.close()