diff --git a/src/quant_engine/research_pipeline.py b/src/quant_engine/research_pipeline.py index eb63623..bfdfb11 100644 --- a/src/quant_engine/research_pipeline.py +++ b/src/quant_engine/research_pipeline.py @@ -87,6 +87,47 @@ class FactorBacktestResult: name="returns", ) + @property + def position_weights(self) -> pd.DataFrame: + """按日末实际股数、收盘估值和账本 NAV 投影资产权重。""" + weights = pd.DataFrame( + 0.0, + index=self.valuation_prices.index.copy(), + columns=self.valuation_prices.columns.copy(), + ) + for date, position in zip( + self.valuation_prices.index, + self.execution.positions, + strict=True, + ): + if position.portfolio_value <= 0: + raise ValueError(f"portfolio value must be positive on {date}") + for asset, shares in position.holdings.items(): + weights.at[date, asset] = ( + shares * float(self.valuation_prices.at[date, asset]) + / position.portfolio_value + ) + return weights + + @property + def cash_weights(self) -> pd.Series: + """返回与实际资产权重使用同一日末 NAV 分母的现金权重。""" + values = [] + for date, position in zip( + self.valuation_prices.index, + self.execution.positions, + strict=True, + ): + if position.portfolio_value <= 0: + raise ValueError(f"portfolio value must be positive on {date}") + values.append(position.cash / position.portfolio_value) + return pd.Series( + values, + index=self.valuation_prices.index.copy(), + dtype=float, + name="cash_weight", + ) + def stats(self, rf: float = 0.0) -> Mapping[str, float]: """复用标准绩效口径计算指标。""" return metrics_summary(self.returns, rf)