diff --git a/src/quant_engine/attribution.py b/src/quant_engine/attribution.py new file mode 100644 index 0000000..cb0c2c8 --- /dev/null +++ b/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, + ) diff --git a/src/quant_engine/research_pipeline.py b/src/quant_engine/research_pipeline.py index 2b52e9d..5e667ac 100644 --- a/src/quant_engine/research_pipeline.py +++ b/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):