diff --git a/README.md b/README.md index b53c885..312eb82 100644 --- a/README.md +++ b/README.md @@ -534,6 +534,41 @@ class MyStrategy(Strategy): 完整 API 参考：[docs/backtest_usage.md](docs/backtest_usage.md) +### 量化因子与组合管理 + +新增三大模块：**因子引擎**（19 个内置因子 + 自定义扩展）、**因子分析**（IC/分层/衰减）、**组合管理**（4 种优化器 + 再平衡引擎）。加上**高级回测增强**：可插拔滑点模型（方根冲击/成交量比例）、执行仿真（TWAP/VWAP/限价单）、归因分析（Brinson + 因子归因）。 + +```python +from easy_tdx.factor import FactorEngine, FactorAnalyzer, preprocess +from easy_tdx.portfolio import RebalanceEngine, FactorWeightedOptimizer +from easy_tdx.backtest import BacktestEngine +from easy_tdx.backtest.slippage import SquareRootSlippage +from easy_tdx.backtest.execution import TWAPExecution + +# 因子研究 +engine = FactorEngine() +factor_data = engine.compute_cross_section(data, ["momentum_20d", "rsi_14"]) +clean = preprocess(factor_data, ["momentum_20d", "rsi_14"]) +forward_returns = engine.compute_forward_returns(data, period=5) +report = FactorAnalyzer(clean, forward_returns).full_report("momentum_20d") +print(f"IC均值={report.mean_ic:.4f} ICIR={report.icir:.4f}") + +# 组合回测 +result = RebalanceEngine( + FactorWeightedOptimizer(), factor_name="momentum_20d", n_stocks=50, cash=1_000_000, +).run(data, start_date=20230101, end_date=20240101) +print(f"年化={result.performance['annual_return']:.2%}") + +# 高级回测（滑点 + 执行仿真） +engine = BacktestEngine( + MyStrategy, cash=1_000_000, + slippage_model=SquareRootSlippage(impact_coeff=0.1), + execution_model=TWAPExecution(n_bars=3), +) +``` + +详细用法和完整工作流示例：**[docs/quantitative-guide.md](docs/quantitative-guide.md)** + ### 策略选股扫描（screen） 把策略翻转成选股器：给定一个策略，扫描全市场找出今天触发买入信号的股票，再对这些信号做历史回测排名。**纯离线数据**，读取本地通达信 `.day` 文件，全市场约 30-60 秒。 @@ -910,6 +945,9 @@ uvicorn.run(app, host="0.0.0.0", port=8000) | `indicator-list` | 列出可用技术指标 | | `backtest` | 回测引擎（加载策略文件，输出绩效报告） | | `portfolio` | 多标的组合回测（共享资金池，均等分配，汇总绩效） | +| `factor list` | 列出所有内置因子 | +| `factor analyze` | 因子分析（IC/分层/衰减） | +| `pfactor backtest` | 组合因子选股回测 | | `run-all` | 批量运行所有策略并排名（绩效排名 + 综合评分 + 可选图表） | | `screen scan` | 策略选股扫描（纯离线，全市场信号扫描） | | `screen rank` | 扫描结果回测排名（按夏普/回撤等指标排序） | @@ -1439,7 +1477,9 @@ src/easy_tdx/ ├── commands/ # 标准协议命令（无 IO） ├── codec/ # price / volume / datetime / frame / bitmap 编解码 ├── chanlun/ # 缠论技术分析（K线合并/分型/笔/线段/中枢/买卖点/背驰） -├── backtest/ # 回测引擎（Strategy基类/向量化引擎/多因子组合/组合回测/绩效分析） +├── factor/ # 因子引擎（Factor ABC/19内置因子/截面计算/因子分析/预处理管道） +├── portfolio/ # 组合管理（4优化器/风险模型/再平衡引擎） +├── backtest/ # 回测引擎（Strategy基类/向量化引擎/多因子组合/滑点模型/执行仿真/归因分析） ├── screen/ # 策略选股扫描（scan信号扫描/rank回测排名/并发扫描/增量缓存） ├── realtime/ # 实时数据推送框架（EventBus/事件驱动/asyncio） ├── web/ # Web API（FastAPI REST + WebSocket） @@ -1471,6 +1511,34 @@ ruff format --check src/ tests/ # format check ## Changelog +### 1.11.1 (2026-06-12) + +**量化因子引擎 + 组合管理 + 高级回测增强** — 三大新模块，补齐从因子研究到组合执行的完整量化链路。 + +**因子引擎（factor/）**： +- `Factor` ABC + 注册表模式（`@register_factor` 装饰器），19 个内置因子 +- `FactorEngine`：单股多因子 / 截面批量 / 远期收益计算 +- 因子类别：动量、波动率、质量、成交量、技术（桥接 MyTT）、缠论（桥接 ChanlunAnalyser）、价值（占位） +- 因子预处理管道：去极值（MAD）、标准化、排名归一化、填充缺失、正交化 +- `FactorAnalyzer`：IC（Spearman）、分层收益（5 组）、换手率、衰减分析、完整报告 + +**组合管理（portfolio/）**： +- 4 种权重优化器：等权、因子加权、风险平价（逆波动率）、均值方差（scipy 可选） +- 风险模型：Ledoit-Wolf 收缩协方差、组合风险分解 +- `RebalanceEngine`：多期调仓回测（周/月/季），100 股整手、佣金+印花税 + +**高级回测增强（backtest/）**： +- 4 种滑点模型：Fixed、Percent、SquareRoot（Almgren-Chriss）、Volume +- 4 种执行仿真：Immediate、TWAP、VWAP、Limit（限价单 + TTL） +- `AttributionAnalyzer`：成本归因、Brinson 归因（配置/选股/交叉）、因子归因 +- 完全向后兼容（`BacktestEngine` 新增 `slippage_model` / `execution_model` 可选参数） + +**CLI**： +- `easy-tdx factor list` / `factor analyze` — 因子列表和分析 +- `easy-tdx pfactor backtest` — 组合因子选股回测 + +**测试**：556 passed, 0 failed（+176 新增） + ### 1.10.5 (2026-06-12) **Web API 全面补齐 + 稳定性修复** — 新增 18 个 REST 端点，Web API 与 CLI 接口覆盖对齐，修复多个生产环境问题。 diff --git a/docs/quantitative-guide.md b/docs/quantitative-guide.md new file mode 100644 index 0000000..ad54f3e --- /dev/null +++ b/docs/quantitative-guide.md @@ -0,0 +1,628 @@ +# 量化因子与组合管理 — 使用指南 + +> 本文档覆盖 easy-tdx v1.11.1 新增的量化计算能力：因子研究、因子分析、组合管理、高级回测（滑点建模/执行仿真/归因分析）。 + +--- + +## 目录 + +- [1. 因子引擎](#1-因子引擎) + - [1.1 内置因子一览](#11-内置因子一览) + - [1.2 单股多因子计算](#12-单股多因子计算) + - [1.3 截面因子计算](#13-截面因子计算) + - [1.4 远期收益计算](#14-远期收益计算) + - [1.5 自定义因子](#15-自定义因子) +- [2. 因子预处理](#2-因子预处理) +- [3. 因子分析](#3-因子分析) +- [4. 组合管理](#4-组合管理) + - [4.1 权重优化器](#41-权重优化器) + - [4.2 风险模型](#42-风险模型) + - [4.3 再平衡引擎](#43-再平衡引擎) +- [5. 高级回测](#5-高级回测) + - [5.1 滑点模型](#51-滑点模型) + - [5.2 执行仿真](#52-执行仿真) + - [5.3 归因分析](#53-归因分析) +- [6. CLI 命令](#6-cli-命令) +- [7. 完整工作流示例](#7-完整工作流示例) + +--- + +## 1. 因子引擎 + +因子引擎（`FactorEngine`）支持单股和多股截面两种计算模式，内置 19 个因子。 + +### 1.1 内置因子一览 + +| 类别 | 因子名 | 说明 | +|------|--------|------| +| **动量** | `momentum_20d` | 20 日收益率 | +| | `momentum_60d` | 60 日收益率 | +| | `reversal_5d` | 5 日反转（负收益） | +| **波动率** | `volatility_20d` | 20 日年化波动率 | +| | `atr_14d` | 14 日平均真实波幅 | +| | `turnover_rate` | 换手率（需 vol 列） | +| **质量** | `sharpe_20d` | 20 日夏普比率 | +| | `max_drawdown_20d` | 20 日最大回撤 | +| | `win_rate_20d` | 20 日上涨天数占比 | +| **成交量** | `obv_trend` | OBV 趋势斜率 | +| | `vol_surge` | 成交量突增倍数 | +| | `amount_ma_ratio` | 成交额 / MA5 比值 | +| **技术** | `macd_hist_signal` | MACD 柱状信号 | +| | `rsi_14` | 14 日 RSI | +| | `boll_position` | 布林带位置（0~1） | +| **缠论** | `chanlun_bi_dir` | 当前笔方向（+1/-1） | +| | `chanlun_mmd` | 最近买卖点（+2/+1/-1/-2） | +| **价值** | `pe_ratio` | 市盈率（占位，返回 NaN） | +| | `pb_ratio` | 市净率（占位，返回 NaN） | + +### 1.2 单股多因子计算 + +```python +from easy_tdx import TdxClient +from easy_tdx.factor import FactorEngine + +client = TdxClient() +df = client.get_security_bars(Market.SH, "600519", KlineCategory.DAY, 0, 300) + +engine = FactorEngine() + +# 计算多个因子 +result = engine.compute_single(df, ["momentum_20d", "volatility_20d", "rsi_14"]) +print(result.tail()) + +# 计算所有内置因子 +result = engine.compute_single(df) # 不传因子名 = 全部 +print(result.columns.tolist()) +``` + +输出 DataFrame 在原始列基础上追加因子列（以因子名命名，前缀 `NaN` 行因窗口不足为 `NaN`）。 + +### 1.3 截面因子计算 + +```python +from easy_tdx import TdxClient +from easy_tdx.factor import FactorEngine + +client = TdxClient() + +# 准备多只股票数据 +stock_pool = ["000001", "000858", "600519", "600036", "601318"] +data = {} +for code in stock_pool: + market = Market.SH if code.startswith("6") else Market.SZ + data[code] = client.get_security_bars(market, code, KlineCategory.DAY, 0, 300) + +engine = FactorEngine() + +# 截面计算：返回 long format（date, code, factor_name...） +factor_data = engine.compute_cross_section( + data, + ["momentum_20d", "volatility_20d", "rsi_14"], +) +print(factor_data.head(10)) + +# 指定日期：只计算某一天的截面 +factor_data = engine.compute_cross_section( + data, ["momentum_20d"], date=20240601, +) +``` + +### 1.4 远期收益计算 + +```python +# 计算未来 5 日收益率（用于因子分析） +forward_returns = engine.compute_forward_returns(data, period=5) +print(forward_returns.head()) +``` + +### 1.5 自定义因子 + +继承 `Factor` 基类，用 `@register_factor` 注册即可自动发现： + +```python +from easy_tdx.factor import Factor, register_factor + +@register_factor +class MyMomentum(Factor): + name = "my_momentum" + description = "自定义动量因子" + window = 20 + + def compute(self, df): + return df["close"].pct_change(self.window) +``` + +注册后直接用名字引用： + +```python +result = engine.compute_single(df, ["my_momentum"]) +``` + +--- + +## 2. 因子预处理 + +6 个纯函数，组合成管道： + +```python +from easy_tdx.factor import preprocess + +# 单因子预处理管道 +clean = preprocess( + factor_data, + factor_names=["momentum_20d"], + steps=["winsorize", "zscore", "fill_missing"], +) +``` + +| 函数 | 说明 | +|------|------| +| `winsorize(df, factor_names, n_sigma=3)` | MAD 去极值 | +| `zscore(df, factor_names)` | 截面标准化 | +| `rank_normalize(df, factor_names)` | 排名归一化 | +| `fill_missing(df, factor_names)` | 填充缺失值 | +| `orthogonalize(df, factor_names, by="market_cap")` | 正交化（去除市值暴露） | +| `preprocess(df, factor_names, steps)` | 组合管道 | + +所有函数自动检测截面数据（有 `date` 列时按日期分组处理）。 + +--- + +## 3. 因子分析 + +```python +from easy_tdx.factor import FactorEngine, FactorAnalyzer, preprocess + +# 1. 计算截面因子 +factor_data = engine.compute_cross_section(data, ["momentum_20d", "rsi_14"]) + +# 2. 预处理 +clean = preprocess(factor_data, ["momentum_20d", "rsi_14"]) + +# 3. 计算远期收益 +forward_returns = engine.compute_forward_returns(data, period=5) + +# 4. 分析 +analyzer = FactorAnalyzer(clean, forward_returns) + +# IC 分析（Spearman 秩相关） +ic_series = analyzer.compute_ic("momentum_20d") +print(f"均值 IC: {ic_series.mean():.4f}, ICIR: {ic_series.mean()/ic_series.std():.4f}") + +# 分层收益（5 组） +quantile_returns = analyzer.compute_quantile_returns("momentum_20d", n_groups=5) +print(quantile_returns.head()) + +# 因子衰减（IC 自相关） +decay = analyzer.compute_decay("momentum_20d", max_lag=10) +print(decay) + +# 完整报告 +report = analyzer.full_report("momentum_20d") +print(f"IC均值={report.mean_ic:.4f} ICIR={report.icir:.4f}") +print(f"多头年化={report.long_only_annual:.2%} 空头年化={report.short_only_annual:.2%}") +print(f"多空夏普={report.long_short_sharpe:.4f} 换手率={report.turnover:.4f}") +``` + +--- + +## 4. 组合管理 + +### 4.1 权重优化器 + +4 种内置优化器： + +```python +from easy_tdx.portfolio import ( + EqualWeightOptimizer, + FactorWeightedOptimizer, + RiskParityOptimizer, + MeanVarianceOptimizer, +) +import pandas as pd + +# 因子分数表（来自 FactorEngine） +scores_df = pd.DataFrame({ + "code": ["000001", "600519", "601318", "000858", "600036"], + "score": [0.8, 0.6, 0.5, 0.3, 0.1], +}) + +# 1. 等权：选前 N 只，等权分配 +opt1 = EqualWeightOptimizer() +weights1 = opt1.optimize(scores_df, n_stocks=3) +# {'000001': 0.333, '600519': 0.333, '601318': 0.333} + +# 2. 因子加权：分数越高权重越大 +opt2 = FactorWeightedOptimizer() +weights2 = opt2.optimize(scores_df, n_stocks=3) +# {'000001': 0.42, '600519': 0.32, '601318': 0.26} + +# 3. 风险平价：按波动率倒数加权 +returns_df = pd.DataFrame(...) # 收益率矩阵 +opt3 = RiskParityOptimizer(returns_df) +weights3 = opt3.optimize(scores_df, n_stocks=3) + +# 4. 均值方差：scipy SLSQP 优化（无 scipy 退化为等权） +opt4 = MeanVarianceOptimizer(returns_df) +weights4 = opt4.optimize(scores_df, n_stocks=3) +``` + +### 4.2 风险模型 + +```python +from easy_tdx.portfolio import RiskModel +import pandas as pd + +risk = RiskModel() + +# 估计协方差矩阵（Ledoit-Wolf 收缩） +returns = pd.DataFrame(...) # N 只股票 × T 天收益率 +cov = risk.estimate_covariance(returns, method="shrinkage", window=60) + +# 组合风险分解 +weights = {"000001": 0.3, "600519": 0.4, "601318": 0.3} +metrics = risk.portfolio_risk(weights, cov) +print(f"年化波动率: {metrics['total_volatility']:.2%}") +print(f"最大风险贡献: {metrics['max_risk_contribution']:.2%}") +print(f"持仓数: {metrics['n_positions']}") +``` + +### 4.3 再平衡引擎 + +```python +from easy_tdx.portfolio import RebalanceEngine, FactorWeightedOptimizer +from easy_tdx import TdxClient + +client = TdxClient() +stock_pool = ["000001", "000858", "600519", "600036", "601318"] +data = {c: client.get_security_bars(..., c, ...) for c in stock_pool} + +# 创建引擎 +engine = RebalanceEngine( + optimizer=FactorWeightedOptimizer(), + factor_name="momentum_20d", # 用哪个因子选股 + n_stocks=3, # 持仓数量 + rebalance_freq="M", # 调仓频率: W/M/Q + commission=0.0003, # 佣金率 + slippage=0.001, # 滑点率 + cash=1_000_000, # 初始资金 +) + +# 运行回测 +result = engine.run(data, start_date=20230101, end_date=20240101) + +# 结果 +print(f"总收益: {result.performance['total_return']:.2%}") +print(f"年化: {result.performance['annual_return']:.2%}") +print(f"最大回撤: {result.performance['max_drawdown']:.2%}") +print(f"夏普: {result.performance['sharpe']:.4f}") +print(f"调仓次数: {len(result.rebalance_dates)}") +print(f"交易笔数: {len(result.trades)}") + +# 权益曲线 +print(result.equity_curve.head()) + +# 持仓历史 +for state in result.states[-5:]: + print(f" {state.date}: 持仓{state.positions_count}只 净值{state.total_value:.0f}") +``` + +--- + +## 5. 高级回测 + +### 5.1 滑点模型 + +4 种可插拔滑点模型，替代原有固定滑点： + +```python +from easy_tdx.backtest import BacktestEngine +from easy_tdx.backtest.slippage import ( + FixedSlippage, + PercentSlippage, + SquareRootSlippage, + VolumeSlippage, +) + +# 1. 固定每股滑点（与旧行为一致） +model1 = FixedSlippage(per_share=0.01) + +# 2. 按金额百分比 +model2 = PercentSlippage(rate=0.001) + +# 3. 方根市场冲击模型（Almgren-Chriss 简化版） +# impact = sigma * sqrt(participation_rate) * price * size * coeff +# A 股量化主流：参与率 >5% 时冲击显著 +model3 = SquareRootSlippage(impact_coeff=0.1) + +# 4. 成交量比例滑点 +model4 = VolumeSlippage(base_bps=10.0) + +# 在 BacktestEngine 中使用 +engine = BacktestEngine( + MyStrategy, + cash=1_000_000, + slippage_model=SquareRootSlippage(impact_coeff=0.1), +) +result = engine.run(df) +``` + +**模型选择建议**： + +| 场景 | 推荐模型 | 参数 | +|------|---------|------| +| 快速原型 | `FixedSlippage` | `per_share=0.01` | +| 中频策略 | `PercentSlippage` | `rate=0.001` | +| 大额订单 | `SquareRootSlippage` | `impact_coeff=0.1` | +| 低流动性股票 | `VolumeSlippage` | `base_bps=10.0` | + +### 5.2 执行仿真 + +4 种执行模型，将单笔信号拆分为多笔子交易： + +```python +from easy_tdx.backtest.execution import ( + ImmediateExecution, + TWAPExecution, + VWAPExecution, + LimitExecution, +) + +# 1. 即时成交（默认，与旧行为一致） +exec1 = ImmediateExecution() + +# 2. TWAP：时间加权平均价格，N 根 K 线均匀拆单 +exec2 = TWAPExecution(n_bars=5) + +# 3. VWAP：成交量加权平均价格，按历史量分布拆单 +exec3 = VWAPExecution(n_bars=5, volume_lookback=20) + +# 4. 限价单：目标价挂单，TTL 内未触发则放弃 +exec4 = LimitExecution(ttl_bars=5) + +# 在 BacktestEngine 中使用 +engine = BacktestEngine( + MyStrategy, + cash=1_000_000, + execution_model=TWAPExecution(n_bars=3), + slippage_model=SquareRootSlippage(), +) +result = engine.run(df) +``` + +**执行模型选择**： + +| 场景 | 推荐模型 | 参数 | +|------|---------|------| +| 小额/快速验证 | `ImmediateExecution` | 默认 | +| 大额建仓/平仓 | `TWAPExecution` | `n_bars=3~5` | +| 追踪 VWAP 基准 | `VWAPExecution` | `n_bars=5` | +| 精确入场价位 | `LimitExecution` | `ttl_bars=5` | + +**TWAP vs VWAP 示例**： + +```python +# TWAP: 300 股拆成 3 笔 100 股，在 bar 1/2/3 以 close 执行 +engine = BacktestEngine( + MyStrategy, cash=100_000, + execution_model=TWAPExecution(n_bars=3), +) + +# VWAP: 按成交量分布拆 300 股 — 成交量大的 bar 分配更多 +engine = BacktestEngine( + MyStrategy, cash=100_000, + execution_model=VWAPExecution(n_bars=3, volume_lookback=20), +) + +# 限价单：在 50 元挂买入，5 根 K 线内 low <= 50 才成交 +class LimitBuyStrategy(Strategy): + def init(self): pass + def next(self): + if self._bar_index == 0: + self.buy(size=100, price=50.0) # 指定限价 + +engine = BacktestEngine( + LimitBuyStrategy, cash=100_000, + execution_model=LimitExecution(ttl_bars=5), +) +``` + +### 5.3 归因分析 + +从回测结果生成归因报告： + +```python +from easy_tdx.backtest import BacktestEngine +from easy_tdx.backtest.attribution import AttributionAnalyzer + +# 运行回测 +engine = BacktestEngine(MyStrategy, cash=1_000_000) +result = engine.run(df) + +# --- 成本归因 --- +analyzer = AttributionAnalyzer(result.trades, result.equity_curve) +cost_report = analyzer.cost_attribution() +print(f"总收益: {cost_report.total_return:.2%}") +print(f"总交易成本: {cost_report.total_trade_cost:.0f} 元") +print(f" 佣金: {cost_report.commission_cost:.0f}") +print(f" 滑点: {cost_report.slippage_cost:.0f}") +print(f" 印花税: {cost_report.stamp_tax_cost:.0f}") + +# --- Brinson 归因（需要基准）--- +import numpy as np +import pandas as pd +# 构造基准曲线（如沪深300） +benchmark = pd.DataFrame({ + "datetime": result.equity_curve["datetime"], + "total": np.linspace(100000, 108000, len(result.equity_curve)), +}) +analyzer = AttributionAnalyzer(result.trades, result.equity_curve, benchmark=benchmark) +brinson_report = analyzer.brinson_attribution() +print(f"配置贡献: {brinson_report.allocation_return:.2%}") +print(f"选股贡献: {brinson_report.selection_return:.2%}") +print(f"交叉效应: {brinson_report.interaction_return:.2%}") + +# --- 因子归因（需要因子数据）--- +exposures = pd.DataFrame({"momentum": [0.5, 0.3, 0.2], "quality": [0.1, -0.1, 0.0]}) +returns = pd.DataFrame({"momentum": [0.05, 0.03, 0.02], "quality": [0.01, -0.02, 0.0]}) +analyzer = AttributionAnalyzer( + result.trades, result.equity_curve, + factor_exposures=exposures, factor_returns=returns, +) +factor_report = analyzer.factor_attribution() +for name, ret in factor_report.factor_returns.items(): + print(f" {name}: {ret:.4f}") +print(f"特质收益: {factor_report.specific_return:.4f}") + +# --- 完整报告（自动选择最佳归因模式）--- +full_report = analyzer.full_report() +``` + +**归因模式优先级**：因子归因 > Brinson 归因 > 成本归因。`full_report()` 自动选择数据最完整的模式。 + +--- + +## 6. CLI 命令 + +```bash +# 列出所有内置因子 +easy-tdx factor list --table + +# 因子分析（需要数据，输出示例代码） +easy-tdx factor analyze momentum_20d + +# 组合因子回测（需要数据，输出示例代码） +easy-tdx pfactor backtest momentum_20d --n-stocks 10 --optimizer factor_weighted +``` + +CLI 命令输出 Python API 示例代码，方便复制使用。完整的因子计算和组合回测建议通过 Python API 完成。 + +--- + +## 7. 完整工作流示例 + +从数据获取到组合回测再到归因分析的完整管道： + +```python +""" +easy-tdx 量化研究完整工作流示例。 + +依赖: pip install easy-tdx +""" + +from easy_tdx import TdxClient, Market, KlineCategory +from easy_tdx.factor import FactorEngine, FactorAnalyzer, preprocess +from easy_tdx.portfolio import RebalanceEngine, FactorWeightedOptimizer +from easy_tdx.backtest import BacktestEngine +from easy_tdx.backtest.slippage import SquareRootSlippage +from easy_tdx.backtest.execution import TWAPExecution +from easy_tdx.backtest.attribution import AttributionAnalyzer + +# ── 1. 数据获取 ────────────────────────────────────── +client = TdxClient() +stock_pool = ["000001", "000858", "600519", "600036", "601318", + "000333", "002415", "601012", "600276", "000568"] + +data = {} +for code in stock_pool: + market = Market.SH if code.startswith("6") else Market.SZ + data[code] = client.get_security_bars( + market, code, KlineCategory.DAY, 0, 500 + ) +print(f"获取 {len(data)} 只股票数据") + +# ── 2. 因子计算 ────────────────────────────────────── +engine = FactorEngine() +factor_data = engine.compute_cross_section( + data, ["momentum_20d", "volatility_20d", "rsi_14"] +) +print(f"截面因子数据: {len(factor_data)} 行") + +# ── 3. 因子预处理 ───────────────────────────────────── +clean = preprocess( + factor_data, + factor_names=["momentum_20d", "volatility_20d", "rsi_14"], + steps=["winsorize", "zscore", "fill_missing"], +) + +# ── 4. 因子分析 ────────────────────────────────────── +forward_returns = engine.compute_forward_returns(data, period=5) + +for factor_name in ["momentum_20d", "volatility_20d", "rsi_14"]: + analyzer = FactorAnalyzer(clean, forward_returns) + report = analyzer.full_report(factor_name) + print(f"\n── {factor_name} ──") + print(f" IC均值: {report.mean_ic:.4f} ICIR: {report.icir:.4f}") + print(f" 多头年化: {report.long_only_annual:.2%}") + print(f" 多空夏普: {report.long_short_sharpe:.4f}") + +# ── 5. 组合回测 ────────────────────────────────────── +rebalancer = RebalanceEngine( + optimizer=FactorWeightedOptimizer(), + factor_name="momentum_20d", + n_stocks=5, + rebalance_freq="M", + cash=1_000_000, +) +result = rebalancer.run(data, start_date=20230101, end_date=20240101) +print(f"\n── 组合回测 ──") +print(f" 总收益: {result.performance['total_return']:.2%}") +print(f" 年化: {result.performance['annual_return']:.2%}") +print(f" 最大回撤: {result.performance['max_drawdown']:.2%}") +print(f" 夏普: {result.performance['sharpe']:.4f}") + +# ── 6. 高级单策略回测（滑点 + 执行仿真）────── +from easy_tdx.backtest import Strategy + +class MomentumStrategy(Strategy): + def init(self): + pass + def next(self): + if self._bar_index < 20: + return + ret = (self.data.close[0] - self.data.close[-20]) / self.data.close[-20] + if ret > 0.05 and self.position["size"] == 0: + self.buy(size=0) + elif ret < -0.03 and self.position["size"] > 0: + self.sell(size=0) + +bt_engine = BacktestEngine( + MomentumStrategy, + cash=500_000, + slippage_model=SquareRootSlippage(impact_coeff=0.1), + execution_model=TWAPExecution(n_bars=3), +) +# 选一只股票做回测 +bt_result = bt_engine.run(data["600519"]) +print(f"\n── 高级回测（600519）──") +print(f" 总收益: {bt_result.performance['total_return']:.2%}") +print(f" 夏普: {bt_result.performance['sharpe']:.4f}") + +# ── 7. 归因分析 ────────────────────────────────────── +att_analyzer = AttributionAnalyzer(bt_result.trades, bt_result.equity_curve) +cost_report = att_analyzer.cost_attribution() +print(f"\n── 成本归因 ──") +print(f" 总交易成本: {cost_report.total_trade_cost:.0f} 元") +print(f" 佣金: {cost_report.commission_cost:.0f}") +print(f" 滑点: {cost_report.slippage_cost:.0f}") +print(f" 印花税: {cost_report.stamp_tax_cost:.0f}") + +print("\n完成。") +client.close() +``` + +--- + +## 向后兼容 + +所有新功能通过可选参数启用，**现有代码零改动**： + +| 现有调用 | 行为 | +|---------|------| +| `BacktestEngine(strategy, slippage=0.01)` | 与旧版完全一致 | +| `BacktestEngine(strategy)` | 无滑点，与旧版一致 | +| `OrderSimulator(df, slippage=0.01)` | 与旧版完全一致 | +| `BacktestEngine(strategy, slippage_model=...)` | 使用新滑点模型 | +| `BacktestEngine(strategy, execution_model=...)` | 使用新执行引擎 | + +新增模块（`factor/`, `portfolio/`, `backtest/slippage.py`, `backtest/execution.py`, `backtest/attribution.py`）为独立新增，不修改任何现有接口。 diff --git a/docs/superpowers/plans/2026-06-12-v1.14.0-slippage-execution.md b/docs/superpowers/plans/2026-06-12-v1.14.0-slippage-execution.md new file mode 100644 index 0000000..a444420 --- /dev/null +++ b/docs/superpowers/plans/2026-06-12-v1.14.0-slippage-execution.md @@ -0,0 +1,1724 @@ +# v1.14.0 滑点模型 + 执行仿真 实施计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** 为回测引擎新增可插拔的滑点模型和执行仿真引擎，同时保持完全向后兼容。 + +**Architecture:** 在 `backtest/` 下新增 `slippage.py` 和 `execution.py`，修改 `orders.py` 和 `engine.py` 集成新模型。所有新功能通过可选参数启用，不破坏现有 API。 + +**Tech Stack:** 纯 numpy/pandas，无新外部依赖。 + +--- + +## 文件结构 + +| 文件 | 操作 | 职责 | +|------|------|------| +| `src/easy_tdx/backtest/slippage.py` | 新增 | SlippageModel ABC + 4 种内置模型 | +| `src/easy_tdx/backtest/execution.py` | 新增 | ExecutionModel ABC + 4 种内置模型 | +| `src/easy_tdx/backtest/orders.py` | 修改 | 集成 SlippageModel | +| `src/easy_tdx/backtest/engine.py` | 修改 | 集成 SlippageModel + ExecutionModel | +| `tests/unit/test_backtest_slippage.py` | 新增 | 滑点模型测试（~15 个） | +| `tests/unit/test_backtest_execution.py` | 新增 | 执行仿真测试（~20 个） | + +--- + +### Task 1: SlippageModel 基类 + FixedSlippage + PercentSlippage + +**Files:** +- Create: `src/easy_tdx/backtest/slippage.py` +- Test: `tests/unit/test_backtest_slippage.py` + +- [ ] **Step 1: Write tests** + +```python +"""滑点模型单元测试。""" +from __future__ import annotations + +import pytest + +from easy_tdx.backtest.slippage import ( + FixedSlippage, + PercentSlippage, + SlippageModel, + SquareRootSlippage, + VolumeSlippage, +) + + +class TestSlippageBase: + """基类验证。""" + + def test_cannot_instantiate_abc(self) -> None: + """不能直接实例化 ABC。""" + with pytest.raises(TypeError): + SlippageModel() # type: ignore[abstract] + + def test_subclass_must_implement_compute(self) -> None: + """子类必须实现 compute。""" + + class BadModel(SlippageModel): + pass + + with pytest.raises(TypeError): + BadModel() # type: ignore[abstract] + + +class TestFixedSlippage: + """固定每股滑点。""" + + def test_zero_per_share(self) -> None: + """per_share=0 时无滑点。""" + model = FixedSlippage(per_share=0.0) + cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY") + assert cost == 0.0 + + def test_basic(self) -> None: + """基本计算：100 股 × 0.01 元/股 = 1.0。""" + model = FixedSlippage(per_share=0.01) + cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY") + assert cost == pytest.approx(1.0) + + def test_large_size(self) -> None: + """大单。""" + model = FixedSlippage(per_share=0.05) + cost = model.compute(price=50.0, size=10000, volume=500000, volatility=0.2, direction="SELL") + assert cost == pytest.approx(500.0) + + def test_direction_irrelevant(self) -> None: + """方向不影响固定滑点。""" + model = FixedSlippage(per_share=0.01) + buy_cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY") + sell_cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="SELL") + assert buy_cost == sell_cost + + +class TestPercentSlippage: + """按成交金额百分比滑点。""" + + def test_zero_rate(self) -> None: + """rate=0 时无滑点。""" + model = PercentSlippage(rate=0.0) + cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY") + assert cost == 0.0 + + def test_basic(self) -> None: + """10元 × 100股 × 0.001 = 1.0。""" + model = PercentSlippage(rate=0.001) + cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY") + assert cost == pytest.approx(1.0) + + def test_high_price(self) -> None: + """高价股。""" + model = PercentSlippage(rate=0.002) + cost = model.compute(price=100.0, size=500, volume=20000, volatility=0.25, direction="BUY") + # 100 × 500 × 0.002 = 100.0 + assert cost == pytest.approx(100.0) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/unit/test_backtest_slippage.py -v -k "TestSlippageBase or TestFixed or TestPercent" --no-header` +Expected: FAIL (import error) + +- [ ] **Step 3: Write implementation** + +```python +"""可插拔滑点模型。""" +from __future__ import annotations + +from abc import ABC, abstractmethod + +import numpy as np + + +class SlippageModel(ABC): + """滑点模型基类。 + + 所有滑点模型必须实现 compute() 方法，返回总滑点成本（金额）。 + """ + + @abstractmethod + def compute( + self, + price: float, + size: float, + volume: float, + volatility: float, + direction: str, + ) -> float: + """计算滑点成本。 + + Args: + price: 成交价格 + size: 订单数量（股） + volume: 当日成交量（股），0 表示无数据 + volatility: 近期年化波动率，0 表示无数据 + direction: 交易方向 BUY / SELL + + Returns: + 总滑点成本（金额，非比率） + """ + ... + + +class FixedSlippage(SlippageModel): + """固定每股滑点（向后兼容）。""" + + def __init__(self, per_share: float = 0.01) -> None: + self._per_share = per_share + + def compute( + self, + price: float, + size: float, + volume: float, + volatility: float, + direction: str, + ) -> float: + return size * self._per_share + + +class PercentSlippage(SlippageModel): + """按成交金额百分比滑点。""" + + def __init__(self, rate: float = 0.001) -> None: + self._rate = rate + + def compute( + self, + price: float, + size: float, + volume: float, + volatility: float, + direction: str, + ) -> float: + return price * size * self._rate +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `python -m pytest tests/unit/test_backtest_slippage.py -v -k "TestSlippageBase or TestFixed or TestPercent" --no-header` +Expected: 9 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/easy_tdx/backtest/slippage.py tests/unit/test_backtest_slippage.py +git commit -m "feat(backtest): add SlippageModel ABC + FixedSlippage + PercentSlippage" +``` + +--- + +### Task 2: SquareRootSlippage + VolumeSlippage + +**Files:** +- Modify: `src/easy_tdx/backtest/slippage.py` +- Modify: `tests/unit/test_backtest_slippage.py` + +- [ ] **Step 1: Write tests** + +在 `test_backtest_slippage.py` 末尾追加： + +```python +class TestSquareRootSlippage: + """方根市场冲击模型。""" + + def test_zero_size(self) -> None: + """size=0 时无冲击。""" + model = SquareRootSlippage(impact_coeff=0.1) + cost = model.compute(price=10.0, size=0, volume=10000, volatility=0.3, direction="BUY") + assert cost == 0.0 + + def test_small_participation_rate(self) -> None: + """低参与率（小单），冲击成本低。""" + model = SquareRootSlippage(impact_coeff=0.1) + # size=100, volume=1000000, participation_rate=0.0001 + cost = model.compute( + price=10.0, size=100, volume=1_000_000, volatility=0.3, direction="BUY" + ) + # σ=0.3, √(0.0001)=0.01, impact = 0.3 × 0.01 × 10 × 100 × 0.1 = 0.03 + assert cost == pytest.approx(0.03) + + def test_high_participation_rate(self) -> None: + """高参与率（大单），冲击成本高。""" + model = SquareRootSlippage(impact_coeff=0.1) + # size=100000, volume=200000, participation_rate=0.5 + cost = model.compute( + price=10.0, size=100_000, volume=200_000, volatility=0.3, direction="BUY" + ) + # 应该显著大于小单 + small_cost = model.compute( + price=10.0, size=100, volume=1_000_000, volatility=0.3, direction="BUY" + ) + assert cost > small_cost * 10 + + def test_zero_volume_fallback(self) -> None: + """volume=0 时退化为 PercentSlippage(rate=0.001)。""" + model = SquareRootSlippage(impact_coeff=0.1) + cost = model.compute(price=10.0, size=100, volume=0, volatility=0.3, direction="BUY") + # 退化为 10 × 100 × 0.001 = 1.0 + assert cost == pytest.approx(1.0) + + def test_zero_volatility_fallback(self) -> None: + """volatility=0 时退化为 PercentSlippage(rate=0.001)。""" + model = SquareRootSlippage(impact_coeff=0.1) + cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.0, direction="BUY") + assert cost == pytest.approx(1.0) + + +class TestVolumeSlippage: + """成交量比例滑点。""" + + def test_zero_size(self) -> None: + """size=0 时无滑点。""" + model = VolumeSlippage(base_bps=10.0) + cost = model.compute(price=10.0, size=0, volume=10000, volatility=0.3, direction="BUY") + assert cost == 0.0 + + def test_basic(self) -> None: + """基本计算。""" + model = VolumeSlippage(base_bps=10.0) + # base_bps=10 → rate=10/10000=0.001 + # participation = 100/10000 = 0.01 + # cost = 0.001 × 0.01 × 10 × 100 = 0.01 + cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY") + assert cost == pytest.approx(0.01) + + def test_high_participation(self) -> None: + """高参与率时滑点高。""" + model = VolumeSlippage(base_bps=10.0) + cost_high = model.compute(price=10.0, size=5000, volume=10000, volatility=0.3, direction="BUY") + cost_low = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY") + assert cost_high > cost_low + + def test_zero_volume_fallback(self) -> None: + """volume=0 时退化为 PercentSlippage(rate=base_bps/10000)。""" + model = VolumeSlippage(base_bps=10.0) + cost = model.compute(price=10.0, size=100, volume=0, volatility=0.3, direction="BUY") + # 退化为 10/10000 × 10 × 100 = 1.0 + assert cost == pytest.approx(1.0) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/unit/test_backtest_slippage.py -v -k "TestSquareRoot or TestVolume" --no-header` +Expected: FAIL (import error) + +- [ ] **Step 3: Write implementation** + +在 `slippage.py` 末尾追加： + +```python +class SquareRootSlippage(SlippageModel): + """方根市场冲击模型（Almgren-Chriss 简化版）。 + + impact = σ × √(participation_rate) × price × size × impact_coeff + + 当 volume=0 或 volatility=0 时退化为 PercentSlippage(rate=0.001)。 + """ + + def __init__(self, impact_coeff: float = 0.1) -> None: + self._impact_coeff = impact_coeff + self._fallback = PercentSlippage(rate=0.001) + + def compute( + self, + price: float, + size: float, + volume: float, + volatility: float, + direction: str, + ) -> float: + if size <= 0: + return 0.0 + if volume <= 0 or volatility <= 0: + return self._fallback.compute(price, size, volume, volatility, direction) + participation_rate = min(size / volume, 1.0) + impact = volatility * np.sqrt(participation_rate) * price * size * self._impact_coeff + return float(impact) + + +class VolumeSlippage(SlippageModel): + """成交量比例滑点。 + + cost = (base_bps / 10000) × (size / volume) × price × size + + 当 volume=0 时退化为 PercentSlippage(rate=base_bps/10000)。 + """ + + def __init__(self, base_bps: float = 10.0) -> None: + self._base_bps = base_bps + self._fallback = PercentSlippage(rate=base_bps / 10000.0) + + def compute( + self, + price: float, + size: float, + volume: float, + volatility: float, + direction: str, + ) -> float: + if size <= 0: + return 0.0 + if volume <= 0: + return self._fallback.compute(price, size, volume, volatility, direction) + rate = self._base_bps / 10000.0 + participation = min(size / volume, 1.0) + return rate * participation * price * size +``` + +- [ ] **Step 4: Run tests** + +Run: `python -m pytest tests/unit/test_backtest_slippage.py -v --no-header` +Expected: 19 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/easy_tdx/backtest/slippage.py tests/unit/test_backtest_slippage.py +git commit -m "feat(backtest): add SquareRootSlippage + VolumeSlippage" +``` + +--- + +### Task 3: OrderSimulator 集成 SlippageModel + +**Files:** +- Modify: `src/easy_tdx/backtest/orders.py` + +- [ ] **Step 1: Write tests** + +在 `tests/unit/test_backtest_orders.py` 末尾追加： + +```python +from easy_tdx.backtest.slippage import FixedSlippage, PercentSlippage, SquareRootSlippage + + +class TestSlippageModelIntegration: + """测试 OrderSimulator 与 SlippageModel 集成。""" + + def test_fixed_slippage_model(self) -> None: + """FixedSlippage 与旧 slippage 参数等价。""" + df = _make_df(10) + sim = OrderSimulator( + df, + execution="next_open", + slippage_model=FixedSlippage(per_share=0.01), + ) + signals = [_buy_signal(0, size=100)] + trades = sim.simulate(signals, cash=20000, position=0) + assert len(trades) == 1 + assert trades[0].slippage == pytest.approx(1.0) # 100 × 0.01 + + def test_percent_slippage_model(self) -> None: + """PercentSlippage 计算。""" + df = _make_df(10) + sim = OrderSimulator( + df, + execution="next_open", + slippage_model=PercentSlippage(rate=0.001), + ) + signals = [_buy_signal(0, size=100)] + trades = sim.simulate(signals, cash=20000, position=0) + assert len(trades) == 1 + # price=101 (next_open), 101 × 100 × 0.001 = 10.1 + assert trades[0].slippage == pytest.approx(10.1) + + def test_slippage_model_overrides_slippage_param(self) -> None: + """slippage_model 优先于 slippage 参数。""" + df = _make_df(10) + sim = OrderSimulator( + df, + execution="next_open", + slippage=999.0, # 应被忽略 + slippage_model=FixedSlippage(per_share=0.01), + ) + signals = [_buy_signal(0, size=100)] + trades = sim.simulate(signals, cash=20000, position=0) + assert len(trades) == 1 + assert trades[0].slippage == pytest.approx(1.0) # 用 model，不是 999*100 + + def test_sell_with_slippage_model(self) -> None: + """卖出时也使用滑点模型。""" + df = _make_df(10) + sim = OrderSimulator( + df, + execution="next_open", + slippage_model=FixedSlippage(per_share=0.02), + ) + signals = [_sell_signal(0, size=100)] + trades = sim.simulate(signals, cash=0, position=200) + assert len(trades) == 1 + assert trades[0].slippage == pytest.approx(2.0) # 100 × 0.02 + + def test_no_slippage_model_uses_old_param(self) -> None: + """不提供 model 时使用旧 slippage 参数（向后兼容）。""" + df = _make_df(10) + sim = OrderSimulator(df, execution="next_open", slippage=0.05) + signals = [_buy_signal(0, size=100)] + trades = sim.simulate(signals, cash=20000, position=0) + assert len(trades) == 1 + assert trades[0].slippage == pytest.approx(5.0) # 100 × 0.05 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/unit/test_backtest_orders.py -v -k "TestSlippageModel" --no-header` +Expected: FAIL (slippage_model param not accepted) + +- [ ] **Step 3: Modify orders.py** + +Changes to `src/easy_tdx/backtest/orders.py`: + +1. Add import: +```python +from __future__ import annotations + +from dataclasses import dataclass, field + +import pandas as pd + +from easy_tdx.backtest.types import Signal, Trade +``` + +2. Add `slippage_model` field to `OrderSimulator`: +```python +@dataclass +class OrderSimulator: + # ... existing fields ... + slippage: float = 0.0 + slippage_model: SlippageModel | None = None # NEW + future_leak_warning: bool = False +``` + +But we can't use `from __future__ import annotations` with dataclass field type hints that reference imported types directly. Since `SlippageModel` is in the same package, use `TYPE_CHECKING`: + +```python +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +import pandas as pd + +from easy_tdx.backtest.types import Signal, Trade + +if TYPE_CHECKING: + from easy_tdx.backtest.slippage import SlippageModel +``` + +3. Replace slippage calculations in `_execute_buy` and `_execute_sell`: + +In `_execute_buy`, replace: +```python +slippage = size * self.slippage +``` +with: +```python +slippage = self._compute_slippage(size, price, False) +``` + +Same in `_execute_sell`: +```python +slippage = self._compute_slippage(size, price, True) +``` + +4. Add helper method: + +```python +def _compute_slippage(self, size: float, price: float, is_sell: bool) -> float: + """计算滑点成本。""" + if self.slippage_model is not None: + volume = self._get_current_volume() + volatility = self._estimate_volatility() + return self.slippage_model.compute( + price=price, size=size, volume=volume, + volatility=volatility, direction="SELL" if is_sell else "BUY", + ) + return size * self.slippage + +def _get_current_volume(self) -> float: + """获取最近一根K线的成交量。""" + if "volume" in self.df.columns and len(self.df) > 0: + return float(self.df["volume"].iloc[-1]) + return 0.0 + +def _estimate_volatility(self) -> float: + """从收盘价估计近期年化波动率。""" + if "close" not in self.df.columns or len(self.df) < 2: + return 0.0 + close = self.df["close"].to_numpy() + returns = np.diff(close) / close[:-1] + if len(returns) < 2: + return 0.0 + daily_vol = float(np.std(returns)) + return daily_vol * np.sqrt(252) +``` + +Also need to add `import numpy as np` at top. + +- [ ] **Step 4: Run all backtest tests** + +Run: `python -m pytest tests/unit/test_backtest_orders.py -v --no-header` +Expected: All existing + 5 new tests pass + +- [ ] **Step 5: Commit** + +```bash +git add src/easy_tdx/backtest/orders.py tests/unit/test_backtest_orders.py +git commit -m "feat(backtest): integrate SlippageModel into OrderSimulator" +``` + +--- + +### Task 4: ExecutionModel 基类 + ImmediateExecution + +**Files:** +- Create: `src/easy_tdx/backtest/execution.py` +- Create: `tests/unit/test_backtest_execution.py` + +- [ ] **Step 1: Write tests** + +```python +"""执行仿真引擎单元测试。""" +from __future__ import annotations + +import pytest + +from easy_tdx.backtest.execution import ExecutionModel, ImmediateExecution +from easy_tdx.backtest.types import Signal + + +def _make_df(n: int = 20) -> "pd.DataFrame": + """构造测试用K线数据。""" + import pandas as pd + data = { + "datetime": [20240101 + i for i in range(n)], + "open": [100.0 + i for i in range(n)], + "close": [101.0 + i for i in range(n)], + "high": [102.0 + i for i in range(n)], + "low": [99.0 + i for i in range(n)], + "volume": [10000] * n, + } + return pd.DataFrame(data) + + +class TestExecutionBase: + """基类验证。""" + + def test_cannot_instantiate_abc(self) -> None: + with pytest.raises(TypeError): + ExecutionModel() # type: ignore[abstract] + + +class TestImmediateExecution: + """即时成交（向后兼容）。""" + + def test_buy_signal(self) -> None: + """买入信号在下一 bar 成交。""" + df = _make_df(10) + model = ImmediateExecution() + signal = Signal(datetime=20240101, direction="BUY", size=100) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=20000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 1 + assert trades[0].direction == "BUY" + assert trades[0].price == 101.0 # next bar open + + def test_sell_signal(self) -> None: + """卖出信号在下一 bar 成交。""" + df = _make_df(10) + model = ImmediateExecution() + signal = Signal(datetime=20240101, direction="SELL", size=100) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=0, position=200, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 1 + assert trades[0].direction == "SELL" + + def test_signal_at_last_bar(self) -> None: + """信号在最后一根K线，无法成交。""" + df = _make_df(10) + model = ImmediateExecution() + signal = Signal(datetime=20240109, direction="BUY", size=100) + trades = model.execute( + signal=signal, df=df, bar_idx=9, + cash=20000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 0 + + def test_with_slippage_model(self) -> None: + """使用滑点模型。""" + from easy_tdx.backtest.slippage import FixedSlippage + df = _make_df(10) + model = ImmediateExecution() + signal = Signal(datetime=20240101, direction="BUY", size=100) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=20000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=FixedSlippage(per_share=0.01), + ) + assert len(trades) == 1 + assert trades[0].slippage == pytest.approx(1.0) + + def test_commission_on_buy(self) -> None: + """买入佣金计算。""" + df = _make_df(10) + model = ImmediateExecution() + signal = Signal(datetime=20240101, direction="BUY", size=100) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=20000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 1 + # 101 × 100 × 0.0003 = 3.03, min_commission=5 → 5.0 + assert trades[0].commission >= 5.0 + + def test_stamp_tax_on_sell(self) -> None: + """卖出印花税。""" + df = _make_df(10) + model = ImmediateExecution() + signal = Signal(datetime=20240101, direction="SELL", size=100) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=0, position=200, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 1 + # commission = max(101×100×0.0003, 5) + 101×100×0.001 = 5 + 10.1 = 15.1 + assert trades[0].commission > 10.0 + + def test_full_position_buy(self) -> None: + """full 模式全仓买入（100股整手）。""" + df = _make_df(10) + model = ImmediateExecution() + signal = Signal(datetime=20240101, direction="BUY", size=0) # size=0 = full + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=20000, position=0, position_mode="full", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 1 + assert trades[0].size == 100 # 20000 / (101 × 1.0003) ≈ 197 → 100 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/unit/test_backtest_execution.py -v -k "TestExecutionBase or TestImmediate" --no-header` +Expected: FAIL (import error) + +- [ ] **Step 3: Write implementation** + +```python +"""可插拔执行仿真引擎。""" +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +import numpy as np +import pandas as pd + +from easy_tdx.backtest.types import Trade + +if TYPE_CHECKING: + from easy_tdx.backtest.slippage import SlippageModel + from easy_tdx.backtest.types import Signal + + +class ExecutionModel(ABC): + """执行仿真基类。 + + 将信号转换为一笔或多笔成交记录。 + """ + + @abstractmethod + def execute( + self, + signal: Signal, + df: pd.DataFrame, + bar_idx: int, + cash: float, + position: float, + position_mode: str, + commission: float, + min_commission: float, + stamp_tax: float, + slippage_model: SlippageModel | None, + ) -> list[Trade]: + """将信号转换为一笔或多笔成交。""" + ... + + def _calc_commission( + self, size: float, price: float, is_sell: bool, + commission: float, min_commission: float, stamp_tax: float, + ) -> float: + """计算手续费。""" + comm = max(size * price * commission, min_commission) + if is_sell: + comm += size * price * stamp_tax + return comm + + def _calc_slippage( + self, size: float, price: float, is_sell: bool, + slippage_model: SlippageModel | None, df: pd.DataFrame, + ) -> float: + """计算滑点。""" + if slippage_model is None: + return 0.0 + volume = float(df["volume"].iloc[-1]) if "volume" in df.columns else 0.0 + volatility = self._estimate_volatility(df) + return slippage_model.compute( + price=price, size=size, volume=volume, + volatility=volatility, + direction="SELL" if is_sell else "BUY", + ) + + def _estimate_volatility(self, df: pd.DataFrame) -> float: + """从收盘价估计近期年化波动率。""" + if "close" not in df.columns or len(df) < 2: + return 0.0 + close = df["close"].to_numpy() + returns = np.diff(close) / close[:-1] + if len(returns) < 2: + return 0.0 + return float(np.std(returns)) * np.sqrt(252) + + def _calc_buy_size( + self, signal_size: float, price: float, cash: float, + position_mode: str, commission: float, + ) -> float: + """计算买入数量。""" + if position_mode == "full" or signal_size == 0: + max_cost = price * (1 + commission) + max_shares = int(cash / max_cost / 100) * 100 + return float(max_shares) + elif position_mode == "percent": + target_value = cash * signal_size + return float(int(target_value / price / 100) * 100) + return signal_size + + +class ImmediateExecution(ExecutionModel): + """即时成交（向后兼容，与现有 OrderSimulator 行为一致）。""" + + def execute( + self, + signal: Signal, + df: pd.DataFrame, + bar_idx: int, + cash: float, + position: float, + position_mode: str, + commission: float, + min_commission: float, + stamp_tax: float, + slippage_model: SlippageModel | None, + ) -> list[Trade]: + exec_idx = bar_idx + 1 + if exec_idx >= len(df): + return [] + + price = float(df["open"].iloc[exec_idx]) + + if signal.direction == "BUY": + size = self._calc_buy_size(signal.size, price, cash, position_mode, commission) + if size <= 0: + return [] + comm = self._calc_commission(size, price, False, commission, min_commission, stamp_tax) + slip = self._calc_slippage(size, price, False, slippage_model, df) + return [Trade( + datetime=int(df["datetime"].iloc[exec_idx]) + if hasattr(df["datetime"].iloc[exec_idx], "strftime") + else int(df["datetime"].iloc[exec_idx]), + direction="BUY", + size=size, + price=price, + commission=comm, + slippage=slip, + )] + + elif signal.direction == "SELL": + size = signal.size if signal.size > 0 else position + if size <= 0: + return [] + if size > position: + size = position + comm = self._calc_commission(size, price, True, commission, min_commission, stamp_tax) + slip = self._calc_slippage(size, price, True, slippage_model, df) + return [Trade( + datetime=int(df["datetime"].iloc[exec_idx]) + if hasattr(df["datetime"].iloc[exec_idx], "strftime") + else int(df["datetime"].iloc[exec_idx]), + direction="SELL", + size=size, + price=price, + commission=comm, + slippage=slip, + )] + + return [] +``` + +- [ ] **Step 4: Run tests** + +Run: `python -m pytest tests/unit/test_backtest_execution.py -v -k "TestExecutionBase or TestImmediate" --no-header` +Expected: 8 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/easy_tdx/backtest/execution.py tests/unit/test_backtest_execution.py +git commit -m "feat(backtest): add ExecutionModel ABC + ImmediateExecution" +``` + +--- + +### Task 5: TWAPExecution + VWAPExecution + +**Files:** +- Modify: `src/easy_tdx/backtest/execution.py` +- Modify: `tests/unit/test_backtest_execution.py` + +- [ ] **Step 1: Write tests** + +在 `test_backtest_execution.py` 追加： + +```python +from easy_tdx.backtest.execution import TWAPExecution, VWAPExecution + + +class TestTWAPExecution: + """时间加权平均价格执行。""" + + def test_split_buy_into_3_bars(self) -> None: + """买入订单拆分为 3 个子订单。""" + df = _make_df(20) + model = TWAPExecution(n_bars=3) + signal = Signal(datetime=20240101, direction="BUY", size=300) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=100000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 3 + total_size = sum(t.size for t in trades) + assert total_size <= 300 # 可能有100股取整损失 + # 每笔在不同 bar 执行 + prices = [t.price for t in trades] + assert prices[0] != prices[1] # 不同 bar 价格不同 + + def test_split_sell_into_2_bars(self) -> None: + """卖出订单拆分为 2 个子订单。""" + df = _make_df(20) + model = TWAPExecution(n_bars=2) + signal = Signal(datetime=20240101, direction="SELL", size=200) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=0, position=500, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 2 + assert sum(t.size for t in trades) == 200.0 + + def test_truncates_at_data_end(self) -> None: + """数据不足 n_bars 时截断。""" + df = _make_df(5) # 只有 5 根 K 线 + model = TWAPExecution(n_bars=10) + signal = Signal(datetime=20240101, direction="BUY", size=1000) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=100000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + # bar_idx=0, 可用 bar 1-4, 最多 4 笔 + assert len(trades) <= 4 + + def test_full_position_mode(self) -> None: + """full 模式下拆分全仓买入。""" + df = _make_df(20) + model = TWAPExecution(n_bars=3) + signal = Signal(datetime=20240101, direction="BUY", size=0) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=60000, position=0, position_mode="full", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 3 + assert all(t.size > 0 for t in trades) + + +class TestVWAPExecution: + """成交量加权平均价格执行。""" + + def test_basic_buy(self) -> None: + """基本买入执行。""" + df = _make_df(20) + model = VWAPExecution(n_bars=3, volume_lookback=10) + signal = Signal(datetime=20240101, direction="BUY", size=300) + trades = model.execute( + signal=signal, df=df, bar_idx=5, + cash=100000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 3 + total_size = sum(t.size for t in trades) + assert total_size <= 300 + + def test_volume_weighted_split(self) -> None: + """成交量大的 bar 分配更多数量。""" + import pandas as pd + df = _make_df(20) + # 让不同 bar 的 volume 不同 + df.loc[6, "volume"] = 50000 # bar 7 volume 很大 + df.loc[7, "volume"] = 50000 + df.loc[8, "volume"] = 50000 + model = VWAPExecution(n_bars=3, volume_lookback=5) + signal = Signal(datetime=20240105, direction="BUY", size=300) + trades = model.execute( + signal=signal, df=df, bar_idx=5, + cash=100000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 3 + # 各笔 size 可能因成交量分布不同而不等 + sizes = [t.size for t in trades] + assert sum(sizes) <= 300 + + def test_truncates_at_data_end(self) -> None: + """数据不足时截断。""" + df = _make_df(5) + model = VWAPExecution(n_bars=10, volume_lookback=3) + signal = Signal(datetime=20240101, direction="BUY", size=1000) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=100000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) <= 4 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/unit/test_backtest_execution.py -v -k "TestTWAP or TestVWAP" --no-header` +Expected: FAIL + +- [ ] **Step 3: Write implementation** + +在 `execution.py` 追加： + +```python +class TWAPExecution(ExecutionModel): + """时间加权平均价格执行。 + + 将订单均匀拆分为 n_bars 份，在连续 n_bars 根 K 线上执行。 + """ + + def __init__(self, n_bars: int = 5) -> None: + self._n_bars = max(1, n_bars) + + def execute( + self, + signal: Signal, + df: pd.DataFrame, + bar_idx: int, + cash: float, + position: float, + position_mode: str, + commission: float, + min_commission: float, + stamp_tax: float, + slippage_model: SlippageModel | None, + ) -> list[Trade]: + if signal.direction == "BUY": + return self._execute_buy( + signal, df, bar_idx, cash, position_mode, + commission, min_commission, stamp_tax, slippage_model, + ) + else: + return self._execute_sell( + signal, df, bar_idx, position, + commission, min_commission, stamp_tax, slippage_model, + ) + + def _execute_buy( + self, signal: Signal, df: pd.DataFrame, bar_idx: int, + cash: float, position_mode: str, commission: float, + min_commission: float, stamp_tax: float, + slippage_model: SlippageModel | None, + ) -> list[Trade]: + # 计算总买入量 + first_price = float(df["open"].iloc[bar_idx + 1]) if bar_idx + 1 < len(df) else 0 + if first_price <= 0: + return [] + total_size = self._calc_buy_size(signal.size, first_price, cash, position_mode, commission) + if total_size <= 0: + return [] + + sub_size = int(total_size / self._n_bars / 100) * 100 + if sub_size <= 0: + sub_size = 100 # 至少 1 手 + + trades: list[Trade] = [] + for i in range(self._n_bars): + exec_idx = bar_idx + 1 + i + if exec_idx >= len(df): + break + price = float(df["close"].iloc[exec_idx]) + actual_size = min(sub_size, total_size - sum(t.size for t in trades)) + actual_size = int(actual_size / 100) * 100 + if actual_size <= 0: + break + comm = self._calc_commission(actual_size, price, False, commission, min_commission, stamp_tax) + slip = self._calc_slippage(actual_size, price, False, slippage_model, df) + dt_raw = df["datetime"].iloc[exec_idx] + dt_int = int(dt_raw.strftime("%Y%m%d")) if hasattr(dt_raw, "strftime") else int(dt_raw) + trades.append(Trade( + datetime=dt_int, direction="BUY", + size=float(actual_size), price=price, + commission=comm, slippage=slip, + )) + return trades + + def _execute_sell( + self, signal: Signal, df: pd.DataFrame, bar_idx: int, + position: float, commission: float, + min_commission: float, stamp_tax: float, + slippage_model: SlippageModel | None, + ) -> list[Trade]: + total_size = signal.size if signal.size > 0 else position + if total_size <= 0: + return [] + + sub_size = int(total_size / self._n_bars / 100) * 100 + if sub_size <= 0: + sub_size = 100 + + trades: list[Trade] = [] + for i in range(self._n_bars): + exec_idx = bar_idx + 1 + i + if exec_idx >= len(df): + break + price = float(df["close"].iloc[exec_idx]) + actual_size = min(sub_size, total_size - sum(t.size for t in trades)) + actual_size = int(actual_size / 100) * 100 + if actual_size <= 0: + break + comm = self._calc_commission(actual_size, price, True, commission, min_commission, stamp_tax) + slip = self._calc_slippage(actual_size, price, True, slippage_model, df) + dt_raw = df["datetime"].iloc[exec_idx] + dt_int = int(dt_raw.strftime("%Y%m%d")) if hasattr(dt_raw, "strftime") else int(dt_raw) + trades.append(Trade( + datetime=dt_int, direction="SELL", + size=float(actual_size), price=price, + commission=comm, slippage=slip, + )) + return trades + + +class VWAPExecution(ExecutionModel): + """成交量加权平均价格执行。 + + 按历史成交量分布比例拆分订单。 + """ + + def __init__(self, n_bars: int = 5, volume_lookback: int = 20) -> None: + self._n_bars = max(1, n_bars) + self._volume_lookback = max(1, volume_lookback) + + def execute( + self, + signal: Signal, + df: pd.DataFrame, + bar_idx: int, + cash: float, + position: float, + position_mode: str, + commission: float, + min_commission: float, + stamp_tax: float, + slippage_model: SlippageModel | None, + ) -> list[Trade]: + if signal.direction == "BUY": + return self._execute_buy( + signal, df, bar_idx, cash, position_mode, + commission, min_commission, stamp_tax, slippage_model, + ) + else: + return self._execute_sell( + signal, df, bar_idx, position, + commission, min_commission, stamp_tax, slippage_model, + ) + + def _get_volume_weights(self, df: pd.DataFrame, bar_idx: int) -> list[float]: + """获取成交量权重分布。""" + start = max(0, bar_idx - self._volume_lookback + 1) + lookback = df.iloc[start:bar_idx + 1] + if "volume" not in lookback.columns or len(lookback) == 0: + return [1.0 / self._n_bars] * self._n_bars + + volumes = lookback["volume"].to_numpy() + total_vol = float(volumes.sum()) + if total_vol <= 0: + return [1.0 / self._n_bars] * self._n_bars + + # 取最近 n_bars 期的平均成交量比例 + weights: list[float] = [] + for i in range(self._n_bars): + idx = max(0, len(volumes) - 1 - (i % max(1, len(volumes)))) + weights.append(float(volumes[idx]) / total_vol) + total_w = sum(weights) + if total_w <= 0: + return [1.0 / self._n_bars] * self._n_bars + return [w / total_w for w in weights] + + def _execute_buy( + self, signal: Signal, df: pd.DataFrame, bar_idx: int, + cash: float, position_mode: str, commission: float, + min_commission: float, stamp_tax: float, + slippage_model: SlippageModel | None, + ) -> list[Trade]: + first_price = float(df["open"].iloc[bar_idx + 1]) if bar_idx + 1 < len(df) else 0 + if first_price <= 0: + return [] + total_size = self._calc_buy_size(signal.size, first_price, cash, position_mode, commission) + if total_size <= 0: + return [] + + weights = self._get_volume_weights(df, bar_idx) + trades: list[Trade] = [] + for i in range(self._n_bars): + exec_idx = bar_idx + 1 + i + if exec_idx >= len(df): + break + price = float(df["close"].iloc[exec_idx]) + w = weights[i] if i < len(weights) else 1.0 / self._n_bars + target = int(total_size * w / 100) * 100 + remaining = total_size - sum(t.size for t in trades) + actual_size = min(target, remaining) + actual_size = int(actual_size / 100) * 100 + if actual_size <= 0: + continue + comm = self._calc_commission(actual_size, price, False, commission, min_commission, stamp_tax) + slip = self._calc_slippage(actual_size, price, False, slippage_model, df) + dt_raw = df["datetime"].iloc[exec_idx] + dt_int = int(dt_raw.strftime("%Y%m%d")) if hasattr(dt_raw, "strftime") else int(dt_raw) + trades.append(Trade( + datetime=dt_int, direction="BUY", + size=float(actual_size), price=price, + commission=comm, slippage=slip, + )) + return trades + + def _execute_sell( + self, signal: Signal, df: pd.DataFrame, bar_idx: int, + position: float, commission: float, + min_commission: float, stamp_tax: float, + slippage_model: SlippageModel | None, + ) -> list[Trade]: + total_size = signal.size if signal.size > 0 else position + if total_size <= 0: + return [] + + weights = self._get_volume_weights(df, bar_idx) + trades: list[Trade] = [] + for i in range(self._n_bars): + exec_idx = bar_idx + 1 + i + if exec_idx >= len(df): + break + price = float(df["close"].iloc[exec_idx]) + w = weights[i] if i < len(weights) else 1.0 / self._n_bars + target = int(total_size * w / 100) * 100 + remaining = total_size - sum(t.size for t in trades) + actual_size = min(target, remaining) + actual_size = int(actual_size / 100) * 100 + if actual_size <= 0: + continue + comm = self._calc_commission(actual_size, price, True, commission, min_commission, stamp_tax) + slip = self._calc_slippage(actual_size, price, True, slippage_model, df) + dt_raw = df["datetime"].iloc[exec_idx] + dt_int = int(dt_raw.strftime("%Y%m%d")) if hasattr(dt_raw, "strftime") else int(dt_raw) + trades.append(Trade( + datetime=dt_int, direction="SELL", + size=float(actual_size), price=price, + commission=comm, slippage=slip, + )) + return trades +``` + +- [ ] **Step 4: Run tests** + +Run: `python -m pytest tests/unit/test_backtest_execution.py -v --no-header` +Expected: All pass + +- [ ] **Step 5: Commit** + +```bash +git add src/easy_tdx/backtest/execution.py tests/unit/test_backtest_execution.py +git commit -m "feat(backtest): add TWAPExecution + VWAPExecution" +``` + +--- + +### Task 6: LimitExecution + +**Files:** +- Modify: `src/easy_tdx/backtest/execution.py` +- Modify: `tests/unit/test_backtest_execution.py` + +- [ ] **Step 1: Write tests** + +追加到 `test_backtest_execution.py`： + +```python +from easy_tdx.backtest.execution import LimitExecution + + +class TestLimitExecution: + """限价单执行。""" + + def test_buy_limit_filled(self) -> None: + """买入限价被触发。""" + df = _make_df(20) + model = LimitExecution(ttl_bars=5) + signal = Signal(datetime=20240101, direction="BUY", size=100, price=100.0) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=20000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 1 + assert trades[0].price == 100.0 + # bar 1: low=100.0, 触发 + assert trades[0].direction == "BUY" + + def test_sell_limit_filled(self) -> None: + """卖出限价被触发。""" + df = _make_df(20) + model = LimitExecution(ttl_bars=5) + signal = Signal(datetime=20240101, direction="SELL", size=100, price=105.0) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=0, position=200, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 1 + assert trades[0].price == 105.0 + + def test_limit_not_triggered(self) -> None: + """限价未触发，返回空。""" + df = _make_df(10) + model = LimitExecution(ttl_bars=3) + # 价格从 100+ 递增，limit=50 不可能触发 + signal = Signal(datetime=20240101, direction="BUY", size=100, price=50.0) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=20000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 0 + + def test_no_price_falls_back_to_immediate(self) -> None: + """无限价时退化为即时执行。""" + df = _make_df(10) + model = LimitExecution(ttl_bars=5) + signal = Signal(datetime=20240101, direction="BUY", size=100, price=None) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=20000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 1 + assert trades[0].price == 101.0 # next bar open + + def test_ttl_expires(self) -> None: + """超出 TTL 后不再尝试。""" + df = _make_df(20) + model = LimitExecution(ttl_bars=2) + # 价格从 99+ 递增，limit=98.0 在 bar 1 (low=100) 和 bar 2 (low=101) 都不触发 + signal = Signal(datetime=20240101, direction="BUY", size=100, price=98.0) + trades = model.execute( + signal=signal, df=df, bar_idx=0, + cash=20000, position=0, position_mode="fixed", + commission=0.0003, min_commission=5.0, stamp_tax=0.001, + slippage_model=None, + ) + assert len(trades) == 0 # TTL=2 bars 内 low 都 > 98 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/unit/test_backtest_execution.py -v -k "TestLimit" --no-header` +Expected: FAIL + +- [ ] **Step 3: Write implementation** + +在 `execution.py` 追加： + +```python +class LimitExecution(ExecutionModel): + """限价单执行。 + + 在目标价位挂单，仅当 bar_low <= price（买入）或 bar_high >= price（卖出）时成交。 + 无限价时退化为 ImmediateExecution。 + """ + + def __init__(self, ttl_bars: int = 5) -> None: + self._ttl_bars = max(1, ttl_bars) + self._fallback = ImmediateExecution() + + def execute( + self, + signal: Signal, + df: pd.DataFrame, + bar_idx: int, + cash: float, + position: float, + position_mode: str, + commission: float, + min_commission: float, + stamp_tax: float, + slippage_model: SlippageModel | None, + ) -> list[Trade]: + if signal.price is None: + return self._fallback.execute( + signal, df, bar_idx, cash, position, position_mode, + commission, min_commission, stamp_tax, slippage_model, + ) + + target_price = signal.price + + for i in range(self._ttl_bars): + exec_idx = bar_idx + 1 + i + if exec_idx >= len(df): + break + row = df.iloc[exec_idx] + triggered = False + if signal.direction == "BUY" and float(row["low"]) <= target_price: + triggered = True + elif signal.direction == "SELL" and float(row["high"]) >= target_price: + triggered = True + + if triggered: + if signal.direction == "BUY": + size = self._calc_buy_size( + signal.size, target_price, cash, position_mode, commission, + ) + if size <= 0: + return [] + comm = self._calc_commission( + size, target_price, False, commission, min_commission, stamp_tax, + ) + slip = self._calc_slippage(size, target_price, False, slippage_model, df) + else: + size = signal.size if signal.size > 0 else position + if size <= 0: + return [] + if size > position: + size = position + comm = self._calc_commission( + size, target_price, True, commission, min_commission, stamp_tax, + ) + slip = self._calc_slippage(size, target_price, True, slippage_model, df) + + dt_raw = row["datetime"] + dt_int = int(dt_raw.strftime("%Y%m%d")) if hasattr(dt_raw, "strftime") else int(dt_raw) + return [Trade( + datetime=dt_int, + direction=signal.direction, + size=float(size), + price=target_price, + commission=comm, + slippage=slip, + )] + + return [] # TTL 内未触发 +``` + +- [ ] **Step 4: Run tests** + +Run: `python -m pytest tests/unit/test_backtest_execution.py -v --no-header` +Expected: All pass + +- [ ] **Step 5: Commit** + +```bash +git add src/easy_tdx/backtest/execution.py tests/unit/test_backtest_execution.py +git commit -m "feat(backtest): add LimitExecution" +``` + +--- + +### Task 7: BacktestEngine 集成 SlippageModel + ExecutionModel + +**Files:** +- Modify: `src/easy_tdx/backtest/engine.py` +- Modify: `tests/unit/test_backtest_engine.py` + +- [ ] **Step 1: Write tests** + +在 `tests/unit/test_backtest_engine.py` 追加： + +```python +from easy_tdx.backtest.slippage import FixedSlippage, SquareRootSlippage +from easy_tdx.backtest.execution import TWAPExecution + + +class TestEngineSlippageModel: + """BacktestEngine 与 SlippageModel 集成。""" + + def test_engine_with_slippage_model(self) -> None: + """引擎使用 SlippageModel 代替固定滑点。""" + class SimpleBuy(Strategy): + def init(self) -> None: + pass + def next(self) -> None: + if self._bar_index == 0: + self.buy(size=100) + + df = _make_df(20) + engine = BacktestEngine( + SimpleBuy, + cash=100000, + slippage_model=FixedSlippage(per_share=0.05), + ) + result = engine.run(df) + # 应产生交易，且有滑点 + buy_trades = result.trades[result.trades["direction"] == "BUY"] + if len(buy_trades) > 0: + assert buy_trades.iloc[0]["slippage"] > 0 + + +class TestEngineExecutionModel: + """BacktestEngine 与 ExecutionModel 集成。""" + + def test_engine_with_twap(self) -> None: + """引擎使用 TWAP 执行。""" + class SimpleBuy(Strategy): + def init(self) -> None: + pass + def next(self) -> None: + if self._bar_index == 0: + self.buy(size=300) + + df = _make_df(20) + engine = BacktestEngine( + SimpleBuy, + cash=100000, + execution_model=TWAPExecution(n_bars=3), + ) + result = engine.run(df) + buy_trades = result.trades[result.trades["direction"] == "BUY"] + assert len(buy_trades) >= 1 # TWAP 产生多笔交易 + + def test_engine_backward_compatible(self) -> None: + """无新参数时行为不变。""" + class SimpleBuy(Strategy): + def init(self) -> None: + pass + def next(self) -> None: + if self._bar_index == 0: + self.buy(size=100) + + df = _make_df(20) + engine = BacktestEngine(SimpleBuy, cash=100000) + result = engine.run(df) + assert len(result.trades) >= 1 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/unit/test_backtest_engine.py -v -k "TestEngineSlippage or TestEngineExecution" --no-header` +Expected: FAIL + +- [ ] **Step 3: Modify engine.py** + +Changes: + +1. Add imports at top of `engine.py`: +```python +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +import pandas as pd + +from easy_tdx.backtest.orders import OrderSimulator +from easy_tdx.backtest.performance import PerformanceAnalyzer +from easy_tdx.backtest.portfolio import PortfolioTracker +from easy_tdx.backtest.strategy import Strategy +from easy_tdx.backtest.types import BacktestResult, Signal, Trade + +if TYPE_CHECKING: + from easy_tdx.backtest.execution import ExecutionModel + from easy_tdx.backtest.slippage import SlippageModel +``` + +2. Add parameters to `BacktestEngine.__init__`: +```python +def __init__( + self, + strategy: type[Strategy] | Strategy, + cash: float = 100000.0, + commission: float = 0.0003, + min_commission: float = 5.0, + stamp_tax: float = 0.001, + slippage: float = 0.0, + execution: str = "next_open", + position_mode: str = "full", + reject_policy: str = "reduce", + benchmark: pd.DataFrame | None = None, + chanlun_level: str | None = None, + slippage_model: SlippageModel | None = None, # NEW + execution_model: ExecutionModel | None = None, # NEW +): + # ... store as self._slippage_model and self._execution_model +``` + +3. Modify `run()` method to support execution_model path: +```python +def run(self, df, chanlun_result=None): + # ... existing validation ... + + signals = self._generate_signals(df, chanlun_result) + + if self._execution_model is not None: + # NEW: use execution model + trades = self._execute_with_model(signals, df) + else: + # EXISTING: use OrderSimulator + simulator = OrderSimulator( + df, execution=self._execution, position_mode=self._position_mode, + reject_policy=self._reject_policy, commission=self._commission, + min_commission=self._min_commission, stamp_tax=self._stamp_tax, + slippage=self._slippage, + slippage_model=self._slippage_model, # NEW + ) + trades = simulator.simulate(signals=signals, cash=self._cash, position=0.0) + + # ... rest unchanged ... +``` + +4. Add `_execute_with_model` method: +```python +def _execute_with_model(self, signals: list[Signal], df: pd.DataFrame) -> list[Trade]: + """Use ExecutionModel to process signals.""" + assert self._execution_model is not None + all_trades: list[Trade] = [] + cash = self._cash + position = 0.0 + + for signal in signals: + bar_idx = self._find_bar_index(df, signal.datetime) + if bar_idx is None: + continue + sub_trades = self._execution_model.execute( + signal=signal, df=df, bar_idx=bar_idx, + cash=cash, position=position, position_mode=self._position_mode, + commission=self._commission, min_commission=self._min_commission, + stamp_tax=self._stamp_tax, + slippage_model=self._slippage_model, + ) + for t in sub_trades: + if not t.rejected: + if t.direction == "BUY": + cash -= t.size * t.price + t.commission + t.slippage + position += t.size + else: + cash += t.size * t.price - t.commission - t.slippage + position -= t.size + all_trades.extend(sub_trades) + return all_trades + +def _find_bar_index(self, df: pd.DataFrame, datetime_val: int) -> int | None: + """Find bar index for a datetime value.""" + dt_col = df["datetime"] + try: + idx = (dt_col == datetime_val).idxmax() if (dt_col == datetime_val).any() else None + if idx is not None: + return int(idx) + except (TypeError, ValueError): + pass + if hasattr(dt_col, "dt"): + dt_ints = dt_col.dt.strftime("%Y%m%d").astype(int) + mask = dt_ints == datetime_val + if mask.any(): + return int(mask.idxmax()) + return None +``` + +5. Also pass `slippage_model` to OrderSimulator in the existing path (when no execution_model). + +- [ ] **Step 4: Run all backtest tests** + +Run: `python -m pytest tests/unit/test_backtest_engine.py -v --no-header` +Expected: All existing + 3 new tests pass + +- [ ] **Step 5: Run full test suite** + +Run: `python -m pytest tests/unit/ -v --no-header -q` +Expected: All tests pass + +- [ ] **Step 6: Commit** + +```bash +git add src/easy_tdx/backtest/engine.py tests/unit/test_backtest_engine.py +git commit -m "feat(backtest): integrate SlippageModel + ExecutionModel into BacktestEngine" +``` + +--- + +### Task 8: 版本号 + 文档更新 + 最终验证 + +**Files:** +- Modify: `pyproject.toml` +- Modify: `CHANGELOG.md` (if exists) + +- [ ] **Step 1: Bump version** + +Update `pyproject.toml` version from `1.13.0` to `1.14.0`. + +- [ ] **Step 2: Run full test suite** + +Run: `python -m pytest tests/unit/ -v --no-header -q` +Expected: All tests pass (498 existing + ~55 new) + +- [ ] **Step 3: Run lint + type check** + +Run: `ruff check src/easy_tdx/backtest/slippage.py src/easy_tdx/backtest/execution.py src/easy_tdx/backtest/orders.py src/easy_tdx/backtest/engine.py` +Run: `ruff format --check src/easy_tdx/backtest/` +Expected: No errors + +- [ ] **Step 4: Final commit** + +```bash +git add pyproject.toml +git commit -m "chore: bump version to v1.14.0" +``` diff --git a/docs/superpowers/plans/2026-06-12-v1.15.0-attribution.md b/docs/superpowers/plans/2026-06-12-v1.15.0-attribution.md new file mode 100644 index 0000000..832a3e2 --- /dev/null +++ b/docs/superpowers/plans/2026-06-12-v1.15.0-attribution.md @@ -0,0 +1,540 @@ +# v1.15.0 归因分析 实施计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** 新增归因分析模块，支持 Brinson 归因（配置 vs 选股）、因子归因、成本归因。 + +**Architecture:** 新增 `backtest/attribution.py`，纯 pandas/numpy 计算，与现有 `FactorEngine` 无缝衔接。 + +**Tech Stack:** 纯 numpy/pandas，无新外部依赖。 + +--- + +## 文件结构 + +| 文件 | 操作 | 职责 | +|------|------|------| +| `src/easy_tdx/backtest/attribution.py` | 新增 | AttributionReport + AttributionAnalyzer | +| `tests/unit/test_backtest_attribution.py` | 新增 | 归因分析测试（~20 个） | + +--- + +### Task 1: AttributionReport + cost_attribution + brinson_attribution + factor_attribution + +**Files:** +- Create: `src/easy_tdx/backtest/attribution.py` +- Create: `tests/unit/test_backtest_attribution.py` + +- [ ] **Step 1: Write implementation** + +```python +"""归因分析模块。""" +from __future__ import annotations + +from dataclasses import dataclass, field + +import numpy as np +import pandas as pd + + +@dataclass +class AttributionReport: + """归因分析报告。""" + + total_return: float = 0.0 + # Brinson 归因 + allocation_return: float = 0.0 + selection_return: float = 0.0 + interaction_return: float = 0.0 + # 因子归因 + factor_returns: dict[str, float] = field(default_factory=dict) + specific_return: float = 0.0 + # 成本归因 + total_trade_cost: float = 0.0 + slippage_cost: float = 0.0 + commission_cost: float = 0.0 + stamp_tax_cost: float = 0.0 + + +class AttributionAnalyzer: + """收益归因分析器。 + + 支持三种归因视角： + 1. 成本归因：分解交易成本的来源（佣金/滑点/印花税） + 2. Brinson 归因：分解超额收益（配置 vs 选股） + 3. 因子归因：分解收益为因子贡献 + 特质收益 + """ + + def __init__( + self, + trades: pd.DataFrame, + equity_curve: pd.DataFrame, + benchmark: pd.DataFrame | None = None, + factor_exposures: pd.DataFrame | None = None, + factor_returns: pd.DataFrame | None = None, + groups: pd.DataFrame | None = None, + ) -> None: + self._trades = trades + self._equity_curve = equity_curve + self._benchmark = benchmark + self._factor_exposures = factor_exposures + self._factor_returns = factor_returns + self._groups = groups + + def cost_attribution(self) -> AttributionReport: + """成本归因：分解交易成本。""" + if self._trades.empty: + return AttributionReport() + + valid = self._trades[~self._trades["rejected"]] if "rejected" in self._trades.columns else self._trades + + slippage_cost = float(valid["slippage"].sum()) if "slippage" in valid.columns else 0.0 + commission_cost = float(valid["commission"].sum()) if "commission" in valid.columns else 0.0 + + # 总成本 = 滑点 + 佣金（佣金内含印花税） + total_trade_cost = slippage_cost + commission_cost + + # 估算印花税（卖出交易 0.1%） + sell_mask = valid["direction"] == "SELL" if "direction" in valid.columns else pd.Series(dtype=bool) + stamp_tax_cost = 0.0 + if sell_mask.any(): + sell_trades = valid[sell_mask] + if "price" in sell_trades.columns and "size" in sell_trades.columns: + stamp_tax_cost = float((sell_trades["price"] * sell_trades["size"] * 0.001).sum()) + + total_return = 0.0 + if not self._equity_curve.empty and "total" in self._equity_curve.columns: + total_arr = self._equity_curve["total"].to_numpy() + if len(total_arr) >= 2 and total_arr[0] > 0: + total_return = float((total_arr[-1] / total_arr[0]) - 1) + + return AttributionReport( + total_return=total_return, + total_trade_cost=total_trade_cost, + slippage_cost=slippage_cost, + commission_cost=commission_cost, + stamp_tax_cost=stamp_tax_cost, + ) + + def brinson_attribution(self) -> AttributionReport: + """Brinson-Hood-Beebower 归因分解。 + + Total = Allocation + Selection + Interaction + R_p = Σ(w_pi × R_pi) 组合收益 + R_b = Σ(w_bi × R_bi) 基准收益 + Allocation = Σ((w_pi - w_bi) × R_bi) + Selection = Σ(w_bi × (R_pi - R_bi)) + Interaction = Σ((w_pi - w_bi) × (R_pi - R_bi)) + + 需要提供 benchmark 参数。 + 如果没有 benchmark，只返回 total_return。 + """ + cost_report = self.cost_attribution() + + if self._benchmark is None: + return cost_report + + # 简化 Brinson：使用 equity_curve 估算 + if self._equity_curve.empty: + return cost_report + + total_arr = self._equity_curve["total"].to_numpy() + if len(total_arr) < 2 or total_arr[0] <= 0: + return cost_report + + portfolio_return = float((total_arr[-1] / total_arr[0]) - 1) + + # 基准收益 + benchmark_return = 0.0 + if "total" in self._benchmark.columns: + bench_arr = self._benchmark["total"].to_numpy() + if len(bench_arr) >= 2 and bench_arr[0] > 0: + benchmark_return = float((bench_arr[-1] / bench_arr[0]) - 1) + + excess_return = portfolio_return - benchmark_return + + # 如果有 groups 信息，按组计算 + allocation = 0.0 + selection = 0.0 + interaction = 0.0 + + if self._groups is not None and not self._groups.empty: + # 按组分解（简化版） + allocation, selection, interaction = self._compute_grouped_brinson( + portfolio_return, benchmark_return, + ) + else: + # 无分组信息时，将全部超额收益归为 selection + selection = excess_return + + return AttributionReport( + total_return=portfolio_return, + allocation_return=allocation, + selection_return=selection, + interaction_return=interaction, + total_trade_cost=cost_report.total_trade_cost, + slippage_cost=cost_report.slippage_cost, + commission_cost=cost_report.commission_cost, + stamp_tax_cost=cost_report.stamp_tax_cost, + ) + + def _compute_grouped_brinson( + self, portfolio_return: float, benchmark_return: float, + ) -> tuple[float, float, float]: + """按组计算 Brinson 归因（简化版）。 + + 当 groups 包含 weight 和 return 列时进行分解。 + """ + if self._groups is None or self._groups.empty: + return 0.0, portfolio_return - benchmark_return, 0.0 + + allocation = 0.0 + selection = 0.0 + interaction = 0.0 + + if "portfolio_weight" in self._groups.columns and "benchmark_weight" in self._groups.columns: + pw = self._groups["portfolio_weight"].to_numpy() + bw = self._groups["benchmark_weight"].to_numpy() + + if "portfolio_return" in self._groups.columns and "benchmark_return" in self._groups.columns: + pr = self._groups["portfolio_return"].to_numpy() + br = self._groups["benchmark_return"].to_numpy() + + allocation = float(np.sum((pw - bw) * br)) + selection = float(np.sum(bw * (pr - br))) + interaction = float(np.sum((pw - bw) * (pr - br))) + + return allocation, selection, interaction + + def factor_attribution(self) -> AttributionReport: + """因子归因分解。 + + R = Σ(β_i × f_i) + α + β_i: 因子暴露度 + f_i: 因子收益率 + α: 特质收益 + + 需要提供 factor_exposures 和 factor_returns。 + """ + cost_report = self.cost_attribution() + + if self._factor_exposures is None or self._factor_returns is None: + return cost_report + + if self._factor_exposures.empty or self._factor_returns.empty: + return cost_report + + # 计算因子贡献 + factor_contributions: dict[str, float] = {} + + # 简化：按列名匹配 + common_factors = set(self._factor_exposures.columns) & set(self._factor_returns.columns) + for factor_name in common_factors: + exposures = self._factor_exposures[factor_name].to_numpy() + returns = self._factor_returns[factor_name].to_numpy() + min_len = min(len(exposures), len(returns)) + if min_len > 0: + contrib = float(np.sum(exposures[:min_len] * returns[:min_len])) + factor_contributions[factor_name] = contrib + + total_factor_return = sum(factor_contributions.values()) + + # 总收益 + total_arr = self._equity_curve["total"].to_numpy() + total_return = 0.0 + if len(total_arr) >= 2 and total_arr[0] > 0: + total_return = float((total_arr[-1] / total_arr[0]) - 1) + + specific_return = total_return - total_factor_return + + return AttributionReport( + total_return=total_return, + factor_returns=factor_contributions, + specific_return=specific_return, + total_trade_cost=cost_report.total_trade_cost, + slippage_cost=cost_report.slippage_cost, + commission_cost=cost_report.commission_cost, + stamp_tax_cost=cost_report.stamp_tax_cost, + ) + + def full_report(self) -> AttributionReport: + """完整归因报告。 + + 按优先级使用： + 1. 因子归因（如果 factor_exposures/factor_returns 可用） + 2. Brinson 归因（如果 benchmark 可用） + 3. 成本归因（始终可用） + """ + if self._factor_exposures is not None and self._factor_returns is not None: + return self.factor_attribution() + if self._benchmark is not None: + return self.brinson_attribution() + return self.cost_attribution() +``` + +- [ ] **Step 2: Write tests** + +```python +"""归因分析单元测试。""" +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from easy_tdx.backtest.attribution import AttributionAnalyzer, AttributionReport + + +def _make_trades( + n_buys: int = 2, n_sells: int = 2, + commission: float = 10.0, slippage: float = 5.0, +) -> pd.DataFrame: + """构造测试交易记录。""" + trades: list[dict[str, object]] = [] + for i in range(n_buys): + trades.append({ + "datetime": 20240101 + i, "direction": "BUY", + "size": 100, "price": 100.0 + i, + "commission": commission, "slippage": slippage, "pnl": 0.0, "rejected": False, + }) + for i in range(n_sells): + trades.append({ + "datetime": 20240110 + i, "direction": "SELL", + "size": 100, "price": 110.0 + i, + "commission": commission, "slippage": slippage, "pnl": 500.0, "rejected": False, + }) + return pd.DataFrame(trades) + + +def _make_equity(initial: float = 100000.0, final: float = 110000.0, n: int = 20) -> pd.DataFrame: + """构造资金曲线。""" + total = np.linspace(initial, final, n) + return pd.DataFrame({ + "datetime": [20240101 + i for i in range(n)], + "total": total, + "cash": total * 0.5, + "position_value": total * 0.5, + }) + + +def _make_benchmark(initial: float = 100000.0, final: float = 105000.0, n: int = 20) -> pd.DataFrame: + """构造基准资金曲线。""" + total = np.linspace(initial, final, n) + return pd.DataFrame({ + "datetime": [20240101 + i for i in range(n)], + "total": total, + }) + + +class TestCostAttribution: + """成本归因。""" + + def test_basic_cost_breakdown(self) -> None: + """基本成本分解。""" + trades = _make_trades(n_buys=2, n_sells=2, commission=10.0, slippage=5.0) + eq = _make_equity() + analyzer = AttributionAnalyzer(trades, eq) + report = analyzer.cost_attribution() + # 4 trades × 10.0 commission = 40.0 + assert report.commission_cost == pytest.approx(40.0) + # 4 trades × 5.0 slippage = 20.0 + assert report.slippage_cost == pytest.approx(20.0) + # total = 60.0 + assert report.total_trade_cost == pytest.approx(60.0) + + def test_total_return(self) -> None: + """总收益计算。""" + trades = _make_trades() + eq = _make_equity(100000.0, 110000.0) + analyzer = AttributionAnalyzer(trades, eq) + report = analyzer.cost_attribution() + assert report.total_return == pytest.approx(0.1) + + def test_empty_trades(self) -> None: + """空交易记录。""" + trades = pd.DataFrame(columns=["datetime", "direction", "size", "price", "commission", "slippage"]) + eq = _make_equity() + analyzer = AttributionAnalyzer(trades, eq) + report = analyzer.cost_attribution() + assert report.total_trade_cost == 0.0 + assert report.slippage_cost == 0.0 + + def test_stamp_tax_estimation(self) -> None: + """印花税估算（卖出 0.1%）。""" + trades = _make_trades(n_buys=0, n_sells=1, commission=0.0, slippage=0.0) + eq = _make_equity() + analyzer = AttributionAnalyzer(trades, eq) + report = analyzer.cost_attribution() + # 卖出 100 股 × 110 元 × 0.001 = 11.0 + assert report.stamp_tax_cost == pytest.approx(11.0) + + +class TestBrinsonAttribution: + """Brinson 归因。""" + + def test_no_benchmark_returns_only_total(self) -> None: + """无基准时只返回总收益。""" + trades = _make_trades() + eq = _make_equity() + analyzer = AttributionAnalyzer(trades, eq, benchmark=None) + report = analyzer.brinson_attribution() + assert report.total_return == pytest.approx(0.1) + assert report.allocation_return == 0.0 + assert report.selection_return == 0.0 + + def test_with_benchmark_selection(self) -> None: + """有基准时超额收益归为 selection。""" + trades = _make_trades() + eq = _make_equity(100000.0, 110000.0) # +10% + bench = _make_benchmark(100000.0, 105000.0) # +5% + analyzer = AttributionAnalyzer(trades, eq, benchmark=bench) + report = analyzer.brinson_attribution() + assert report.total_return == pytest.approx(0.1) + # excess = 10% - 5% = 5%, all attributed to selection + assert report.selection_return == pytest.approx(0.05) + + def test_with_groups_decomposition(self) -> None: + """有分组时进行 Brinson 三因子分解。""" + trades = _make_trades() + eq = _make_equity(100000.0, 110000.0) + bench = _make_benchmark(100000.0, 105000.0) + groups = pd.DataFrame({ + "portfolio_weight": [0.6, 0.4], + "benchmark_weight": [0.5, 0.5], + "portfolio_return": [0.15, 0.05], + "benchmark_return": [0.10, 0.0], + }) + analyzer = AttributionAnalyzer(trades, eq, benchmark=bench, groups=groups) + report = analyzer.brinson_attribution() + # Allocation = (0.6-0.5)*0.10 + (0.4-0.5)*0.0 = 0.01 + assert report.allocation_return == pytest.approx(0.01) + # Selection = 0.5*(0.15-0.10) + 0.5*(0.05-0.0) = 0.05 + assert report.selection_return == pytest.approx(0.05) + # Interaction = (0.1)*0.05 + (-0.1)*0.05 = 0.0 + assert report.interaction_return == pytest.approx(0.0) + + +class TestFactorAttribution: + """因子归因。""" + + def test_no_factors_returns_only_cost(self) -> None: + """无因子数据时只返回成本归因。""" + trades = _make_trades() + eq = _make_equity() + analyzer = AttributionAnalyzer(trades, eq) + report = analyzer.factor_attribution() + assert report.factor_returns == {} + assert report.specific_return == 0.0 + + def test_basic_factor_decomposition(self) -> None: + """基本因子分解。""" + trades = _make_trades() + eq = _make_equity(100000.0, 110000.0) + exposures = pd.DataFrame({ + "momentum": [0.5, 0.3, 0.2], + "volatility": [0.1, -0.1, 0.0], + }) + returns = pd.DataFrame({ + "momentum": [0.05, 0.03, 0.02], + "volatility": [0.01, -0.02, 0.0], + }) + analyzer = AttributionAnalyzer( + trades, eq, + factor_exposures=exposures, factor_returns=returns, + ) + report = analyzer.factor_attribution() + # momentum: sum(0.5*0.05, 0.3*0.03, 0.2*0.02) = 0.025+0.009+0.004 = 0.038 + assert report.factor_returns["momentum"] == pytest.approx(0.038) + # volatility: sum(0.1*0.01, -0.1*-0.02, 0*0) = 0.001+0.002+0 = 0.003 + assert report.factor_returns["volatility"] == pytest.approx(0.003) + # total_return = 0.1 + # specific = 0.1 - 0.038 - 0.003 = 0.059 + assert report.specific_return == pytest.approx(0.059) + + def test_empty_factor_data(self) -> None: + """空因子数据。""" + trades = _make_trades() + eq = _make_equity() + exposures = pd.DataFrame() + returns = pd.DataFrame() + analyzer = AttributionAnalyzer( + trades, eq, + factor_exposures=exposures, factor_returns=returns, + ) + report = analyzer.factor_attribution() + assert report.factor_returns == {} + + +class TestFullReport: + """完整报告。""" + + def test_prefers_factor_over_brinson(self) -> None: + """有因子数据时优先使用因子归因。""" + trades = _make_trades() + eq = _make_equity(100000.0, 110000.0) + bench = _make_benchmark(100000.0, 105000.0) + exposures = pd.DataFrame({"momentum": [0.5]}) + returns = pd.DataFrame({"momentum": [0.05]}) + analyzer = AttributionAnalyzer( + trades, eq, benchmark=bench, + factor_exposures=exposures, factor_returns=returns, + ) + report = analyzer.full_report() + assert "momentum" in report.factor_returns + assert report.specific_return != 0.0 # 因子归因有 specific + + def test_falls_back_to_cost_only(self) -> None: + """无基准无因子时只返回成本归因。""" + trades = _make_trades() + eq = _make_equity() + analyzer = AttributionAnalyzer(trades, eq) + report = analyzer.full_report() + assert report.total_trade_cost > 0 + assert report.factor_returns == {} + assert report.allocation_return == 0.0 +``` + +- [ ] **Step 3: Run tests** + +```bash +python -m pytest tests/unit/test_backtest_attribution.py -v --no-header +``` + +- [ ] **Step 4: ruff check** + +```bash +ruff check src/easy_tdx/backtest/attribution.py tests/unit/test_backtest_attribution.py +ruff format --check src/easy_tdx/backtest/attribution.py tests/unit/test_backtest_attribution.py +``` + +- [ ] **Step 5: Full test suite** + +```bash +python -m pytest tests/unit/ -q --no-header +``` + +- [ ] **Step 6: Commit** + +```bash +git add src/easy_tdx/backtest/attribution.py tests/unit/test_backtest_attribution.py +git commit -m "feat(backtest): add AttributionAnalyzer with Brinson, factor, cost attribution" +``` + +--- + +### Task 2: 版本号 bump + 最终验证 + +- [ ] **Step 1**: Update pyproject.toml version from `1.14.0` to `1.15.0` + +- [ ] **Step 2**: Run full test suite + +```bash +python -m pytest tests/unit/ -q --no-header +``` + +- [ ] **Step 3**: Commit + +```bash +git add pyproject.toml +git commit -m "chore: bump version to v1.15.0" +``` diff --git a/docs/superpowers/specs/2026-06-12-advanced-backtest-design.md b/docs/superpowers/specs/2026-06-12-advanced-backtest-design.md new file mode 100644 index 0000000..d0c1837 --- /dev/null +++ b/docs/superpowers/specs/2026-06-12-advanced-backtest-design.md @@ -0,0 +1,355 @@ +# 高级回测增强 — 设计文档 + +> **日期**: 2026-06-12 +> **版本**: v1.0 +> **前置**: 方案 A（v1.11.0–v1.13.0）已完成 +> **范围**: 方案 B — 滑点建模、执行仿真、归因分析 +> **目标市场**: 纯 A 股 + +## 1. 背景与目标 + +easy-tdx 的回测引擎（`BacktestEngine` + `OrderSimulator`）已支持基础信号→撮合→绩效管道，但成本建模过于简单（固定每股滑点），执行假设过于理想（瞬间成交），且无收益归因能力。 + +本设计在**不破坏现有 API** 的前提下，新增三个核心能力： + +1. **可插拔滑点模型** — 从固定滑点升级为市场冲击模型（方根模型、成交量比例等） +2. **执行仿真引擎** — 支持大额订单拆分（TWAP/VWAP）、限价单等真实执行方式 +3. **归因分析** — Brinson 归因（配置 vs 选股）、因子归因（收益分解为因子贡献） + +## 2. 模块总览 + +``` +src/easy_tdx/backtest/ +├── slippage.py # 新增：可插拔滑点模型（4 种） +├── execution.py # 新增：执行仿真引擎（4 种） +├── attribution.py # 新增：归因分析（Brinson + 因子 + 成本） +├── engine.py # 修改：接入 slippage_model / execution_model +├── orders.py # 修改：用 SlippageModel 替代固定滑点 +├── performance.py # 不变 +├── strategy.py # 不变 +├── types.py # 修改：新增 AttributionReport +├── portfolio.py # 不变 +├── portfolio_engine.py # 不变 +└── combo.py # 不变 +``` + +### 依赖关系 + +``` +Signal → ExecutionModel → SlippageModel → Trade + ↓ + AttributionAnalyzer → AttributionReport +``` + +## 3. 滑点建模（`slippage.py`） + +### 3.1 基类 + +```python +class SlippageModel(ABC): + """滑点模型基类。""" + + @abstractmethod + def compute( + self, + price: float, + size: float, + volume: float, + volatility: float, + direction: str, + ) -> float: + """返回总滑点成本（金额）。 + + Args: + price: 成交价 + size: 订单数量（股） + volume: 当日成交量（股），0 表示无数据 + volatility: 近期年化波动率，0 表示无数据 + direction: BUY / SELL + """ + ... +``` + +### 3.2 四种内置模型 + +| 模型 | 公式 | 适用场景 | +|------|------|---------| +| `FixedSlippage(per_share=0.01)` | `size × per_share` | 向后兼容，快速原型 | +| `PercentSlippage(rate=0.001)` | `price × size × rate` | 按成交金额百分比 | +| `SquareRootSlippage(impact_coeff=0.1)` | `σ × √(Q/V) × price × Q × coeff` | A 股量化主流，参与率高时冲击大 | +| `VolumeSlippage(base_bps=10.0)` | `base_bps/10000 × (size/volume) × price × size` | 基于成交量比例，流动性差时成本高 | + +### 3.3 SquareRootSlippage 详解 + +``` +participation_rate = size / volume # 参与率 +impact = volatility × √(participation_rate) × price × size × impact_coeff +``` + +- 当 `volume=0` 或 `volatility=0` 时，退化为 `PercentSlippage(rate=0.001)` +- `impact_coeff` 默认 0.1，对应 A 股中小盘股的经验值 +- 参与率 > 5% 时冲击成本显著增大（√ 函数的自然效果） + +### 3.4 集成点 + +`OrderSimulator` 新增参数： + +```python +slippage_model: SlippageModel | None = None +``` + +当 `slippage_model` 非空时，忽略原有 `self.slippage` 参数，调用 `slippage_model.compute()` 计算滑点。 + +`BacktestEngine` 透传： + +```python +BacktestEngine(strategy, slippage_model=SquareRootSlippage()) +``` + +当同时提供 `slippage_model` 和 `slippage` 时，`slippage_model` 优先。 + +## 4. 执行仿真（`execution.py`） + +### 4.1 基类 + +```python +class ExecutionModel(ABC): + """执行仿真基类。""" + + @abstractmethod + def execute( + self, + signal: Signal, + df: pd.DataFrame, + bar_idx: int, + cash: float, + position: float, + position_mode: str, + commission: float, + min_commission: float, + stamp_tax: float, + slippage_model: SlippageModel | None, + ) -> list[Trade]: + """将信号转换为一笔或多笔成交记录。""" + ... +``` + +### 4.2 四种内置模型 + +| 模型 | 行为 | 适用场景 | +|------|------|---------| +| `ImmediateExecution` | 现有行为，下一 bar 即时成交 | 向后兼容 | +| `TWAPExecution(n_bars=5)` | 将订单均匀拆分为 N 份，在连续 N bar 执行 | 大额订单分批建仓 | +| `VWAPExecution(n_bars=5, volume_lookback=20)` | 按历史成交量分布比例拆分 | 追踪 VWAP 基准 | +| `LimitExecution(ttl_bars=5)` | 限价挂单，仅当价格触及才成交 | 精确入场价位控制 | + +### 4.3 TWAPExecution 详解 + +```python +class TWAPExecution(ExecutionModel): + def __init__(self, n_bars: int = 5) -> None: + self.n_bars = n_bars + + def execute(self, signal, df, bar_idx, ...): + sub_size = total_size / n_bars + trades = [] + for i in range(n_bars): + exec_bar = bar_idx + 1 + i + if exec_bar >= len(df): + break # 超出数据范围，剩余未执行 + price = df["close"].iloc[exec_bar] # 按 close 执行 + trade = self._make_trade(signal, sub_size, price, exec_bar, ...) + trades.append(trade) + return trades +``` + +- 买入时使用 `position_mode` 确定总数量，然后均匀拆分 +- 卖出时直接拆分持仓 +- 每笔子交易独立计算佣金和滑点 +- 100 股整手约束：每笔子交易向下取整到 100 的倍数 + +### 4.4 VWAPExecution 详解 + +```python +class VWAPExecution(ExecutionModel): + def __init__(self, n_bars: int = 5, volume_lookback: int = 20) -> None: ... + + def execute(self, signal, df, bar_idx, ...): + # 取最近 volume_lookback 根 K 线的成交量分布 + lookback = df.iloc[max(0, bar_idx - volume_lookback):bar_idx + 1] + avg_volumes = [] + for i in range(n_bars): + offset = i % len(lookback) + avg_volumes.append(float(lookback["volume"].iloc[-(offset + 1)])) + total_vol = sum(avg_volumes) + weights = [v / total_vol for v in avg_volumes] + # 按 weights 拆分订单 + ... +``` + +### 4.5 LimitExecution 详解 + +```python +class LimitExecution(ExecutionModel): + def __init__(self, ttl_bars: int = 5) -> None: + self.ttl_bars = ttl_bars # 限价单有效期（bar 数） + + def execute(self, signal, df, bar_idx, ...): + if signal.price is None: + # 无限价，退化为即时执行 + return ImmediateExecution().execute(...) + target_price = signal.price + trades = [] + for i in range(self.ttl_bars): + exec_bar = bar_idx + 1 + i + if exec_bar >= len(df): + break + row = df.iloc[exec_bar] + if signal.direction == "BUY" and row["low"] <= target_price: + trades.append(self._make_trade(signal, size, target_price, exec_bar, ...)) + break + elif signal.direction == "SELL" and row["high"] >= target_price: + trades.append(self._make_trade(signal, size, target_price, exec_bar, ...)) + break + return trades # 可能返回空列表（限价未触发） +``` + +### 4.6 集成点 + +`BacktestEngine` 新增参数： + +```python +execution_model: ExecutionModel | None = None +``` + +当 `execution_model` 非空时，信号处理从执行模型走，不走原有 `_resolve_exec_index` / `_get_price`。 + +**关键：执行模型产生多笔 Trade，需要修正 `BacktestEngine._generate_signals` 的信号循环逻辑**。 + +现有逻辑： + +```python +for signal in signals: + trades = simulator.simulate([signal], cash, position) +``` + +新逻辑（当 execution_model 存在时）： + +```python +for signal in signals: + sub_trades = execution_model.execute(signal, df, bar_idx, cash, position, ...) + all_trades.extend(sub_trades) +``` + +## 5. 归因分析（`attribution.py`） + +### 5.1 数据结构 + +```python +@dataclass +class AttributionReport: + """归因分析报告。""" + # 总收益 + total_return: float + # Brinson 归因 + allocation_return: float + selection_return: float + interaction_return: float + # 因子归因 + factor_returns: dict[str, float] + specific_return: float + # 成本归因 + total_trade_cost: float + slippage_cost: float + commission_cost: float + stamp_tax_cost: float +``` + +### 5.2 AttributionAnalyzer + +```python +class AttributionAnalyzer: + """收益归因分析器。""" + + def __init__( + self, + trades: pd.DataFrame, + equity_curve: pd.DataFrame, + benchmark: pd.DataFrame | None = None, + factor_exposures: pd.DataFrame | None = None, + factor_returns: pd.DataFrame | None = None, + groups: pd.DataFrame | None = None, + ) -> None: ... + + def brinson_attribution(self) -> AttributionReport: + """Brinson-Hood-Beebower 归因分解。 + + Total = Allocation + Selection + Interaction + R_p = Σ(w_pi × R_pi) # 组合收益 + R_b = Σ(w_bi × R_bi) # 基准收益 + Allocation = Σ((w_pi - w_bi) × R_bi) + Selection = Σ(w_bi × (R_pi - R_bi)) + Interaction = Σ((w_pi - w_bi) × (R_pi - R_bi)) + """ + + def factor_attribution(self) -> AttributionReport: + """因子归因分解。 + + R = Σ(β_i × f_i) + α + β_i: 因子暴露度 + f_i: 因子收益率 + α: 特质收益 + """ + + def cost_attribution(self) -> AttributionReport: + """成本归因：分解佣金/滑点/印花税。""" + + def full_report(self) -> AttributionReport: + """完整归因报告。""" +``` + +### 5.3 与现有模块衔接 + +- `trades` 参数直接来自 `BacktestResult.trades` +- `equity_curve` 来自 `BacktestResult.equity_curve` +- `factor_exposures` / `factor_returns` 来自 `FactorEngine`（v1.11.0 已实现） +- `groups` 可用于 Brinson 分组（如行业分类），可选 + +## 6. 向后兼容策略 + +| 现有调用 | 行为 | +|---------|------| +| `BacktestEngine(strategy, slippage=0.01)` | 与现有行为完全一致 | +| `BacktestEngine(strategy)` | 无滑点，与现有行为一致 | +| `BacktestEngine(strategy, slippage_model=SquareRootSlippage())` | 使用新滑点模型 | +| `BacktestEngine(strategy, execution_model=TWAPExecution())` | 使用新执行引擎 | +| `OrderSimulator(df, slippage=0.01)` | 与现有行为完全一致 | +| `OrderSimulator(df, slippage_model=FixedSlippage(0.01))` | 等价 | + +**不变更的文件**: `strategy.py`, `performance.py`, `portfolio.py`, `combo.py` + +## 7. 版本计划 + +### v1.14.0 — 滑点 + 执行 + +- `slippage.py`: SlippageModel ABC + 4 种模型 +- `execution.py`: ExecutionModel ABC + 4 种模型 +- `orders.py`: 集成 SlippageModel +- `engine.py`: 集成 SlippageModel + ExecutionModel +- `types.py`: 无变更（Trade/Signal 已够用） +- 测试: ~35 个 + +### v1.15.0 — 归因分析 + +- `attribution.py`: AttributionAnalyzer + AttributionReport +- `types.py`: 新增 AttributionReport +- `performance.py`: 可选集成 AttributionAnalyzer +- CLI: `easy-tdx backtest attribution` 命令 +- 测试: ~20 个 + +## 8. 不做的事 + +- **订单簿仿真**：A 股 Level-2 数据获取困难，回测中用成交量比例代理 +- **融资融券**：需要额外保证金模型，超出当前范围 +- **期指/期权对冲**：超出纯 A 股范围 +- **高频仿真**：当前是日线级别回测，微秒级仿真不适用 diff --git a/pyproject.toml b/pyproject.toml index caac6e4..09f4f8f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "easy-tdx" -version = "1.15.0" +version = "1.11.1" description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步" readme = "README.md" requires-python = ">=3.10" diff --git a/src/easy_tdx/cli/__init__.py b/src/easy_tdx/cli/__init__.py index 6049d7f..9eb96f3 100644 --- a/src/easy_tdx/cli/__init__.py +++ b/src/easy_tdx/cli/__init__.py @@ -20,13 +20,13 @@ from .cmd_capital import capital_flow from .cmd_chanlun import chanlun from .cmd_ex import ex from .cmd_factor import factor -from .cmd_pfactor import pfactor from .cmd_finance import f10, fund_flow from .cmd_indicator import indicator, indicator_list from .cmd_info import server_info, symbol_info from .cmd_kline import kline from .cmd_monitor import market_stat, unusual from .cmd_offline import offline +from .cmd_pfactor import pfactor from .cmd_quote import quote, quote_list from .cmd_run_all import run_all from .cmd_tick import tick diff --git a/tests/unit/test_factor_analysis.py b/tests/unit/test_factor_analysis.py index 3fb2256..e8eed25 100644 --- a/tests/unit/test_factor_analysis.py +++ b/tests/unit/test_factor_analysis.py @@ -4,7 +4,6 @@ from __future__ import annotations import numpy as np import pandas as pd -import pytest from easy_tdx.factor.analysis import FactorAnalyzer, FactorReport diff --git a/tests/unit/test_factor_transform.py b/tests/unit/test_factor_transform.py index 0a02a22..37d1fe7 100644 --- a/tests/unit/test_factor_transform.py +++ b/tests/unit/test_factor_transform.py @@ -4,7 +4,6 @@ from __future__ import annotations import numpy as np import pandas as pd -import pytest from easy_tdx.factor.transform import ( fill_missing, diff --git a/tests/unit/test_portfolio_optimizer.py b/tests/unit/test_portfolio_optimizer.py index fc86ff9..b5f48d5 100644 --- a/tests/unit/test_portfolio_optimizer.py +++ b/tests/unit/test_portfolio_optimizer.py @@ -57,7 +57,11 @@ class TestRiskParity: assert abs(sum(w.values()) - 1.0) < 1e-6 def test_with_volatility_column(self): - scores = pd.DataFrame({"code": ["A", "B", "C"], "score": [1.0, 1.0, 1.0], "volatility": [0.1, 0.2, 0.4]}) + scores = pd.DataFrame({ + "code": ["A", "B", "C"], + "score": [1.0, 1.0, 1.0], + "volatility": [0.1, 0.2, 0.4], + }) w = RiskParityOptimizer().optimize(scores, n_stocks=3) assert w["A"] > w["C"] diff --git a/tests/unit/test_portfolio_rebalance.py b/tests/unit/test_portfolio_rebalance.py index 1fb03e5..e58f229 100644 --- a/tests/unit/test_portfolio_rebalance.py +++ b/tests/unit/test_portfolio_rebalance.py @@ -3,7 +3,6 @@ from __future__ import annotations import numpy as np import pandas as pd -import pytest from easy_tdx.portfolio.optimizer import EqualWeightOptimizer, FactorWeightedOptimizer from easy_tdx.portfolio.rebalance import RebalanceEngine @@ -27,7 +26,11 @@ def _make_market(n_stocks: int = 10, n_days: int = 120, seed: int = 42) -> dict[ class TestRebalanceEngine: def test_basic_run(self): - engine = RebalanceEngine(optimizer=EqualWeightOptimizer(), factor_name="momentum_20d", n_stocks=5, rebalance_freq="M", cash=1_000_000) + engine = RebalanceEngine( + optimizer=EqualWeightOptimizer(), + factor_name="momentum_20d", n_stocks=5, + rebalance_freq="M", cash=1_000_000, + ) result = engine.run(_make_market(), start_date=20240101, end_date=20240430) assert len(result.states) > 0 assert len(result.rebalance_dates) > 0 @@ -35,7 +38,10 @@ class TestRebalanceEngine: assert "total_return" in result.performance def test_with_factor_weighted(self): - engine = RebalanceEngine(optimizer=FactorWeightedOptimizer(), factor_name="momentum_20d", n_stocks=5) + engine = RebalanceEngine( + optimizer=FactorWeightedOptimizer(), + factor_name="momentum_20d", n_stocks=5, + ) result = engine.run(_make_market(), start_date=20240101, end_date=20240430) assert len(result.states) > 0 @@ -44,12 +50,16 @@ class TestRebalanceEngine: assert result.performance["total_return"] == 0.0 def test_equity_curve_dates_sorted(self): - result = RebalanceEngine(optimizer=EqualWeightOptimizer(), rebalance_freq="M").run(_make_market(), start_date=20240101, end_date=20240430) + result = RebalanceEngine( + optimizer=EqualWeightOptimizer(), rebalance_freq="M", + ).run(_make_market(), start_date=20240101, end_date=20240430) dates = result.equity_curve["datetime"].tolist() assert dates == sorted(dates) def test_trades_recorded(self): - engine = RebalanceEngine(optimizer=EqualWeightOptimizer(), n_stocks=3, rebalance_freq="M") + engine = RebalanceEngine( + optimizer=EqualWeightOptimizer(), n_stocks=3, rebalance_freq="M", + ) result = engine.run(_make_market(), start_date=20240101, end_date=20240430) assert len(result.trades) > 0 assert "BUY" in result.trades["direction"].values