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>
102 lines
3.6 KiB
Python
102 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import pandas as pd # type: ignore[import-untyped]
|
|
|
|
from .orders import Position, Side, Trade
|
|
|
|
Signal = Side # alias semantico
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BacktestResult:
|
|
equity_curve: pd.Series
|
|
returns: pd.Series
|
|
trades: list[Trade]
|
|
|
|
|
|
class BacktestEngine:
|
|
"""Engine event-driven sincrono: itera bar per bar, applica segnali con
|
|
delay di 1 bar (segnale a t -> eseguito a t+1 open) per evitare lookahead.
|
|
|
|
Position sizing: 1 unit per posizione. Fees applicati su entry+exit.
|
|
Niente leva, niente liquidation, niente funding (semplificazione Phase 1).
|
|
"""
|
|
|
|
def __init__(self, fees_bp: float = 5.0) -> None:
|
|
self.fees_bp = fees_bp
|
|
|
|
def run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult:
|
|
signals = signals.reindex(ohlcv.index).ffill().fillna(Side.FLAT)
|
|
|
|
# Esecuzione con delay 1: segnale a t-1 esegue a open di t.
|
|
shifted = [Side.FLAT, *list(signals.iloc[:-1])]
|
|
executed_side = pd.Series(shifted, index=ohlcv.index, dtype=object)
|
|
|
|
position: Position | None = None
|
|
position_entry_ts: pd.Timestamp | None = None
|
|
trades: list[Trade] = []
|
|
equity = 0.0
|
|
equity_history: list[float] = []
|
|
returns_history: list[float] = []
|
|
prev_equity = 0.0
|
|
|
|
for ts, row in ohlcv.iterrows():
|
|
target_side = executed_side.loc[ts]
|
|
current_side = position.side if position else Side.FLAT
|
|
|
|
if target_side != current_side:
|
|
if position is not None:
|
|
assert position_entry_ts is not None
|
|
trade = Trade(
|
|
entry_ts=position_entry_ts,
|
|
exit_ts=ts,
|
|
side=position.side,
|
|
size=position.size,
|
|
entry_price=position.entry_price,
|
|
exit_price=float(row["open"]),
|
|
fees_bp=self.fees_bp,
|
|
)
|
|
trades.append(trade)
|
|
equity += trade.net_pnl
|
|
position = None
|
|
position_entry_ts = None
|
|
if target_side in (Side.LONG, Side.SHORT):
|
|
position = Position(
|
|
side=target_side, entry_price=float(row["open"]), size=1.0
|
|
)
|
|
position_entry_ts = ts
|
|
|
|
mark = float(row["close"])
|
|
mtm = position.unrealized_pnl(mark) if position else 0.0
|
|
current_equity = equity + mtm
|
|
equity_history.append(current_equity)
|
|
returns_history.append(current_equity - prev_equity)
|
|
prev_equity = current_equity
|
|
|
|
if position is not None:
|
|
assert position_entry_ts is not None
|
|
last_ts = ohlcv.index[-1]
|
|
last_close = float(ohlcv["close"].iloc[-1])
|
|
trade = Trade(
|
|
entry_ts=position_entry_ts,
|
|
exit_ts=last_ts,
|
|
side=position.side,
|
|
size=position.size,
|
|
entry_price=position.entry_price,
|
|
exit_price=last_close,
|
|
fees_bp=self.fees_bp,
|
|
)
|
|
trades.append(trade)
|
|
equity += trade.net_pnl
|
|
equity_history[-1] = equity
|
|
if len(returns_history) >= 2:
|
|
returns_history[-1] = equity - equity_history[-2]
|
|
|
|
return BacktestResult(
|
|
equity_curve=pd.Series(equity_history, index=ohlcv.index, name="equity"),
|
|
returns=pd.Series(returns_history, index=ohlcv.index, name="returns"),
|
|
trades=trades,
|
|
)
|