158 lines
4.9 KiB
Python
158 lines
4.9 KiB
Python
"""Factor-score portfolio construction and backtest integration contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from quant_engine.backtest import run_weight_backtest
|
|
from quant_engine.portfolio_construction import (
|
|
equal_weight,
|
|
scores_to_target_weights,
|
|
scores_to_weight_table,
|
|
select_top_k,
|
|
)
|
|
|
|
|
|
def test_select_top_k_ignores_nan_and_breaks_ties_by_input_order() -> None:
|
|
scores = pd.Series([1.0, 1.0, np.nan, 0.5], index=["B", "A", "C", "D"])
|
|
|
|
selected = select_top_k(scores, top_k=2)
|
|
|
|
assert selected.tolist() == ["B", "A"]
|
|
|
|
|
|
def test_select_top_k_can_select_lowest_scores() -> None:
|
|
scores = pd.Series([3.0, 1.0, 2.0], index=["A", "B", "C"])
|
|
|
|
selected = select_top_k(scores, top_k=2, largest=False)
|
|
|
|
assert selected.tolist() == ["B", "C"]
|
|
|
|
|
|
def test_equal_weight_allocates_requested_gross_exposure() -> None:
|
|
result = equal_weight(pd.Index(["A", "B", "C"]), gross_exposure=0.9)
|
|
|
|
pd.testing.assert_series_equal(
|
|
result,
|
|
pd.Series([0.3, 0.3, 0.3], index=["A", "B", "C"], name="weight"),
|
|
)
|
|
|
|
|
|
def test_equal_weight_returns_empty_float_series_for_no_assets() -> None:
|
|
result = equal_weight(pd.Index([], dtype=object))
|
|
|
|
assert result.empty
|
|
assert result.dtype == float
|
|
assert result.name == "weight"
|
|
|
|
|
|
def test_scores_to_target_weights_keeps_full_universe_with_zero_for_unselected() -> None:
|
|
scores = pd.Series([0.2, 0.8, 0.5], index=["A", "B", "C"])
|
|
|
|
result = scores_to_target_weights(scores, top_k=2)
|
|
|
|
pd.testing.assert_series_equal(
|
|
result,
|
|
pd.Series([0.0, 0.5, 0.5], index=scores.index, name="weight"),
|
|
)
|
|
|
|
|
|
def test_scores_to_target_weights_divides_exposure_over_available_scores() -> None:
|
|
scores = pd.Series([1.0, np.nan, 0.5], index=["A", "B", "C"])
|
|
|
|
result = scores_to_target_weights(scores, top_k=5, gross_exposure=0.8)
|
|
|
|
pd.testing.assert_series_equal(
|
|
result,
|
|
pd.Series([0.4, 0.0, 0.4], index=scores.index, name="weight"),
|
|
)
|
|
|
|
|
|
def test_scores_to_weight_table_constructs_each_rebalance_independently() -> None:
|
|
dates = pd.to_datetime(["2026-01-05", "2026-01-07"])
|
|
scores = pd.DataFrame(
|
|
{"A": [3.0, 1.0], "B": [2.0, 3.0], "C": [1.0, 2.0]},
|
|
index=dates,
|
|
)
|
|
|
|
result = scores_to_weight_table(scores, top_k=2)
|
|
|
|
expected = pd.DataFrame(
|
|
{"A": [0.5, 0.0], "B": [0.5, 0.5], "C": [0.0, 0.5]},
|
|
index=dates,
|
|
)
|
|
pd.testing.assert_frame_equal(result, expected)
|
|
|
|
changed_future = scores.copy()
|
|
changed_future.iloc[1] = [100.0, -100.0, 0.0]
|
|
changed_result = scores_to_weight_table(changed_future, top_k=2)
|
|
pd.testing.assert_series_equal(result.iloc[0], changed_result.iloc[0])
|
|
|
|
|
|
def test_effective_holding_weights_flow_into_weight_backtest() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=3, freq="B")
|
|
effective_weights = pd.DataFrame(
|
|
{"A": [1.0, 0.0], "B": [0.0, 1.0]},
|
|
index=dates[[0, 2]],
|
|
)
|
|
stock_returns = pd.DataFrame(
|
|
{"A": [0.10, 0.0, 0.0], "B": [0.0, 0.0, 0.20]},
|
|
index=dates,
|
|
)
|
|
|
|
result = run_weight_backtest(effective_weights, stock_returns)
|
|
|
|
pd.testing.assert_series_equal(result.nav, pd.Series([1.1, 1.1, 1.32], index=dates))
|
|
pd.testing.assert_frame_equal(result.weights, effective_weights)
|
|
|
|
|
|
@pytest.mark.parametrize("top_k", [0, -1])
|
|
def test_portfolio_construction_rejects_non_positive_top_k(top_k: int) -> None:
|
|
scores = pd.Series([1.0], index=["A"])
|
|
|
|
with pytest.raises(ValueError, match="top_k must be positive"):
|
|
select_top_k(scores, top_k=top_k)
|
|
|
|
|
|
@pytest.mark.parametrize("gross_exposure", [-0.1, np.inf, np.nan])
|
|
def test_equal_weight_rejects_invalid_gross_exposure(gross_exposure: float) -> None:
|
|
with pytest.raises(ValueError, match="gross_exposure"):
|
|
equal_weight(pd.Index(["A"]), gross_exposure=gross_exposure)
|
|
|
|
|
|
def test_portfolio_construction_rejects_duplicate_assets() -> None:
|
|
duplicate_scores = pd.Series([1.0, 2.0], index=["A", "A"])
|
|
|
|
with pytest.raises(ValueError, match="unique asset labels"):
|
|
scores_to_target_weights(duplicate_scores, top_k=1)
|
|
|
|
|
|
def test_weight_table_rejects_duplicate_rebalance_dates() -> None:
|
|
duplicate_date = pd.Timestamp("2026-01-05")
|
|
scores = pd.DataFrame(
|
|
{"A": [1.0, 2.0]},
|
|
index=[duplicate_date, duplicate_date],
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="unique rebalance dates"):
|
|
scores_to_weight_table(scores, top_k=1)
|
|
|
|
|
|
def test_weight_table_rejects_unsorted_rebalance_dates() -> None:
|
|
scores = pd.DataFrame(
|
|
{"A": [1.0, 2.0]},
|
|
index=pd.to_datetime(["2026-01-07", "2026-01-05"]),
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="chronological order"):
|
|
scores_to_weight_table(scores, top_k=1)
|
|
|
|
|
|
def test_weight_table_rejects_non_numeric_scores() -> None:
|
|
scores = pd.DataFrame({"A": ["high"], "B": ["low"]})
|
|
|
|
with pytest.raises(TypeError, match="numeric"):
|
|
scores_to_weight_table(scores, top_k=1)
|