easy_tdx_max/docs/superpowers/plans/2026-06-12-v1.15.0-attribution.md

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

"""归因分析模块。"""
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

"""归因分析单元测试。"""
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

python -m pytest tests/unit/test_backtest_attribution.py -v --no-header

• Step 4: ruff check

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

python -m pytest tests/unit/ -q --no-header

• Step 6: Commit

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

python -m pytest tests/unit/ -q --no-header

• Step 3: Commit

git add pyproject.toml
git commit -m "chore: bump version to v1.15.0"