feat(backtest): event-driven engine with 1-bar exec delay

Engine sincrono bar-per-bar con delay 1: segnale a t-1 esegue a open di t
per evitare lookahead. Position sizing 1 unit, fees su entry+exit,
mark-to-market su close, chiusura forzata posizione open a fine serie.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-09 19:15:05 +02:00
parent 36e05233d0
commit 9301ab9051
2 changed files with 157 additions and 0 deletions
+56
View File
@@ -0,0 +1,56 @@
import numpy as np
import pandas as pd
import pytest
from multi_swarm.backtest.engine import BacktestEngine
from multi_swarm.backtest.orders import Side
@pytest.fixture
def trending_ohlcv() -> pd.DataFrame:
idx = pd.date_range("2024-01-01", periods=100, freq="1h", tz="UTC")
close = np.linspace(100, 120, 100)
df = pd.DataFrame(
{"open": close, "high": close + 0.5, "low": close - 0.5, "close": close, "volume": 1.0},
index=idx,
)
return df
def test_engine_no_signals_zero_pnl(trending_ohlcv: pd.DataFrame) -> None:
signals = pd.Series([Side.FLAT] * len(trending_ohlcv), index=trending_ohlcv.index)
engine = BacktestEngine(fees_bp=5.0)
result = engine.run(trending_ohlcv, signals)
assert result.equity_curve.iloc[-1] == pytest.approx(0.0)
assert len(result.trades) == 0
def test_engine_long_in_uptrend_makes_profit(trending_ohlcv: pd.DataFrame) -> None:
signals = pd.Series([Side.LONG] * len(trending_ohlcv), index=trending_ohlcv.index)
engine = BacktestEngine(fees_bp=5.0)
result = engine.run(trending_ohlcv, signals)
assert result.equity_curve.iloc[-1] > 0
assert len(result.trades) == 1
assert result.trades[0].side == Side.LONG
def test_engine_position_flips_on_side_change(trending_ohlcv: pd.DataFrame) -> None:
half = len(trending_ohlcv) // 2
signals = pd.Series(
[Side.LONG] * half + [Side.SHORT] * (len(trending_ohlcv) - half),
index=trending_ohlcv.index,
)
engine = BacktestEngine(fees_bp=5.0)
result = engine.run(trending_ohlcv, signals)
assert len(result.trades) == 2
assert result.trades[0].side == Side.LONG
assert result.trades[1].side == Side.SHORT
def test_engine_fees_are_subtracted(trending_ohlcv: pd.DataFrame) -> None:
signals = pd.Series([Side.LONG] * len(trending_ohlcv), index=trending_ohlcv.index)
engine_no_fees = BacktestEngine(fees_bp=0.0)
engine_fees = BacktestEngine(fees_bp=10.0)
r1 = engine_no_fees.run(trending_ohlcv, signals)
r2 = engine_fees.run(trending_ohlcv, signals)
assert r1.equity_curve.iloc[-1] > r2.equity_curve.iloc[-1]