Compare commits

...
3 Commits
4 changed files with 88 additions and 1 deletions
+2
View File
@@ -124,6 +124,8 @@ print(factor_backtest.returns)
print(factor_backtest.stats())
print(factor_backtest.execution.ledger_frame)
print(factor_backtest.execution.trades_frame)
print(factor_backtest.position_weights) # 实际日末资产权重
print(factor_backtest.cash_weights)
# 所有分析都以实际成交后的 Ledger 为事实源,不直接使用目标权重伪造结果。
attribution = factor_backtest.return_attribution()
@@ -9,6 +9,7 @@
- 贡献闭合到成本后日收益并显式暴露 residual;
- 严格日期对齐的 TE / IR / alpha / beta;
- 标签安全且可分组的 Euler component risk。
- 从 Ledger 股数和收盘估值投影的实际资产 / 现金权重。
## Branch stack
@@ -27,7 +28,7 @@ Riskfolio-Lib 和 PyPortfolioOpt 只作为时间语义、Ledger、相对指标
## Verification
- `pytest -q --cov=src --cov-report=term-missing`: 513 passed,9 个既有 SciPy warning,91% coverage;
- `pytest -q --cov=src --cov-report=term-missing`: 514 passed,9 个既有 SciPy warning,91% coverage;
- `mypy --strict src/`: 15 source files passed;
- 变更范围 `ruff check`: passed;
- 全仓 Ruff:仅 13 个既有 `tests/governance/*` PT009;
+41
View File
@@ -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)
+43
View File
@@ -291,3 +291,46 @@ def test_factor_backtest_exposes_net_benchmark_metrics() -> None:
assert relative["n_observations"] == len(result.returns)
assert relative["tracking_error"] > 0
def test_factor_backtest_projects_actual_close_weights_from_ledger() -> None:
dates = _calendar()
scores = pd.DataFrame({"A": [1.0], "B": [0.0]}, index=dates[:1])
opens = pd.DataFrame(
{"A": [10.0, 10.0, 10.0, 10.0], "B": [20.0, 20.0, 20.0, 20.0]},
index=dates,
)
closes = pd.DataFrame(
{"A": [10.0, 11.0, 12.0, 12.0], "B": [20.0, 20.0, 20.0, 20.0]},
index=dates,
)
result = run_factor_backtest_research(
scores,
opens,
closes,
top_k=1,
gross_exposure=0.5,
execution_price_field="open",
valuation_price_field="close",
initial_cash=1_000.0,
config=ExecutionConfig(
commission_bps=0,
stamp_tax_bps=0,
slippage_bps=0,
min_trade_amount=0,
),
)
weights = result.position_weights
cash = result.cash_weights
assert weights.index.equals(result.nav.index)
assert weights.columns.tolist() == ["A", "B"]
assert weights.loc[dates[0]].sum() == 0.0
assert cash.loc[dates[0]] == 1.0
assert weights.loc[dates[1], "A"] == pytest.approx(550.0 / 1_050.0)
pd.testing.assert_series_equal(
weights.sum(axis=1) + cash,
pd.Series(1.0, index=dates),
check_names=False,
)