tickreplay 0.1.0: tick-accurate MT5 backtesting engine + research toolkit
Memory-mapped tickstores, a nogil numba replay kernel validated trade-for-trade against an independent reference implementation, purged walk-forward CV, an append-only FDR registry, tick-exact labels, and online changepoint detection. 129 tests, 4 runnable examples on synthetic data, CI for linux/macos.
This commit is contained in:
danielmjrr
@@ -0,0 +1,81 @@
|
||||
"""Replay throughput on synthetic ticks.
|
||||
|
||||
Run: python examples/benchmark.py [--ticks 20000000]
|
||||
"""
|
||||
import argparse
|
||||
import statistics
|
||||
import time
|
||||
|
||||
import numpy as np
|
||||
|
||||
import tickreplay as tr
|
||||
|
||||
TICKS_PER_DAY = 200_000
|
||||
WARMUP_TICKS = 100_000
|
||||
|
||||
|
||||
def build_store(n_ticks, seed):
|
||||
# generate_ticks skips weekends, so ask for enough calendar days to cover it.
|
||||
days = int(np.ceil(n_ticks / TICKS_PER_DAY * 7 / 5)) + 3
|
||||
time_ns, bid, ask = tr.synth.generate_ticks(
|
||||
days=days, ticks_per_day=TICKS_PER_DAY, seed=seed)
|
||||
return tr.TickStore(time_ns[:n_ticks], bid[:n_ticks], ask[:n_ticks])
|
||||
|
||||
|
||||
def ma_cross_signal(bars):
|
||||
fast = bars["close"].rolling(10).mean()
|
||||
slow = bars["close"].rolling(50).mean()
|
||||
regime = np.sign(fast - slow).fillna(0.0).to_numpy()
|
||||
return np.where(regime != np.r_[0.0, regime[:-1]], regime, 0).astype("int8")
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Replay throughput on synthetic ticks.")
|
||||
parser.add_argument("--ticks", type=int, default=20_000_000,
|
||||
help="store size; 20M ticks needs about 1 GB of RAM")
|
||||
parser.add_argument("--timeframe", default="M5", choices=sorted(tr.TIMEFRAMES),
|
||||
help="bar timeframe the signal is built on")
|
||||
parser.add_argument("--seed", type=int, default=1, help="synthetic data seed")
|
||||
args = parser.parse_args()
|
||||
|
||||
t0 = time.perf_counter()
|
||||
store = build_store(args.ticks, args.seed)
|
||||
gen_ms = (time.perf_counter() - t0) * 1e3
|
||||
|
||||
config = tr.Config(sl=3.0, tp=6.0, max_hold="4h", cost=0.05)
|
||||
|
||||
# Compile the kernel on a small slice so JIT time stays out of the measurement.
|
||||
warm = tr.TickStore(store.time_ns[:WARMUP_TICKS], store.bid[:WARMUP_TICKS],
|
||||
store.ask[:WARMUP_TICKS])
|
||||
warm_bars = warm.bars(args.timeframe)
|
||||
tr.replay(warm, warm_bars, ma_cross_signal(warm_bars), config)
|
||||
|
||||
t0 = time.perf_counter()
|
||||
bars = store.bars(args.timeframe)
|
||||
bars_ms = (time.perf_counter() - t0) * 1e3
|
||||
signal = ma_cross_signal(bars)
|
||||
|
||||
timings = []
|
||||
for _ in range(3):
|
||||
t0 = time.perf_counter()
|
||||
result = tr.replay(store, bars, signal, config)
|
||||
timings.append((time.perf_counter() - t0) * 1e3)
|
||||
median_ms = statistics.median(timings)
|
||||
|
||||
# The kernel only walks ticks while a position is open, so quote the covered
|
||||
# fraction alongside the headline rate.
|
||||
visited = int((result.arrays["exit_idx"] - result.arrays["entry_idx"]).sum())
|
||||
|
||||
print(f"ticks {len(store):>14,}")
|
||||
print(f"{args.timeframe} bars{'':<10}{len(bars):>14,}")
|
||||
print(f"trades {len(result):>14,}")
|
||||
print(f"ticks in position{visited:>14,} ({visited / max(len(store), 1):.1%})")
|
||||
print(f"generate {gen_ms:>14,.1f} ms")
|
||||
print(f"derive bars {bars_ms:>14,.1f} ms")
|
||||
print(f"replay (median) {median_ms:>14,.1f} ms of {[round(x, 1) for x in timings]}")
|
||||
print(f"kernel rate {visited / median_ms / 1e3:>14,.1f} M ticks/s scanned")
|
||||
print(f"end-to-end rate {len(store) / median_ms / 1e3:>14,.1f} M store ticks/s")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,46 @@
|
||||
"""N-bar breakout on M5 bars, exiting on the opposite breakout.
|
||||
|
||||
Run: python examples/breakout.py
|
||||
"""
|
||||
import numpy as np
|
||||
|
||||
import tickreplay as tr
|
||||
|
||||
LOOKBACK = 20
|
||||
|
||||
|
||||
def main():
|
||||
time_ns, bid, ask = tr.synth.generate_ticks(days=30, seed=11)
|
||||
store = tr.TickStore(time_ns, bid, ask)
|
||||
bars = store.bars("M5")
|
||||
|
||||
# shift(1) keeps the current bar out of its own reference range.
|
||||
prior_high = bars["high"].shift(1).rolling(LOOKBACK).max()
|
||||
prior_low = bars["low"].shift(1).rolling(LOOKBACK).min()
|
||||
state = np.where(bars["close"] > prior_high, 1,
|
||||
np.where(bars["close"] < prior_low, -1, 0)).astype("int8")
|
||||
|
||||
# Fire on the bar that breaks out, not on every bar that stays beyond the range.
|
||||
signal = np.where(state != np.r_[np.int8(0), state[:-1]], state, 0).astype("int8")
|
||||
|
||||
config = tr.Config(sl=4.0, tp=8.0, max_hold="8h", exit_on_opposite=True, cost=0.05)
|
||||
result = tr.replay(store, bars, signal, config)
|
||||
|
||||
print(f"ticks {len(store):>12,}")
|
||||
print(f"M5 bars {len(bars):>12,}")
|
||||
print(f"signals {int((signal != 0).sum()):>12,} ({LOOKBACK}-bar breakout)")
|
||||
print("config sl=4.0 tp=8.0 max_hold=8h exit_on_opposite=True cost=0.05")
|
||||
print()
|
||||
for key, value in result.summary().items():
|
||||
print(f"{key:<14}{value:>12.4f}" if isinstance(value, float)
|
||||
else f"{key:<14}{value:>12,}")
|
||||
|
||||
print()
|
||||
print("exit reasons")
|
||||
counts = result.trades["exit_reason"].value_counts()
|
||||
for reason, count in counts.items():
|
||||
print(f" {reason:<10}{count:>6,}{count / counts.sum():>8.1%}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,41 @@
|
||||
"""Moving-average cross on M5 bars, replayed tick by tick.
|
||||
|
||||
Run: python examples/ma_cross.py
|
||||
"""
|
||||
import numpy as np
|
||||
|
||||
import tickreplay as tr
|
||||
|
||||
|
||||
def main():
|
||||
time_ns, bid, ask = tr.synth.generate_ticks(days=30, seed=7)
|
||||
store = tr.TickStore(time_ns, bid, ask)
|
||||
bars = store.bars("M5")
|
||||
|
||||
fast = bars["close"].rolling(10).mean()
|
||||
slow = bars["close"].rolling(50).mean()
|
||||
|
||||
# Regime is +1 while fast is above slow. The signal fires only on the bar where
|
||||
# the regime flips, so one cross is one entry instead of one entry per bar.
|
||||
regime = np.sign(fast - slow).fillna(0.0).to_numpy()
|
||||
previous = np.r_[0.0, regime[:-1]]
|
||||
signal = np.where(regime != previous, regime, 0).astype("int8")
|
||||
|
||||
# signal[b] is built from closes up to and including bar b, so it is knowable
|
||||
# only once bar b has closed. No extra shift is needed: the engine enters at the
|
||||
# first tick after the signal bar closes.
|
||||
config = tr.Config(sl=3.0, tp=6.0, max_hold="4h", cost=0.05)
|
||||
result = tr.replay(store, bars, signal, config)
|
||||
|
||||
print(f"ticks {len(store):>12,}")
|
||||
print(f"M5 bars {len(bars):>12,}")
|
||||
print(f"signals {int((signal != 0).sum()):>12,}")
|
||||
print("config sl=3.0 tp=6.0 max_hold=4h cost=0.05 (price units)")
|
||||
print()
|
||||
for key, value in result.summary().items():
|
||||
print(f"{key:<14}{value:>12.4f}" if isinstance(value, float)
|
||||
else f"{key:<14}{value:>12,}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,59 @@
|
||||
"""Sweep exit rules, then test the winner honestly.
|
||||
|
||||
Picking the best cell of a grid and reporting its p-value as-is is the standard way
|
||||
to manufacture an edge. This example sweeps, then thins overlapping trades before
|
||||
testing, and reads the result against a selection-adjusted threshold.
|
||||
|
||||
Run: python examples/sweep_and_validate.py
|
||||
"""
|
||||
import numpy as np
|
||||
|
||||
import tickreplay as tr
|
||||
|
||||
HOUR_NS = 3_600 * 10**9
|
||||
|
||||
|
||||
def main():
|
||||
time_ns, bid, ask = tr.synth.generate_ticks(days=60, seed=3)
|
||||
store = tr.TickStore(time_ns, bid, ask)
|
||||
bars = store.bars("M5")
|
||||
|
||||
fast = bars["close"].rolling(10).mean()
|
||||
slow = bars["close"].rolling(50).mean()
|
||||
regime = np.sign(fast - slow).fillna(0.0).to_numpy()
|
||||
signal = np.where(regime != np.r_[0.0, regime[:-1]], regime, 0).astype("int8")
|
||||
|
||||
base = {"max_hold": "4h", "cost": 0.05}
|
||||
grid = [{"sl": sl, "tp": tp} for sl in (2.0, 3.0, 4.0) for tp in (4.0, 6.0, 8.0)]
|
||||
table = tr.sweep_exits(store, bars, signal, grid, base=base)
|
||||
table = table.sort_values("total_net", ascending=False).reset_index(drop=True)
|
||||
|
||||
print(f"{len(grid)} exit configs over {len(store):,} ticks\n")
|
||||
print(table[["sl", "tp", "n_trades", "total_net", "hit_rate",
|
||||
"max_drawdown"]].to_string(index=False, float_format="%.3f"))
|
||||
|
||||
best = table.iloc[0]
|
||||
result = tr.replay(store, bars, signal,
|
||||
tr.Config(sl=best["sl"], tp=best["tp"], **base))
|
||||
pnl = result.arrays["pnl_net"]
|
||||
|
||||
# Trades whose holding windows overlap are correlated; a t-stat over all of them
|
||||
# is inflated. Test a non-overlapping subset spaced by the max hold time.
|
||||
keep = tr.research.stats.thin_by_gap(result.arrays["entry_time_ns"], 4 * HOUR_NS)
|
||||
t_stat, p_value = tr.research.stats.one_sided_positive(pnl[keep])
|
||||
threshold = 0.05 / len(grid)
|
||||
|
||||
print(f"\nbest by total_net: sl={best['sl']} tp={best['tp']}")
|
||||
print(f" trades {len(pnl):>8,} -> {int(keep.sum()):,} after thinning")
|
||||
print(f" one-sided t {t_stat:>8.3f}")
|
||||
print(f" p (mean > 0) {p_value:>8.4f}")
|
||||
print(f" threshold (0.05/{len(grid)}) {threshold:>8.4f}")
|
||||
print(f"\n verdict: {'NOT significant' if p_value > threshold else 'passes'}"
|
||||
" once selection over the grid is accounted for.")
|
||||
print(" Expected: this is a synthetic random walk with costs, so there is no edge"
|
||||
"\n to find. A pass here would mean the procedure is broken, not that the"
|
||||
"\n strategy works.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user