diff --git a/.env.example b/.env.example new file mode 100644 index 00000000..ddc18030 --- /dev/null +++ b/.env.example @@ -0,0 +1,15 @@ +# Copy this file to .env and fill in your own values. .env is git-ignored; +# never commit real key IDs or private keys. +# +# Create an API key at https://kalshi.com/account/profile (production) or in the +# demo dashboard, then store the downloaded private key OUTSIDE this repository +# and point the *_KEYFILE variable at it with an absolute path. + +# --- Demo environment (demo-api.kalshi.co) --- +DEMO_KEYID=00000000-0000-0000-0000-000000000000 +DEMO_KEYFILE=/absolute/path/to/kalshi-demo-private-key.pem + +# --- Production environment (api.elections.kalshi.com) --- +# Leave these unset until you intend to trade with real funds. +PROD_KEYID= +PROD_KEYFILE= diff --git a/clients.py b/clients.py index ee0ba2da..33326984 100644 --- a/clients.py +++ b/clients.py @@ -1,25 +1,49 @@ -import requests +"""HTTP and WebSocket clients for the Kalshi trade API. + +Authentication uses an RSA key pair. Every request carries the API key ID, a +Unix-millisecond timestamp, and an RSA-PSS/SHA-256 signature computed over the +concatenation ``timestamp + method + path``. The signed path must not include +the query string, which is why it is stripped in :meth:`request_headers`. +""" + import base64 +import json import time -from typing import Any, Dict, Optional -from datetime import datetime, timedelta from enum import Enum -import json - -from requests.exceptions import HTTPError +from typing import Any, Dict, Optional, Tuple -from cryptography.hazmat.primitives import serialization, hashes +import requests +import websockets +from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import padding, rsa -from cryptography.exceptions import InvalidSignature +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry + +# Default socket timeout applied to every HTTP request, as (connect, read) +# seconds. Without an explicit timeout, ``requests`` blocks forever: a stalled +# server or a silently dropped connection hangs the calling thread with no +# recovery path. +DEFAULT_TIMEOUT: Tuple[float, float] = (5.0, 30.0) + +# Retry policy for transient transport and server-side failures. POST is +# deliberately excluded from the retried methods because order placement and +# cancellation are not safe to replay blindly. +RETRY_TOTAL = 3 +RETRY_BACKOFF_FACTOR = 0.5 +RETRY_STATUS_FORCELIST = (429, 500, 502, 503, 504) + +# Minimum spacing between two consecutive API calls, in seconds. +RATE_LIMIT_INTERVAL_SECONDS = 0.1 -import websockets class Environment(Enum): DEMO = "demo" PROD = "prod" + class KalshiBaseClient: """Base client class for interacting with the Kalshi API.""" + def __init__( self, key_id: str, @@ -29,14 +53,31 @@ def __init__( """Initializes the client with the provided API key and private key. Args: - key_id (str): Your Kalshi API key ID. - private_key (rsa.RSAPrivateKey): Your RSA private key. - environment (Environment): The API environment to use (DEMO or PROD). + key_id: Your Kalshi API key ID. + private_key: Your RSA private key. + environment: The API environment to use (DEMO or PROD). + + Raises: + ValueError: If ``key_id`` is empty, if ``private_key`` is missing, + or if ``environment`` is not a known :class:`Environment`. """ + if not key_id: + raise ValueError( + "key_id must be a non-empty string; check that the " + "DEMO_KEYID / PROD_KEYID environment variable is set" + ) + if private_key is None: + raise ValueError("private_key must be a loaded RSA private key") + self.key_id = key_id self.private_key = private_key self.environment = environment - self.last_api_call = datetime.now() + + # Monotonic clock rather than wall-clock time: it never jumps backwards + # on an NTP correction or a DST transition, so the rate limiter can + # neither be pushed into an unbounded sleep nor be tricked into skipping + # its delay entirely. + self.last_api_call = time.monotonic() if self.environment == Environment.DEMO: self.HTTP_BASE_URL = "https://demo-api.kalshi.co" @@ -45,108 +86,185 @@ def __init__( self.HTTP_BASE_URL = "https://api.elections.kalshi.com" self.WS_BASE_URL = "wss://api.elections.kalshi.com" else: - raise ValueError("Invalid environment") + raise ValueError(f"Invalid environment: {environment!r}") - def request_headers(self, method: str, path: str) -> Dict[str, Any]: - """Generates the required authentication headers for API requests.""" - current_time_milliseconds = int(time.time() * 1000) - timestamp_str = str(current_time_milliseconds) - # Remove query params from path - path_parts = path.split('?') + def request_headers(self, method: str, path: str) -> Dict[str, str]: + """Generates the required authentication headers for API requests. - msg_string = timestamp_str + method + path_parts[0] + Args: + method: HTTP method, for example ``"GET"``. Normalised to upper case + so that the signed string always matches what the server + recomputes. + path: Request path, optionally including a query string. + + Returns: + The four headers the Kalshi API requires, plus the JSON content type. + """ + timestamp_str = str(int(time.time() * 1000)) + + # Sign the path only. Everything from the first "?" onwards is excluded, + # because the Kalshi signature specification covers the path component + # and not the query string. + path_without_query = path.split("?", 1)[0] + + msg_string = timestamp_str + method.upper() + path_without_query signature = self.sign_pss_text(msg_string) - headers = { + return { "Content-Type": "application/json", "KALSHI-ACCESS-KEY": self.key_id, "KALSHI-ACCESS-SIGNATURE": signature, "KALSHI-ACCESS-TIMESTAMP": timestamp_str, } - return headers def sign_pss_text(self, text: str) -> str: - """Signs the text using RSA-PSS and returns the base64 encoded signature.""" - message = text.encode('utf-8') + """Signs ``text`` with RSA-PSS/SHA-256 and base64-encodes the result. + + Raises: + ValueError: If the key cannot produce a signature, for example + because it is too short for a SHA-256 digest plus salt, or + because the object passed in is not an RSA private key. + """ + message = text.encode("utf-8") try: signature = self.private_key.sign( message, padding.PSS( mgf=padding.MGF1(hashes.SHA256()), - salt_length=padding.PSS.DIGEST_LENGTH + salt_length=padding.PSS.DIGEST_LENGTH, ), - hashes.SHA256() + hashes.SHA256(), ) - return base64.b64encode(signature).decode('utf-8') - except InvalidSignature as e: - raise ValueError("RSA sign PSS failed") from e + except (ValueError, TypeError) as exc: + # ``sign`` raises ValueError for an unusable key/padding combination + # and TypeError when the object is not an RSA private key. It never + # raises InvalidSignature, which is verification-only, so the + # previous handler for it could never fire. The original exception is + # chained so the root cause stays visible in the traceback. + raise ValueError("RSA-PSS signing failed") from exc + return base64.b64encode(signature).decode("utf-8") + class KalshiHttpClient(KalshiBaseClient): """Client for handling HTTP connections to the Kalshi API.""" + def __init__( self, key_id: str, private_key: rsa.RSAPrivateKey, environment: Environment = Environment.DEMO, + timeout: Tuple[float, float] = DEFAULT_TIMEOUT, ): + """Initializes the HTTP client. + + Args: + key_id: Your Kalshi API key ID. + private_key: Your RSA private key. + environment: The API environment to use (DEMO or PROD). + timeout: ``(connect, read)`` timeout in seconds for every request. + """ super().__init__(key_id, private_key, environment) self.host = self.HTTP_BASE_URL + self.timeout = timeout self.exchange_url = "/trade-api/v2/exchange" self.markets_url = "/trade-api/v2/markets" self.portfolio_url = "/trade-api/v2/portfolio" + # One pooled session keeps TLS connections warm across calls and carries + # the retry policy. Only GET and DELETE are replayed on a transient + # failure; replaying POST could duplicate an order. + self.session = requests.Session() + adapter = HTTPAdapter( + max_retries=Retry( + total=RETRY_TOTAL, + backoff_factor=RETRY_BACKOFF_FACTOR, + status_forcelist=RETRY_STATUS_FORCELIST, + allowed_methods=frozenset({"GET", "DELETE"}), + # Let ``raise_if_bad_response`` surface the final status so the + # caller keeps seeing a requests.HTTPError with a real response + # body attached, rather than a urllib3 MaxRetryError. + raise_on_status=False, + ) + ) + self.session.mount("https://", adapter) + + def close(self) -> None: + """Releases the pooled HTTP connections held by this client.""" + self.session.close() + + def __enter__(self) -> "KalshiHttpClient": + return self + + def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: + self.close() + + def rate_limit(self) -> None: - """Built-in rate limiter to prevent exceeding API rate limits.""" - THRESHOLD_IN_MILLISECONDS = 100 - now = datetime.now() - threshold_in_microseconds = 1000 * THRESHOLD_IN_MILLISECONDS - threshold_in_seconds = THRESHOLD_IN_MILLISECONDS / 1000 - if now - self.last_api_call < timedelta(microseconds=threshold_in_microseconds): - time.sleep(threshold_in_seconds) - self.last_api_call = datetime.now() + """Built-in rate limiter to prevent exceeding API rate limits. + + Sleeps only the time still outstanding before the next call is allowed. + The previous implementation slept the full 100 ms threshold even when + 99 ms had already elapsed, which roughly halved achievable throughput. + """ + elapsed = time.monotonic() - self.last_api_call + if elapsed < RATE_LIMIT_INTERVAL_SECONDS: + time.sleep(RATE_LIMIT_INTERVAL_SECONDS - elapsed) + self.last_api_call = time.monotonic() def raise_if_bad_response(self, response: requests.Response) -> None: """Raises an HTTPError if the response status code indicates an error.""" - if response.status_code not in range(200, 299): + # A successful status is 200-299 inclusive. The previous + # ``range(200, 299)`` membership test excluded 299, so a 299 response + # would have been reported as a failure. + if not 200 <= response.status_code < 300: response.raise_for_status() def post(self, path: str, body: dict) -> Any: """Performs an authenticated POST request to the Kalshi API.""" self.rate_limit() - response = requests.post( + response = self.session.post( self.host + path, json=body, - headers=self.request_headers("POST", path) + headers=self.request_headers("POST", path), + timeout=self.timeout, ) self.raise_if_bad_response(response) return response.json() - def get(self, path: str, params: Dict[str, Any] = {}) -> Any: - """Performs an authenticated GET request to the Kalshi API.""" + def get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Any: + """Performs an authenticated GET request to the Kalshi API. + + ``params`` defaults to ``None`` rather than ``{}``: a mutable default is + created once at function-definition time and shared by every call, so + any caller that mutated it would corrupt all subsequent requests. + """ self.rate_limit() - response = requests.get( + response = self.session.get( self.host + path, headers=self.request_headers("GET", path), - params=params + params=params or {}, + timeout=self.timeout, ) self.raise_if_bad_response(response) return response.json() - def delete(self, path: str, params: Dict[str, Any] = {}) -> Any: + def delete(self, path: str, params: Optional[Dict[str, Any]] = None) -> Any: """Performs an authenticated DELETE request to the Kalshi API.""" self.rate_limit() - response = requests.delete( + response = self.session.delete( self.host + path, headers=self.request_headers("DELETE", path), - params=params + params=params or {}, + timeout=self.timeout, ) self.raise_if_bad_response(response) return response.json() + def get_balance(self) -> Dict[str, Any]: """Retrieves the account balance.""" - return self.get(self.portfolio_url + '/balance') + return self.get(self.portfolio_url + "/balance") def get_exchange_status(self) -> Dict[str, Any]: """Retrieves the exchange status.""" @@ -162,18 +280,21 @@ def get_trades( ) -> Dict[str, Any]: """Retrieves trades based on provided filters.""" params = { - 'ticker': ticker, - 'limit': limit, - 'cursor': cursor, - 'max_ts': max_ts, - 'min_ts': min_ts, + "ticker": ticker, + "limit": limit, + "cursor": cursor, + "max_ts": max_ts, + "min_ts": min_ts, } - # Remove None values + # Drop unset filters so the query string only carries what the caller + # actually asked for. params = {k: v for k, v in params.items() if v is not None} - return self.get(self.markets_url + '/trades', params=params) + return self.get(self.markets_url + "/trades", params=params) + class KalshiWebSocketClient(KalshiBaseClient): """Client for handling WebSocket connections to the Kalshi API.""" + def __init__( self, key_id: str, @@ -181,54 +302,84 @@ def __init__( environment: Environment = Environment.DEMO, ): super().__init__(key_id, private_key, environment) - self.ws = None + self.ws: Optional[Any] = None self.url_suffix = "/trade-api/ws/v2" self.message_id = 1 # Add counter for message IDs - async def connect(self): + async def connect(self) -> None: """Establishes a WebSocket connection using authentication.""" host = self.WS_BASE_URL + self.url_suffix auth_headers = self.request_headers("GET", self.url_suffix) - async with websockets.connect(host, additional_headers=auth_headers) as websocket: + # ping_interval / ping_timeout make a half-open connection detectable: + # without keepalive pings a silently dropped TCP session looks identical + # to an idle one and the consumer waits forever for the next message. + async with websockets.connect( + host, + additional_headers=auth_headers, + ping_interval=20, + ping_timeout=20, + close_timeout=10, + ) as websocket: self.ws = websocket - await self.on_open() - await self.handler() + try: + await self.on_open() + await self.handler() + finally: + # Clear the reference on the way out so a later send cannot be + # attempted on a closed socket. + self.ws = None + - async def on_open(self): + async def on_open(self) -> None: """Callback when WebSocket connection is opened.""" print("WebSocket connection opened.") await self.subscribe_to_tickers() - async def subscribe_to_tickers(self): + async def subscribe_to_tickers(self) -> None: """Subscribe to ticker updates for all markets.""" + if self.ws is None: + raise RuntimeError("subscribe_to_tickers called before connect()") subscription_message = { "id": self.message_id, "cmd": "subscribe", - "params": { - "channels": ["ticker"] - } + "params": {"channels": ["ticker"]}, } await self.ws.send(json.dumps(subscription_message)) self.message_id += 1 - async def handler(self): + async def handler(self) -> None: """Handle incoming messages.""" + if self.ws is None: + raise RuntimeError("handler called before connect()") try: async for message in self.ws: await self.on_message(message) - except websockets.ConnectionClosed as e: - await self.on_close(e.code, e.reason) - except Exception as e: - await self.on_error(e) + except websockets.ConnectionClosed as exc: + await self.on_close(exc.code, exc.reason) + except Exception as exc: + # Report the failure, then re-raise. Swallowing it here made every + # unexpected error look like a clean shutdown: ``connect()`` returned + # normally and the caller had no way to distinguish a closed stream + # from a crashed consumer. + await self.on_error(exc) + raise - async def on_message(self, message): + async def on_message(self, message: Any) -> None: """Callback for handling incoming messages.""" print("Received message:", message) - async def on_error(self, error): + async def on_error(self, error: Exception) -> None: """Callback for handling errors.""" print("WebSocket error:", error) - async def on_close(self, close_status_code, close_msg): + async def on_close( + self, close_status_code: Optional[int], close_msg: Optional[str] + ) -> None: """Callback when WebSocket connection is closed.""" - print("WebSocket connection closed with code:", close_status_code, "and message:", close_msg) \ No newline at end of file + print( + "WebSocket connection closed with code:", + close_status_code, + "and message:", + close_msg, + ) + diff --git a/main.py b/main.py index 822403a7..eec66a39 100644 --- a/main.py +++ b/main.py @@ -1,44 +1,115 @@ +"""Example entry point for the Kalshi starter client. + +Reads the API key ID and the private key path from the environment, fetches the +account balance over HTTP, then opens a WebSocket subscription. The work is +inside ``main()`` so that importing this module has no side effects: the previous +version issued a network call and blocked on a WebSocket at import time. +""" + +import asyncio import os -from dotenv import load_dotenv + from cryptography.hazmat.primitives import serialization -import asyncio +from cryptography.hazmat.primitives.asymmetric import rsa +from dotenv import load_dotenv + +from clients import Environment, KalshiHttpClient, KalshiWebSocketClient + +# Toggle the environment here. DEMO is the default so that a misconfigured run +# cannot place real orders against the production exchange. +ENV = Environment.DEMO + + +def load_credentials(env: Environment) -> tuple: + """Resolves the key ID and private key for ``env`` from the environment. + + Returns: + A ``(key_id, private_key)`` tuple. -from clients import KalshiHttpClient, KalshiWebSocketClient, Environment + Raises: + RuntimeError: If either environment variable is unset, or if the key + file is missing or cannot be parsed. The variable names are named + explicitly; the key material itself is never echoed. + """ + prefix = "DEMO" if env == Environment.DEMO else "PROD" + key_id_var = f"{prefix}_KEYID" + key_file_var = f"{prefix}_KEYFILE" -# Load environment variables -load_dotenv() -env = Environment.DEMO # toggle environment here -KEYID = os.getenv('DEMO_KEYID') if env == Environment.DEMO else os.getenv('PROD_KEYID') -KEYFILE = os.getenv('DEMO_KEYFILE') if env == Environment.DEMO else os.getenv('PROD_KEYFILE') + key_id = os.getenv(key_id_var) + key_file = os.getenv(key_file_var) -try: - with open(KEYFILE, "rb") as key_file: - private_key = serialization.load_pem_private_key( - key_file.read(), - password=None # Provide the password if your key is encrypted + # Check both variables before touching the filesystem. Previously a missing + # KEYFILE reached ``open(None, "rb")``, which raises TypeError rather than + # FileNotFoundError, so the friendly "Private key file not found" message + # could never be reached and the operator saw an opaque type error instead. + required = ((key_id_var, key_id), (key_file_var, key_file)) + missing = [name for name, value in required if not value] + if missing: + raise RuntimeError( + f"Missing required environment variable(s): {', '.join(missing)}. " + f"Copy .env.example to .env and fill it in." ) -except FileNotFoundError: - raise FileNotFoundError(f"Private key file not found at {KEYFILE}") -except Exception as e: - raise Exception(f"Error loading private key: {str(e)}") - -# Initialize the HTTP client -client = KalshiHttpClient( - key_id=KEYID, - private_key=private_key, - environment=env -) - -# Get account balance -balance = client.get_balance() -print("Balance:", balance) - -# Initialize the WebSocket client -ws_client = KalshiWebSocketClient( - key_id=KEYID, - private_key=private_key, - environment=env -) - -# Connect via WebSocket -asyncio.run(ws_client.connect()) \ No newline at end of file + + try: + with open(key_file, "rb") as handle: + private_key = serialization.load_pem_private_key( + handle.read(), + password=None, # Provide the password if your key is encrypted. + ) + except FileNotFoundError as exc: + raise RuntimeError( + f"Private key file not found at the path given by {key_file_var}" + ) from exc + except (ValueError, TypeError) as exc: + # ValueError covers malformed PEM and an encrypted key with no password; + # TypeError covers a password supplied for an unencrypted key. The + # message deliberately omits the file contents so key material is never + # written to logs. + raise RuntimeError( + f"Could not parse the private key referenced by {key_file_var}: " + f"{type(exc).__name__}" + ) from exc + + if not isinstance(private_key, rsa.RSAPrivateKey): + # Kalshi request signing is RSA-PSS only. Failing here produces a clear + # message instead of a confusing signature rejection from the server. + raise RuntimeError( + f"The key referenced by {key_file_var} is a " + f"{type(private_key).__name__}, but an RSA private key is required." + ) + + return key_id, private_key + + +def main() -> None: + """Fetches the balance over HTTP, then streams ticker updates.""" + load_dotenv() + key_id, private_key = load_credentials(ENV) + + # The context manager closes the pooled HTTP connections on the way out, + # including when get_balance raises. + with KalshiHttpClient( + key_id=key_id, + private_key=private_key, + environment=ENV, + ) as client: + balance = client.get_balance() + print("Balance:", balance) + + ws_client = KalshiWebSocketClient( + key_id=key_id, + private_key=private_key, + environment=ENV, + ) + + try: + asyncio.run(ws_client.connect()) + except KeyboardInterrupt: + # Ctrl-C during a long-lived subscription is an ordinary way to stop, + # not a failure, so exit quietly instead of dumping a traceback. + print("Interrupted; closing WebSocket.") + + +if __name__ == "__main__": + main() + diff --git a/requirements.txt b/requirements.txt index f6d3832e..4ad99d46 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,23 @@ -requests==2.32.3 +# Pins raised to the minimum versions that clear every advisory reported by +# `pip-audit -r requirements.txt`. Against the pre-change pins that run reported +# 25 rows, which de-duplicate to 15 distinct advisories across 4 packages; after +# this change the same command reports "No known vulnerabilities found". +# Fix versions come from that report, not from guesswork: +# requests 2.32.3 -> PYSEC-2026-1872 (fix 2.32.4), PYSEC-2026-2275 (fix 2.33.0) +# urllib3 2.3.0 -> PYSEC-2026-141/1994/1996/1997/1998/1999 (highest fix 2.7.0) +# cryptography 44.0.2 -> PYSEC-2026-35/2141/3552/3553/3554, GHSA-537c-gmf6-5ccf +# (highest fix 50.0.0) +# python-dotenv 1.0.1 -> PYSEC-2026-2270 (fix 1.2.2) +# Re-run `pip-audit -r requirements.txt` after editing this file. +requests==2.33.0 python-dateutil==2.9.0.post0 -cryptography==44.0.2 -urllib3==2.3.0 -python-dotenv==1.0.1 +cryptography==50.0.0 +urllib3==2.7.0 +python-dotenv==1.2.2 websockets==14.1 -datetime==5.5 + +# The `datetime` PyPI distribution (previously pinned at 5.5) was removed. It is +# a third-party package that shadows nothing in this codebase: `clients.py` and +# `main.py` only need the standard library, and carrying it invites confusion +# with the built-in `datetime` module while adding an unnecessary dependency and +# its transitive `zope.interface` requirement to the supply chain.