feat(backtest): v1.15.0 — attribution analysis

- AttributionReport dataclass with Brinson/factor/cost fields
- AttributionAnalyzer with 3 attribution modes:
  - Cost attribution (commission/slippage/stamp tax breakdown)
  - Brinson attribution (allocation + selection + interaction)
  - Factor attribution (factor contribution + specific return)
- Priority: factor > Brinson > cost in full_report()
- 556 tests passing (12 new)

Co-Authored-By: Claude <noreply@anthropic.com>

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "easy-tdx"
-version = "1.14.0"
+version = "1.15.0"
 description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步"
 readme = "README.md"
 requires-python = ">=3.10"

src/easy_tdx/backtest/attribution.py

@@ -0,0 +1,217 @@
+"""归因分析模块。"""
+
+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 归因分解。"""
+        cost_report = self.cost_attribution()
+
+        if self._benchmark is None:
+            return cost_report
+
+        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)
+
+        if self._groups is not None and not self._groups.empty:
+            allocation, selection, interaction = self._compute_grouped_brinson(
+                portfolio_return,
+                benchmark_return,
+            )
+        else:
+            excess_return = portfolio_return - benchmark_return
+            allocation = 0.0
+            selection = excess_return
+            interaction = 0.0
+
+        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 归因。"""
+        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:
+        """因子归因分解。"""
+        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:
+        """完整归因报告。"""
+        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()

tests/unit/test_backtest_attribution.py

@@ -0,0 +1,211 @@
+"""归因分析单元测试。"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from easy_tdx.backtest.attribution import AttributionAnalyzer
+
+
+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()
+        assert report.commission_cost == pytest.approx(40.0)
+        assert report.slippage_cost == pytest.approx(20.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
+
+    def test_stamp_tax_estimation(self) -> None:
+        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()
+        assert report.stamp_tax_cost == pytest.approx(11.0)
+
+
+class TestBrinsonAttribution:
+    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
+
+    def test_with_benchmark_selection(self) -> None:
+        trades = _make_trades()
+        eq = _make_equity(100000.0, 110000.0)
+        bench = _make_benchmark(100000.0, 105000.0)
+        analyzer = AttributionAnalyzer(trades, eq, benchmark=bench)
+        report = analyzer.brinson_attribution()
+        assert report.total_return == pytest.approx(0.1)
+        assert report.selection_return == pytest.approx(0.05)
+
+    def test_with_groups_decomposition(self) -> None:
+        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()
+        assert report.allocation_return == pytest.approx(0.01)
+        assert report.selection_return == pytest.approx(0.05)
+        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 == {}
+
+    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()
+        assert report.factor_returns["momentum"] == pytest.approx(0.038)
+        assert report.factor_returns["volatility"] == pytest.approx(0.003)
+        assert report.specific_return == pytest.approx(0.059)
+
+    def test_empty_factor_data(self) -> None:
+        trades = _make_trades()
+        eq = _make_equity()
+        analyzer = AttributionAnalyzer(
+            trades,
+            eq,
+            factor_exposures=pd.DataFrame(),
+            factor_returns=pd.DataFrame(),
+        )
+        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
+
+    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