115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
"""单一回测引擎（fast/strict 两档，同一代码路径、同一撮合逻辑）。
|
||
|
||
不做"快层 vectorbt + 真层自研"双引擎——用户必然信任更快的那层，
|
||
一旦快层省略 T+1/费用，两套结论背离即反复扯皮"以谁为准"。
|
||
这里用开关控制规则是否注入，引擎只有一份。
|
||
|
||
数据流：
|
||
bars -> DataFrame -> strategy.compute(指标) -> 逐 bar 喂 strategy.on_bar(broker)
|
||
-> broker 产生 fills -> 每根 bar 记录 equity -> metrics
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
|
||
import pandas as pd
|
||
|
||
from ..domain import Bar
|
||
from .broker import PaperBroker
|
||
from .metrics import compute_metrics
|
||
|
||
|
||
@dataclass
|
||
class BacktestConfig:
|
||
initial_cash: float = 1000000.0
|
||
fast_mode: bool = False # True: 关 T+1/费用，交互试探
|
||
bars_per_year: int = 252 # 日线 252；分钟级另算
|
||
|
||
|
||
def run_backtest(bars: list[Bar], strategy, cfg: BacktestConfig | None = None) -> dict:
|
||
cfg = cfg or BacktestConfig()
|
||
if not bars:
|
||
return _empty_result()
|
||
|
||
df = pd.DataFrame(
|
||
[{"ts": b.ts, "open": b.open, "high": b.high, "low": b.low,
|
||
"close": b.close, "volume": b.volume} for b in bars]
|
||
).sort_values("ts").reset_index(drop=True)
|
||
|
||
# 指标预计算（由策略持有，单一事实源）
|
||
ind = strategy.compute(df["close"], df["high"], df["low"])
|
||
indicator_cols = list(ind.columns)
|
||
df = pd.concat([df, ind], axis=1)
|
||
|
||
broker = PaperBroker(
|
||
initial_cash=cfg.initial_cash,
|
||
enable_costs=not cfg.fast_mode,
|
||
enable_t_plus_1=not cfg.fast_mode,
|
||
)
|
||
|
||
equity_values = []
|
||
for i in range(len(df)):
|
||
row = df.iloc[i]
|
||
strategy.on_bar(i, row, broker)
|
||
equity_values.append(broker.equity(row["close"]))
|
||
broker.release_t_plus_1()
|
||
|
||
equity = pd.Series(equity_values, index=df["ts"], name="equity")
|
||
metrics = compute_metrics(equity, cfg.bars_per_year)
|
||
metrics["num_trades"] = len(broker.fills)
|
||
metrics["win_rate"] = _win_rate(broker.fills)
|
||
|
||
return {
|
||
"df": df,
|
||
"indicator_cols": indicator_cols,
|
||
"fills": broker.fills,
|
||
"equity": equity,
|
||
"metrics": metrics,
|
||
"final_cash": broker.cash,
|
||
"final_position": broker.position,
|
||
}
|
||
|
||
|
||
def _win_rate(fills) -> float:
|
||
"""FIFO 配对计算卖出胜率。"""
|
||
from collections import deque
|
||
buys: deque = deque()
|
||
wins = 0
|
||
total_sells = 0
|
||
for f in fills:
|
||
if f.side.value == "buy":
|
||
buys.append((f.price, f.qty))
|
||
else: # sell
|
||
remaining = f.qty
|
||
total_sells += 1
|
||
profitable = True
|
||
while remaining > 0 and buys:
|
||
buy_price, buy_qty = buys[0]
|
||
if f.price >= buy_price:
|
||
pass
|
||
else:
|
||
profitable = False
|
||
take = min(remaining, buy_qty)
|
||
buy_qty -= take
|
||
remaining -= take
|
||
if buy_qty <= 0:
|
||
buys.popleft()
|
||
else:
|
||
buys[0] = (buy_price, buy_qty)
|
||
if profitable and f.qty > 0:
|
||
wins += 1
|
||
return wins / total_sells if total_sells else 0.0
|
||
|
||
|
||
def _empty_result() -> dict:
|
||
return {
|
||
"df": pd.DataFrame(),
|
||
"indicator_cols": [],
|
||
"fills": [],
|
||
"equity": pd.Series(dtype=float),
|
||
"metrics": {"total_return": 0.0, "max_drawdown": 0.0, "sharpe": 0.0,
|
||
"volatility": 0.0, "num_trades": 0, "win_rate": 0.0},
|
||
"final_cash": 0.0,
|
||
"final_position": 0.0,
|
||
}
|