diff --git a/tests/test_risk.py b/tests/test_risk.py index e69272b..8160e6e 100644 --- a/tests/test_risk.py +++ b/tests/test_risk.py @@ -10,12 +10,114 @@ from quant_engine.risk import ( ComponentRiskResult, CovarianceSnapshot, component_var, + estimate_covariance_snapshot, labeled_component_risk, marginal_risk_contribution, risk_contribution, ) +def test_estimate_covariance_snapshot_is_complete_case_and_reproducible() -> None: + dates = pd.date_range("2026-01-05", periods=6, freq="B") + returns = pd.DataFrame( + { + "A": [0.01, 0.02, 0.03, 0.04, 0.05, 99.0], + "B": [0.02, 0.01, np.nan, 0.03, 0.04, -99.0], + }, + index=dates, + ) + as_of = dates[4] + + snapshot = estimate_covariance_snapshot( + returns, + as_of_date=as_of, + lookback_sessions=4, + min_observations=3, + data_snapshot_id="market-returns-20260109-v1", + return_frequency="1d", + periods_per_year=252, + ) + + expected_window = returns.loc[:as_of].tail(4) + expected = expected_window.dropna(how="any").cov() + pd.testing.assert_frame_equal(snapshot.covariance, expected) + assert snapshot.snapshot_id.startswith("sample-cov-v1:") + assert snapshot.as_of_date == as_of.date() + assert snapshot.method == "sample" + assert snapshot.window_start_date == expected_window.index[0].date() + assert snapshot.window_end_date == as_of.date() + assert snapshot.observations == 3 + assert snapshot.lookback_sessions == 4 + assert snapshot.missing_policy == "complete_case" + assert snapshot.data_snapshot_id == "market-returns-20260109-v1" + assert len(snapshot.input_sha256) == 64 + + future_changed = returns.copy() + future_changed.loc[dates[-1], :] = [1_000_000.0, -1_000_000.0] + repeated = estimate_covariance_snapshot( + future_changed, + as_of_date=as_of, + lookback_sessions=4, + min_observations=3, + data_snapshot_id="market-returns-20260109-v1", + return_frequency="1d", + periods_per_year=252, + ) + assert repeated.snapshot_id == snapshot.snapshot_id + pd.testing.assert_frame_equal(repeated.covariance, snapshot.covariance) + + +def test_covariance_snapshot_identity_captures_data_and_estimator_contract() -> None: + dates = pd.date_range("2026-01-05", periods=4, freq="B") + returns = pd.DataFrame( + {"A": [0.01, 0.02, -0.01, 0.03], "B": [0.02, -0.01, 0.01, 0.04]}, + index=dates, + ) + base = estimate_covariance_snapshot( + returns, + as_of_date=dates[-1], + lookback_sessions=4, + min_observations=3, + data_snapshot_id="snapshot-a", + ) + different_source = estimate_covariance_snapshot( + returns, + as_of_date=dates[-1], + lookback_sessions=4, + min_observations=3, + data_snapshot_id="snapshot-b", + ) + + assert base.snapshot_id != different_source.snapshot_id + assert base.covariance.equals(different_source.covariance) + + +def test_estimate_covariance_snapshot_rejects_ambiguous_or_insufficient_history() -> None: + dates = pd.date_range("2026-01-05", periods=4, freq="B") + returns = pd.DataFrame( + {"A": [0.01, np.nan, 0.03, 0.04], "B": [0.02, 0.01, np.nan, 0.03]}, + index=dates, + ) + + with pytest.raises(ValueError, match="complete observations"): + estimate_covariance_snapshot( + returns, + as_of_date=dates[-1], + lookback_sessions=4, + min_observations=3, + data_snapshot_id="snapshot-a", + ) + + with pytest.raises(ValueError, match="strictly increasing"): + estimate_covariance_snapshot( + returns.iloc[::-1], + as_of_date=dates[-1], + lookback_sessions=4, + min_observations=2, + data_snapshot_id="snapshot-a", + ) + + def test_covariance_snapshot_is_validated_and_immutable_by_interface() -> None: covariance = pd.DataFrame( [[0.04, 0.01], [0.01, 0.09]],