From ecf6b4e4bf5fe161bdb485af31c51cd5764f7a80 Mon Sep 17 00:00:00 2001
From: ao gong <41768719+ageorge156@users.noreply.github.com>
Date: Fri, 21 Aug 2026 21:22:36 +0800
Subject: [PATCH] test: add red factor-to-weights portfolio contract

---
 tests/test_portfolio_construction.py | 137 +++++++++++++++++++++++++++
 1 file changed, 137 insertions(+)
 create mode 100644 tests/test_portfolio_construction.py

diff --git a/tests/test_portfolio_construction.py b/tests/test_portfolio_construction.py
new file mode 100644
index 0000000..43cd606
--- /dev/null
+++ b/tests/test_portfolio_construction.py
@@ -0,0 +1,137 @@
+"""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_factor_scores_flow_directly_into_weight_backtest() -> None:
+    dates = pd.date_range("2026-01-05", periods=3, freq="B")
+    scores = pd.DataFrame(
+        {"A": [2.0, 0.0], "B": [1.0, 3.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,
+    )
+
+    weights = scores_to_weight_table(scores, top_k=1)
+    result = run_weight_backtest(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, 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_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)