From 40006717230ea3ec23a01d6c5f12420fb317357a Mon Sep 17 00:00:00 2001 From: Benjamin Adovasio Date: Mon, 9 Feb 2026 18:28:05 -0500 Subject: [PATCH] added kalshi_client.py --- kalshi_client.py | 148 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 148 insertions(+) create mode 100644 kalshi_client.py diff --git a/kalshi_client.py b/kalshi_client.py new file mode 100644 index 0000000..90f1824 --- /dev/null +++ b/kalshi_client.py @@ -0,0 +1,148 @@ +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)