"""Stable research-run artifact contracts for downstream persistence.""" from __future__ import annotations import json from datetime import date import pandas as pd import pytest from quant_engine.artifact import ( RESEARCH_ARTIFACT_SCHEMA_VERSION, ResearchRunArtifact, build_research_run_artifact, ) from quant_engine.execution import ExecutionConfig from quant_engine.research_pipeline import FactorBacktestResult, run_factor_backtest_research from quant_engine.risk import CovarianceSnapshot def _backtest_result() -> FactorBacktestResult: dates = pd.date_range("2026-01-05", periods=4, freq="B") scores = pd.DataFrame( {"A": [2.0, 0.0], "B": [1.0, 3.0]}, index=dates[:2], ) opens = pd.DataFrame( {"A": [10.0, 10.0, 15.0, 15.0], "B": [20.0, 20.0, 20.0, 21.0]}, index=dates, ) closes = pd.DataFrame( {"A": [10.0, 12.0, 15.0, 15.0], "B": [20.0, 20.0, 18.0, 21.0]}, index=dates, ) return run_factor_backtest_research( scores, opens, closes, top_k=1, 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, ), ) def _build( result: FactorBacktestResult, *, parameters: dict[str, object] | None = None, risk_snapshots: dict[date, CovarianceSnapshot] | None = None, ) -> ResearchRunArtifact: benchmark = pd.Series( [0.0, 0.01, -0.01, 0.02], index=result.returns.index, name="benchmark_return", ) return build_research_run_artifact( result, run_id="run-20260105-a", strategy_id="alpha-top1", strategy_name="Alpha Top 1", strategy_version="1.0.0", engine_version="1.2.0", code_revision="3b1ad07", data_snapshot_id="qtdb-pro-20260108-v1", calendar="CN-A", timezone="Asia/Shanghai", started_at="2026-01-08T10:00:00+08:00", finished_at="2026-01-08T10:01:00+08:00", parameters=parameters or {"top_k": 1, "lag_sessions": 1}, benchmark_id="000300.SH", benchmark_returns=benchmark, risk_snapshots=risk_snapshots, ) def test_research_artifact_projects_versioned_queryable_fact_tables() -> None: result = _backtest_result() artifact = _build(result) assert artifact.schema_version == RESEARCH_ARTIFACT_SCHEMA_VERSION assert artifact.run.loc[0, "run_id"] == "run-20260105-a" assert artifact.run.loc[0, "benchmark_alignment_policy"] == "exact_session_index" assert artifact.nav["run_id"].unique().tolist() == ["run-20260105-a"] assert artifact.nav["pnl_pct"].tolist() == pytest.approx(result.returns.tolist()) assert artifact.nav["benchmark_return"].tolist() == pytest.approx( [0.0, 0.01, -0.01, 0.02] ) assert artifact.signals.columns.tolist() == [ "run_id", "signal_date", "execution_date", "asset_id", "factor_score", "target_weight", ] first_signal = artifact.signals[ artifact.signals["signal_date"] == result.factor_scores.index[0].date() ] assert first_signal.set_index("asset_id").loc["A", "factor_score"] == 2.0 assert first_signal.set_index("asset_id").loc["A", "target_weight"] == 1.0 assert first_signal["execution_date"].unique().tolist() == [ result.schedule.signal_to_execution.iloc[0].date() ] assert set(artifact.trades["side"]) == {"buy", "sell"} assert artifact.trades["trade_id"].is_unique assert artifact.trades["trade_id"].str.startswith("run-20260105-a:").all() assert artifact.trades["signal_id"].str.startswith("run-20260105-a:signal:").all() assert {"security", "cash"}.issubset(set(artifact.positions["asset_type"])) assert artifact.positions.groupby("trade_date")["weight"].sum().tolist() == pytest.approx( [1.0, 1.0, 1.0, 1.0] ) assert set(artifact.attribution.columns) == { "run_id", "trade_date", "asset_id", "overnight", "intraday", "asset_total", } assert artifact.attribution_daily["residual"].abs().max() < 1e-12 assert artifact.risk.empty assert artifact.risk.columns.tolist() == [ "run_id", "trade_date", "asset_id", "weight", "marginal_risk", "component_risk", "risk_contribution", "covariance_snapshot_id", "covariance_as_of_date", "risk_measure", "return_frequency", "periods_per_year", ] assert artifact.performance.loc[0, "n_trades"] == len(artifact.trades) assert artifact.performance.loc[0, "ir"] == pytest.approx( result.benchmark_stats(pd.Series([0.0, 0.01, -0.01, 0.02], index=result.returns.index))[ "information_ratio" ] ) assert "sortino" in artifact.performance.columns def test_research_artifact_projects_annualized_risk_from_actual_positions() -> None: result = _backtest_result() trade_date = result.position_weights.index[-1].date() covariance = pd.DataFrame( [[0.0001, 0.00002], [0.00002, 0.0004]], index=["A", "B"], columns=["A", "B"], ) snapshot = CovarianceSnapshot( snapshot_id="cov-20260107-v1", as_of_date="2026-01-07", covariance=covariance, return_frequency="1d", periods_per_year=252, data_snapshot_id="qtdb-pro-20260108-v1", ) artifact = _build(result, risk_snapshots={trade_date: snapshot}) risk = artifact.risk.set_index("asset_id") expected_weights = result.position_weights.loc[pd.Timestamp(trade_date)] assert artifact.schema_version == "1.1.0" assert risk.index.tolist() == ["A", "B"] assert risk["weight"].tolist() == pytest.approx(expected_weights.tolist()) assert risk["covariance_snapshot_id"].unique().tolist() == ["cov-20260107-v1"] assert risk["covariance_as_of_date"].unique().tolist() == [date(2026, 1, 7)] assert risk["risk_measure"].unique().tolist() == ["annualized_volatility"] assert risk["return_frequency"].unique().tolist() == ["1d"] assert risk["periods_per_year"].unique().tolist() == [252] assert risk["component_risk"].sum() == pytest.approx((0.0004 * 252) ** 0.5) assert risk["risk_contribution"].sum() == pytest.approx(1.0) def test_research_artifact_rejects_risk_from_a_different_data_snapshot() -> None: result = _backtest_result() trade_date = result.position_weights.index[-1].date() covariance = pd.DataFrame( [[0.0001, 0.0], [0.0, 0.0004]], index=["A", "B"], columns=["A", "B"], ) with pytest.raises(ValueError, match="data lineage differs"): _build( result, risk_snapshots={ trade_date: CovarianceSnapshot( snapshot_id="foreign-covariance", as_of_date="2026-01-07", covariance=covariance, return_frequency="1d", periods_per_year=252, data_snapshot_id="different-market-snapshot", ) }, ) def test_research_artifact_rejects_future_or_misaligned_risk_snapshots() -> None: result = _backtest_result() trade_date = result.position_weights.index[-1].date() covariance = pd.DataFrame( [[0.0001, 0.0], [0.0, 0.0004]], index=["A", "B"], columns=["A", "B"], ) with pytest.raises(ValueError, match="must not be after trade date"): _build( result, risk_snapshots={ trade_date: CovarianceSnapshot( snapshot_id="future-covariance", as_of_date="2026-01-09", covariance=covariance, return_frequency="1d", periods_per_year=252, data_snapshot_id="qtdb-pro-20260108-v1", ) }, ) with pytest.raises(ValueError, match="same asset labels"): _build( result, risk_snapshots={ trade_date: CovarianceSnapshot( snapshot_id="incomplete-universe", as_of_date="2026-01-07", covariance=covariance.loc[["B"], ["B"]], return_frequency="1d", periods_per_year=252, data_snapshot_id="qtdb-pro-20260108-v1", ) }, ) def test_research_artifact_serialization_and_hashes_are_deterministic() -> None: result = _backtest_result() first = _build(result, parameters={"top_k": 1, "lag_sessions": 1}) second = _build(result, parameters={"lag_sessions": 1, "top_k": 1}) assert first.run.loc[0, "config_hash"] == second.run.loc[0, "config_hash"] assert first.content_sha256 == second.content_sha256 assert first.manifest() == second.manifest() decoded = json.loads(first.canonical_json()) assert decoded["schema_version"] == RESEARCH_ARTIFACT_SCHEMA_VERSION assert decoded["tables"]["nav"][0]["trade_date"] == "2026-01-05" leaked_copy = first.nav leaked_copy.loc[0, "nav"] = -999.0 assert first.nav.loc[0, "nav"] != -999.0 assert first.content_sha256 == second.content_sha256 def test_research_artifact_requires_complete_reproducibility_identity() -> None: result = _backtest_result() with pytest.raises(ValueError, match="code_revision"): build_research_run_artifact( result, run_id="run-1", strategy_id="alpha-top1", strategy_name="Alpha Top 1", strategy_version="1.0.0", engine_version="1.2.0", code_revision="", data_snapshot_id="snapshot-1", calendar="CN-A", timezone="Asia/Shanghai", started_at="2026-01-08T10:00:00+08:00", finished_at="2026-01-08T10:01:00+08:00", parameters={}, ) def test_research_artifact_requires_benchmark_identity_and_returns_together() -> None: result = _backtest_result() with pytest.raises(ValueError, match="benchmark_id and benchmark_returns"): build_research_run_artifact( result, run_id="run-1", strategy_id="alpha-top1", strategy_name="Alpha Top 1", strategy_version="1.0.0", engine_version="1.2.0", code_revision="3b1ad07", data_snapshot_id="snapshot-1", calendar="CN-A", timezone="Asia/Shanghai", started_at="2026-01-08T10:00:00+08:00", finished_at="2026-01-08T10:01:00+08:00", parameters={}, benchmark_id="000300.SH", )