feat: add ledger-backed daily return attribution

src/quant_engine/attribution.py

@@ -0,0 +1,141 @@
+"""Post-execution daily return attribution derived from the portfolio ledger.
+
+The ledger is the source of truth: previous-close holdings explain overnight
+PnL, current-close holdings explain intraday PnL, and actual execution costs
+remain a separate contribution.  Target weights and factor scores are not
+accepted here because they are intentions rather than realized positions.
+"""
+
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+
+import pandas as pd
+
+from quant_engine.execution import ExecutionSimulationResult
+
+__all__ = ["DailyReturnAttribution", "compute_daily_return_attribution"]
+
+
+@dataclass(frozen=True, slots=True, eq=False)
+class DailyReturnAttribution:
+    """Auditable decomposition of each net portfolio return."""
+
+    overnight: pd.DataFrame
+    intraday: pd.DataFrame
+    transaction_cost: pd.Series
+    residual: pd.Series
+    total_return: pd.Series
+
+    @property
+    def asset_contributions(self) -> pd.DataFrame:
+        """Return the combined overnight and intraday contribution by asset."""
+        return self.overnight + self.intraday
+
+    @property
+    def explained_return(self) -> pd.Series:
+        """Return asset contributions plus execution costs, before residual."""
+        explained = self.asset_contributions.sum(axis=1) + self.transaction_cost
+        return explained.rename("explained_return")
+
+
+def _validate_prices(
+    execution: ExecutionSimulationResult,
+    execution_prices: pd.DataFrame,
+    valuation_prices: pd.DataFrame,
+) -> pd.DatetimeIndex:
+    if not isinstance(execution_prices, pd.DataFrame):
+        raise TypeError("execution_prices must be a pandas DataFrame")
+    if not isinstance(valuation_prices, pd.DataFrame):
+        raise TypeError("valuation_prices must be a pandas DataFrame")
+    if not isinstance(execution_prices.index, pd.DatetimeIndex):
+        raise TypeError("execution_prices must use a DatetimeIndex")
+    if not execution_prices.index.equals(valuation_prices.index):
+        raise ValueError("execution and valuation prices must use matching trading calendars")
+    if not execution_prices.columns.equals(valuation_prices.columns):
+        raise ValueError("execution and valuation prices must use matching asset labels")
+
+    ledger_index = pd.DatetimeIndex(pd.Timestamp(position.date) for position in execution.positions)
+    if not ledger_index.equals(execution_prices.index):
+        raise ValueError("ledger and price histories must use matching trading calendars")
+    if len(execution.positions) != len(execution.daily_executions):
+        raise ValueError("ledger positions and executions must have matching lengths")
+    return execution_prices.index.copy()
+
+
+def _price_for_held_asset(
+    prices: pd.DataFrame,
+    date: pd.Timestamp,
+    asset: str,
+    stage: str,
+) -> float:
+    if asset not in prices.columns:
+        raise ValueError(f"missing {stage} price for held asset {asset} on {date}")
+    price = float(prices.at[date, asset])
+    if not math.isfinite(price) or price <= 0:
+        raise ValueError(f"invalid {stage} price for held asset {asset} on {date}")
+    return price
+
+
+def compute_daily_return_attribution(
+    execution: ExecutionSimulationResult,
+    execution_prices: pd.DataFrame,
+    valuation_prices: pd.DataFrame,
+) -> DailyReturnAttribution:
+    """Decompose net daily returns using realized pre/post-execution holdings.
+
+    For each session, previous-close shares earn the move from the previous
+    close to the current execution price; current-close shares earn the move
+    from execution price to current close.  Actual commissions, stamp tax and
+    slippage are divided by the same previous NAV denominator.  ``residual``
+    exposes any failure of those components to close to the ledger return.
+    """
+    index = _validate_prices(execution, execution_prices, valuation_prices)
+    columns = execution_prices.columns.copy()
+    overnight = pd.DataFrame(0.0, index=index.copy(), columns=columns)
+    intraday = pd.DataFrame(0.0, index=index.copy(), columns=columns)
+    cost = pd.Series(0.0, index=index.copy(), name="transaction_cost")
+
+    previous_holdings: dict[str, float] = {}
+    previous_nav = execution.initial_cash
+    for row_number, (date, position, daily) in enumerate(
+        zip(index, execution.positions, execution.daily_executions, strict=True)
+    ):
+        if previous_nav <= 0 or not math.isfinite(previous_nav):
+            raise ValueError(f"previous portfolio value must be positive and finite on {date}")
+
+        for asset, shares in previous_holdings.items():
+            execution_price = _price_for_held_asset(
+                execution_prices, date, asset, "execution"
+            )
+            previous_close = _price_for_held_asset(
+                valuation_prices, index[row_number - 1], asset, "previous valuation"
+            )
+            overnight.at[date, asset] = shares * (execution_price - previous_close) / previous_nav
+
+        for asset, shares in position.holdings.items():
+            execution_price = _price_for_held_asset(
+                execution_prices, date, asset, "execution"
+            )
+            close_price = _price_for_held_asset(valuation_prices, date, asset, "valuation")
+            intraday.at[date, asset] = shares * (close_price - execution_price) / previous_nav
+
+        cost.at[date] = -sum(item.total_cost for item in daily.executions) / previous_nav
+        previous_holdings = position.holdings
+        previous_nav = position.portfolio_value
+
+    total_return = pd.Series(
+        execution.daily_returns.to_numpy(copy=True),
+        index=index.copy(),
+        name="total_return",
+    )
+    explained = (overnight + intraday).sum(axis=1) + cost
+    residual = (total_return - explained).rename("residual")
+    return DailyReturnAttribution(
+        overnight=overnight,
+        intraday=intraday,
+        transaction_cost=cost,
+        residual=residual,
+        total_return=total_return,
+    )

src/quant_engine/research_pipeline.py

@@ -16,6 +16,7 @@ import numpy as np
 import pandas as pd
 from pandas.api.types import is_numeric_dtype
 
+from quant_engine.attribution import DailyReturnAttribution, compute_daily_return_attribution
 from quant_engine.execution import (
     ExecutionConfig,
     ExecutionSimulationResult,
@@ -90,6 +91,14 @@ class FactorBacktestResult:
         """复用标准绩效口径计算指标。"""
         return metrics_summary(self.returns, rf)
 
+    def return_attribution(self) -> DailyReturnAttribution:
+        """从实际成交后持仓与账本生成逐日净收益归因。"""
+        return compute_daily_return_attribution(
+            self.execution,
+            self.execution_prices,
+            self.valuation_prices,
+        )
+
 
 def _validate_datetime_index(index: pd.Index, name: str) -> pd.DatetimeIndex:
     if not isinstance(index, pd.DatetimeIndex):