152 lines
4.6 KiB
Python
152 lines
4.6 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 (
|
|
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_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 == ()
|