""" Trade microstructure analytics — aggressor classification and markout curves. Lee-Ready algorithm for trade direction classification, plus forward markout analysis: what happens to mid price N seconds after a trade of a given type. """ from __future__ import annotations import numpy as np # ── Aggressor classification ───────────────────────────────── def classify_lee_ready( trade_px: float, mid_at_trade: float, bid_at_trade: float | None = None, ask_at_trade: float | None = None, ) -> str: """Lee-Ready: trade above mid = buy, below mid = sell. At mid: compare to previous tick (quote rule) — if unavailable, compare to bid/ask (trade at bid = sell, at ask = buy). """ if trade_px > mid_at_trade: return "buy" elif trade_px < mid_at_trade: return "sell" else: if ask_at_trade is not None and trade_px >= ask_at_trade: return "buy" if bid_at_trade is not None and trade_px <= bid_at_trade: return "sell" return "unknown" def classify_bulk_lee_ready( trades: list[dict], mids: list[float] | None = None, bids: list[float] | None = None, asks: list[float] | None = None, ) -> list[str]: """Classify a list of trades using Lee-Ready. trades: [{"px": float, ...}, ...] mids: optional list of mid prices at each trade time bids/asks: optional best bid/ask at each trade time """ results = [] for i, trade in enumerate(trades): px = float(trade.get("px", 0)) mid = float(mids[i]) if mids and i < len(mids) else px bid = float(bids[i]) if bids and i < len(bids) else None ask = float(asks[i]) if asks and i < len(asks) else None results.append(classify_lee_ready(px, mid, bid, ask)) return results # ── Markout curves ─────────────────────────────────────────── def compute_markouts( trades: list[dict], mid_prices: list[float], trade_times: list[int], # ms since epoch horizons_ms: list[int] | None = None, ) -> dict: """For each trade, compute mid-price change at specified horizons. Returns: {"buys": {horizon_ms: [markout_values...]}, "sells": {...}, ...} """ if horizons_ms is None: horizons_ms = [100, 500, 1000, 5000, 10000, 30000, 60000] results: dict[str, dict[int, list[float]]] = { "buy": {h: [] for h in horizons_ms}, "sell": {h: [] for h in horizons_ms}, } bids_at_trade = [] asks_at_trade = [] mids_at_trade = [] for i, (trade, mid) in enumerate(zip(trades, mid_prices)): mids_at_trade.append(mid) bids_at_trade.append(mid * 0.9995 if mid > 0 else 0) asks_at_trade.append(mid * 1.0005 if mid > 0 else 0) sides = classify_bulk_lee_ready(trades, mids_at_trade, bids_at_trade, asks_at_trade) for i, (trade, side, t0) in enumerate(zip(trades, sides, trade_times)): base_mid = mid_prices[i] if i < len(mid_prices) else 0 if base_mid <= 0: continue for horizon in horizons_ms: target_ts = t0 + horizon future_mid = base_mid for j in range(i + 1, len(mid_prices)): if trade_times[j] >= target_ts: future_mid = mid_prices[j] break else: if len(mid_prices) > i + 1: future_mid = mid_prices[-1] markout = (future_mid - base_mid) / base_mid * 10000 # bps if side in ("buy", "sell"): results[side][horizon].append(markout) return results def markout_summary( markouts: dict[str, dict[int, list[float]]], ) -> dict: """Summarize markout curves with mean, std, t-stat.""" summary = {} for side in ("buy", "sell"): summary[side] = {} for horizon, vals in markouts.get(side, {}).items(): if not vals: summary[side][horizon] = {"mean": 0, "std": 0, "t_stat": 0, "count": 0} continue a = np.array(vals, dtype=float) a = a[np.isfinite(a)] mean = float(np.mean(a)) std = float(np.std(a, ddof=1)) t_stat = mean / std * np.sqrt(len(a)) if std > 0 else 0 summary[side][horizon] = { "mean_bps": round(mean, 2), "std_bps": round(std, 2), "t_stat": round(t_stat, 3), "count": len(a), } return summary # ── Trade metrics ──────────────────────────────────────────── def trade_volume_profile( trades: list[dict], n_buckets: int = 20, ) -> dict: """Volume profile: trade count and volume by size bucket.""" sizes = [float(t.get("sz", 0)) for t in trades if float(t.get("sz", 0)) > 0] if not sizes: return {"buckets": [], "counts": [], "volumes": []} min_sz, max_sz = min(sizes), max(sizes) if min_sz == max_sz: buckets = [min_sz] else: buckets = np.linspace(min_sz, max_sz, n_buckets + 1).tolist() counts = [0] * n_buckets volumes = [0.0] * n_buckets for sz in sizes: for b in range(n_buckets): if buckets[b] <= sz < buckets[b + 1] or (b == n_buckets - 1 and sz == buckets[b + 1]): counts[b] += 1 volumes[b] += sz break return { "buckets": [round((buckets[i] + buckets[i + 1]) / 2, 6) for i in range(n_buckets)], "counts": counts, "volumes": [round(v, 6) for v in volumes], } def trade_arrival_rate( trade_times_ms: list[int], window_sec: int = 60, ) -> dict: """Trade arrival intensity (trades per second) over rolling windows.""" if not trade_times_ms: return {"mean_rate": 0, "max_rate": 0, "burst_count": 0, "rates": []} t0 = trade_times_ms[0] rates = [] burst_count = 0 window_ms = window_sec * 1000 for start in range(t0, trade_times_ms[-1], window_ms): end = start + window_ms count = sum(1 for t in trade_times_ms if start <= t < end) rate = count / window_sec rates.append(rate) if rate > rates[-2] * 3 if len(rates) > 1 else rate > 10: burst_count += 1 a = np.array(rates, dtype=float) if rates else np.array([0.0]) return { "mean_rate": round(float(np.mean(a)), 3), "max_rate": round(float(np.max(a)), 3), "std_rate": round(float(np.std(a)), 3), "burst_count": burst_count, "rates": [round(r, 3) for r in rates[-100:]], }