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)