docs: add quantitative guide, update README + CHANGELOG, bump v1.11.1

Co-Authored-By: Claude <noreply@anthropic.com>
2026-09-12 15:44:15 +08:00 · 2026-06-12 22:12:19 +08:00
parent bfefadf70b
commit a6ed0eac16
11 changed files with 3338 additions and 11 deletions
@@ -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"
+```
@@ -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 股范围
+- **高频仿真**：当前是日线级别回测，微秒级仿真不适用