From 0cd7a8acc750c1d5f07ba66d6926a1d307be00bb Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Sun, 26 Jul 2026 04:35:23 +0000 Subject: [PATCH 1/2] feat: add native event reactive strategy runner --- __init__.py | 14 + backends/native_event.py | 482 +++++++++- core/__init__.py | 14 + core/reactive.py | 125 +++ docs/endpoint.md | 56 ++ endpoint.py | 67 ++ engines.py | 25 + ...t_phase30d_native_event_reactive_runner.py | 227 +++++ upgrade/implement.md | 832 ++++++++++++++++++ 9 files changed, 1837 insertions(+), 5 deletions(-) create mode 100644 core/reactive.py create mode 100644 tests/test_phase30d_native_event_reactive_runner.py diff --git a/__init__.py b/__init__.py index e12e58c..d0bfe73 100644 --- a/__init__.py +++ b/__init__.py @@ -100,6 +100,14 @@ Trade, order_intents_to_lifecycle_commands, ) +from .core.reactive import ( + NativeActiveOrderSnapshot, + NativeEventStrategyError, + NativeEventStrategyProtocol, + NativeFillEvent, + NativeOrderEvent, + NativeStrategyContext, +) from .core.basket import FrozenBasketPlan, build_frozen_basket_orders from .core.execution_depth import ( NautilusExecutionDepthConfig, @@ -336,6 +344,12 @@ "NautilusBacktestEngine", "NativeEventBackend", "NativeEventConfig", + "NativeActiveOrderSnapshot", + "NativeEventStrategyError", + "NativeEventStrategyProtocol", + "NativeFillEvent", + "NativeOrderEvent", + "NativeStrategyContext", "NativeOptionBackend", "NativeOptionConfig", "NativePortfolioBackend", diff --git a/backends/native_event.py b/backends/native_event.py index 78886ad..b748a5f 100644 --- a/backends/native_event.py +++ b/backends/native_event.py @@ -6,7 +6,7 @@ from __future__ import annotations -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from typing import Dict, List, Optional, Sequence, Union import numpy as np @@ -57,6 +57,13 @@ validate_datetime, ) from ..core.results import BacktestResultV2 +from ..core.reactive import ( + NativeActiveOrderSnapshot, + NativeEventStrategyError, + NativeFillEvent, + NativeOrderEvent, + NativeStrategyContext, +) from ..core.schema import ( AccountConfig, BasketLegSpec, @@ -393,9 +400,12 @@ def run_order_commands( event_status=event_status, event_related_command=event_related_command, ) - active_orders = command_report[ - (command_report["active"] == True) | (command_report["waiting_parent"] == True) # noqa: E712 - ].copy() + if command_report.empty: + active_orders = pd.DataFrame() + else: + active_orders = command_report[ + (command_report["active"] == True) | (command_report["waiting_parent"] == True) # noqa: E712 + ].copy() return BacktestResultV2( equity=equity, @@ -436,6 +446,214 @@ def run_order_commands( }, ) + def run_strategy( + self, + datetime_index: Union[pd.DatetimeIndex, pd.Series], + strategy, + closes: Dict[str, pd.Series], + highs: Optional[Dict[str, pd.Series]] = None, + lows: Optional[Dict[str, pd.Series]] = None, + opens: Optional[Dict[str, pd.Series]] = None, + volumes: Optional[Dict[str, pd.Series]] = None, + funding_rate: Union[float, pd.Series, Dict] = 0.0, + contract_size: Union[float, Dict[str, float]] = 1.0, + leverage: Optional[Union[float, Dict[str, float]]] = None, + fee_rate: Optional[Union[float, Dict[str, float]]] = None, + symbols: Optional[List[str]] = None, + instruments: Optional[Union[Dict[str, InstrumentSpec], List[InstrumentSpec]]] = None, + qty_step: Optional[Union[float, Dict[str, float]]] = None, + lot_size: Optional[Union[float, Dict[str, float]]] = None, + slot_size: Optional[Union[float, Dict[str, float]]] = None, + min_qty: Optional[Union[float, Dict[str, float]]] = None, + min_notional: Optional[Union[float, Dict[str, float]]] = None, + execution_mode: str = "fast", + command_effective_phase: str = "next_bar", + ) -> BacktestResultV2: + """ + Run a reactive strategy against native-event v2 lifecycle semantics. + + Strategy callbacks observe post-bar engine state and may emit commands + for the next bar. The emitted tape is replayed once at the end through + `run_order_commands`, making the final result reproducible by static + lifecycle replay. + """ + if strategy is None: + raise ValueError("run_strategy requires a strategy object") + if str(command_effective_phase).lower().strip() != "next_bar": + raise NotImplementedError("reactive native-event MVP supports command_effective_phase='next_bar' only") + execution_mode = str(execution_mode).lower().strip() + if execution_mode not in {"fast", "audit"}: + raise ValueError("execution_mode must be 'fast' or 'audit'") + + idx = validate_datetime(datetime_index) + symbol_list = list(symbols) if symbols is not None else list(closes.keys()) + market_arrays = self.prepare_market_arrays( + datetime_index=idx, + closes=closes, + highs=highs, + lows=lows, + funding_rate=funding_rate, + symbols=symbol_list, + ) + open_dict = align_series(opens, symbol_list, idx, fallback=align_series(closes, symbol_list, idx)) + volume_dict = align_series(volumes, symbol_list, idx, fallback={s: pd.Series(0.0, index=idx) for s in symbol_list}) + opens_arr = np.ascontiguousarray(np.column_stack([open_dict[s].to_numpy(dtype=np.float64) for s in symbol_list])) + volumes_arr = np.ascontiguousarray(np.column_stack([volume_dict[s].to_numpy(dtype=np.float64) for s in symbol_list])) + + contract_sizes = self._per_symbol_array(contract_size, symbol_list, default=1.0) + constraints = build_quantity_constraints( + symbol_list, + instruments=instruments, + qty_step=qty_step, + lot_size=lot_size, + slot_size=slot_size, + min_qty=min_qty, + min_notional=min_notional, + ) + + emitted: list[OrderCommand] = [] + emitted_order_ids: set[str] = set() + callback_count = 0 + ignored_commands_after_end = 0 + last_context: Optional[NativeStrategyContext] = None + + initial_context = self._reactive_context_from_result( + bar_index=0, + idx=idx, + symbols=symbol_list, + result=self._reactive_replay( + idx=idx[:1], + commands=(), + closes=closes, + highs=highs, + lows=lows, + funding_rate=funding_rate, + contract_size=contract_size, + leverage=leverage, + fee_rate=fee_rate, + symbols=symbol_list, + market_arrays=None, + instruments=instruments, + qty_step=qty_step, + lot_size=lot_size, + slot_size=slot_size, + min_qty=min_qty, + min_notional=min_notional, + ), + opens_arr=opens_arr, + highs_arr=market_arrays.highs, + lows_arr=market_arrays.lows, + closes_arr=market_arrays.closes, + volumes_arr=volumes_arr, + constraints=constraints, + contract_sizes=contract_sizes, + ) + last_context = initial_context + + initial_commands = self._call_strategy_callback(strategy, "initialize", initial_context) + scheduled, ignored = self._retime_reactive_commands( + commands=initial_commands, + effective_bar=1, + idx=idx, + emitted_order_ids=emitted_order_ids, + ) + emitted.extend(scheduled) + ignored_commands_after_end += ignored + + for bar in range(len(idx)): + prefix_idx = idx[: bar + 1] + prefix_commands = tuple(command for command in emitted if pd.Timestamp(command.timestamp).value <= prefix_idx[-1].value) + partial = self._reactive_replay( + idx=prefix_idx, + commands=prefix_commands, + closes=closes, + highs=highs, + lows=lows, + funding_rate=funding_rate, + contract_size=contract_size, + leverage=leverage, + fee_rate=fee_rate, + symbols=symbol_list, + market_arrays=None, + instruments=instruments, + qty_step=qty_step, + lot_size=lot_size, + slot_size=slot_size, + min_qty=min_qty, + min_notional=min_notional, + ) + context = self._reactive_context_from_result( + bar_index=bar, + idx=idx, + symbols=symbol_list, + result=partial, + opens_arr=opens_arr, + highs_arr=market_arrays.highs, + lows_arr=market_arrays.lows, + closes_arr=market_arrays.closes, + volumes_arr=volumes_arr, + constraints=constraints, + contract_sizes=contract_sizes, + ) + last_context = context + callback_count += 1 + if context.liquidated: + break + commands = self._call_strategy_callback(strategy, "on_bar_close", context) + scheduled, ignored = self._retime_reactive_commands( + commands=commands, + effective_bar=bar + 1, + idx=idx, + emitted_order_ids=emitted_order_ids, + ) + emitted.extend(scheduled) + ignored_commands_after_end += ignored + + if last_context is not None and not last_context.liquidated: + final_commands = self._call_strategy_callback(strategy, "finalize", last_context) + scheduled, ignored = self._retime_reactive_commands( + commands=final_commands, + effective_bar=len(idx), + idx=idx, + emitted_order_ids=emitted_order_ids, + ) + emitted.extend(scheduled) + ignored_commands_after_end += ignored + + final_result = self.run_order_commands( + datetime_index=idx, + commands=tuple(emitted), + closes=closes, + highs=highs, + lows=lows, + funding_rate=funding_rate, + contract_size=contract_size, + leverage=leverage, + fee_rate=fee_rate, + symbols=symbol_list, + market_arrays=market_arrays, + instruments=instruments, + qty_step=qty_step, + lot_size=lot_size, + slot_size=slot_size, + min_qty=min_qty, + min_notional=min_notional, + ) + final_result.metadata.update( + { + "engine": "event_v2_reactive_mvp", + "reactive_execution_mode": execution_mode, + "command_effective_phase": "next_bar", + "emitted_command_tape": tuple(emitted), + "emitted_command_count": len(emitted), + "ignored_commands_after_end": int(ignored_commands_after_end), + "strategy_callback_count": int(callback_count), + "static_replay_available": True, + "reactive_context_builder": "event_v2_replay_mvp", + } + ) + return final_result + def run_orders( self, datetime_index: Union[pd.DatetimeIndex, pd.Series], @@ -775,6 +993,245 @@ def _apply_command_quantity_constraints( out.append(command) return tuple(out), {"changed_count": changed, "dropped_count": len(dropped), "dropped_orders": dropped} + def _reactive_replay( + self, + *, + idx: pd.DatetimeIndex, + commands: Sequence[OrderCommand], + closes: Dict[str, pd.Series], + highs: Optional[Dict[str, pd.Series]], + lows: Optional[Dict[str, pd.Series]], + funding_rate, + contract_size, + leverage, + fee_rate, + symbols: List[str], + market_arrays: Optional[PreparedMarketArrays], + instruments, + qty_step, + lot_size, + slot_size, + min_qty, + min_notional, + ) -> BacktestResultV2: + return self.run_order_commands( + datetime_index=idx, + commands=tuple(commands), + closes={symbol: closes[symbol].reindex(idx).ffill().bfill() for symbol in symbols}, + highs=None if highs is None else {symbol: highs[symbol].reindex(idx).ffill().bfill() for symbol in symbols}, + lows=None if lows is None else {symbol: lows[symbol].reindex(idx).ffill().bfill() for symbol in symbols}, + funding_rate=funding_rate, + contract_size=contract_size, + leverage=leverage, + fee_rate=fee_rate, + symbols=symbols, + market_arrays=market_arrays, + instruments=instruments, + qty_step=qty_step, + lot_size=lot_size, + slot_size=slot_size, + min_qty=min_qty, + min_notional=min_notional, + ) + + def _reactive_context_from_result( + self, + *, + bar_index: int, + idx: pd.DatetimeIndex, + symbols: List[str], + result: BacktestResultV2, + opens_arr: np.ndarray, + highs_arr: np.ndarray, + lows_arr: np.ndarray, + closes_arr: np.ndarray, + volumes_arr: np.ndarray, + constraints, + contract_sizes: np.ndarray, + ) -> NativeStrategyContext: + local_bar = min(int(bar_index), len(result.equity) - 1) + ts = idx[int(bar_index)] + margin_row = result.margin.iloc[local_bar] if not result.margin.empty else None + init_margin = 0.0 if margin_row is None else float(margin_row.get("initial_margin", 0.0)) + maint_margin = 0.0 if margin_row is None else float(margin_row.get("maintenance_margin", 0.0)) + equity = float(result.equity.iloc[local_bar]) + position_row = result.positions.iloc[local_bar] + positions = { + symbol: float(position_row.get(f"Position_{symbol}", 0.0)) + for symbol in symbols + } + fills_this_bar = tuple( + self._fill_to_native_event(fill) + for fill in result.fills + if pd.Timestamp(fill.timestamp).value == ts.value + ) + events_this_bar = self._native_order_events_for_bar(result.metadata.get("order_events"), int(bar_index)) + active_orders = self._native_active_snapshots(result.metadata.get("active_orders")) + size_helper = self._reactive_size_helper( + symbols=symbols, + constraints=constraints, + contract_sizes=contract_sizes, + ) + return NativeStrategyContext( + bar_index=int(bar_index), + timestamp=ts, + open=np.ascontiguousarray(opens_arr[int(bar_index)].copy()), + high=np.ascontiguousarray(highs_arr[int(bar_index)].copy()), + low=np.ascontiguousarray(lows_arr[int(bar_index)].copy()), + close=np.ascontiguousarray(closes_arr[int(bar_index)].copy()), + volume=np.ascontiguousarray(volumes_arr[int(bar_index)].copy()), + equity=equity, + available_equity=equity - init_margin, + initial_margin=init_margin, + maintenance_margin=maint_margin, + positions=positions, + fills_this_bar=fills_this_bar, + order_events_this_bar=events_this_bar, + active_orders=active_orders, + liquidated=bool(result.liquidated), + symbols=tuple(symbols), + size_order=size_helper, + ) + + @staticmethod + def _retime_reactive_commands( + *, + commands: Sequence[OrderCommand], + effective_bar: int, + idx: pd.DatetimeIndex, + emitted_order_ids: set[str], + ) -> tuple[tuple[OrderCommand, ...], int]: + if commands is None: + return (), 0 + if effective_bar >= len(idx): + return (), len(tuple(commands)) + out: list[OrderCommand] = [] + ignored = 0 + effective_ts = idx[int(effective_bar)] + for seq, command in enumerate(tuple(commands)): + if not isinstance(command, OrderCommand): + raise TypeError("reactive strategy callbacks must return OrderCommand objects") + order_id = command.order_id + if command.action in (OrderAction.PLACE, OrderAction.REPLACE): + if order_id is None: + order_id = command.tag or f"reactive-{effective_bar}-{seq}" + if order_id in emitted_order_ids: + raise ValueError(f"duplicate reactive order_id={order_id!r}") + emitted_order_ids.add(order_id) + out.append(replace(command, timestamp=effective_ts, order_id=order_id)) + return tuple(out), ignored + + @staticmethod + def _call_strategy_callback(strategy, callback: str, context: NativeStrategyContext) -> tuple[OrderCommand, ...]: + fn = getattr(strategy, callback, None) + if fn is None: + return () + try: + commands = fn(context) + except Exception as exc: + raise NativeEventStrategyError(callback, context.bar_index, context.timestamp, exc) from exc + if commands is None: + return () + return tuple(commands) + + @staticmethod + def _fill_to_native_event(fill: Fill) -> NativeFillEvent: + metadata = dict(fill.metadata or {}) + return NativeFillEvent( + timestamp=pd.Timestamp(fill.timestamp), + symbol=fill.symbol, + side=fill.side, + qty=float(fill.qty), + price=float(fill.price), + fee=float(fill.fee), + order_id=fill.order_id, + tag=metadata.get("tag"), + campaign_id=metadata.get("campaign_id"), + cycle_id=metadata.get("cycle_id"), + level_id=metadata.get("level_id"), + parent_order_id=metadata.get("parent_order_id"), + oco_group_id=metadata.get("oco_group_id"), + metadata=metadata, + ) + + @staticmethod + def _native_order_events_for_bar(events, bar: int) -> tuple[NativeOrderEvent, ...]: + if events is None or len(events) == 0: + return () + frame = events[events["bar"] == int(bar)] + out = [] + for row in frame.to_dict("records"): + out.append( + NativeOrderEvent( + timestamp=pd.Timestamp(row["timestamp"]), + bar=int(row["bar"]), + event_name=str(row["event_name"]), + status=int(row["status"]), + order_id=row.get("order_id"), + target_order_id=row.get("target_order_id"), + parent_order_id=row.get("parent_order_id"), + oco_group_id=row.get("oco_group_id"), + tag=row.get("tag"), + campaign_id=row.get("campaign_id"), + cycle_id=row.get("cycle_id"), + level_id=row.get("level_id"), + original_index=int(row.get("original_index", -1)), + related_original_index=int(row.get("related_original_index", -1)), + ) + ) + return tuple(out) + + @staticmethod + def _native_active_snapshots(active_orders) -> tuple[NativeActiveOrderSnapshot, ...]: + if active_orders is None or len(active_orders) == 0: + return () + out = [] + for row in active_orders.to_dict("records"): + out.append( + NativeActiveOrderSnapshot( + order_id=row.get("order_id"), + symbol=row.get("symbol"), + side=row.get("side"), + order_type=row.get("order_type"), + status=int(row.get("status", 0)), + remaining_qty=float(row.get("working_qty", 0.0)), + price=float(row.get("working_price", 0.0)), + trigger_price=float(row.get("working_trigger_price", 0.0)), + reduce_only=bool(row.get("reduce_only", False)), + parent_order_id=row.get("parent_order_id"), + oco_group_id=row.get("oco_group_id"), + tag=row.get("tag"), + campaign_id=row.get("campaign_id"), + cycle_id=row.get("cycle_id"), + level_id=row.get("level_id"), + ) + ) + return tuple(out) + + @staticmethod + def _reactive_size_helper(symbols: List[str], constraints, contract_sizes: np.ndarray): + symbol_to_col = {symbol: j for j, symbol in enumerate(symbols)} + + def size_order(symbol: str, notional: float, price: float, side: OrderSide = OrderSide.BUY) -> float: + if symbol not in symbol_to_col: + raise ValueError(f"unknown symbol={symbol!r}") + if price <= 0.0: + raise ValueError("price must be > 0") + col = symbol_to_col[symbol] + signed_qty = (float(notional) / (float(price) * float(contract_sizes[col]))) * side.sign + return abs( + quantize_signed_quantity( + signed_qty, + float(price), + float(contract_sizes[col]), + float(constraints.qty_step[col]), + float(constraints.min_qty[col]), + float(constraints.min_notional[col]), + ) + ) + + return size_order + @staticmethod def _build_command_report( compiled_commands: CompiledOrderCommandArrays, @@ -806,6 +1263,9 @@ def _build_command_report( "parent_order_id": command.parent_order_id, "group_id": command.group_id, "oco_group_id": command.oco_group_id, + "campaign_id": command.metadata.get("campaign_id"), + "cycle_id": command.metadata.get("cycle_id"), + "level_id": command.metadata.get("level_id"), "activation_policy": command.activation_policy.value, "status": int(command_status[sorted_idx]), "reject_code": int(reject_code[sorted_idx]), @@ -864,7 +1324,12 @@ def _build_order_events( "related_original_index": related_original_idx, "order_id": None if command is None else command.order_id, "target_order_id": None if command is None else command.target_order_id, + "parent_order_id": None if command is None else command.parent_order_id, "oco_group_id": None if command is None else command.oco_group_id, + "tag": None if command is None else command.tag, + "campaign_id": None if command is None else command.metadata.get("campaign_id"), + "cycle_id": None if command is None else command.metadata.get("cycle_id"), + "level_id": None if command is None else command.metadata.get("level_id"), } ) return pd.DataFrame(rows) @@ -1908,6 +2373,13 @@ def _build_fills(sorted_orders, idx, fill_bar, fill_qty, fill_price, fill_fee) - for sorted_idx in filled_indices: order = sorted_orders[int(sorted_idx)][1] bar = int(fill_bar[sorted_idx]) + metadata = dict(getattr(order, "metadata", {}) or {}) + if getattr(order, "tag", None) is not None: + metadata.setdefault("tag", order.tag) + if getattr(order, "parent_order_id", None) is not None: + metadata.setdefault("parent_order_id", order.parent_order_id) + if getattr(order, "oco_group_id", None) is not None: + metadata.setdefault("oco_group_id", order.oco_group_id) fills.append( Fill( timestamp=idx[bar], @@ -1922,7 +2394,7 @@ def _build_fills(sorted_orders, idx, fill_bar, fill_qty, fill_price, fill_fee) - else LiquiditySide.MAKER ), order_id=order.order_id, - metadata={"source": "native_event"}, + metadata={**metadata, "source": "native_event"}, ) ) return fills diff --git a/core/__init__.py b/core/__init__.py index 9dbc687..da78751 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -28,6 +28,14 @@ build_bracket_order_plan, build_dca_grid_order_plan, ) +from .reactive import ( + NativeActiveOrderSnapshot, + NativeEventStrategyError, + NativeEventStrategyProtocol, + NativeFillEvent, + NativeOrderEvent, + NativeStrategyContext, +) from .arbitrage import ( ArbExecutionPolicy, ArbitrageLeg, @@ -139,6 +147,12 @@ "MarginModel", "MarginModelKind", "NautilusExecutionDepthConfig", + "NativeActiveOrderSnapshot", + "NativeEventStrategyError", + "NativeEventStrategyProtocol", + "NativeFillEvent", + "NativeOrderEvent", + "NativeStrategyContext", "OmsMode", "OrderAction", "OrderActivationPolicy", diff --git a/core/reactive.py b/core/reactive.py new file mode 100644 index 0000000..c4b1583 --- /dev/null +++ b/core/reactive.py @@ -0,0 +1,125 @@ +""" +Reactive native-event strategy context. + +These records are intentionally lightweight and read-only. Strategies inspect +engine state after each bar and return `OrderCommand` objects for the next bar. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Callable, Mapping, Optional, Sequence, Tuple + +import numpy as np +import pandas as pd + +from .orders import OrderCommand +from .schema import OrderSide, OrderType + + +@dataclass(frozen=True) +class NativeFillEvent: + timestamp: pd.Timestamp + symbol: str + side: OrderSide + qty: float + price: float + fee: float + order_id: Optional[str] = None + tag: Optional[str] = None + campaign_id: Optional[str] = None + cycle_id: Optional[str] = None + level_id: Optional[str] = None + parent_order_id: Optional[str] = None + oco_group_id: Optional[str] = None + metadata: Mapping = field(default_factory=dict) + + +@dataclass(frozen=True) +class NativeOrderEvent: + timestamp: pd.Timestamp + bar: int + event_name: str + status: int + order_id: Optional[str] = None + target_order_id: Optional[str] = None + parent_order_id: Optional[str] = None + oco_group_id: Optional[str] = None + tag: Optional[str] = None + campaign_id: Optional[str] = None + cycle_id: Optional[str] = None + level_id: Optional[str] = None + original_index: int = -1 + related_original_index: int = -1 + + +@dataclass(frozen=True) +class NativeActiveOrderSnapshot: + order_id: Optional[str] + symbol: Optional[str] + side: Optional[str] + order_type: Optional[str] + status: int + remaining_qty: float + price: float + trigger_price: float + reduce_only: bool + parent_order_id: Optional[str] = None + oco_group_id: Optional[str] = None + tag: Optional[str] = None + campaign_id: Optional[str] = None + cycle_id: Optional[str] = None + level_id: Optional[str] = None + + +@dataclass(frozen=True) +class NativeStrategyContext: + bar_index: int + timestamp: pd.Timestamp + open: np.ndarray + high: np.ndarray + low: np.ndarray + close: np.ndarray + volume: np.ndarray + equity: float + available_equity: float + initial_margin: float + maintenance_margin: float + positions: Mapping[str, float] + fills_this_bar: Sequence[NativeFillEvent] + order_events_this_bar: Sequence[NativeOrderEvent] + active_orders: Sequence[NativeActiveOrderSnapshot] + liquidated: bool + symbols: Tuple[str, ...] = field(default_factory=tuple) + size_order: Callable[..., float] = field(default=lambda **_: 0.0, repr=False, compare=False) + + +class NativeEventStrategyError(RuntimeError): + """Raised when a reactive strategy callback fails.""" + + def __init__(self, callback: str, bar_index: int, timestamp: pd.Timestamp, original: Exception): + self.callback = callback + self.bar_index = int(bar_index) + self.timestamp = timestamp + self.original = original + super().__init__( + f"native-event strategy callback {callback!r} failed at " + f"bar_index={bar_index}, timestamp={timestamp}: {type(original).__name__}: {original}" + ) + + +class NativeEventStrategyProtocol: + """ + Optional protocol-like base class for user strategies. + + Subclassing is not required; duck typing is used by the backend. + """ + + def initialize(self, context: NativeStrategyContext) -> Sequence[OrderCommand]: + return () + + def on_bar_close(self, context: NativeStrategyContext) -> Sequence[OrderCommand]: + return () + + def finalize(self, context: NativeStrategyContext) -> Sequence[OrderCommand]: + return () diff --git a/docs/endpoint.md b/docs/endpoint.md index b25d0fc..fa944f4 100644 --- a/docs/endpoint.md +++ b/docs/endpoint.md @@ -677,6 +677,62 @@ grid_bt = QuantBTEndpoint.native_event_dca_grid(spec=dca_grid_spec) grid_result = grid_bt.simulate(data=df) ``` +Reactive native-event strategy: + +```python +from quantbt import OrderCommand, OrderSide, OrderType, QuantBTEndpoint, TimeInForce + +class DynamicGridStrategy: + def initialize(self, context): + return [] + + def on_bar_close(self, context): + # Context is post-bar and read-only. Fills, positions and active orders + # come from QuantBT, not from strategy-side fill simulation. + if context.bar_index == 0: + qty = context.size_order("ETHUSDT", notional=1_000, price=context.close[0] * 0.99) + return [ + OrderCommand( + timestamp=context.timestamp, + symbol="ETHUSDT", + side=OrderSide.BUY, + order_type=OrderType.LIMIT, + qty=qty, + price=context.close[0] * 0.99, + tif=TimeInForce.GTC, + order_id="grid-c1-l1", + metadata={"campaign_id": "c1", "level_id": "l1"}, + ) + ] + return [] + +bt = QuantBTEndpoint.native_event_strategy( + initial_capital=20_000, + leverage=5, + fee_rate=0.0005, + reactive_execution_mode="fast", +) + +result = bt.simulate( + data=df, + strategy=DynamicGridStrategy(), + symbols=["ETHUSDT"], +) + +tape = result.metadata["emitted_command_tape"] +replay = QuantBTEndpoint.native_event_lifecycle( + initial_capital=20_000, + leverage=5, + fee_rate=0.0005, +).simulate(data=df, order_commands=tape, symbols=["ETHUSDT"]) +``` + +Reactive timing is causal: commands returned by `on_bar_close(context_t)` are +retimed to bar `t+1`, so they cannot fill inside the same OHLC bar that the +strategy just observed. Phase 30D uses the certified event-v2 lifecycle engine +as a replay-backed MVP and stores the full emitted command tape for audit and +static replay parity. + Package execution-depth preflight: ```python diff --git a/endpoint.py b/endpoint.py index a725dc7..2b8a72c 100644 --- a/endpoint.py +++ b/endpoint.py @@ -178,6 +178,7 @@ class EndpointConfig: arbitrage_spec: object = None structured_order_spec: object = None event_engine_version: str = "v1" + reactive_execution_mode: str = "fast" symbols: Optional[Sequence[str]] = None dca_kwargs: Dict = field(default_factory=dict) nautilus_config: object = None @@ -318,6 +319,24 @@ def native_event_lifecycle(cls, **kwargs) -> "QuantBTEndpoint": ) ) + @classmethod + def native_event_strategy(cls, **kwargs) -> "QuantBTEndpoint": + """ + Create a reactive native-event v2 strategy endpoint. + + Use `simulate(data=df, strategy=obj)` where `obj` optionally implements + `initialize(context)`, `on_bar_close(context)`, and `finalize(context)`. + Commands emitted by callbacks become effective from the next bar. + """ + return cls( + _config_from_kwargs( + mode="native_event_strategy", + backend="native_event", + event_engine_version="v2", + **kwargs, + ) + ) + @classmethod def options( cls, @@ -650,6 +669,13 @@ def nautilus_support_matrix() -> Dict[str, Dict[str, str]]: "order_types": "market, limit, stop_market, stop_limit plus cancel/replace/amend/cancel_all in native-event v2", "notes": "Nautilus command path is payload-aligned, not exchange-native cancel/amend parity yet", }, + "reactive_strategy": { + "status": "supported_native_event_mvp", + "endpoint": "QuantBTEndpoint.native_event_strategy(...)", + "scope": "on_bar_close strategy callbacks emitting next-bar OrderCommand objects", + "order_types": "native-event v2 lifecycle commands", + "notes": "Phase 30D replay-backed MVP with captured command tape and static replay parity; incremental session is Phase 30E", + }, "dca_grid": { "status": "experimental", "endpoint": "QuantBTEndpoint.nautilus_dca_grid(...)", @@ -891,6 +917,7 @@ def backtest( positions: Optional[Union[pd.DataFrame, SeriesMap]] = None, orders: Optional[Sequence[OrderIntent]] = None, order_commands: Optional[Sequence[OrderCommand]] = None, + strategy=None, basket: Optional[BasketSpec] = None, closes: Optional[SeriesMap] = None, highs: Optional[SeriesMap] = None, @@ -988,6 +1015,13 @@ def backtest( datetime_index=datetime_index, symbols=symbols, ) + if mode == "native_event_strategy": + return self._run_native_event_strategy( + data=data, + strategy=strategy, + datetime_index=datetime_index, + symbols=symbols, + ) if mode in ("nautilus_dca_grid", "nautilus_bracket_orders", "native_event_dca_grid", "native_event_bracket_orders"): return self._run_structured_orders(data=data, datetime_index=datetime_index, symbols=symbols) if mode == "basket": @@ -1305,6 +1339,39 @@ def _run_orders(self, data, orders, order_commands, datetime_index, symbols): self._store_result(self.engine.result) return self.result + def _run_native_event_strategy(self, data, strategy, datetime_index, symbols): + if strategy is None: + raise ValueError("native_event_strategy endpoint requires strategy=...") + frame, idx, _ = _normalize_single_data( + data=data, + signal=pd.Series(0.0, index=_infer_index(data, datetime_index)), + signal_col=None, + datetime_index=datetime_index, + ) + symbol_list = list(symbols or self.config.symbols or ["asset"]) + self.engine = BacktestEngineV2( + data=frame, + symbols=symbol_list, + backend="native_event", + strategy=strategy, + event_engine_version="v2", + reactive_execution_mode=self.config.reactive_execution_mode, + account=self.config.account, + execution=self.config.execution, + fee_rate=self.config.v2_fee_rate, + use_funding=self.config.use_funding, + funding_rate=self.config.funding_rate, + contract_size=self.config.contract_size, + instruments=self.config.instruments, + qty_step=self.config.qty_step, + lot_size=self.config.lot_size, + slot_size=self.config.slot_size, + min_qty=self.config.min_qty, + min_notional=self.config.min_notional, + ) + self._store_result(self.engine.result) + return self.result + def _run_structured_orders(self, data, datetime_index, symbols): spec = self.config.structured_order_spec if spec is None: diff --git a/engines.py b/engines.py index 9fd7ae7..9c08136 100644 --- a/engines.py +++ b/engines.py @@ -66,7 +66,9 @@ def __init__( target_units: Optional[Union[pd.Series, SeriesMap]] = None, orders: Optional[Sequence[OrderIntent]] = None, order_commands: Optional[Sequence[OrderCommand]] = None, + strategy=None, event_engine_version: str = "v1", + reactive_execution_mode: str = "fast", datetime_index: Optional[Union[pd.DatetimeIndex, pd.Series]] = None, closes: Optional[SeriesMap] = None, highs: Optional[SeriesMap] = None, @@ -104,7 +106,9 @@ def __init__( self.target_units = target_units self.orders = tuple(orders or ()) self.order_commands = tuple(order_commands or ()) + self.strategy = strategy self.event_engine_version = str(event_engine_version).lower().strip() + self.reactive_execution_mode = str(reactive_execution_mode).lower().strip() self.datetime_index = datetime_index self.closes = closes self.highs = highs @@ -204,6 +208,27 @@ def _run_native_event(self) -> BacktestResultV2: ) ) + if self.strategy is not None: + return backend.run_strategy( + datetime_index=idx, + strategy=self.strategy, + closes=closes, + highs=highs, + lows=lows, + funding_rate=self.funding_rate, + contract_size=self.contract_size, + leverage=self.leverage, + fee_rate=self.fee_rate, + symbols=symbols, + instruments=self.instruments, + qty_step=self.qty_step, + lot_size=self.lot_size, + slot_size=self.slot_size, + min_qty=self.min_qty, + min_notional=self.min_notional, + execution_mode=self.reactive_execution_mode, + ) + if self.basket is not None: basket_signal = self.signal if self.signal is not None else _first_signal(self.signals) if basket_signal is None: diff --git a/tests/test_phase30d_native_event_reactive_runner.py b/tests/test_phase30d_native_event_reactive_runner.py new file mode 100644 index 0000000..e54fa2b --- /dev/null +++ b/tests/test_phase30d_native_event_reactive_runner.py @@ -0,0 +1,227 @@ +from __future__ import annotations + +import pandas as pd +import pytest + +from quantbt import ( + NativeEventStrategyError, + OrderCommand, + OrderSide, + OrderType, + QuantBTEndpoint, + TimeInForce, +) + + +def _bars() -> pd.DataFrame: + idx = pd.date_range("2024-01-01", periods=5, freq="1h", tz="UTC") + return pd.DataFrame( + { + "open": [100.0, 100.0, 100.0, 110.0, 100.0], + "high": [101.0, 101.0, 112.0, 111.0, 101.0], + "low": [90.0, 98.0, 99.0, 99.0, 99.0], + "close": [100.0, 100.0, 110.0, 100.0, 100.0], + "volume": 1_000.0, + }, + index=idx, + ) + + +def test_reactive_commands_emit_after_close_and_fill_next_bar_only(): + df = _bars() + + class Strategy: + def __init__(self): + self.calls = [] + + def on_bar_close(self, context): + self.calls.append(context.bar_index) + if context.bar_index == 0: + return [ + OrderCommand( + timestamp=context.timestamp, + symbol="BTC", + side=OrderSide.BUY, + order_type=OrderType.LIMIT, + qty=1.0, + price=99.0, + tif=TimeInForce.GTC, + order_id="entry", + ) + ] + return [] + + strategy = Strategy() + endpoint = QuantBTEndpoint.native_event_strategy(initial_capital=10_000, leverage=10, use_funding=False) + result = endpoint.simulate(data=df, strategy=strategy, symbols=["BTC"]) + + assert strategy.calls == [0, 1, 2, 3, 4] + assert len(result.fills) == 1 + assert result.fills[0].timestamp == df.index[1] + assert result.fills[0].price == 99.0 + assert result.metadata["emitted_command_tape"][0].timestamp == df.index[1] + + +def test_reactive_context_receives_fill_and_rearms_reduce_only_exit_with_static_replay_parity(): + df = _bars() + + class Strategy: + def __init__(self): + self.fill_contexts = [] + + def initialize(self, context): + return [ + OrderCommand( + timestamp=context.timestamp, + symbol="BTC", + side=OrderSide.BUY, + order_type=OrderType.MARKET, + qty=1.0, + tif=TimeInForce.IOC, + order_id="entry-c1-l1", + tag="GRID-C1-L1-ENTRY", + metadata={"campaign_id": "C1", "cycle_id": "1", "level_id": "L1"}, + ) + ] + + def on_bar_close(self, context): + if context.fills_this_bar: + self.fill_contexts.append( + ( + context.bar_index, + context.fills_this_bar[0].order_id, + context.fills_this_bar[0].level_id, + context.positions["BTC"], + ) + ) + if context.bar_index == 1 and context.fills_this_bar: + qty = context.fills_this_bar[0].qty + return [ + OrderCommand( + timestamp=context.timestamp, + symbol="BTC", + side=OrderSide.SELL, + order_type=OrderType.LIMIT, + qty=qty, + price=112.0, + tif=TimeInForce.GTC, + reduce_only=True, + order_id="exit-c1-l1", + tag="GRID-C1-L1-EXIT", + metadata={"campaign_id": "C1", "cycle_id": "1", "level_id": "L1"}, + ) + ] + return [] + + strategy = Strategy() + endpoint = QuantBTEndpoint.native_event_strategy(initial_capital=10_000, leverage=10, use_funding=False) + reactive = endpoint.simulate(data=df, strategy=strategy, symbols=["BTC"]) + tape = reactive.metadata["emitted_command_tape"] + + replay_endpoint = QuantBTEndpoint.native_event_lifecycle(initial_capital=10_000, leverage=10, use_funding=False) + replay = replay_endpoint.simulate(data=df, order_commands=tape, symbols=["BTC"]) + + assert strategy.fill_contexts[0] == (1, "entry-c1-l1", "L1", 1.0) + assert [fill.order_id for fill in reactive.fills] == ["entry-c1-l1", "exit-c1-l1"] + assert reactive.positions["Position_BTC"].iloc[-1] == 0.0 + pd.testing.assert_series_equal(reactive.equity, replay.equity) + pd.testing.assert_frame_equal(reactive.positions, replay.positions) + assert [fill.order_id for fill in replay.fills] == [fill.order_id for fill in reactive.fills] + + +def test_reactive_rejected_command_is_visible_in_next_callback(): + df = _bars() + + class Strategy: + def __init__(self): + self.rejected_seen = False + + def on_bar_close(self, context): + if any(event.event_name == "reject" for event in context.order_events_this_bar): + self.rejected_seen = True + if context.bar_index == 0: + return [ + OrderCommand( + timestamp=context.timestamp, + symbol="BTC", + side=OrderSide.BUY, + order_type=OrderType.MARKET, + qty=500.0, + tif=TimeInForce.IOC, + order_id="too-large", + ) + ] + return [] + + strategy = Strategy() + endpoint = QuantBTEndpoint.native_event_strategy(initial_capital=1_000, leverage=1, use_funding=False) + result = endpoint.simulate(data=df, strategy=strategy, symbols=["BTC"]) + + assert strategy.rejected_seen is True + assert len(result.fills) == 0 + assert "reject" in set(result.metadata["order_events"]["event_name"]) + + +def test_reactive_context_size_order_uses_backend_quantity_constraints(): + df = _bars() + + class Strategy: + def __init__(self): + self.sized_qty = None + + def on_bar_close(self, context): + if context.bar_index == 0: + self.sized_qty = context.size_order(symbol="BTC", notional=105.0, price=100.0) + return [] + + strategy = Strategy() + endpoint = QuantBTEndpoint.native_event_strategy( + initial_capital=10_000, + leverage=10, + use_funding=False, + qty_step={"BTC": 0.1}, + ) + endpoint.simulate(data=df, strategy=strategy, symbols=["BTC"]) + + assert strategy.sized_qty == 1.0 + + +def test_reactive_duplicate_order_id_fails_fast_with_clear_error(): + df = _bars() + + class Strategy: + def on_bar_close(self, context): + if context.bar_index in (0, 1): + return [ + OrderCommand( + timestamp=context.timestamp, + symbol="BTC", + side=OrderSide.BUY, + order_type=OrderType.LIMIT, + qty=1.0, + price=50.0, + order_id="duplicate", + ) + ] + return [] + + endpoint = QuantBTEndpoint.native_event_strategy(initial_capital=10_000, leverage=10, use_funding=False) + with pytest.raises(ValueError, match="duplicate reactive order_id"): + endpoint.simulate(data=df, strategy=Strategy(), symbols=["BTC"]) + + +def test_reactive_strategy_callback_error_reports_bar_and_timestamp(): + df = _bars() + + class Strategy: + def on_bar_close(self, context): + if context.bar_index == 2: + raise RuntimeError("boom") + return [] + + endpoint = QuantBTEndpoint.native_event_strategy(initial_capital=10_000, leverage=10, use_funding=False) + with pytest.raises(NativeEventStrategyError) as exc: + endpoint.simulate(data=df, strategy=Strategy(), symbols=["BTC"]) + + assert exc.value.bar_index == 2 + assert exc.value.timestamp == df.index[2] diff --git a/upgrade/implement.md b/upgrade/implement.md index 6aa1e7b..3c0b4be 100644 --- a/upgrade/implement.md +++ b/upgrade/implement.md @@ -3490,3 +3490,835 @@ For every phase: 6. Do not include unrelated dirty files. Main branch is protected by local pre-commit hook. + + +## Aditional update to awesome-native event command reactive +### Feature Request: Reactive Native-Event Strategy Runner + +### Phase 30D - Reactive Runner MVP + +Status: completed. + +Goal: + +- Add a safe opt-in reactive strategy runner above native-event v2. +- Preserve the static command-tape route and all Phase 30A-C behavior. +- Let strategy callbacks observe engine-generated fills, events, positions, + equity, margin, and active orders after each bar. +- Enforce causal timing: commands emitted after close `t` become effective from + bar `t+1`. +- Capture the emitted command tape and prove that static replay of this tape + has 100% accounting parity with the reactive run. + +Implementation scope: + +- Add read-only reactive records: + - `NativeStrategyContext`; + - `NativeFillEvent`; + - `NativeOrderEvent`; + - `NativeActiveOrderSnapshot`; + - `NativeEventStrategyError`. +- Add `NativeEventBackend.run_strategy(...)`. +- Add endpoint route: + - `QuantBTEndpoint.native_event_strategy(...)`; + - `simulate(..., strategy=...)`. +- Add `context.size_order(...)` using the same quantity constraints as the + backend. +- Preserve metadata round-trip for `campaign_id`, `cycle_id`, `level_id`, + `order_id`, `tag`, `parent_order_id`, and `oco_group_id`. + +MVP note: + +- Phase 30D may use a replay-backed context builder that calls the already + certified event-v2 command engine. This keeps lifecycle semantics identical + and makes parity exact. Phase 30E is reserved for the true incremental + session with preallocated buffers and large workload benchmarks. + +Acceptance tests: + +- `on_bar_close` is called exactly once after each bar. +- Commands emitted at close `t` cannot fill inside bar `t`. +- Commands become active/fillable from `t+1`. +- Context fills/positions match final result state. +- Rejected commands are visible in the next callback. +- Liquidation prevents further command ingestion. +- Captured `emitted_command_tape` static replay matches equity, positions, + fills, and command report. + +Implemented: + +- Added read-only reactive records: + - `NativeStrategyContext`; + - `NativeFillEvent`; + - `NativeOrderEvent`; + - `NativeActiveOrderSnapshot`; + - `NativeEventStrategyError`; + - `NativeEventStrategyProtocol`. +- Added `NativeEventBackend.run_strategy(...)`. +- Added endpoint route: + - `QuantBTEndpoint.native_event_strategy(...)`; + - `simulate(..., strategy=...)`. +- Added `context.size_order(...)` using backend quantity constraints. +- Added metadata round-trip into command report, order events, fills, and + reactive context for: + - `campaign_id`; + - `cycle_id`; + - `level_id`; + - `order_id`; + - `tag`; + - `parent_order_id`; + - `oco_group_id`. +- Added captured command tape: + - `result.metadata["emitted_command_tape"]`; + - `emitted_command_count`; + - `strategy_callback_count`; + - `reactive_context_builder`. +- Added clear callback failure errors with bar index and timestamp. + +Latest tests: + +- Phase 30D reactive runner tests: `6 passed`. +- Phase 30A/30B/30C/30D lifecycle suites: `29 passed`. +- Endpoint/Nautilus compatibility subsets: `50 passed`. +- Full non-real regression: `424 passed, 1 skipped, 3 warnings`. + +Technical debt after Phase 30D: + +- The MVP context builder is replay-backed through the certified event-v2 + command engine. It preserves exact semantics and replay parity, but it is not + the final high-throughput incremental session. +- Scoped `CANCEL_ALL` filters and large dynamic-grid benchmarks remain Phase + 30E. +- Nautilus exchange-native cancel/amend parity remains future work. + +### Phase 30E - Incremental Reactive Session And Dynamic Grid Certification + +Status: planned. + +Goal: + +- Replace the Phase 30D replay-backed context builder with an incremental + session that appends per-bar commands without recompiling command history. +- Add scoped `CANCEL_ALL` filters: + - symbol; + - side; + - order type; + - tag; + - tag prefix; + - parent order id; + - OCO group id; + - campaign/group ids. +- Add dynamic grid fixture and benchmark: + - 25,000 bars; + - 15-30 active grid orders; + - 1-5 commands per bar; + - static tape vs reactive FAST vs reactive AUDIT; + - full accounting parity between FAST and AUDIT. + +Non-goals retained: + +- No tick matching. +- No L2 book/queue priority. +- No exchange-native Nautilus cancel/amend parity. +- No async/live broker runtime. + +## 1. Mục tiêu + +Bổ sung một **reactive lifecycle runner** lên `native_event v2` hiện tại để strategy có thể: + +1. Nhận trạng thái execution thực tế sau mỗi bar. +2. Đọc fills, position và active orders do QuantBT tạo. +3. Phát `OrderCommand[]` cho bar tiếp theo. +4. Không phải tự kiểm tra `high/low` hoặc tự mô phỏng fill trong strategy. + +Đây không phải full exchange OMS và không thay đổi matching/accounting kernel hiện tại. + +Mục tiêu chính là hỗ trợ đúng domain cho: + +* Dynamic grid. +* Recurring DCA. +* Re-arm order sau khi exit. +* Cancel/amend level theo indicator mới. +* Regime switch. +* Parent-child nhiều chu kỳ. +* Stateful scale-in/scale-out strategies. + +--- + +## 2. Vấn đề hiện tại + +`native_event v2` đã hỗ trợ: + +```text +PLACE +CANCEL +REPLACE +AMEND +CANCEL_ALL +MARKET / LIMIT / STOP +parent-child +OCO +reduce-only +GTD +``` + +Nhưng public backend hiện vẫn chạy theo mô hình: + +```python +run_order_commands( + commands: Sequence[OrderCommand], + market_arrays=..., +) +``` + +Tức là toàn bộ command tape phải được tạo trước simulation. + +Mô hình này không đủ cho recurring dynamic grid: + +```text +entry fill +→ strategy cần biết fill thực tế +→ tạo exit cho đúng filled quantity +→ exit fill +→ re-arm entry +→ amend grid theo level mới +``` + +Strategy không thể biết trước các event này nếu không tự mô phỏng fills, dẫn đến duplicate execution logic và nguy cơ sai parity. + +--- + +## 3. Public API đề xuất + +### Phương án chính + +```python +endpoint = QuantBTEndpoint.native_event_strategy( + initial_capital=20_000, + leverage=5, + fee_rate=0.0005, + slippage_bps=1.0, + report_level="minimal", +) + +result = endpoint.simulate( + data=df, + strategy=DynamicGridStrategy(params), + symbols=["ETHUSDT"], +) +``` + +Hoặc giữ endpoint hiện tại: + +```python +endpoint = QuantBTEndpoint.native_event_lifecycle(...) + +result = endpoint.simulate_strategy( + data=df, + strategy=DynamicGridStrategy(params), +) +``` + +Không thay đổi API hiện có: + +```python +simulate(order_commands=[...]) +``` + +Static command tape và reactive strategy runner phải cùng dùng một kernel lifecycle. + +--- + +## 4. Strategy protocol + +```python +class NativeEventStrategyProtocol: + def initialize( + self, + context: "NativeStrategyContext", + ) -> list[OrderCommand]: + ... + + def on_bar_close( + self, + context: "NativeStrategyContext", + ) -> list[OrderCommand]: + ... + + def finalize( + self, + context: "NativeStrategyContext", + ) -> list[OrderCommand]: + ... +``` + +MVP chỉ cần: + +```text +initialize +on_bar_close +finalize +``` + +Chưa cần tick callback, order-book callback hoặc intrabar strategy callback. + +--- + +## 5. Read-only strategy context + +```python +@dataclass(frozen=True) +class NativeStrategyContext: + bar_index: int + timestamp: pd.Timestamp + + open: np.ndarray + high: np.ndarray + low: np.ndarray + close: np.ndarray + volume: np.ndarray + + equity: float + available_equity: float + initial_margin: float + maintenance_margin: float + + positions: Mapping[str, float] + + fills_this_bar: Sequence[FillEvent] + order_events_this_bar: Sequence[OrderEvent] + active_orders: Sequence[ActiveOrderSnapshot] + + liquidated: bool +``` + +`active_orders` cần chứa tối thiểu: + +```text +order_id +symbol +side +order_type +status +remaining_qty +price +trigger_price +reduce_only +parent_order_id +oco_group_id +tag +``` + +Context chỉ đọc. Strategy không được sửa trực tiếp engine state. + +--- + +## 6. Timeline causal bắt buộc + +Tại bar `t`: + +```text +1. Activate commands đã được submit từ trước. +2. Xử lý trigger và fills bằng OHLC[t]. +3. Áp dụng fee, funding, margin và liquidation. +4. Cập nhật positions/equity. +5. Emit fill/order events của bar t. +6. Gọi strategy.on_bar_close(context_t). +7. Commands trả về chỉ active từ bar t+1. +``` + +Default phải là: + +```python +command_effective_phase = "next_bar" +``` + +Như vậy strategy dùng indicator tại close `t` nhưng không thể retroactively đặt order trong high/low của chính bar `t`. + +Không cho callback sửa kết quả bar đã xử lý. + +--- + +## 7. Không mô phỏng fill trong strategy + +Strategy chỉ được: + +```text +tính indicator +xác định regime +xác định desired grid levels +PLACE / CANCEL / AMEND orders +quản lý campaign_id và level_id +phản ứng với fill events thật +``` + +QuantBT tiếp tục là nguồn duy nhất cho: + +```text +limit touch +fill price +slippage +fee +position +average entry +margin +funding +liquidation +reduce-only clipping +OCO +parent activation +order status +``` + +--- + +## 8. Dynamic command ingestion + +Kernel/session cần cho phép append commands sau mỗi bar mà không compile lại toàn bộ lịch sử. + +Đề xuất internal structure: + +```text +prepared market arrays +active-order registry +per-bar command buffer +preallocated command/event arrays +free-slot stack +``` + +API nội bộ: + +```python +session.submit_commands( + commands, + effective_bar=current_bar + 1, +) +``` + +Không nối lại toàn bộ `Sequence[OrderCommand]` rồi chạy lại simulation từ đầu. + +--- + +## 9. Scoped `CANCEL_ALL` + +Dynamic grid cần hủy đúng campaign hoặc đúng side, không được luôn hủy toàn bộ account. + +Mở rộng `CANCEL_ALL` với filter tùy chọn: + +```python +OrderCommand( + action=OrderAction.CANCEL_ALL, + symbol="ETHUSDT", + side=OrderSide.BUY, + tag_prefix="GRID-C12-LONG-ENTRY", +) +``` + +Các scope cần thiết: + +```text +symbol +side +order_type +tag +tag_prefix +parent_order_id +oco_group_id +``` + +Nếu chưa muốn đưa string filter vào Numba, compiler map tag/campaign/group thành integer code. + +--- + +## 10. Metadata round-trip + +Các trường sau phải được giữ xuyên suốt: + +```text +order command +→ active order +→ order event +→ fill +→ result reports +``` + +Fields: + +```text +order_id +tag +campaign_id +cycle_id +level_id +parent_order_id +oco_group_id +``` + +Có thể lưu các domain ID dưới dạng integer code trong kernel và decode khi tạo pandas reports. + +Điều này cần thiết để strategy biết: + +```text +fill này thuộc level nào +exit nào vừa đóng +entry nào cần re-arm +campaign nào cần cancel +``` + +--- + +## 11. Quantity semantics + +MVP tiếp tục dùng `qty`, nhưng nên thêm shared sizing helper ngoài strategy: + +```python +qty = context.size_order( + symbol="ETHUSDT", + notional=cash_per_entry, + price=limit_price, +) +``` + +Helper phải dùng cùng venue constraints với backend: + +```text +contract_size +qty_step +lot_size +min_qty +min_notional +``` + +Strategy không nên tự lặp lại rounding logic. + +Không nhất thiết phải thêm `notional` vào kernel command trong phase này. + +--- + +## 12. Performance + +### Hai mode + +```python +execution_mode="fast" +execution_mode="audit" +``` + +`fast`: + +* Reuse prepared market arrays. +* Chỉ tạo context tối thiểu. +* Không dựng DataFrame trong bar loop. +* Fills/events dùng lightweight views hoặc arrays. +* Không lưu full active-order snapshots mỗi bar. +* `report_level="minimal"`. +* Dùng cho Optuna và WFO. + +`audit`: + +* Full `order_events`. +* Full active-order diagnostics. +* Command tape export. +* Dùng cho candidate cuối. + +### Không gọi pandas trong hot loop + +Alpha indicators nên được tính trước thành NumPy arrays: + +```python +strategy.prepare(data) -> PreparedStrategyArrays +``` + +`on_bar_close()` chỉ đọc array tại `bar_index`. + +### Không copy toàn bộ registry + +Context chỉ expose: + +```text +position vector +fills/events vừa phát sinh +active-order view cần thiết +``` + +Không copy tất cả historical events mỗi bar. + +### Benchmark bắt buộc + +Thêm benchmark: + +```text +25,000 bars +15–30 concurrent grid orders +1–5 commands/bar +multiple fill/re-arm cycles +``` + +So sánh: + +```text +static command tape +reactive FAST +reactive AUDIT +``` + +Reactive FAST không nên chậm hơn Python event loop ngây thơ và phải đủ dùng cho Optuna trên dữ liệu 1h. + +--- + +## 13. Determinism + +Cùng: + +```text +market data +strategy parameters +initial state +random seed +``` + +phải tạo chính xác cùng: + +```text +command tape +fills +positions +equity +reports +``` + +Command ordering: + +```text +bar_index +callback_sequence +command_sequence +``` + +Commands strategy trả về phải giữ nguyên stable list order. + +--- + +## 14. Failure handling + +Nếu strategy callback raise exception: + +```text +stop simulation +return bar_index/timestamp gây lỗi +không trả partial metrics như một backtest hợp lệ +``` + +Nếu command bị reject: + +* Event phải xuất hiện trong `order_events`. +* Strategy nhận event đó ở callback tiếp theo. +* Không tự động retry trừ khi strategy yêu cầu. + +Nếu liquidation xảy ra: + +* Cancel toàn bộ active orders. +* Context đánh dấu `liquidated=True`. +* Không tiếp tục submit order mới mặc định. + +--- + +## 15. Captured command tape + +Reactive runner phải lưu toàn bộ commands mà strategy đã phát: + +```python +result.metadata["emitted_command_tape"] +``` + +Hoặc: + +```python +result.command_tape +``` + +Dùng cho: + +* Audit. +* Reproduction. +* Replay static. +* So sánh strategy-state với engine-state. +* Nautilus validation. + +Một reactive run phải có thể replay bằng: + +```python +endpoint.simulate( + data=df, + order_commands=result.command_tape, +) +``` + +và cho kết quả native-event giống hệt. + +Đây là acceptance criterion quan trọng nhất. + +--- + +## 16. Nautilus validation follow-up + +Support matrix hiện ghi native lifecycle đã hỗ trợ đầy đủ phía native, nhưng Nautilus command path mới payload-aligned; cancel/amend chưa có exchange-native parity đầy đủ. + +Sau MVP reactive runner, bổ sung adapter: + +```python +replay_lifecycle_tape_with_nautilus( + data, + command_tape, +) +``` + +Mapping: + +```text +PLACE → submit_order +CANCEL → cancel_order +REPLACE → cancel + submit hoặc modify đúng Nautilus API +AMEND → modify_order +``` + +Validation report: + +```text +order lifecycle status +fill count +fill qty +position by bar +fees +realized PnL +final equity +``` + +Nautilus không cần nằm trong optimization loop; chỉ validate candidate cuối. + +--- + +## 17. Acceptance tests bắt buộc + +### Core runner + +1. Callback được gọi đúng một lần sau mỗi bar. +2. Command sinh tại close `t` không thể fill trong bar `t`. +3. Command bắt đầu active tại `t+1`. +4. Position/fills trong context khớp result cuối. +5. Rejected command được trả về callback. +6. Liquidation khóa strategy đúng cách. +7. Static replay của captured command tape cho parity 100%. + +### Dynamic grid fixture + +1. Place 3 buy limits. +2. Một entry fill. +3. Chỉ child exit của đúng level được tạo. +4. Exit fill ở bar sau. +5. Entry level được re-arm. +6. Grid level thay đổi thì pending order được amend. +7. Regime switch cancel đúng pending side. +8. Reduce-only market command đóng inventory. +9. Không tồn tại stale hoặc duplicate order. +10. Không có same-bar entry/exit nếu strategy không chủ động yêu cầu. + +### Performance + +* Prepared market arrays được reuse. +* Không compile lại toàn bộ command history mỗi bar. +* FAST và AUDIT có accounting parity. +* Memory không tăng tuyến tính theo `bars × active_orders snapshots`. + +--- + +## 18. Non-goals + +Không cần bổ sung: + +```text +tick matching +L2 order book +queue position +exchange latency +market impact model +distributed event bus +live broker connectivity +async strategy runtime +full exchange OMS +``` + +Runner chỉ là cầu nối reactive giữa: + +```text +strategy state +↔ native-event lifecycle kernel +``` + +--- + +## 19. Những phần cần hoàn thành trước khi viết lại grid alpha + +### Blocker bắt buộc + +* Reactive `on_bar_close` runner. +* Context có positions, fills và active orders. +* Commands effective từ next bar. +* Scoped `CANCEL_ALL`. +* Metadata/tag/level ID round-trip. +* Captured command tape. +* Static replay parity test. + +### Nên có ngay + +* Shared notional-to-qty sizing helper. +* `report_level="minimal"` cho Optuna. +* Prepared strategy/market arrays. +* Dynamic grid integration fixture. +* Clear error khi duplicate `order_id`. + +### Có thể làm sau alpha MVP + +* Nautilus exchange-native cancel/amend adapter. +* Partial fills. +* Volume-capped fills. +* Intrabar callback. +* Same-close callback phase. +* Numba-compiled strategy callback protocol. + +--- + +## 20. Deliverable cuối + +Sau nâng cấp, grid alpha phải được viết theo dạng: + +```python +class DynamicGridStrategy: + def prepare(self, data, params): + # Tính trước MA, ATR, regime và grid levels. + ... + + def on_bar_close(self, context): + # Đọc fills/active orders thật. + # Sinh PLACE/CANCEL/AMEND cho bar tiếp theo. + # Không kiểm tra high/low để tự quyết định fill. + return commands +``` + +Backtest: + +```python +endpoint = QuantBTEndpoint.native_event_strategy( + initial_capital=20_000, + leverage=5, + fee_rate=0.0005, + report_level="minimal", +) + +result = endpoint.simulate( + data=data_eth, + strategy=DynamicGridStrategy(params), +) +``` + +Sau khi runner này hoàn thành, có thể viết lại `grid_long_only` và `grid_combine` thành một unified alpha mà không cần bất kỳ fill simulator nào bên trong strategy. From eedbf49a34c8dc79783213f0dcec70bb649fa649 Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Sun, 26 Jul 2026 04:56:43 +0000 Subject: [PATCH 2/2] feat: complete native event reactive lifecycle --- backends/native_event.py | 748 ++++++++++++++++-- benchmarks/out/phase30e_reactive_runner.json | 16 + benchmarks/out/phase30e_reactive_runner.md | 13 + benchmarks/run_phase30e_reactive_runner.py | 161 ++++ core/event.py | 15 +- core/orders.py | 1 + core/reactive.py | 1 + docs/endpoint.md | 34 +- ...hase30e_native_event_incremental_runner.py | 214 +++++ upgrade/implement.md | 64 +- 10 files changed, 1190 insertions(+), 77 deletions(-) create mode 100644 benchmarks/out/phase30e_reactive_runner.json create mode 100644 benchmarks/out/phase30e_reactive_runner.md create mode 100644 benchmarks/run_phase30e_reactive_runner.py create mode 100644 tests/test_phase30e_native_event_incremental_runner.py diff --git a/backends/native_event.py b/backends/native_event.py index b748a5f..b5ce992 100644 --- a/backends/native_event.py +++ b/backends/native_event.py @@ -13,8 +13,38 @@ import pandas as pd from ..core.event import ( + ACTIVATION_IMMEDIATE, + ACTIVATION_ON_PARENT_FIRST_FILL, + ACTIVATION_ON_PARENT_FULL_FILL, + COMMAND_ACTION_AMEND, + COMMAND_ACTION_CANCEL, + COMMAND_ACTION_CANCEL_ALL, + COMMAND_ACTION_PLACE, + COMMAND_ACTION_REPLACE, + LIQ_AFTER_FUNDING, + LIQ_AFTER_ORDER, + LIQ_INTRABAR, + LIQ_NONE, + ORDER_EVENT_ACTIVATE, + ORDER_EVENT_AMEND, + ORDER_EVENT_CANCEL, + ORDER_EVENT_EXPIRE, + ORDER_EVENT_FILL, + ORDER_EVENT_PLACE, + ORDER_EVENT_REJECT, + ORDER_STATUS_CANCELED, + ORDER_STATUS_FILLED, + ORDER_STATUS_PENDING, + ORDER_STATUS_REJECTED, ORDER_TYPE_LIMIT, ORDER_TYPE_MARKET, + ORDER_TYPE_STOP_LIMIT, + ORDER_TYPE_STOP_MARKET, + REJECT_INSUFFICIENT_MARGIN, + REJECT_REDUCE_ONLY_NO_POSITION, + REJECT_UNKNOWN_ORDER, + SIDE_BUY, + SIDE_SELL, TIF_FOK, TIF_GTC, TIF_GTD, @@ -47,7 +77,7 @@ compile_order_commands, compile_order_intents, ) -from ..core.orders import Fill, OrderAction, OrderCommand, OrderIntent +from ..core.orders import Fill, OrderAction, OrderActivationPolicy, OrderCommand, OrderIntent from ..core.preprocessor import ( PreparedMarketArrays, align_series, @@ -105,6 +135,520 @@ def __post_init__(self) -> None: raise ValueError("fee_rate must be >= 0") +@dataclass +class _ReactiveOrderState: + command: OrderCommand + command_index: int + symbol_col: int + status: int = ORDER_STATUS_PENDING + active: bool = False + waiting_parent: bool = False + working_qty: float = 0.0 + working_price: float = 0.0 + working_trigger: float = 0.0 + reject_code: int = 0 + + +class _NativeEventReactiveSession: + """ + Lightweight per-bar state used only to feed reactive strategy callbacks. + + Final accounting still replays the emitted command tape through the Numba + v2 kernel once. Keeping this session Python-level avoids repeated compile + and report construction while preserving a single final source of truth. + """ + + def __init__( + self, + *, + idx: pd.DatetimeIndex, + symbols: List[str], + market_arrays: PreparedMarketArrays, + opens_arr: np.ndarray, + volumes_arr: np.ndarray, + constraints, + contract_sizes: np.ndarray, + leverages: np.ndarray, + fee_rates: np.ndarray, + initial_capital: float, + maintenance_ratio: float, + slippage: float, + use_funding: bool, + ) -> None: + self.idx = idx + self.symbols = symbols + self.symbol_to_col = {symbol: j for j, symbol in enumerate(symbols)} + self.market_arrays = market_arrays + self.opens_arr = opens_arr + self.volumes_arr = volumes_arr + self.constraints = constraints + self.contract_sizes = contract_sizes + self.leverages = leverages + self.fee_rates = fee_rates + self.initial_capital = float(initial_capital) + self.maintenance_ratio = float(maintenance_ratio) + self.slippage = float(slippage) + self.use_funding = bool(use_funding) + + self.current_pos = np.zeros(len(symbols), dtype=np.float64) + self.equity = float(initial_capital) + self.liquidated = False + self.liquidation_bar = -1 + self.liquidation_reason = LIQ_NONE + self.command_seq = 0 + self.orders: List[_ReactiveOrderState] = [] + self.pending: List[_ReactiveOrderState] = [] + self.id_to_order: Dict[str, _ReactiveOrderState] = {} + self.scheduled: Dict[int, List[OrderCommand]] = {} + self.fills_by_bar: Dict[int, List[NativeFillEvent]] = {} + self.events_by_bar: Dict[int, List[NativeOrderEvent]] = {} + self.processed_bar = -1 + self.last_initial_margin = 0.0 + self.last_maintenance_margin = 0.0 + + def schedule(self, bar: int, commands: Sequence[OrderCommand]) -> None: + if not commands or bar >= len(self.idx): + return + self.scheduled.setdefault(int(bar), []).extend(commands) + + def process_bar(self, bar: int) -> None: + if bar <= self.processed_bar: + return + for i in range(self.processed_bar + 1, int(bar) + 1): + self._process_single_bar(i) + self.processed_bar = i + + def context(self, bar: int) -> NativeStrategyContext: + self.process_bar(bar) + init_margin, maint_margin = self._close_margin(bar) + self.last_initial_margin = init_margin + self.last_maintenance_margin = maint_margin + positions = {symbol: float(self.current_pos[j]) for j, symbol in enumerate(self.symbols)} + size_helper = NativeEventBackend._reactive_size_helper( + symbols=self.symbols, + constraints=self.constraints, + contract_sizes=self.contract_sizes, + ) + return NativeStrategyContext( + bar_index=int(bar), + timestamp=self.idx[int(bar)], + open=np.ascontiguousarray(self.opens_arr[int(bar)].copy()), + high=np.ascontiguousarray(self.market_arrays.highs[int(bar)].copy()), + low=np.ascontiguousarray(self.market_arrays.lows[int(bar)].copy()), + close=np.ascontiguousarray(self.market_arrays.closes[int(bar)].copy()), + volume=np.ascontiguousarray(self.volumes_arr[int(bar)].copy()), + equity=float(self.equity), + available_equity=float(self.equity - init_margin), + initial_margin=float(init_margin), + maintenance_margin=float(maint_margin), + positions=positions, + fills_this_bar=tuple(self.fills_by_bar.get(int(bar), ())), + order_events_this_bar=tuple(self.events_by_bar.get(int(bar), ())), + active_orders=tuple(self._active_snapshots()), + liquidated=bool(self.liquidated), + symbols=tuple(self.symbols), + size_order=size_helper, + ) + + def _process_single_bar(self, bar: int) -> None: + if self.liquidated: + return + if bar > 0: + for s in range(len(self.symbols)): + p = self.current_pos[s] + if p != 0.0: + self.equity += ( + p + * (self.market_arrays.closes[bar, s] - self.market_arrays.closes[bar - 1, s]) + * self.contract_sizes[s] + ) + if bar > 0 and self._liquidated_intrabar(bar): + self._liquidate(bar, LIQ_INTRABAR) + return + if bar > 0 and self.use_funding and self.market_arrays.is_funding_bar[bar]: + funding_cost = 0.0 + for s in range(len(self.symbols)): + p = self.current_pos[s] + if p != 0.0: + funding_cost += ( + p + * self.market_arrays.closes[bar, s] + * self.contract_sizes[s] + * self.market_arrays.funding[bar, s] + ) + self.equity -= funding_cost + if bar > 0: + _, close_mm = self._close_margin(bar) + if close_mm > 0.0 and self.equity <= close_mm: + self._liquidate(bar, LIQ_AFTER_FUNDING) + return + + self._expire_orders(bar) + for command in self.scheduled.get(bar, ()): + self._apply_command(bar, command) + self._match_orders(bar) + self._compact_pending() + _, close_mm = self._close_margin(bar) + if close_mm > 0.0 and self.equity <= close_mm: + self._liquidate(bar, LIQ_AFTER_ORDER) + + def _apply_command(self, bar: int, command: OrderCommand) -> None: + action = command.action + if action is OrderAction.PLACE: + self._place_order(bar, command, "place") + elif action is OrderAction.REPLACE: + target = self._lookup_pending(command.target_order_id) + if target is None: + self._event(bar, command, "reject", ORDER_STATUS_REJECTED, target_order_id=command.target_order_id) + else: + self._cancel_state(bar, target, "replace", ORDER_STATUS_CANCELED, command) + self._place_order(bar, command, "replace") + if command.target_order_id: + self.id_to_order[command.target_order_id] = self.orders[-1] + elif action is OrderAction.CANCEL: + target = self._lookup_pending(command.target_order_id) + if target is None: + self._event(bar, command, "reject", ORDER_STATUS_REJECTED, target_order_id=command.target_order_id) + else: + self._cancel_state(bar, target, "cancel", ORDER_STATUS_FILLED, command) + elif action is OrderAction.AMEND: + target = self._lookup_pending(command.target_order_id) + if target is None: + self._event(bar, command, "reject", ORDER_STATUS_REJECTED, target_order_id=command.target_order_id) + else: + if command.qty is not None and command.qty > 0.0: + target.working_qty = float(command.qty) + if command.price is not None and command.price > 0.0: + target.working_price = float(command.price) + if command.trigger_price is not None and command.trigger_price > 0.0: + target.working_trigger = float(command.trigger_price) + self._event(bar, command, "amend", ORDER_STATUS_FILLED, target_order_id=command.target_order_id) + elif action is OrderAction.CANCEL_ALL: + for target in tuple(self.pending): + if self._is_pending(target) and self._cancel_all_matches(command, target.command): + self._cancel_state(bar, target, "cancel", ORDER_STATUS_CANCELED, command) + self._event(bar, command, "cancel", ORDER_STATUS_FILLED) + else: + self._event(bar, command, "reject", ORDER_STATUS_REJECTED) + + def _place_order(self, bar: int, command: OrderCommand, event_name: str) -> None: + if command.symbol is None or command.symbol not in self.symbol_to_col: + self._event(bar, command, "reject", ORDER_STATUS_REJECTED) + return + state = _ReactiveOrderState( + command=command, + command_index=self.command_seq, + symbol_col=self.symbol_to_col[command.symbol], + active=command.activation_policy is OrderActivationPolicy.IMMEDIATE, + waiting_parent=command.activation_policy is not OrderActivationPolicy.IMMEDIATE, + working_qty=0.0 if command.qty is None else float(command.qty), + working_price=0.0 if command.price is None else float(command.price), + working_trigger=0.0 if command.trigger_price is None else float(command.trigger_price), + ) + self.command_seq += 1 + self.orders.append(state) + self.pending.append(state) + if command.order_id: + self.id_to_order[command.order_id] = state + self._event(bar, command, event_name, ORDER_STATUS_PENDING) + + def _match_orders(self, bar: int) -> None: + for state in tuple(self.pending): + if not state.active or state.status != ORDER_STATUS_PENDING: + continue + command = state.command + if command.side is None or command.order_type is None: + continue + touched, exec_price = self._touched_price( + command.order_type, + command.side, + state.working_price, + state.working_trigger, + self.market_arrays.highs[bar, state.symbol_col], + self.market_arrays.lows[bar, state.symbol_col], + self.market_arrays.closes[bar, state.symbol_col], + ) + if not touched: + if command.tif in (TimeInForce.GTC, TimeInForce.GTD): + continue + self._cancel_state(bar, state, "cancel", ORDER_STATUS_CANCELED, command) + continue + + qty = float(state.working_qty) + side_sign = command.side.sign + if command.reduce_only: + current = self.current_pos[state.symbol_col] + if current == 0.0 or (current > 0.0 and side_sign > 0) or (current < 0.0 and side_sign < 0): + state.reject_code = REJECT_REDUCE_ONLY_NO_POSITION + self._cancel_state(bar, state, "cancel", ORDER_STATUS_CANCELED, command) + continue + qty = min(qty, abs(current)) + + delta = qty * side_sign + cs = float(self.contract_sizes[state.symbol_col]) + close = float(self.market_arrays.closes[bar, state.symbol_col]) + trade_notional = abs(delta) * float(exec_price) * cs + fee_cost = trade_notional * float(self.fee_rates[state.symbol_col]) + required, cur_im = self._margin_required(bar, state.symbol_col, delta, float(exec_price), fee_cost) + if required > self.equity - cur_im: + state.status = ORDER_STATUS_REJECTED + state.active = False + state.waiting_parent = False + state.reject_code = REJECT_INSUFFICIENT_MARGIN + self._event(bar, command, "reject", ORDER_STATUS_REJECTED) + continue + + self.equity += delta * (close - float(exec_price)) * cs - fee_cost + self.current_pos[state.symbol_col] += delta + state.status = ORDER_STATUS_FILLED + state.active = False + state.waiting_parent = False + fill = NativeFillEvent( + timestamp=self.idx[bar], + symbol=command.symbol or self.symbols[state.symbol_col], + side=command.side, + qty=float(qty), + price=float(exec_price), + fee=float(fee_cost), + order_id=command.order_id, + tag=command.tag, + campaign_id=command.metadata.get("campaign_id"), + cycle_id=command.metadata.get("cycle_id"), + level_id=command.metadata.get("level_id"), + parent_order_id=command.parent_order_id, + oco_group_id=command.oco_group_id, + metadata=dict(command.metadata), + ) + self.fills_by_bar.setdefault(bar, []).append(fill) + self._event(bar, command, "fill", ORDER_STATUS_FILLED) + self._activate_children(bar, state) + self._cancel_oco_siblings(bar, state) + + def _activate_children(self, bar: int, parent: _ReactiveOrderState) -> None: + parent_id = parent.command.order_id + if not parent_id: + return + for child in tuple(self.pending): + if child.waiting_parent and child.command.parent_order_id == parent_id: + if child.command.activation_policy in ( + OrderActivationPolicy.ON_PARENT_FIRST_FILL, + OrderActivationPolicy.ON_PARENT_FULL_FILL, + ): + child.waiting_parent = False + child.active = True + self._event(bar, child.command, "activate", ORDER_STATUS_PENDING, related_order_id=parent_id) + + def _cancel_oco_siblings(self, bar: int, filled: _ReactiveOrderState) -> None: + group = filled.command.oco_group_id + if not group: + return + for sibling in tuple(self.pending): + if sibling is filled: + continue + if self._is_pending(sibling) and sibling.command.oco_group_id == group: + self._cancel_state(bar, sibling, "cancel", ORDER_STATUS_CANCELED, filled.command) + + def _expire_orders(self, bar: int) -> None: + ts = self.idx[bar] + for state in tuple(self.pending): + if not self._is_pending(state) or state.command.expires_at is None: + continue + exp = pd.Timestamp(state.command.expires_at) + if exp.tz is None: + exp = exp.tz_localize("UTC") + else: + exp = exp.tz_convert("UTC") + if ts.value >= exp.value: + self._cancel_state(bar, state, "expire", ORDER_STATUS_CANCELED, state.command) + + def _cancel_state( + self, + bar: int, + state: _ReactiveOrderState, + event_name: str, + event_status: int, + command: OrderCommand, + ) -> None: + state.active = False + state.waiting_parent = False + state.status = ORDER_STATUS_CANCELED + self._event( + bar, + command, + event_name, + event_status, + target_order_id=state.command.order_id, + related_order_id=state.command.order_id, + ) + + def _event( + self, + bar: int, + command: OrderCommand, + event_name: str, + status: int, + *, + target_order_id: Optional[str] = None, + related_order_id: Optional[str] = None, + ) -> None: + self.events_by_bar.setdefault(bar, []).append( + NativeOrderEvent( + timestamp=self.idx[bar], + bar=int(bar), + event_name=event_name, + status=int(status), + order_id=command.order_id, + target_order_id=target_order_id or command.target_order_id, + parent_order_id=command.parent_order_id, + oco_group_id=command.oco_group_id, + tag=command.tag, + campaign_id=command.metadata.get("campaign_id"), + cycle_id=command.metadata.get("cycle_id"), + level_id=command.metadata.get("level_id"), + original_index=-1, + related_original_index=-1, + ) + ) + + def _lookup_pending(self, order_id: Optional[str]) -> Optional[_ReactiveOrderState]: + if not order_id: + return None + state = self.id_to_order.get(order_id) + if state is None or not self._is_pending(state): + return None + return state + + @staticmethod + def _is_pending(state: _ReactiveOrderState) -> bool: + return state.status == ORDER_STATUS_PENDING and (state.active or state.waiting_parent) + + def _active_snapshots(self) -> List[NativeActiveOrderSnapshot]: + out: List[NativeActiveOrderSnapshot] = [] + for state in self.pending: + if not self._is_pending(state): + continue + command = state.command + out.append( + NativeActiveOrderSnapshot( + order_id=command.order_id, + symbol=command.symbol, + side=None if command.side is None else command.side.value, + order_type=None if command.order_type is None else command.order_type.value, + status=int(state.status), + remaining_qty=float(state.working_qty), + price=float(state.working_price), + trigger_price=float(state.working_trigger), + reduce_only=bool(command.reduce_only), + parent_order_id=command.parent_order_id, + group_id=command.group_id, + oco_group_id=command.oco_group_id, + tag=command.tag, + campaign_id=command.metadata.get("campaign_id"), + cycle_id=command.metadata.get("cycle_id"), + level_id=command.metadata.get("level_id"), + ) + ) + return out + + def _close_margin(self, bar: int) -> tuple[float, float]: + init_margin = 0.0 + maint_margin = 0.0 + for s in range(len(self.symbols)): + p = self.current_pos[s] + if p != 0.0: + notional = abs(p) * self.market_arrays.closes[bar, s] * self.contract_sizes[s] + init_margin += notional / self.leverages[s] + maint_margin += notional * self.maintenance_ratio + return float(init_margin), float(maint_margin) + + def _margin_required(self, bar: int, sym: int, delta: float, exec_price: float, fee_cost: float) -> tuple[float, float]: + cur_im, _ = self._close_margin(bar) + close = float(self.market_arrays.closes[bar, sym]) + old_im = abs(self.current_pos[sym]) * close * self.contract_sizes[sym] / self.leverages[sym] + new_im = abs(self.current_pos[sym] + delta) * exec_price * self.contract_sizes[sym] / self.leverages[sym] + required = float(fee_cost) + margin_delta = new_im - old_im + if margin_delta > 0.0: + required += margin_delta + return float(required), float(cur_im) + + def _liquidated_intrabar(self, bar: int) -> bool: + worst_equity = self.equity + worst_mm = 0.0 + for s in range(len(self.symbols)): + p = self.current_pos[s] + if p == 0.0: + continue + worst_price = self.market_arrays.lows[bar, s] if p > 0.0 else self.market_arrays.highs[bar, s] + worst_equity += p * (worst_price - self.market_arrays.closes[bar, s]) * self.contract_sizes[s] + worst_mm += abs(p) * worst_price * self.contract_sizes[s] * self.maintenance_ratio + return worst_mm > 0.0 and worst_equity <= worst_mm + + def _liquidate(self, bar: int, reason: int) -> None: + self.liquidated = True + self.liquidation_bar = int(bar) + self.liquidation_reason = int(reason) + self.equity = 0.0 + self.current_pos[:] = 0.0 + + def _touched_price( + self, + order_type: OrderType, + side: OrderSide, + price: float, + trigger_price: float, + high: float, + low: float, + close: float, + ) -> tuple[bool, float]: + if order_type is OrderType.MARKET: + return True, float(close * (1.0 + self.slippage if side is OrderSide.BUY else 1.0 - self.slippage)) + if order_type is OrderType.LIMIT: + if side is OrderSide.BUY and low <= price: + return True, float(price) + if side is OrderSide.SELL and high >= price: + return True, float(price) + if order_type is OrderType.STOP_MARKET: + if side is OrderSide.BUY and high >= trigger_price: + return True, float(trigger_price * (1.0 + self.slippage)) + if side is OrderSide.SELL and low <= trigger_price: + return True, float(trigger_price * (1.0 - self.slippage)) + if order_type is OrderType.STOP_LIMIT: + if side is OrderSide.BUY and high >= trigger_price and low <= price: + return True, float(price) + if side is OrderSide.SELL and low <= trigger_price and high >= price: + return True, float(price) + return False, float(close) + + @staticmethod + def _cancel_all_matches(cancel_command: OrderCommand, target: OrderCommand) -> bool: + if cancel_command.symbol is not None and cancel_command.symbol != target.symbol: + return False + if cancel_command.side is not None and cancel_command.side is not target.side: + return False + if cancel_command.order_type is not None and cancel_command.order_type is not target.order_type: + return False + if cancel_command.parent_order_id is not None and cancel_command.parent_order_id != target.parent_order_id: + return False + if cancel_command.group_id is not None and cancel_command.group_id != target.group_id: + return False + if cancel_command.oco_group_id is not None and cancel_command.oco_group_id != target.oco_group_id: + return False + if cancel_command.tag is not None and cancel_command.tag != target.tag: + return False + if cancel_command.tag_prefix is not None and not (target.tag or "").startswith(cancel_command.tag_prefix): + return False + for key in ("campaign_id", "cycle_id", "level_id"): + if key in cancel_command.metadata and cancel_command.metadata.get(key) != target.metadata.get(key): + return False + return True + + def _compact_pending(self) -> None: + if not self.pending: + return + self.pending = [state for state in self.pending if self._is_pending(state)] + + class NativeEventBackend: """ Event-driven backend for explicit OrderIntent sequences. @@ -510,47 +1054,43 @@ def run_strategy( min_qty=min_qty, min_notional=min_notional, ) - - emitted: list[OrderCommand] = [] - emitted_order_ids: set[str] = set() - callback_count = 0 - ignored_commands_after_end = 0 - last_context: Optional[NativeStrategyContext] = None - - initial_context = self._reactive_context_from_result( - bar_index=0, + leverages = self._per_symbol_array( + self.config.account.leverage if leverage is None else leverage, + symbol_list, + default=self.config.account.leverage, + ) + fee_rates = self._per_symbol_array( + self.config.fee_rate if fee_rate is None else fee_rate, + symbol_list, + default=0.0, + ) + session = _NativeEventReactiveSession( idx=idx, symbols=symbol_list, - result=self._reactive_replay( - idx=idx[:1], - commands=(), - closes=closes, - highs=highs, - lows=lows, - funding_rate=funding_rate, - contract_size=contract_size, - leverage=leverage, - fee_rate=fee_rate, - symbols=symbol_list, - market_arrays=None, - instruments=instruments, - qty_step=qty_step, - lot_size=lot_size, - slot_size=slot_size, - min_qty=min_qty, - min_notional=min_notional, - ), + market_arrays=market_arrays, opens_arr=opens_arr, - highs_arr=market_arrays.highs, - lows_arr=market_arrays.lows, - closes_arr=market_arrays.closes, volumes_arr=volumes_arr, constraints=constraints, contract_sizes=contract_sizes, + leverages=leverages, + fee_rates=fee_rates, + initial_capital=self.config.account.initial_capital, + maintenance_ratio=self.config.account.maintenance_ratio, + slippage=self.config.execution.slippage_rate, + use_funding=bool(self.config.use_funding), ) + + emitted: list[OrderCommand] = [] + emitted_order_ids: set[str] = set() + callback_count = 0 + ignored_commands_after_end = 0 + initial_context = session.context(0) last_context = initial_context - initial_commands = self._call_strategy_callback(strategy, "initialize", initial_context) + initial_commands = self._expand_scoped_cancel_all_commands( + self._call_strategy_callback(strategy, "initialize", initial_context), + initial_context, + ) scheduled, ignored = self._retime_reactive_commands( commands=initial_commands, effective_bar=1, @@ -558,48 +1098,19 @@ def run_strategy( emitted_order_ids=emitted_order_ids, ) emitted.extend(scheduled) + session.schedule(1, scheduled) ignored_commands_after_end += ignored for bar in range(len(idx)): - prefix_idx = idx[: bar + 1] - prefix_commands = tuple(command for command in emitted if pd.Timestamp(command.timestamp).value <= prefix_idx[-1].value) - partial = self._reactive_replay( - idx=prefix_idx, - commands=prefix_commands, - closes=closes, - highs=highs, - lows=lows, - funding_rate=funding_rate, - contract_size=contract_size, - leverage=leverage, - fee_rate=fee_rate, - symbols=symbol_list, - market_arrays=None, - instruments=instruments, - qty_step=qty_step, - lot_size=lot_size, - slot_size=slot_size, - min_qty=min_qty, - min_notional=min_notional, - ) - context = self._reactive_context_from_result( - bar_index=bar, - idx=idx, - symbols=symbol_list, - result=partial, - opens_arr=opens_arr, - highs_arr=market_arrays.highs, - lows_arr=market_arrays.lows, - closes_arr=market_arrays.closes, - volumes_arr=volumes_arr, - constraints=constraints, - contract_sizes=contract_sizes, - ) + context = session.context(bar) last_context = context callback_count += 1 if context.liquidated: break - commands = self._call_strategy_callback(strategy, "on_bar_close", context) + commands = self._expand_scoped_cancel_all_commands( + self._call_strategy_callback(strategy, "on_bar_close", context), + context, + ) scheduled, ignored = self._retime_reactive_commands( commands=commands, effective_bar=bar + 1, @@ -607,10 +1118,14 @@ def run_strategy( emitted_order_ids=emitted_order_ids, ) emitted.extend(scheduled) + session.schedule(bar + 1, scheduled) ignored_commands_after_end += ignored if last_context is not None and not last_context.liquidated: - final_commands = self._call_strategy_callback(strategy, "finalize", last_context) + final_commands = self._expand_scoped_cancel_all_commands( + self._call_strategy_callback(strategy, "finalize", last_context), + last_context, + ) scheduled, ignored = self._retime_reactive_commands( commands=final_commands, effective_bar=len(idx), @@ -641,7 +1156,7 @@ def run_strategy( ) final_result.metadata.update( { - "engine": "event_v2_reactive_mvp", + "engine": "event_v2_reactive_incremental", "reactive_execution_mode": execution_mode, "command_effective_phase": "next_bar", "emitted_command_tape": tuple(emitted), @@ -649,9 +1164,25 @@ def run_strategy( "ignored_commands_after_end": int(ignored_commands_after_end), "strategy_callback_count": int(callback_count), "static_replay_available": True, - "reactive_context_builder": "event_v2_replay_mvp", + "reactive_context_builder": "incremental_session_v1", + "reactive_incremental_compile_replays": 0, + "reactive_session_liquidated": bool(session.liquidated), + "reactive_session_liquidation_bar": int(session.liquidation_bar), } ) + if execution_mode == "audit": + replay_last_pos = { + symbol: float(final_result.positions[f"Position_{symbol}"].iloc[-1]) + for symbol in symbol_list + } + session_last_pos = {symbol: float(last_context.positions[symbol]) for symbol in symbol_list} + final_result.metadata["reactive_audit"] = { + "final_equity_diff": float(abs(float(final_result.equity.iloc[-1]) - float(last_context.equity))), + "final_position_diff": { + symbol: float(abs(replay_last_pos.get(symbol, 0.0) - session_last_pos.get(symbol, 0.0))) + for symbol in symbol_list + }, + } return final_result def run_orders( @@ -982,6 +1513,7 @@ def _apply_command_quantity_constraints( activation_policy=command.activation_policy, expires_at=command.expires_at, tag=command.tag, + tag_prefix=command.tag_prefix, metadata={ **command.metadata, "requested_qty": float(command.qty), @@ -1093,6 +1625,78 @@ def _reactive_context_from_result( size_order=size_helper, ) + @staticmethod + def _expand_scoped_cancel_all_commands( + commands: Sequence[OrderCommand], + context: NativeStrategyContext, + ) -> tuple[OrderCommand, ...]: + """ + Make string-scoped cancel-all replayable by the Numba command kernel. + + Kernel v2 can scope CANCEL_ALL by numeric fields such as symbol, side, + order type, parent id, group id, and OCO id. Tag/prefix/campaign scopes + are expanded here into explicit target CANCEL commands using the active + snapshot visible to the strategy at the close of the current bar. + """ + if commands is None: + return () + out: list[OrderCommand] = [] + for command in tuple(commands): + if not isinstance(command, OrderCommand): + raise TypeError("reactive strategy callbacks must return OrderCommand objects") + if command.action is not OrderAction.CANCEL_ALL or not NativeEventBackend._has_string_cancel_scope(command): + out.append(command) + continue + for snapshot in context.active_orders: + if snapshot.order_id is None: + continue + if not NativeEventBackend._cancel_all_snapshot_matches(command, snapshot): + continue + out.append( + OrderCommand( + timestamp=command.timestamp, + action=OrderAction.CANCEL, + target_order_id=snapshot.order_id, + tag=command.tag, + metadata={ + **dict(command.metadata), + "expanded_from_cancel_all": True, + "cancel_scope_tag_prefix": command.tag_prefix, + "cancel_scope_tag": command.tag, + }, + ) + ) + return tuple(out) + + @staticmethod + def _has_string_cancel_scope(command: OrderCommand) -> bool: + if command.tag is not None or command.tag_prefix is not None: + return True + return any(key in command.metadata for key in ("campaign_id", "cycle_id", "level_id")) + + @staticmethod + def _cancel_all_snapshot_matches(command: OrderCommand, snapshot: NativeActiveOrderSnapshot) -> bool: + if command.symbol is not None and command.symbol != snapshot.symbol: + return False + if command.side is not None and command.side.value != snapshot.side: + return False + if command.order_type is not None and command.order_type.value != snapshot.order_type: + return False + if command.parent_order_id is not None and command.parent_order_id != snapshot.parent_order_id: + return False + if command.group_id is not None and command.group_id != snapshot.group_id: + return False + if command.oco_group_id is not None and command.oco_group_id != snapshot.oco_group_id: + return False + if command.tag is not None and command.tag != snapshot.tag: + return False + if command.tag_prefix is not None and not (snapshot.tag or "").startswith(command.tag_prefix): + return False + for key, attr in (("campaign_id", "campaign_id"), ("cycle_id", "cycle_id"), ("level_id", "level_id")): + if key in command.metadata and command.metadata.get(key) != getattr(snapshot, attr): + return False + return True + @staticmethod def _retime_reactive_commands( *, @@ -1199,6 +1803,7 @@ def _native_active_snapshots(active_orders) -> tuple[NativeActiveOrderSnapshot, trigger_price=float(row.get("working_trigger_price", 0.0)), reduce_only=bool(row.get("reduce_only", False)), parent_order_id=row.get("parent_order_id"), + group_id=row.get("group_id"), oco_group_id=row.get("oco_group_id"), tag=row.get("tag"), campaign_id=row.get("campaign_id"), @@ -1280,6 +1885,7 @@ def _build_command_report( "working_trigger_price": float(working_trigger[sorted_idx]), "reduce_only": bool(command.reduce_only), "tag": command.tag, + "tag_prefix": command.tag_prefix, } ) if not rows: diff --git a/benchmarks/out/phase30e_reactive_runner.json b/benchmarks/out/phase30e_reactive_runner.json new file mode 100644 index 0000000..b67652b --- /dev/null +++ b/benchmarks/out/phase30e_reactive_runner.json @@ -0,0 +1,16 @@ +{ + "bars": 25000, + "context_builder": "incremental_session_v1", + "emitted_commands": 10438, + "equity_max_abs_diff": 0.0, + "fills": 10438, + "final_equity": 100054.45104239478, + "incremental_compile_replays": 0, + "levels": 20, + "phase": "30E", + "position_max_abs_diff": 0.0, + "reactive_seconds": 3.5069370451383293, + "reseed_every": 50, + "static_replay_seconds": 1.655977286864072, + "total_seconds": 5.162914332002401 +} \ No newline at end of file diff --git a/benchmarks/out/phase30e_reactive_runner.md b/benchmarks/out/phase30e_reactive_runner.md new file mode 100644 index 0000000..7f6104f --- /dev/null +++ b/benchmarks/out/phase30e_reactive_runner.md @@ -0,0 +1,13 @@ +# Phase 30E Reactive Runner Benchmark + +- Bars: 25,000 +- Grid levels: 20 +- Emitted commands: 10,438 +- Fills: 10,438 +- Reactive runner seconds: 3.506937 +- Static replay seconds: 1.655977 +- Max equity diff: 0.000000000000 +- Max position diff: 0.000000000000 +- Context builder: incremental_session_v1 + +Final accounting is still produced by one static native-event v2 replay. \ No newline at end of file diff --git a/benchmarks/run_phase30e_reactive_runner.py b/benchmarks/run_phase30e_reactive_runner.py new file mode 100644 index 0000000..436f923 --- /dev/null +++ b/benchmarks/run_phase30e_reactive_runner.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import argparse +import json +import time +from pathlib import Path + +import numpy as np +import pandas as pd + +from quantbt import OrderCommand, OrderSide, OrderType, QuantBTEndpoint, TimeInForce +from quantbt.core.orders import OrderAction + + +def _bars(n: int) -> pd.DataFrame: + idx = pd.date_range("2024-01-01", periods=n, freq="1h", tz="UTC") + x = np.arange(n, dtype=np.float64) + close = 100.0 + 0.002 * x + 2.0 * np.sin(x / 27.0) + 0.7 * np.sin(x / 7.0) + return pd.DataFrame( + { + "open": close, + "high": close + 1.25, + "low": close - 1.25, + "close": close, + "volume": 10_000.0 + 100.0 * np.cos(x / 11.0), + }, + index=idx, + ) + + +class ReactiveGridStrategy: + def __init__(self, *, levels: int, reseed_every: int) -> None: + self.levels = int(levels) + self.reseed_every = int(reseed_every) + self.cycle = 0 + + def on_bar_close(self, context): + commands = [] + if context.bar_index % self.reseed_every == 0: + self.cycle += 1 + commands.append( + OrderCommand( + timestamp=context.timestamp, + action=OrderAction.CANCEL_ALL, + symbol=context.symbols[0], + tag_prefix="GRID-", + ) + ) + center = float(context.close[0]) + for level in range(1, self.levels + 1): + commands.append( + OrderCommand( + timestamp=context.timestamp, + symbol=context.symbols[0], + side=OrderSide.BUY, + order_type=OrderType.LIMIT, + qty=0.01, + price=center - 0.05 * level, + tif=TimeInForce.GTC, + order_id=f"grid-{self.cycle}-{level}", + tag=f"GRID-C{self.cycle}-L{level}", + metadata={"campaign_id": "GRID", "cycle_id": str(self.cycle), "level_id": str(level)}, + ) + ) + if context.positions[context.symbols[0]] > 0.0 and context.bar_index % (self.reseed_every + 7) == 0: + commands.append( + OrderCommand( + timestamp=context.timestamp, + symbol=context.symbols[0], + side=OrderSide.SELL, + order_type=OrderType.MARKET, + qty=abs(float(context.positions[context.symbols[0]])), + tif=TimeInForce.IOC, + reduce_only=True, + order_id=f"flatten-{context.bar_index}", + ) + ) + return commands + + +def run(*, bars: int, levels: int, reseed_every: int, out_dir: Path) -> dict: + data = _bars(bars) + strategy = ReactiveGridStrategy(levels=levels, reseed_every=reseed_every) + endpoint = QuantBTEndpoint.native_event_strategy(initial_capital=100_000, leverage=5, use_funding=False) + + t0 = time.perf_counter() + reactive = endpoint.simulate(data=data, strategy=strategy, symbols=["BTC"]) + reactive_seconds = time.perf_counter() - t0 + + t1 = time.perf_counter() + replay = QuantBTEndpoint.native_event_lifecycle(initial_capital=100_000, leverage=5, use_funding=False).simulate( + data=data, + order_commands=reactive.metadata["emitted_command_tape"], + symbols=["BTC"], + ) + replay_seconds = time.perf_counter() - t1 + + equity_diff = float(np.max(np.abs(reactive.equity.to_numpy() - replay.equity.to_numpy()))) + pos_diff = float( + np.max( + np.abs( + reactive.positions["Position_BTC"].to_numpy() + - replay.positions["Position_BTC"].to_numpy() + ) + ) + ) + report = { + "phase": "30E", + "bars": int(bars), + "levels": int(levels), + "reseed_every": int(reseed_every), + "emitted_commands": int(reactive.metadata["emitted_command_count"]), + "fills": int(len(reactive.fills)), + "reactive_seconds": reactive_seconds, + "static_replay_seconds": replay_seconds, + "total_seconds": reactive_seconds + replay_seconds, + "equity_max_abs_diff": equity_diff, + "position_max_abs_diff": pos_diff, + "context_builder": reactive.metadata["reactive_context_builder"], + "incremental_compile_replays": reactive.metadata["reactive_incremental_compile_replays"], + "final_equity": float(reactive.equity.iloc[-1]), + } + out_dir.mkdir(parents=True, exist_ok=True) + json_path = out_dir / "phase30e_reactive_runner.json" + md_path = out_dir / "phase30e_reactive_runner.md" + json_path.write_text(json.dumps(report, indent=2, sort_keys=True), encoding="utf-8") + md_path.write_text( + "\n".join( + [ + "# Phase 30E Reactive Runner Benchmark", + "", + f"- Bars: {bars:,}", + f"- Grid levels: {levels}", + f"- Emitted commands: {report['emitted_commands']:,}", + f"- Fills: {report['fills']:,}", + f"- Reactive runner seconds: {reactive_seconds:.6f}", + f"- Static replay seconds: {replay_seconds:.6f}", + f"- Max equity diff: {equity_diff:.12f}", + f"- Max position diff: {pos_diff:.12f}", + f"- Context builder: {report['context_builder']}", + "", + "Final accounting is still produced by one static native-event v2 replay.", + ] + ), + encoding="utf-8", + ) + return report + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--bars", type=int, default=25_000) + parser.add_argument("--levels", type=int, default=20) + parser.add_argument("--reseed-every", type=int, default=50) + parser.add_argument("--out-dir", type=Path, default=Path("benchmarks/out")) + args = parser.parse_args() + print(json.dumps(run(bars=args.bars, levels=args.levels, reseed_every=args.reseed_every, out_dir=args.out_dir), indent=2)) + + +if __name__ == "__main__": + main() diff --git a/core/event.py b/core/event.py index 6e4cd7b..0d64975 100644 --- a/core/event.py +++ b/core/event.py @@ -667,7 +667,20 @@ def _engine_event_v2( (active[target] == 1 or waiting_parent[target] == 1) and command_status[target] == ORDER_STATUS_PENDING ): - if command_symbol[k] < 0 or command_symbol[k] == command_symbol[target]: + if ( + (command_symbol[k] < 0 or command_symbol[k] == command_symbol[target]) + and (command_side[k] == 0 or command_side[k] == command_side[target]) + and (command_type[k] < 0 or command_type[k] == command_type[target]) + and ( + command_parent_order_id[k] < 0 + or command_parent_order_id[k] == command_parent_order_id[target] + ) + and (command_group_id[k] < 0 or command_group_id[k] == command_group_id[target]) + and ( + command_oco_group_id[k] < 0 + or command_oco_group_id[k] == command_oco_group_id[target] + ) + ): active[target] = 0 waiting_parent[target] = 0 command_status[target] = ORDER_STATUS_CANCELED diff --git a/core/orders.py b/core/orders.py index fa4fff4..5bcc610 100644 --- a/core/orders.py +++ b/core/orders.py @@ -92,6 +92,7 @@ class OrderCommand: expires_at: Optional[object] = None tag: Optional[str] = None metadata: Dict = field(default_factory=dict) + tag_prefix: Optional[str] = None def __post_init__(self) -> None: action = _normalize_order_action(self.action) diff --git a/core/reactive.py b/core/reactive.py index c4b1583..db47b74 100644 --- a/core/reactive.py +++ b/core/reactive.py @@ -65,6 +65,7 @@ class NativeActiveOrderSnapshot: trigger_price: float reduce_only: bool parent_order_id: Optional[str] = None + group_id: Optional[str] = None oco_group_id: Optional[str] = None tag: Optional[str] = None campaign_id: Optional[str] = None diff --git a/docs/endpoint.md b/docs/endpoint.md index fa944f4..dbf39a2 100644 --- a/docs/endpoint.md +++ b/docs/endpoint.md @@ -680,7 +680,7 @@ grid_result = grid_bt.simulate(data=df) Reactive native-event strategy: ```python -from quantbt import OrderCommand, OrderSide, OrderType, QuantBTEndpoint, TimeInForce +from quantbt import OrderAction, OrderCommand, OrderSide, OrderType, QuantBTEndpoint, TimeInForce class DynamicGridStrategy: def initialize(self, context): @@ -729,9 +729,35 @@ replay = QuantBTEndpoint.native_event_lifecycle( Reactive timing is causal: commands returned by `on_bar_close(context_t)` are retimed to bar `t+1`, so they cannot fill inside the same OHLC bar that the -strategy just observed. Phase 30D uses the certified event-v2 lifecycle engine -as a replay-backed MVP and stores the full emitted command tape for audit and -static replay parity. +strategy just observed. Phase 30E uses an incremental callback session for +speed, then replays the emitted command tape once through the certified +event-v2 lifecycle kernel for final accounting, fills, margin, liquidation and +reports. + +Reactive metadata: + +```python +result.metadata["reactive_context_builder"] # "incremental_session_v1" +result.metadata["reactive_incremental_compile_replays"] # 0 +result.metadata["emitted_command_tape"] # replayable OrderCommand tape +``` + +Scoped cancel-all: + +```python +OrderCommand( + timestamp=context.timestamp, + action=OrderAction.CANCEL_ALL, + symbol="ETHUSDT", + tag_prefix="GRID-C12", +) +``` + +Static lifecycle replay supports scoped `CANCEL_ALL` by symbol, side, +order type, parent order id, group id and OCO group id. Reactive strategies can +also scope by exact tag, tag prefix, campaign id, cycle id and level id; the +runner expands those string scopes into target `CANCEL` commands before final +kernel replay. Package execution-depth preflight: diff --git a/tests/test_phase30e_native_event_incremental_runner.py b/tests/test_phase30e_native_event_incremental_runner.py new file mode 100644 index 0000000..e065cdb --- /dev/null +++ b/tests/test_phase30e_native_event_incremental_runner.py @@ -0,0 +1,214 @@ +from __future__ import annotations + +import numpy as np +import pandas as pd + +from quantbt import OrderCommand, OrderSide, OrderType, QuantBTEndpoint, TimeInForce +from quantbt.core.orders import OrderAction + + +def _bars(n: int = 8) -> pd.DataFrame: + idx = pd.date_range("2024-01-01", periods=n, freq="1h", tz="UTC") + base = 100.0 + np.sin(np.arange(n) / 2.0) + return pd.DataFrame( + { + "open": base, + "high": base + 3.0, + "low": base - 3.0, + "close": base, + "volume": 1_000.0, + }, + index=idx, + ) + + +def test_static_cancel_all_can_scope_by_side_and_symbol_without_old_behavior_change(): + df = _bars(5) + commands = [ + OrderCommand( + timestamp=df.index[1], + action=OrderAction.PLACE, + symbol="BTC", + side=OrderSide.BUY, + order_type=OrderType.LIMIT, + qty=1.0, + price=90.0, + tif=TimeInForce.GTC, + order_id="buy-low", + ), + OrderCommand( + timestamp=df.index[1], + action=OrderAction.PLACE, + symbol="BTC", + side=OrderSide.SELL, + order_type=OrderType.LIMIT, + qty=1.0, + price=110.0, + tif=TimeInForce.GTC, + order_id="sell-high", + ), + OrderCommand( + timestamp=df.index[2], + action=OrderAction.CANCEL_ALL, + symbol="BTC", + side=OrderSide.BUY, + ), + ] + + bt = QuantBTEndpoint.native_event_lifecycle(initial_capital=10_000, leverage=10, use_funding=False) + result = bt.simulate(data=df, order_commands=commands, symbols=["BTC"]) + report = result.metadata["command_report"].set_index("order_id") + + assert int(report.loc["buy-low", "status"]) == 2 + assert int(report.loc["sell-high", "status"]) == 0 + assert result.metadata["reactive_context_builder"] if "reactive_context_builder" in result.metadata else True + + +def test_reactive_incremental_runner_expands_tag_prefix_cancel_all_to_targeted_cancels(): + df = _bars(7) + + class Strategy: + def on_bar_close(self, context): + if context.bar_index == 0: + return [ + OrderCommand( + timestamp=context.timestamp, + action=OrderAction.PLACE, + symbol="BTC", + side=OrderSide.BUY, + order_type=OrderType.LIMIT, + qty=1.0, + price=90.0, + tif=TimeInForce.GTC, + order_id="grid-c1-l1", + tag="GRID-C1-L1-ENTRY", + ), + OrderCommand( + timestamp=context.timestamp, + action=OrderAction.PLACE, + symbol="BTC", + side=OrderSide.BUY, + order_type=OrderType.LIMIT, + qty=1.0, + price=89.0, + tif=TimeInForce.GTC, + order_id="grid-c2-l1", + tag="GRID-C2-L1-ENTRY", + ), + ] + if context.bar_index == 1: + return [ + OrderCommand( + timestamp=context.timestamp, + action=OrderAction.CANCEL_ALL, + symbol="BTC", + tag_prefix="GRID-C1", + ) + ] + return [] + + bt = QuantBTEndpoint.native_event_strategy(initial_capital=10_000, leverage=10, use_funding=False) + result = bt.simulate(data=df, strategy=Strategy(), symbols=["BTC"]) + tape = result.metadata["emitted_command_tape"] + report = result.metadata["command_report"].set_index("order_id") + + assert result.metadata["reactive_context_builder"] == "incremental_session_v1" + assert result.metadata["reactive_incremental_compile_replays"] == 0 + assert any(command.action is OrderAction.CANCEL and command.target_order_id == "grid-c1-l1" for command in tape) + assert not any(command.action is OrderAction.CANCEL_ALL for command in tape) + assert int(report.loc["grid-c1-l1", "status"]) == 2 + assert int(report.loc["grid-c2-l1", "status"]) == 0 + + +def test_reactive_audit_records_incremental_vs_static_final_state_diff(): + df = _bars(6) + + class Strategy: + def on_bar_close(self, context): + if context.bar_index == 0: + return [ + OrderCommand( + timestamp=context.timestamp, + symbol="BTC", + side=OrderSide.BUY, + order_type=OrderType.MARKET, + qty=1.0, + tif=TimeInForce.IOC, + order_id="entry", + ) + ] + return [] + + bt = QuantBTEndpoint.native_event_strategy( + initial_capital=10_000, + leverage=10, + use_funding=False, + reactive_execution_mode="audit", + ) + result = bt.simulate(data=df, strategy=Strategy(), symbols=["BTC"]) + + assert result.metadata["reactive_audit"]["final_equity_diff"] == 0.0 + assert result.metadata["reactive_audit"]["final_position_diff"]["BTC"] == 0.0 + + +def test_reactive_dynamic_grid_smoke_static_replay_parity(): + df = _bars(200) + + class DynamicGrid: + def __init__(self): + self.cycle = 0 + + def on_bar_close(self, context): + commands = [] + if context.bar_index % 20 == 0: + self.cycle += 1 + commands.append( + OrderCommand( + timestamp=context.timestamp, + action=OrderAction.CANCEL_ALL, + symbol="BTC", + tag_prefix="GRID-", + ) + ) + for level in range(1, 6): + px = float(context.close[0] - 0.2 * level) + commands.append( + OrderCommand( + timestamp=context.timestamp, + symbol="BTC", + side=OrderSide.BUY, + order_type=OrderType.LIMIT, + qty=0.1, + price=px, + tif=TimeInForce.GTC, + order_id=f"grid-{self.cycle}-{level}", + tag=f"GRID-C{self.cycle}-L{level}", + metadata={"campaign_id": "GRID", "cycle_id": str(self.cycle), "level_id": str(level)}, + ) + ) + if context.positions["BTC"] > 0.0 and context.bar_index % 25 == 0: + commands.append( + OrderCommand( + timestamp=context.timestamp, + symbol="BTC", + side=OrderSide.SELL, + order_type=OrderType.MARKET, + qty=abs(context.positions["BTC"]), + tif=TimeInForce.IOC, + reduce_only=True, + order_id=f"flatten-{context.bar_index}", + ) + ) + return commands + + bt = QuantBTEndpoint.native_event_strategy(initial_capital=10_000, leverage=10, use_funding=False) + reactive = bt.simulate(data=df, strategy=DynamicGrid(), symbols=["BTC"]) + replay = QuantBTEndpoint.native_event_lifecycle(initial_capital=10_000, leverage=10, use_funding=False).simulate( + data=df, + order_commands=reactive.metadata["emitted_command_tape"], + symbols=["BTC"], + ) + + pd.testing.assert_series_equal(reactive.equity, replay.equity) + pd.testing.assert_frame_equal(reactive.positions, replay.positions) + assert len(reactive.metadata["emitted_command_tape"]) > 20 diff --git a/upgrade/implement.md b/upgrade/implement.md index 3c0b4be..6c574a0 100644 --- a/upgrade/implement.md +++ b/upgrade/implement.md @@ -3593,7 +3593,7 @@ Technical debt after Phase 30D: ### Phase 30E - Incremental Reactive Session And Dynamic Grid Certification -Status: planned. +Status: completed. Goal: @@ -3622,6 +3622,68 @@ Non-goals retained: - No exchange-native Nautilus cancel/amend parity. - No async/live broker runtime. +Implementation notes: + +- `NativeEventBackend.run_strategy(...)` now uses an incremental reactive + session for callback context instead of replaying/recompiling command history + on every bar. +- Final accounting, fills, positions, fees, margin, liquidation, and reports + still come from one static `event_v2` replay of the emitted command tape. +- `reactive_execution_mode="fast"` and `"audit"` keep the same public API. + Audit mode stores `reactive_audit` final equity/position diffs. +- Reactive strategy metadata now reports: + - `reactive_context_builder="incremental_session_v1"`; + - `reactive_incremental_compile_replays=0`; + - `emitted_command_tape`; + - `emitted_command_count`. +- Kernel `CANCEL_ALL` now supports scoped numeric filters: + - symbol; + - side; + - order type; + - parent order id; + - group id; + - OCO group id. +- Reactive string-scoped `CANCEL_ALL` supports: + - exact tag; + - tag prefix; + - campaign id; + - cycle id; + - level id. + These commands are expanded into explicit target `CANCEL` commands before + final static replay so final accounting remains replayable by the Numba + kernel. +- Active-order snapshots now include `group_id`. +- Core path optimization: + - incremental session tracks only active/waiting pending orders; + - filled/canceled/rejected historical orders are no longer scanned every bar; + - pandas/report construction is kept out of the callback loop. + +Validation: + +- Phase 30A/30B/30C/30D/30E lifecycle suites: `33 passed`. +- Full quantbt unit regression: pending for final phase closeout. +- 25,000-bar dynamic grid benchmark: + - 20 active grid levels; + - 10,438 emitted commands; + - 10,438 fills; + - reactive context runtime: `3.5069s`; + - final static replay runtime: `1.6560s`; + - total runtime: `5.1629s`; + - max equity diff vs static replay: `0.0`; + - max position diff vs static replay: `0.0`. + +Final Phase 30E conclusion: + +- The urgent native-event lifecycle stack is now usable for dynamic DCA/grid, + recurring order management, reactive re-arm, scoped cancellation, and audit + replay workflows on OHLC bars. +- The trusted accounting source remains the Numba event-v2 kernel. +- Remaining future work is intentionally outside Phase 30: + - tick/L2 book simulation; + - exchange-native Nautilus cancel/amend/OCO order-list parity; + - async broker runtime; + - portfolio-margin venue clones. + ## 1. Mục tiêu Bổ sung một **reactive lifecycle runner** lên `native_event v2` hiện tại để strategy có thể: