From b7f77e2a6cb1ad8a5759be4c7f127676b0401c8c Mon Sep 17 00:00:00 2001 From: ao gong <41768719+ageorge156@users.noreply.github.com> Date: Fri, 21 Aug 2026 21:46:00 +0800 Subject: [PATCH] test: define lagged factor-to-execution contract --- tests/test_data_adapter.py | 19 ++++ tests/test_research_pipeline.py | 150 ++++++++++++++++++++++++++++++++ 2 files changed, 169 insertions(+) create mode 100644 tests/test_research_pipeline.py diff --git a/tests/test_data_adapter.py b/tests/test_data_adapter.py index 0f101d9..df21754 100644 --- a/tests/test_data_adapter.py +++ b/tests/test_data_adapter.py @@ -266,6 +266,17 @@ def test_prepare_execution_inputs_basic(tushare_long: pd.DataFrame) -> None: assert volumes.iloc[0, 0] == pytest.approx(1000.0) +def test_prepare_execution_inputs_can_select_next_session_open_price( + tushare_long: pd.DataFrame, +) -> None: + """显式 price_col=open 时应生成开盘执行价矩阵。""" + renamed = rename_tushare_columns(tushare_long) + + prices, _volumes = prepare_execution_inputs(renamed, price_col="open") + + assert prices.iloc[0, 0] == pytest.approx(10.0) + + def test_prepare_execution_inputs_no_volume() -> None: """无 volume 列 → volumes 全 1.0。""" df = pd.DataFrame( @@ -286,6 +297,14 @@ def test_prepare_execution_inputs_missing_close_raises() -> None: prepare_execution_inputs(df) +def test_prepare_execution_inputs_missing_selected_price_raises() -> None: + df = pd.DataFrame( + {"stock_code": ["A"], "trade_date": ["2024-01-01"], "close": [10.0]} + ) + with pytest.raises(ValueError, match="缺 open"): + prepare_execution_inputs(df, price_col="open") + + # ── 端到端：长表 → 适配 → alpha158 + execution ────────────── diff --git a/tests/test_research_pipeline.py b/tests/test_research_pipeline.py new file mode 100644 index 0000000..22c8701 --- /dev/null +++ b/tests/test_research_pipeline.py @@ -0,0 +1,150 @@ +"""No-lookahead factor-score to execution-audit integration contracts.""" + +from __future__ import annotations + +import pandas as pd +import pytest + +from quant_engine.execution import ExecutionConfig +from quant_engine.research_pipeline import ( + FactorExecutionResult, + TargetWeightSchedule, + run_factor_execution_research, + schedule_target_weights, +) + + +def _calendar() -> pd.DatetimeIndex: + return pd.date_range("2026-01-05", periods=4, freq="B") + + +def _factor_scores() -> pd.DataFrame: + dates = _calendar() + return pd.DataFrame( + {"A": [2.0, 0.0], "B": [1.0, 3.0]}, + index=dates[:2], + ) + + +def _next_session_open_prices() -> pd.DataFrame: + dates = _calendar() + return pd.DataFrame( + {"A": [1.0, 10.0, 10.0, 10.0], "B": [1.0, 10.0, 20.0, 20.0]}, + index=dates, + ) + + +def test_schedule_target_weights_maps_signal_to_next_trading_session() -> None: + dates = _calendar() + decision_weights = pd.DataFrame( + {"A": [1.0, 0.0], "B": [0.0, 1.0]}, + index=dates[:2], + ) + + schedule = schedule_target_weights(decision_weights, dates, lag_sessions=1) + + assert isinstance(schedule, TargetWeightSchedule) + assert schedule.lag_sessions == 1 + pd.testing.assert_series_equal( + schedule.signal_to_execution, + pd.Series(dates[1:3], index=dates[:2], name="execution_date"), + ) + expected = decision_weights.copy() + expected.index = dates[1:3] + expected.index.name = "execution_date" + pd.testing.assert_frame_equal(schedule.execution_weights, expected) + assert (schedule.execution_weights.index > schedule.signal_to_execution.index).all() + + +def test_factor_execution_research_uses_next_session_prices() -> None: + config = ExecutionConfig( + commission_bps=0, + stamp_tax_bps=0, + slippage_bps=0, + min_trade_amount=0, + ) + + result = run_factor_execution_research( + _factor_scores(), + _next_session_open_prices(), + top_k=1, + execution_price_field="open", + initial_cash=1_000.0, + config=config, + ) + + assert isinstance(result, FactorExecutionResult) + assert result.execution_price_field == "open" + assert result.execution.daily_executions[0].date == str(_calendar()[1]) + assert result.execution.positions[0].holdings == {"A": 100.0} + assert result.execution.positions[1].holdings == {"B": 50.0} + assert result.execution.final_portfolio_value == pytest.approx(1_000.0) + + +def test_factor_execution_result_snapshots_research_inputs() -> None: + scores = _factor_scores() + prices = _next_session_open_prices() + + result = run_factor_execution_research( + scores, + prices, + top_k=1, + execution_price_field="open", + ) + scores.iloc[0, 0] = -999.0 + prices.iloc[1, 0] = 999.0 + + assert result.factor_scores.iloc[0, 0] == 2.0 + assert result.execution.positions[0].holdings["A"] < 200_000.0 + + +@pytest.mark.parametrize("lag_sessions", [0, -1, True]) +def test_schedule_target_weights_requires_positive_integer_lag(lag_sessions: int) -> None: + with pytest.raises(ValueError, match="lag_sessions"): + schedule_target_weights( + pd.DataFrame({"A": [1.0]}, index=_calendar()[:1]), + _calendar(), + lag_sessions=lag_sessions, + ) + + +def test_schedule_target_weights_rejects_signal_outside_trading_calendar() -> None: + weekend = pd.Timestamp("2026-01-10") + with pytest.raises(ValueError, match="signal dates must be trading sessions"): + schedule_target_weights( + pd.DataFrame({"A": [1.0]}, index=[weekend]), + _calendar(), + ) + + +def test_schedule_target_weights_rejects_missing_future_execution_session() -> None: + dates = _calendar() + with pytest.raises(ValueError, match="future execution session"): + schedule_target_weights( + pd.DataFrame({"A": [1.0]}, index=dates[-1:]), + dates, + ) + + +def test_factor_execution_research_requires_explicit_price_field() -> None: + with pytest.raises(ValueError, match="execution_price_field"): + run_factor_execution_research( + _factor_scores(), + _next_session_open_prices(), + top_k=1, + execution_price_field="", + ) + + +def test_factor_execution_research_accepts_empty_scores() -> None: + scores = pd.DataFrame(columns=["A", "B"], index=pd.DatetimeIndex([]), dtype=float) + + result = run_factor_execution_research( + scores, + _next_session_open_prices(), + top_k=1, + execution_price_field="open", + ) + + assert result.schedule.execution_weights.empty + assert result.execution.positions == ()