337 lines
11 KiB
Python
337 lines
11 KiB
Python
"""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 (
|
|
FactorBacktestResult,
|
|
FactorExecutionResult,
|
|
TargetWeightSchedule,
|
|
run_factor_backtest_research,
|
|
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_prices.loc[_calendar()[1], "A"] == 10.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 == ()
|
|
|
|
|
|
def test_factor_backtest_research_runs_signal_to_daily_performance_without_lookahead() -> None:
|
|
"""信号日保持现金，下一日开盘成交后才参与当日收盘收益。"""
|
|
dates = _calendar()
|
|
scores = pd.DataFrame({"A": [2.0], "B": [1.0]}, index=dates[:1])
|
|
opens = pd.DataFrame(
|
|
{"A": [1.0, 10.0, 10.0, 10.0], "B": [1.0, 20.0, 20.0, 20.0]},
|
|
index=dates,
|
|
)
|
|
closes = pd.DataFrame(
|
|
{"A": [500.0, 11.0, 12.0, 12.0], "B": [500.0, 20.0, 20.0, 20.0]},
|
|
index=dates,
|
|
)
|
|
config = ExecutionConfig(
|
|
commission_bps=0,
|
|
stamp_tax_bps=0,
|
|
slippage_bps=0,
|
|
min_trade_amount=0,
|
|
)
|
|
|
|
result = run_factor_backtest_research(
|
|
scores,
|
|
execution_prices=opens,
|
|
valuation_prices=closes,
|
|
top_k=1,
|
|
execution_price_field="open",
|
|
valuation_price_field="close",
|
|
initial_cash=1_000.0,
|
|
config=config,
|
|
)
|
|
|
|
assert isinstance(result, FactorBacktestResult)
|
|
assert result.execution_price_field == "open"
|
|
assert result.valuation_price_field == "close"
|
|
pd.testing.assert_series_equal(
|
|
result.nav,
|
|
pd.Series([1.0, 1.1, 1.2, 1.2], index=dates, name="nav"),
|
|
)
|
|
pd.testing.assert_series_equal(
|
|
result.returns,
|
|
pd.Series([0.0, 0.1, 1.2 / 1.1 - 1.0, 0.0], index=dates, name="returns"),
|
|
)
|
|
assert result.stats()["n_days"] == 4
|
|
assert result.execution.daily_executions[0].executions == ()
|
|
assert result.execution.daily_executions[1].executions[0].price == 10.0
|
|
|
|
|
|
def test_factor_backtest_result_snapshots_both_price_semantics() -> None:
|
|
scores = pd.DataFrame({"A": [1.0]}, index=_calendar()[:1])
|
|
opens = pd.DataFrame({"A": [10.0, 10.0, 10.0, 10.0]}, index=_calendar())
|
|
closes = pd.DataFrame({"A": [10.0, 11.0, 12.0, 13.0]}, index=_calendar())
|
|
|
|
result = run_factor_backtest_research(
|
|
scores,
|
|
execution_prices=opens,
|
|
valuation_prices=closes,
|
|
top_k=1,
|
|
execution_price_field="open",
|
|
valuation_price_field="close",
|
|
)
|
|
opens.iloc[1, 0] = 999.0
|
|
closes.iloc[1, 0] = 999.0
|
|
|
|
assert result.execution_prices.iloc[1, 0] == 10.0
|
|
assert result.valuation_prices.iloc[1, 0] == 11.0
|
|
|
|
|
|
def test_factor_backtest_research_requires_matching_daily_calendars() -> None:
|
|
scores = pd.DataFrame({"A": [1.0]}, index=_calendar()[:1])
|
|
opens = pd.DataFrame({"A": [10.0, 10.0, 10.0, 10.0]}, index=_calendar())
|
|
closes = pd.DataFrame({"A": [10.0, 11.0, 12.0]}, index=_calendar()[:3])
|
|
|
|
with pytest.raises(ValueError, match="matching trading calendars"):
|
|
run_factor_backtest_research(
|
|
scores,
|
|
execution_prices=opens,
|
|
valuation_prices=closes,
|
|
top_k=1,
|
|
execution_price_field="open",
|
|
valuation_price_field="close",
|
|
)
|
|
|
|
|
|
def test_factor_backtest_starts_at_first_signal_instead_of_price_warmup() -> None:
|
|
"""因子预热行情不能作为空仓日混入研究绩效区间。"""
|
|
dates = pd.date_range("2026-01-05", periods=5, freq="B")
|
|
scores = pd.DataFrame({"A": [1.0]}, index=dates[2:3])
|
|
opens = pd.DataFrame({"A": [1.0, 1.0, 1.0, 10.0, 10.0]}, index=dates)
|
|
closes = pd.DataFrame({"A": [100.0, 200.0, 300.0, 11.0, 12.0]}, index=dates)
|
|
config = ExecutionConfig(
|
|
commission_bps=0,
|
|
stamp_tax_bps=0,
|
|
slippage_bps=0,
|
|
min_trade_amount=0,
|
|
)
|
|
|
|
result = run_factor_backtest_research(
|
|
scores,
|
|
execution_prices=opens,
|
|
valuation_prices=closes,
|
|
top_k=1,
|
|
execution_price_field="open",
|
|
valuation_price_field="close",
|
|
initial_cash=1_000.0,
|
|
config=config,
|
|
)
|
|
|
|
assert result.nav.index.equals(dates[2:])
|
|
pd.testing.assert_series_equal(
|
|
result.nav,
|
|
pd.Series([1.0, 1.1, 1.2], index=dates[2:], name="nav"),
|
|
)
|
|
assert result.stats()["n_days"] == 3
|
|
|
|
|
|
def test_factor_backtest_exposes_net_benchmark_metrics() -> None:
|
|
dates = _calendar()
|
|
scores = pd.DataFrame({"A": [1.0]}, index=dates[:1])
|
|
prices = pd.DataFrame({"A": [10.0, 10.0, 11.0, 11.0]}, index=dates)
|
|
result = run_factor_backtest_research(
|
|
scores,
|
|
prices,
|
|
prices,
|
|
top_k=1,
|
|
execution_price_field="open",
|
|
valuation_price_field="close",
|
|
config=ExecutionConfig(
|
|
commission_bps=0,
|
|
stamp_tax_bps=0,
|
|
slippage_bps=0,
|
|
min_trade_amount=0,
|
|
),
|
|
)
|
|
benchmark = pd.Series([0.0, 0.01, -0.01, 0.0], index=dates)
|
|
|
|
relative = result.benchmark_stats(benchmark)
|
|
|
|
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,
|
|
)
|