Adriano
9301ab9051
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>
57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
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]
|