cfdaily
241 lines
7.4 KiB
Python
241 lines
7.4 KiB
Python
"""
|
|
BacktestRunner - 回测引擎
|
|
|
|
胶水层：取数据 → 生成信号 → 模拟交易 → 出报告
|
|
|
|
自带简易回测引擎，不依赖 vnpy，降低使用门槛。
|
|
保留 vnpy 入口供高级功能切换。
|
|
|
|
Usage:
|
|
from data_platform import DataCatalog
|
|
from data_platform.backtest_runner import BacktestRunner
|
|
from data_platform.strategy_base import BaseStrategy
|
|
|
|
class MyStrategy(BaseStrategy):
|
|
def generate_signals(self, data):
|
|
...
|
|
|
|
runner = BacktestRunner(DataCatalog())
|
|
result = runner.run(MyStrategy(), "600519", "20250101", "20251231")
|
|
print(result.summary())
|
|
"""
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional, List, Dict
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
|
|
from data_platform.catalog import DataCatalog
|
|
from data_platform.strategy_base import BaseStrategy
|
|
from data_platform.backtest_report import BacktestReport
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class Trade:
|
|
"""单笔交易记录"""
|
|
entry_date: pd.Timestamp
|
|
exit_date: Optional[pd.Timestamp]
|
|
entry_price: float
|
|
exit_price: Optional[float]
|
|
direction: int # 1=多, -1=空
|
|
shares: int
|
|
profit: Optional[float] = None
|
|
profit_pct: Optional[float] = None
|
|
|
|
|
|
@dataclass
|
|
class BacktestResult:
|
|
"""回测结果"""
|
|
strategy_name: str
|
|
code: str
|
|
start_date: pd.Timestamp
|
|
end_date: pd.Timestamp
|
|
initial_capital: float
|
|
final_capital: float
|
|
total_return: float
|
|
annual_return: float
|
|
max_drawdown: float
|
|
sharpe_ratio: float
|
|
win_rate: float
|
|
total_trades: int
|
|
trades: List[Trade] = field(default_factory=list)
|
|
equity_curve: Optional[pd.Series] = None
|
|
|
|
def summary(self) -> str:
|
|
return (
|
|
f"策略: {self.strategy_name} | 股票: {self.code}\n"
|
|
f"区间: {self.start_date.date()} ~ {self.end_date.date()}\n"
|
|
f"总收益率: {self.total_return:.2%} | 年化: {self.annual_return:.2%}\n"
|
|
f"最大回撤: {self.max_drawdown:.2%} | 夏普: {self.sharpe_ratio:.2f}\n"
|
|
f"胜率: {self.win_rate:.2%} | 交易次数: {self.total_trades}"
|
|
)
|
|
|
|
|
|
class BacktestRunner:
|
|
"""
|
|
回测引擎
|
|
|
|
一条命令完成：获取数据 → 运行策略 → 生成回测结果
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
catalog: DataCatalog,
|
|
commission_rate: float = 0.0003,
|
|
slippage: float = 0.001,
|
|
):
|
|
self.catalog = catalog
|
|
self.commission_rate = commission_rate
|
|
self.slippage = slippage
|
|
|
|
def run(
|
|
self,
|
|
strategy: BaseStrategy,
|
|
code: str,
|
|
start: str,
|
|
end: str,
|
|
initial_capital: float = 1_000_000,
|
|
) -> BacktestResult:
|
|
"""
|
|
运行单只股票回测
|
|
|
|
Args:
|
|
strategy: 策略实例（BaseStrategy 子类）
|
|
code: 股票代码
|
|
start: 起始日期 YYYYMMDD
|
|
end: 结束日期 YYYYMMDD
|
|
initial_capital: 初始资金
|
|
|
|
Returns:
|
|
BacktestResult
|
|
"""
|
|
# 1. 获取数据
|
|
data = self.catalog.get_daily(code, start=start, end=end)
|
|
if len(data) < 20:
|
|
raise ValueError(f"数据不足：{code} 仅 {len(data)} 行")
|
|
|
|
# 2. 生成信号
|
|
signals = strategy.generate_signals(data)
|
|
if "signal" not in signals.columns:
|
|
raise ValueError(f"策略 {strategy.name} 未生成 signal 列")
|
|
|
|
# 3. 模拟交易
|
|
trades, equity = self._simulate(signals, initial_capital)
|
|
|
|
# 4. 计算指标
|
|
return self._build_result(
|
|
strategy.name, code, signals["date"].iloc[0], signals["date"].iloc[-1],
|
|
initial_capital, equity, trades
|
|
)
|
|
|
|
def run_batch(
|
|
self,
|
|
strategy: BaseStrategy,
|
|
codes: List[str],
|
|
start: str,
|
|
end: str,
|
|
initial_capital: float = 1_000_000,
|
|
) -> Dict[str, BacktestResult]:
|
|
"""批量回测多只股票"""
|
|
results = {}
|
|
for code in codes:
|
|
try:
|
|
results[code] = self.run(strategy, code, start, end, initial_capital)
|
|
except Exception as e:
|
|
logger.warning("回测 %s 失败: %s", code, e)
|
|
return results
|
|
|
|
# ------------------------------------------------------------------
|
|
# 内部方法
|
|
# ------------------------------------------------------------------
|
|
|
|
def _simulate(self, data: pd.DataFrame, capital: float):
|
|
"""模拟交易，返回 (trades, equity_series)"""
|
|
position = 0 # 当前持仓股数
|
|
entry_price = 0.0
|
|
entry_date = None
|
|
trades = []
|
|
equity = []
|
|
|
|
for _, row in data.iterrows():
|
|
price = row["close"]
|
|
|
|
if row["signal"] == 1 and position == 0:
|
|
# 买入
|
|
cost = price * (1 + self.slippage)
|
|
commission = capital * 0.99 * self.commission_rate # 用99%资金买入
|
|
shares = int(capital * 0.99 / (cost * (1 + self.commission_rate)))
|
|
shares = shares // 100 * 100 # 整手
|
|
if shares > 0:
|
|
position = shares
|
|
entry_price = cost
|
|
entry_date = row["date"]
|
|
capital -= shares * cost + shares * cost * self.commission_rate
|
|
|
|
elif row["signal"] == -1 and position > 0:
|
|
# 卖出
|
|
sell_price = price * (1 - self.slippage)
|
|
proceeds = position * sell_price * (1 - self.commission_rate)
|
|
capital += proceeds
|
|
trades.append(Trade(
|
|
entry_date=entry_date,
|
|
exit_date=row["date"],
|
|
entry_price=entry_price,
|
|
exit_price=sell_price,
|
|
direction=1,
|
|
shares=position,
|
|
profit=proceeds - position * entry_price,
|
|
profit_pct=(sell_price / entry_price - 1),
|
|
))
|
|
position = 0
|
|
|
|
# 当日权益
|
|
equity.append(capital + position * price)
|
|
|
|
return trades, pd.Series(equity, index=data.index)
|
|
|
|
def _build_result(self, name, code, start, end, capital, equity, trades):
|
|
"""构建回测结果"""
|
|
final = equity.iloc[-1]
|
|
total_return = final / capital - 1
|
|
days = (end - start).days or 1
|
|
annual_return = (1 + total_return) ** (252 / max(days, 1)) - 1
|
|
|
|
# 最大回撤
|
|
cummax = equity.cummax()
|
|
drawdown = (equity - cummax) / cummax
|
|
max_drawdown = drawdown.min()
|
|
|
|
# 夏普比率
|
|
daily_returns = equity.pct_change().dropna()
|
|
sharpe = (
|
|
daily_returns.mean() / daily_returns.std() * np.sqrt(252)
|
|
if len(daily_returns) > 1 and daily_returns.std() > 0
|
|
else 0.0
|
|
)
|
|
|
|
# 胜率
|
|
wins = sum(1 for t in trades if t.profit and t.profit > 0)
|
|
total = len(trades)
|
|
|
|
return BacktestResult(
|
|
strategy_name=name,
|
|
code=code,
|
|
start_date=start,
|
|
end_date=end,
|
|
initial_capital=capital,
|
|
final_capital=final,
|
|
total_return=total_return,
|
|
annual_return=annual_return,
|
|
max_drawdown=max_drawdown,
|
|
sharpe_ratio=sharpe,
|
|
win_rate=wins / total if total > 0 else 0.0,
|
|
total_trades=total,
|
|
trades=trades,
|
|
equity_curve=equity,
|
|
)
|