m5-backtest-engine
- Date
- Language
- Python
Notes
How the repository is arranged
Repository map
The C++ implementation under core handles event replay, book state, execution, scheduling, and risk, with replay_engine.hpp defining the replay interface. The python/scripts directory contains a CSV converter. The root Python scripts were carried over from the original project. README.md records the build steps and current limits, while docs and stress_tests contain usage notes and checks.
Replay data contract
Market data is converted into timestamped binary events before the engine reads it. The archive records exchange time, local time, sequence, symbol, event type, and a fixed-size payload. generate_data.py creates synthetic archives, and strategy1.py shows how CSV bars become binary streams.
Strategy to engine flow
Strategy1 observes public trades for a leader symbol, keeps rolling price and volume windows, and calculates a shock score after both windows are warm. A threshold crossing submits an IOC market order for the responder symbol. The engine then reports fills through the strategy callback.
Reproducibility checks
The CLI hashes the strategy file and stores the configuration and run metadata with the JSON report. The stress_tests directory contains determinism_test.py, edge-case, liquidation, synthetic-data, and performance checks so the engine can be examined from more than one normal path.
Current limits
The Python package folder has been removed. The root CLI and some scripts still depend on imports, compiled libraries, and paths from the original project. Data and compiled libraries are not included. Check the README and current source before running documentation examples. The bar backtester’s sharpe field is a sign-based score, not a calculated Sharpe ratio.
Source files
m5-backtest-engine/1"""Strategy1: An example trading strategy replayed using the C++ engine.
2
3Execute directly via:
4 python -m backtest.strategy1 3m
5"""
6
7from __future__ import annotations
8
9import argparse
10import logging
11import os
12import struct
13import sys
14import time
15from datetime import UTC, datetime, timedelta
16from pathlib import Path
17from typing import Any
18
19import pandas as pd
20
21# Add the src folder and scripts folder to the python path
22sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
23sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
24
25try:
26 import _bootstrap # noqa: F401
27except ModuleNotFoundError:
28 pass
29
30import m5_engine as qe
31
32from backtesting.data_downloader import download_binance_klines
33
34logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
35logger = logging.getLogger(__name__)
36
37class Strategy1(qe.Strategy):
38 """
39 Submit responder orders when leader price and volume changes cross a threshold.
40 """
41
42 def __init__(
43 self,
44 engine: qe.ReplayEngine,
45 leader_id: int,
46 responder_id: int,
47 return_window: int = 5,
48 volume_window: int = 20,
49 shock_threshold: float = 1.0,
50 qty_per_trade: float = 1000.0,
51 ):
52 super().__init__()
53 self.engine = engine
54 self.leader_id = leader_id
55 self.responder_id = responder_id
56 self.return_window = return_window
57 self.volume_window = volume_window
58 self.shock_threshold = shock_threshold
59 self.qty_per_trade = qty_per_trade
60
61 # Queues to hold leader price & volume history
62 from collections import deque
63 self.leader_prices: deque[float] = deque(maxlen=return_window)
64 self.leader_volumes: deque[float] = deque(maxlen=volume_window)
65 self.current_order_id = 1
66
67 def on_init(self, registry: Any):
68 self.registry = registry
69
70 def on_market_event(self, event: Any, book: Any):
71 """Called for every trade and depth update chronologically."""
72 if event.type == qe.EventType.PUBLIC_TRADE:
73 trade = event.trade
74 if event.symbol_id == self.leader_id:
75 self._update_leader(trade)
76
77 def _update_leader(self, trade: Any):
78 if trade.price <= 0:
79 return
80
81 # Check if queues are warmed up
82 if (
83 len(self.leader_prices) == self.return_window
84 and len(self.leader_volumes) == self.volume_window
85 ):
86 ref_price = self.leader_prices[0]
87 avg_vol = sum(self.leader_volumes) / self.volume_window
88
89 if avg_vol > 0:
90 import math
91 volume_shock = (trade.qty / avg_vol) - 1.0
92 trailing_return = math.log(trade.price / ref_price) if ref_price > 0 else 0
93
94 # Scaled shock score
95 shock_score = (trailing_return / 0.01) * (1.0 + max(0.0, volume_shock))
96
97 if abs(shock_score) >= self.shock_threshold:
98 self._fire_responder_order(shock_score)
99
100 self.leader_prices.append(trade.price)
101 self.leader_volumes.append(trade.qty)
102
103 def _fire_responder_order(self, shock_score: float):
104 """Submit a market IOC order for the responder asset."""
105 order = qe.OrderSubmitPayload()
106 order.order_id = self.current_order_id
107 self.current_order_id += 1
108
109 order.qty = int(self.qty_per_trade)
110 order.side = qe.Side.BUY if shock_score > 0 else qe.Side.SELL
111 order.type = qe.OrderType.MARKET
112 order.tif = qe.TimeInForce.IOC
113 order.price = 0 # Market orders use 0 limit price fallback in our engine
114
115 self.engine.submit_order(self.responder_id, order)
116
117 def on_fill(self, fill: Any, symbol_id: int):
118 """Called automatically when an order is filled."""
119 logger.info(
120 "[FILL] Symbol: %d, Side: %s, Qty: %.2f, Price: %.2f",
121 symbol_id,
122 fill.side.name,
123 fill.fill_qty / 1000.0, # unscale qty
124 fill.fill_price / 100.0, # unscale price
125 )
126
127def parse_duration(duration_str: str) -> timedelta:
128 duration_str = duration_str.strip().lower()
129 unit = duration_str[-1]
130 value = int(duration_str[:-1])
131 if unit == "m":
132 return timedelta(days=value * 30)
133 elif unit == "y":
134 return timedelta(days=value * 365)
135 elif unit == "d":
136 return timedelta(days=value)
137 else:
138 raise ValueError(f"Unsupported duration: {duration_str}")
139
140def pack_event(exchange_ts: int, sequence_id: int, symbol_id: int, event_type: int, payload: bytes) -> bytes:
141 header = struct.pack("<qqQHBB", exchange_ts, exchange_ts, sequence_id, symbol_id, event_type, 0)
142 return header + payload + b"\x00" * (100 - len(payload))
143
144def convert_klines_to_binary(df: pd.DataFrame, output_path: str, symbol_id: int) -> int:
145 events = []
146 seq = 0
147 for row in df.itertuples(index=False):
148 ts_ns = int(pd.to_datetime(row.timestamp).timestamp() * 1_000_000_000)
149 price = int(float(row.close) * 100)
150 qty = int(float(row.volume) * 1000)
151 payload = struct.pack("<qqB", price, qty, 0)
152 event = pack_event(ts_ns, seq, symbol_id, 2, payload)
153 events.append((ts_ns, event))
154 seq += 1
155 events.sort(key=lambda x: x[0])
156 os.makedirs(os.path.dirname(output_path), exist_ok=True)
157 with open(output_path, "wb") as f:
158 f.write(struct.pack("<8sIIHH", b"CREARCHV", 1, len(events), 1, 0))
159 for _, event_bytes in events:
160 f.write(event_bytes)
161 return len(events)
162
163def run_strategy1_backtest(duration: str, interval: str = "1h") -> int:
164 delta = parse_duration(duration)
165 end_time = datetime.now(tz=UTC)
166 start_time = end_time - delta
167 five_years_ago = end_time - timedelta(days=5 * 365)
168
169 # 1. Acquire cache
170 btc_master = download_binance_klines("BTCUSDT", interval, five_years_ago, end_time, "data/study")
171 eth_master = download_binance_klines("ETHUSDT", interval, five_years_ago, end_time, "data/study")
172
173 # 2. Filter data
174 btc_df = pd.read_csv(btc_master)
175 eth_df = pd.read_csv(eth_master)
176 btc_df["dt"] = pd.to_datetime(btc_df["timestamp"], utc=True)
177 eth_df["dt"] = pd.to_datetime(eth_df["timestamp"], utc=True)
178 btc_filtered = btc_df[(btc_df["dt"] >= start_time) & (btc_df["dt"] <= end_time)]
179 eth_filtered = eth_df[(eth_df["dt"] >= start_time) & (eth_df["dt"] <= end_time)]
180
181 # 3. Convert to binary crearchv
182 btc_bin = "data/study/BTCUSDT_strategy1.bin"
183 eth_bin = "data/study/ETHUSDT_strategy1.bin"
184 convert_klines_to_binary(btc_filtered, btc_bin, symbol_id=1)
185 convert_klines_to_binary(eth_filtered, eth_bin, symbol_id=2)
186
187 # 4. Engine setup
188 registry = qe.SymbolRegistry()
189
190 # BTC
191 btc_meta = qe.SymbolMeta()
192 btc_meta.symbol_id = 1
193 btc_meta.symbol_name = "BTCUSDT"
194 btc_meta.price_scale = 100
195 btc_meta.qty_scale = 1000
196 btc_meta.tick_size = 1
197 btc_meta.lot_size = 1
198 btc_meta.maker_fee_bps = 2
199 btc_meta.taker_fee_bps = 5
200 registry.register_symbol(btc_meta)
201
202 # ETH
203 eth_meta = qe.SymbolMeta()
204 eth_meta.symbol_id = 2
205 eth_meta.symbol_name = "ETHUSDT"
206 eth_meta.price_scale = 100
207 eth_meta.qty_scale = 1000
208 eth_meta.tick_size = 1
209 eth_meta.lot_size = 1
210 eth_meta.maker_fee_bps = 2
211 eth_meta.taker_fee_bps = 5
212 registry.register_symbol(eth_meta)
213
214 lat = qe.LatencyConfig()
215 lat.submission_ns = 35_000_000
216 lat.cancellation_ns = 30_000_000
217
218 engine = qe.ReplayEngine(registry, lat, qe.QueueConfig(), qe.RiskLimits(), 100_000_00)
219
220 # Attach the strategy.
221 # Use five price observations, twenty volume observations, and a 0.1 shock threshold.
222 strategy = Strategy1(engine, leader_id=1, responder_id=2, return_window=5, volume_window=20, shock_threshold=0.1)
223 engine.set_strategy(strategy)
224
225 # 6. Stream readers
226 btc_reader = qe.BinaryArchiveReader(btc_bin)
227 eth_reader = qe.BinaryArchiveReader(eth_bin)
228 engine.add_stream(btc_reader)
229 engine.add_stream(eth_reader)
230
231 logger.info("Executing C++ Backtest for Strategy1...")
232 start_perf = time.perf_counter()
233 engine.run()
234 end_perf = time.perf_counter()
235
236 stats = engine.stats()
237 risk = engine.risk()
238
239 print("\n" + "=" * 45)
240 print(" C++ STRATEGY1 BACKTEST SIMULATION ")
241 print("=" * 45)
242 print(f"Target lookback: {duration}")
243 print(f"Time Range: {start_time.strftime('%Y-%m-%d')} to {end_time.strftime('%Y-%m-%d')}")
244 print(f"Engine Replay time: {end_perf - start_perf:.4f}s")
245 print(f"Events Replayed: {stats.events_processed:,}")
246 print(f"Order Fills: {stats.fill_events}")
247 print("Initial Capital: $100,000.00")
248 print(f"Ending Balance: ${risk.account().balance / 100:.2f}")
249 print(f"Realized P&L: ${risk.total_realized_pnl() / 100:.2f}")
250 print(f"Unrealized P&L: ${risk.total_unrealized_pnl() / 100:.2f}")
251 print("=" * 45 + "\n")
252 return 0
253
254if __name__ == "__main__":
255 parser = argparse.ArgumentParser()
256 parser.add_argument("duration", nargs="?", default="3m")
257 args = parser.parse_args()
258 sys.exit(run_strategy1_backtest(args.duration))
259