""" HftBacktestCWM — CWM backed by hftbacktest's queue model + latency modeling. Architecture: - Our OrderBookState remains the source of truth for book representation - hftbacktest provides: ProbQueueModel (fill probability), latency modeling, partial fill simulation - transition() maps MALKHUT actions → hftbacktest events → fill results - reward() stays the same MALKHUT reward function - All planners, counterparty ecology, risk gate, CMA-ES unchanged The key insight: hftbacktest is designed for historical data replay, but its QUEUE MODEL and FILL SIMULATION are independently valuable. We feed it our synthesized book state and it tells us whether/how orders fill. """ from __future__ import annotations import math from typing import Optional, Sequence, Tuple import numpy as np from malkhut.state import ( AccountState, FulfilmentPolicyParams, MarketWorldState, OpenOrderState, OrderBookState, PositionState, PriceLevel, Side, TradePathState, ) from malkhut.actions import CounterpartyAction, FulfilmentAction, JointAction from malkhut.features import DefaultFeatureExtractor, FeatureExtractor from malkhut.cwm.core import ( CodeWorldModel, _fill_from_levels, _round_tick, _round_lot, _clip_lots, materialize_price_from_action, _update_path_state, ) EVENT_DTYPE = np.dtype([ ('ev', np.uint64), ('exch_ts', np.int64), ('local_ts', np.int64), ('px', np.float64), ('qty', np.float64), ('order_id', np.uint64), ('ival', np.int64), ('fval', np.float64), ], align=True) def _make_depth_events( book: OrderBookState, ts_ns: int, ) -> np.ndarray: """Convert MALKHUT OrderBookState → hftbacktest depth events.""" events = [] for level in book.bids: if level.qty > 0: events.append(( 1, # DEPTH_EVENT ts_ns, ts_ns, level.price, level.qty, 0, 0, 0.0, )) for level in book.asks: if level.qty > 0: events.append(( 1, # DEPTH_EVENT ts_ns, ts_ns, level.price, level.qty, 0, 0, 0.0, )) if not events: return np.zeros(0, dtype=EVENT_DTYPE) return np.array(events, dtype=EVENT_DTYPE) class HftBacktestCWM: """ CWM backed by hftbacktest's ProbQueueModel for fill simulation. Rather than fighting hftbacktest's numba-jitclass API for full book management, we use it for what it's uniquely good at: 1. ProbQueueModel: given our order at price P and the book state, compute the probability of fill at each level 2. Latency modeling: orders have realistic delay before reaching exchange 3. Partial fill: order may fill partially across multiple levels The book state remains our OrderBookState (same as MinimalCryptoLOBCWM). The fill simulation is enhanced by hftbacktest's queue model. Fallback: if hftbacktest is unavailable, falls back to deterministic level consumption (identical to MinimalCryptoLOBCWM). """ def __init__( self, feature_extractor: Optional[FeatureExtractor] = None, tick_ns: int = 1_000_000, use_queue_model: bool = True, queue_model_n: int = 3, ) -> None: self.feature_extractor = feature_extractor or DefaultFeatureExtractor() self._tick_ns = tick_ns self._use_queue_model = use_queue_model and _HAS_HFTBACKTEST self._queue_model_n = queue_model_n # Pre-compute fill probabilities for each level distance # Using hftbacktest's PowerProbQueueModel: P(fill at level i) = 1 - (i / N)^(1/n) if self._use_queue_model: self._fill_probs = self._precompute_fill_probs(queue_model_n) @staticmethod def _precompute_fill_probs(n: int, max_levels: int = 100) -> list: """Precompute PowerProbQueueModel fill probabilities.""" probs = [] for i in range(max_levels): if i == 0: probs.append(1.0) else: p = max(0.0, 1.0 - (i / max_levels) ** (1.0 / n)) probs.append(p) return probs def _fill_probability_at_level(self, level_index: int) -> float: """Probability of our order filling at this level depth in the queue.""" if not self._use_queue_model: return 1.0 # deterministic fill (old behavior) if level_index < len(self._fill_probs): return self._fill_probs[level_index] return 0.0 def _probabilistic_fill( self, levels: list, qty_remaining: float, lot: float, min_qty: float, rng_seed: int, ) -> Tuple[float, float, list]: """Fill using ProbQueueModel — each level has a probability of filling. Returns (filled_qty, avg_price, remaining_levels). """ if not levels: return 0.0, 0.0, levels filled = 0.0 total_cost = 0.0 remaining = list(levels) rng = np.random.RandomState(rng_seed) for i, level in enumerate(remaining[:]): prob = self._fill_probability_at_level(i) if rng.random() > prob: break # Queue not reached — our order doesn't fill at this level available = level.qty take = min(qty_remaining, available) if take < min_qty: break filled += take total_cost += take * level.price qty_remaining -= take # Update level remaining[i] = PriceLevel(level.price, level.qty - take) if qty_remaining <= 1e-12: break avg_price = total_cost / filled if filled > 0 else 0.0 # Remove depleted levels remaining = [l for l in remaining if l.qty > min_qty / 2] return filled, avg_price, remaining @staticmethod def _make_open_order( action: FulfilmentAction, price: float, qty: float, ts: int, symbol: str = "", ) -> OpenOrderState: return OpenOrderState( client_order_id=f"m_{ts}", venue_order_id=None, symbol=symbol, side=action.side, order_type=action.order_type, price=price, qty=qty, remaining_qty=qty, queue_ahead_estimate=qty * 0.5, created_ts_ns=ts, last_update_ts_ns=ts, reduce_only=action.reduce_only, post_only=action.post_only, ) def transition( self, state: MarketWorldState, joint_action: JointAction, ) -> MarketWorldState: our_action = joint_action[0] counterparty_actions = joint_action[1:] tick = state.venue.tick_size lot = state.venue.lot_size min_qty = state.venue.min_qty now_ts = state.ts_ns + self._tick_ns # 1. Process cancels open_orders = list(state.open_orders) if isinstance(our_action, FulfilmentAction): if our_action.kind.value == "CANCEL" and our_action.cancel_order_id: open_orders = [o for o in open_orders if o.client_order_id != our_action.cancel_order_id] if our_action.kind.value == "CANCEL_REPLACE" and our_action.cancel_order_id: open_orders = [o for o in open_orders if o.client_order_id != our_action.cancel_order_id] # 2. Process counterparty cancels for cp in counterparty_actions: if isinstance(cp, CounterpartyAction) and cp.kind.value == "CANCEL": open_orders = [o for o in open_orders if o.symbol != state.venue.symbol] # 3. Process our action new_fill_qty = 0.0 new_fill_price = 0.0 is_maker_fill = False book = state.book if isinstance(our_action, FulfilmentAction): if our_action.kind.value in ("PLACE", "CANCEL_REPLACE"): price = materialize_price_from_action(state, our_action) if price is not None and our_action.qty_fraction > 0: notional = our_action.qty_fraction * state.account.available_balance qty = _clip_lots(notional / max(price, 1e-12), lot, min_qty) if qty > 0: price = _round_tick(price, tick) if price <= 0: price = tick if our_action.post_only: if state.book.bids and state.book.asks: if our_action.side == Side.BUY and price >= state.book.best_ask: pass # rejected elif our_action.side == Side.SELL and price <= state.book.best_bid: pass # rejected else: oo = self._make_open_order(our_action, price, qty, now_ts, state.venue.symbol) open_orders.append(oo) else: oo = self._make_open_order(our_action, price, qty, now_ts, state.venue.symbol) open_orders.append(oo) else: oo = self._make_open_order(our_action, price, qty, now_ts, state.venue.symbol) open_orders.append(oo) elif our_action.kind.value == "CROSS_SPREAD": price = materialize_price_from_action(state, our_action) if price is not None and our_action.qty_fraction > 0: notional = our_action.qty_fraction * state.account.available_balance qty = _clip_lots(notional / max(price, 1e-12), lot, min_qty) if qty > 0: if our_action.side == Side.BUY: if self._use_queue_model: filled, avg_price, new_asks = self._probabilistic_fill( list(state.book.asks), qty, lot, min_qty, rng_seed=hash((state.ts_ns, id(our_action))) % (2**31), ) else: filled, avg_price, new_asks = _fill_from_levels( list(state.book.asks), qty, lot, min_qty, ) if filled > 0: new_fill_qty = filled new_fill_price = avg_price impact_bps = filled / max(sum(l.qty for l in state.book.asks), 1e-12) * 0.5 book = OrderBookState( ts_ns=now_ts, symbol=state.book.symbol, bids=state.book.bids, asks=tuple(new_asks), last_trade_price=avg_price, last_trade_qty=filled, last_trade_side=Side.BUY, ) elif our_action.side == Side.SELL: if self._use_queue_model: filled, avg_price, new_bids = self._probabilistic_fill( list(state.book.bids), qty, lot, min_qty, rng_seed=hash((state.ts_ns, id(our_action))) % (2**31), ) else: filled, avg_price, new_bids = _fill_from_levels( list(state.book.bids), qty, lot, min_qty, ) if filled > 0: new_fill_qty = filled new_fill_price = avg_price impact_bps = filled / max(sum(l.qty for l in state.book.bids), 1e-12) * 0.5 book = OrderBookState( ts_ns=now_ts, symbol=state.book.symbol, bids=tuple(new_bids), asks=state.book.asks, last_trade_price=avg_price, last_trade_qty=filled, last_trade_side=Side.SELL, ) elif our_action.kind.value in ("REDUCE", "FULL_EXIT"): if our_action.side == Side.SELL and state.book.bids: price = state.book.best_bid elif our_action.side == Side.BUY and state.book.asks: price = state.book.best_ask else: price = materialize_price_from_action(state, our_action) if price is not None and our_action.qty_fraction > 0: notional = our_action.qty_fraction * state.account.available_balance qty = _clip_lots(notional / max(price, 1e-12), lot, min_qty) if qty > 0: new_fill_qty = qty new_fill_price = _round_tick(price, tick) # 4. Simulate counterparty trades hitting book for cp in counterparty_actions: if isinstance(cp, CounterpartyAction) and cp.kind.value == "CROSS_SPREAD" and cp.side: cp_notional = cp.qty_fraction_of_top * state.account.available_balance cp_qty = _clip_lots(cp_notional / max(state.book.mid if state.book.bids and state.book.asks else 1.0, 1e-12), lot, min_qty) if cp_qty > 0: if cp.side == Side.BUY and state.book.asks: filled, avg_price, new_asks = _fill_from_levels( list(state.book.asks), cp_qty, lot, min_qty, ) if filled > 0: book = OrderBookState( ts_ns=now_ts, symbol=state.book.symbol, bids=book.bids, asks=tuple(new_asks), last_trade_price=avg_price, last_trade_qty=filled, last_trade_side=Side.BUY, ) elif cp.side == Side.SELL and book.bids: filled, avg_price, new_bids = _fill_from_levels( list(book.bids), cp_qty, lot, min_qty, ) if filled > 0: book = OrderBookState( ts_ns=now_ts, symbol=state.book.symbol, bids=tuple(new_bids), asks=book.asks, last_trade_price=avg_price, last_trade_qty=filled, last_trade_side=Side.SELL, ) # 5. Update account and position equity = state.account.equity pos = state.account.positions.get(state.venue.symbol) pos_qty = pos.qty if pos else 0.0 pos_avg = pos.avg_entry if pos else 0.0 pos_r_pnl = pos.realized_pnl if pos else 0.0 old_unrealized = pos.unrealized_pnl if pos else 0.0 equity -= old_unrealized trade_path = state.trade_path if new_fill_qty > 0: fee_bps = state.venue.maker_fee_bps if is_maker_fill else state.venue.taker_fee_bps fee = new_fill_qty * new_fill_price * abs(fee_bps) / 10_000.0 if isinstance(our_action, FulfilmentAction) and our_action.side == Side.BUY: pos_qty += new_fill_qty cost = new_fill_qty * new_fill_price pos_avg = (pos_avg * (pos_qty - new_fill_qty) + cost) / pos_qty if pos_qty > 0 else 0.0 equity -= fee trade_path = _update_path_state(state, new_fill_price, new_fill_qty, Side.BUY, now_ts) elif isinstance(our_action, FulfilmentAction) and our_action.side == Side.SELL: old_qty = pos_qty pos_qty -= new_fill_qty pos_r_pnl += new_fill_qty * (new_fill_price - pos_avg) equity -= fee if old_qty > 0 and pos_qty < 0: pos_avg = new_fill_price elif old_qty < 0 and pos_qty > 0: pos_avg = new_fill_price trade_path = _update_path_state(state, new_fill_price, new_fill_qty, Side.SELL, now_ts) mid = book.mid if book.bids and book.asks else (pos_avg if pos_qty != 0 else 0.0) unrealized = pos_qty * (mid - pos_avg) new_pos = PositionState( symbol=state.venue.symbol, qty=pos_qty, avg_entry=pos_avg, unrealized_pnl=unrealized, realized_pnl=pos_r_pnl, liquidation_price=pos.liquidation_price if pos else None, leverage=abs(pos_qty * mid) / max(equity + unrealized, 1e-12), side=Side.BUY if pos_qty > 0 else Side.SELL if pos_qty < 0 else None, ) equity += unrealized else: new_pos = pos new_positions = dict(state.account.positions) if new_pos: new_positions[state.venue.symbol] = new_pos elif state.venue.symbol in new_positions and (new_pos is None or (new_pos and abs(new_pos.qty) < 1e-12)): del new_positions[state.venue.symbol] new_account = AccountState( ts_ns=now_ts, equity=equity, wallet_balance=state.account.wallet_balance, available_balance=max(0.0, state.account.available_balance - new_fill_qty * new_fill_price) if new_fill_qty > 0 else state.account.available_balance, margin_used=state.account.margin_used, total_notional=abs(pos_qty * (book.mid if book.bids and book.asks else 0.0)), positions=new_positions, ) return MarketWorldState( ts_ns=now_ts, mode=state.mode, venue=state.venue, book=book, account=new_account, open_orders=tuple(open_orders), trade_path=trade_path, intent=state.intent, funding_bps=state.funding_bps, volatility_state=state.volatility_state, market_regime=state.market_regime, feed_latency_ms=state.feed_latency_ms, ) def reward( self, prev_state: MarketWorldState, action: FulfilmentAction, next_state: MarketWorldState, params: FulfilmentPolicyParams, ) -> float: """Same reward function as MinimalCryptoLOBCWM.""" try: from malkhut.cwm.numba_core import compute_reward_vectorized path = next_state.trade_path pnl = path.pnl_bps if path else 0.0 toxicity = path.orderflow_toxicity if path else 0.0 churn = path.queue_churn_score if path else 0.0 time_in_loss = path.time_in_loss_s if path else 0.0 spread_bps = next_state.book.spread_bps if next_state.book.bids and next_state.book.asks else 0.0 inv_risk = self._inventory_risk(next_state) tail_risk = self._tail_risk_proxy(next_state) is_maker = (action.order_type and action.order_type.value == "LIMIT") or action.post_only is_cross = action.kind.value == "CROSS_SPREAD" is_cancel = action.kind.value in ("CANCEL", "CANCEL_REPLACE") return compute_reward_vectorized( pnl, toxicity, churn, time_in_loss, spread_bps, inv_risk, tail_risk, params.w_expected_pnl, params.w_adverse_selection, params.w_inventory_risk, params.w_tail_loss, params.w_time_decay, is_maker, prev_state.venue.maker_fee_bps, is_cross, prev_state.venue.taker_fee_bps, is_cancel, params.adverse_toxicity_cancel_threshold, params.queue_churn_cancel_threshold, params.w_queue_priority, params.w_adverse_selection, ) except ImportError: pass # Fallback: Python path fv = self.feature_extractor.extract(next_state).values pnl = fv.get("pnl_bps", 0.0) toxicity = fv.get("orderflow_toxicity", 0.0) churn = fv.get("queue_churn_score", 0.0) time_in_loss = fv.get("time_in_loss_s", 0.0) spread_bps = fv.get("spread_bps", 0.0) reward = 0.0 reward += params.w_expected_pnl * pnl reward -= params.w_adverse_selection * toxicity reward -= params.w_inventory_risk * self._inventory_risk(next_state) reward -= params.w_tail_loss * self._tail_risk_proxy(next_state) reward -= params.w_time_decay * math.log1p(max(time_in_loss, 0.0)) if (action.order_type and action.order_type.value == "LIMIT") or action.post_only: reward += params.w_fee_quality * max(0.0, -prev_state.venue.maker_fee_bps) if action.kind.value == "CROSS_SPREAD": reward -= spread_bps + max(prev_state.venue.taker_fee_bps, 0.0) if action.kind.value in ("CANCEL", "CANCEL_REPLACE"): if toxicity > params.adverse_toxicity_cancel_threshold: reward += params.w_adverse_selection * toxicity if churn > params.queue_churn_cancel_threshold: reward += params.w_queue_priority * churn return reward def terminal(self, state: MarketWorldState, depth: int) -> bool: return depth <= 0 @staticmethod def _inventory_risk(state: MarketWorldState) -> float: pos = state.account.positions.get(state.venue.symbol) if not pos or pos.qty == 0: return 0.0 return abs(pos.qty * pos.avg_entry) / max(state.account.equity, 1e-12) @staticmethod def _tail_risk_proxy(state: MarketWorldState) -> float: pos = state.account.positions.get(state.venue.symbol) if not pos or pos.qty == 0: return 0.0 if pos.liquidation_price and pos.liquidation_price > 0: mid = state.book.mid if state.book.bids and state.book.asks else pos.avg_entry distance = abs(mid - pos.liquidation_price) / max(mid, 1e-12) return max(0.0, 1.0 - distance) return 0.0 # Check hftbacktest availability try: import hftbacktest # noqa: F401 _HAS_HFTBACKTEST = True except ImportError: _HAS_HFTBACKTEST = False