File size: 3,374 Bytes
d1cf250 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | """Pure order-book reconstruction for Kalshi binary markets.
Kalshi exposes a market's book as two arrays of *bids*:
- ``yes``: resting bids to buy YES contracts, ``[price_cents, count]``
- ``no`` : resting bids to buy NO contracts, ``[price_cents, count]``
A NO bid at price ``p`` is economically an offer to *sell YES* at ``100 - p``.
So the YES book is:
- YES bid side = the ``yes`` array directly (best bid = highest yes price)
- YES ask side = the ``no`` array mapped to ``ask = 100 - no_price``
(best ask = ``100 - highest no price``)
Prices are integer cents in [1, 99]; the book is kept in cents and features
are returned in dollars (0-1). No I/O and no network here on purpose — the
reconstruction logic is unit-tested independently of the live collector.
"""
from dataclasses import dataclass, field
@dataclass
class KalshiOrderBook:
market_ticker: str
# price_cents -> total resting contract count
yes_bids: dict[int, int] = field(default_factory=dict)
no_bids: dict[int, int] = field(default_factory=dict)
def apply_snapshot(self, yes, no):
"""Replace the whole book. ``yes``/``no`` are lists of ``[price, count]``."""
self.yes_bids = {int(p): int(c) for p, c in yes if int(c) > 0}
self.no_bids = {int(p): int(c) for p, c in no if int(c) > 0}
def apply_delta(self, price, delta, side):
"""Apply an incremental count change to one price level on one side."""
book = self.yes_bids if side == "yes" else self.no_bids
price = int(price)
new_count = book.get(price, 0) + int(delta)
if new_count > 0:
book[price] = new_count
else:
book.pop(price, None)
def features(self):
"""Microstructure features for the current top of book.
Returns ``None`` when the book is one-sided or crossed — those states
are transient artifacts, not tradeable prices, and would only add
noise to a training set.
"""
if not self.yes_bids or not self.no_bids:
return None
best_yes = max(self.yes_bids)
best_no = max(self.no_bids)
bid = best_yes # highest price to SELL a YES into
ask = 100 - best_no # lowest price to BUY a YES from
if bid > ask:
return None # crossed book — drop it
bid_size = self.yes_bids[best_yes]
ask_size = self.no_bids[best_no]
total = bid_size + ask_size
# Microprice: the size-weighted fair value. Heavy bid size pulls it
# toward the ask — a classic short-horizon directional predictor and
# the core feature this whole collection effort exists to study.
microprice = (bid * ask_size + ask * bid_size) / total
return {
"market_ticker": self.market_ticker,
"best_bid": bid / 100,
"best_ask": ask / 100,
"bid_size": bid_size,
"ask_size": ask_size,
"mid": (bid + ask) / 200,
"spread": (ask - bid) / 100,
"microprice": microprice / 100,
"imbalance": bid_size / total,
"yes_depth": sum(self.yes_bids.values()),
"no_depth": sum(self.no_bids.values()),
"n_yes_levels": len(self.yes_bids),
"n_no_levels": len(self.no_bids),
}
|