feat(backtest): add core data types (Signal/Trade/Position/BacktestResult)

- Add Signal dataclass for trading signals with optional price/stop_loss/take_profit
- Add Trade dataclass for executed trades with commission/slippage/pnl/rejected
- Add Position dataclass for position snapshots (long/short/flat)
- Add BacktestResult dataclass with to_dict()/to_json()/summary() methods
- Add comprehensive unit tests (13 test cases, 100% pass)
- All code passes mypy strict, ruff lint+format checks

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
GitHub
2026-06-09 16:43:44 +08:00
co-authored by Claude Opus 4.8
parent 5f14c44791
commit f37b75ea42
3 changed files with 405 additions and 0 deletions
+1
View File
@@ -0,0 +1 @@
"""easy_tdx.backtest — 向量化策略回测引擎(纯计算,零网络依赖)。"""
+137
View File
@@ -0,0 +1,137 @@
"""回测引擎核心数据类型定义。
使用纯 dataclass + 类型注解,保持 mypy strict 兼容。
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from typing import Any, Literal
import pandas as pd
# ── 交易信号 ────────────────────────────────────────────────────────────────
@dataclass
class Signal:
"""策略产生的交易信号。
Attributes:
datetime: 信号时间(Unix timestamp 毫秒)
direction: 交易方向
size: 交易数量(0 = 全仓/清仓)
price: 限价(None = 市价单)
stop_loss: 止损价(None = 不设置)
take_profit: 止盈价(None = 不设置)
"""
datetime: int
direction: Literal["BUY", "SELL"]
size: float
price: float | None = None
stop_loss: float | None = None
take_profit: float | None = None
# ── 成交记录 ────────────────────────────────────────────────────────────────
@dataclass
class Trade:
"""已成交记录。
Attributes:
datetime: 成交时间(Unix timestamp 毫秒)
direction: 交易方向
size: 成交数量
price: 成交价格
commission: 手续费
slippage: 滑点成本
pnl: 已实现盈亏(仅平仓时计算)
rejected: 是否被拒绝(资金不足/不允许做空等)
"""
datetime: int
direction: Literal["BUY", "SELL"]
size: float
price: float
commission: float
slippage: float
pnl: float = 0.0
rejected: bool = False
# ── 持仓快照 ────────────────────────────────────────────────────────────────
@dataclass
class Position:
"""持仓快照。
Attributes:
datetime: 快照时间(Unix timestamp 毫秒)
size: 持仓数量(正=多头,负=空头,0=空仓)
avg_price: 平均持仓成本
market_value: 市值
unrealized_pnl: 未实现盈亏
"""
datetime: int
size: float
avg_price: float
market_value: float
unrealized_pnl: float
# ── 回测结果 ────────────────────────────────────────────────────────────────
@dataclass
class BacktestResult:
"""回测完整结果。
Attributes:
performance: 绩效指标字典(总收益率、夏普比率、最大回撤等)
equity_curve: 资金曲线 DataFrameindex=datetime, columns=equity/drawdown等)
trades: 成交记录 DataFrame
positions: 持仓快照 DataFrame
config: 配置参数字典
"""
performance: dict[str, float]
equity_curve: pd.DataFrame
trades: pd.DataFrame
positions: pd.DataFrame
config: dict[str, Any]
def to_dict(self) -> dict[str, Any]:
"""将结果转换为 JSON 兼容字典。
DataFrame 转为 records 列表格式。
"""
return {
"performance": self.performance,
"equity_curve": self.equity_curve.to_dict(orient="records"),
"trades": self.trades.to_dict(orient="records"),
"positions": self.positions.to_dict(orient="records"),
"config": self.config,
}
def to_json(self) -> str:
"""将结果序列化为 JSON 字符串。"""
return json.dumps(self.to_dict(), ensure_ascii=False, indent=2)
def summary(self) -> None:
"""打印回测概要(标准输出)。"""
print("=== 回测绩效概要 ===")
for key, value in self.performance.items():
if isinstance(value, float):
print(f"{key}: {value:.4f}")
else:
print(f"{key}: {value}")
print(f"\n成交记录数: {len(self.trades)}")
print(f"持仓快照数: {len(self.positions)}")
print(f"资金曲线点数: {len(self.equity_curve)}")