fix: enforce chronological rebalance scores

README.md

@@ -23,6 +23,7 @@
 - `indicators` — 50+ 技术指标（MACD / KDJ / 布林 / ATR / ADX / 等）
 - `data_adapter` — 桥接 qtdb_pro 长表与新模块（rename / long-wide / 复权 / vwap 代理）
 - `backtest` — weight-based 多日仿真（rebalance_table / compute_nav / compare_to_benchmark）
+- `portfolio_construction` — 多期因子分数 → Top-K → 等权目标权重表
 - `metrics` — 绩效（年化收益 / 波动率 / Sharpe / 最大回撤 / Calmar）
 - `factor_library` — 通用方法（turnover / winsorize / IC / OLS / jb_test）
 - `portfolio_decomp` — 组合分解（risk_parity / mean_variance / 因子归因）
@@ -61,6 +62,7 @@ from quant_engine.execution import (
     ExecutionConfig, simulate_with_daily_data, compute_realized_pnl,
 )
 from quant_engine.backtest import run_weight_backtest
+from quant_engine.portfolio_construction import scores_to_weight_table
 from quant_engine.indicators import macd, bollinger, kdj
 from quant_engine.data_adapter import (
     long_to_wide, wide_to_long, rename_tushare_columns,
@@ -74,7 +76,12 @@ df = load_qtdb_daily(["000001.SZ"], "2024-01-01", with_adj=True)
 prices, volumes = prepare_execution_inputs(df)
 result = simulate_with_daily_data(prices, initial_cash=1_000_000.0)
 
-# 权重回测 → 稳定结果对象 → 绩效/基准分析
+# 多期因子分数 → Top-K 等权组合 → 稳定回测结果
+rebalance_weights = scores_to_weight_table(
+    factor_scores,
+    top_k=20,
+    gross_exposure=1.0,
+)
 backtest = run_weight_backtest(
     weights=rebalance_weights,
     stock_returns=daily_returns,

src/quant_engine/portfolio_construction.py

@@ -0,0 +1,104 @@
+"""因子分数到目标权重的轻量组合构建闭环。"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+from pandas.api.types import is_numeric_dtype
+
+__all__ = [
+    "select_top_k",
+    "equal_weight",
+    "scores_to_target_weights",
+    "scores_to_weight_table",
+]
+
+
+def _validate_top_k(top_k: int) -> None:
+    if isinstance(top_k, bool) or not isinstance(top_k, int) or top_k <= 0:
+        raise ValueError("top_k must be positive")
+
+
+def _validate_gross_exposure(gross_exposure: float) -> None:
+    if not np.isfinite(gross_exposure) or gross_exposure < 0:
+        raise ValueError("gross_exposure must be finite and non-negative")
+
+
+def _validate_score_series(scores: pd.Series) -> None:
+    if not isinstance(scores, pd.Series):
+        raise TypeError(f"scores must be a pandas Series, got {type(scores).__name__}")
+    if not scores.index.is_unique:
+        raise ValueError("scores must contain unique asset labels")
+    if not is_numeric_dtype(scores.dtype):
+        raise TypeError("scores must contain numeric values")
+
+
+def select_top_k(scores: pd.Series, top_k: int, *, largest: bool = True) -> pd.Index:
+    """稳定选择最高或最低的 K 个有效因子分数。"""
+    _validate_top_k(top_k)
+    _validate_score_series(scores)
+    valid_scores = scores.dropna()
+    ordered = valid_scores.sort_values(ascending=not largest, kind="mergesort")
+    return ordered.iloc[:top_k].index.copy()
+
+
+def equal_weight(assets: pd.Index, *, gross_exposure: float = 1.0) -> pd.Series:
+    """在已选资产间等权分配指定总敞口。"""
+    _validate_gross_exposure(gross_exposure)
+    if not assets.is_unique:
+        raise ValueError("assets must contain unique asset labels")
+    if assets.empty:
+        return pd.Series(index=assets.copy(), dtype=float, name="weight")
+    weight = gross_exposure / len(assets)
+    return pd.Series(weight, index=assets.copy(), dtype=float, name="weight")
+
+
+def scores_to_target_weights(
+    scores: pd.Series,
+    top_k: int,
+    *,
+    gross_exposure: float = 1.0,
+    largest: bool = True,
+) -> pd.Series:
+    """把单期因子分数转换为完整股票池目标权重。"""
+    _validate_score_series(scores)
+    selected = select_top_k(scores, top_k, largest=largest)
+    selected_weights = equal_weight(selected, gross_exposure=gross_exposure)
+    result = pd.Series(0.0, index=scores.index.copy(), dtype=float, name="weight")
+    result.loc[selected_weights.index] = selected_weights
+    return result
+
+
+def scores_to_weight_table(
+    scores: pd.DataFrame,
+    top_k: int,
+    *,
+    gross_exposure: float = 1.0,
+    largest: bool = True,
+) -> pd.DataFrame:
+    """逐调仓日独立构建目标权重表，避免使用未来分数。"""
+    if not isinstance(scores, pd.DataFrame):
+        raise TypeError(f"scores must be a pandas DataFrame, got {type(scores).__name__}")
+    _validate_top_k(top_k)
+    _validate_gross_exposure(gross_exposure)
+    if not scores.index.is_unique:
+        raise ValueError("scores must contain unique rebalance dates")
+    if not scores.index.is_monotonic_increasing:
+        raise ValueError("scores rebalance dates must be in chronological order")
+    if not scores.columns.is_unique:
+        raise ValueError("scores must contain unique asset labels")
+    if not all(is_numeric_dtype(dtype) for dtype in scores.dtypes):
+        raise TypeError("scores must contain numeric values")
+    if scores.empty:
+        return pd.DataFrame(index=scores.index.copy(), columns=scores.columns.copy(), dtype=float)
+
+    rows = [
+        scores_to_target_weights(
+            row,
+            top_k,
+            gross_exposure=gross_exposure,
+            largest=largest,
+        ).to_numpy()
+        for _, row in scores.iterrows()
+    ]
+    return pd.DataFrame(rows, index=scores.index.copy(), columns=scores.columns.copy(), dtype=float)

tests/test_portfolio_construction.py

@@ -0,0 +1,158 @@
+"""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_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)