Simon
273aee2772
- cross_sectional_zscore / composite_score / build_factor_panels 从 local_engine 迁入 quant/composite.py；新增统一入口 build_score_panel(daily, factor_specs) - local_engine re-export 保持旧引用兼容；selection/engine 的评分面板构建均指向 composite —— 选股与回测的复合分实现收敛于一处 - 回归：quant/eval/research/selection 一致性/qlib 引擎测试全过；全量 pytest 通过
127 lines
4.9 KiB
Python
127 lines
4.9 KiB
Python
"""ResearchSpec 校验与因子评估（IC/RankIC/分层）测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
from app.domain.entities.research import (
|
|
CostSpec,
|
|
FactorSpec,
|
|
ResearchSpec,
|
|
SelectionSpec,
|
|
UniverseSpec,
|
|
)
|
|
from app.quant.composite import composite_score, cross_sectional_zscore
|
|
from app.quant.evaluation import run_factor_test
|
|
from app.quant.factors import compute_factor
|
|
from app.quant.local_engine import rebalance_dates
|
|
from pydantic import ValidationError
|
|
|
|
from conftest_quant import synthetic_daily
|
|
|
|
|
|
def _spec(start: date = date(2024, 3, 1), end: date = date(2024, 12, 31), **kw) -> ResearchSpec:
|
|
base = dict(
|
|
type="backtest",
|
|
universe=UniverseSpec(exclude_st=False, min_listing_days=0),
|
|
factors=[FactorSpec(name="momentum_20")],
|
|
selection=SelectionSpec(top_n=10),
|
|
rebalance="monthly",
|
|
period=(start, end),
|
|
)
|
|
base.update(kw)
|
|
return ResearchSpec(**base)
|
|
|
|
|
|
class TestSpecValidation:
|
|
def test_inverted_period_rejected(self) -> None:
|
|
with pytest.raises(ValidationError, match="start < end"):
|
|
_spec(start=date(2024, 12, 1), end=date(2024, 1, 1))
|
|
|
|
def test_duplicate_factors_rejected(self) -> None:
|
|
with pytest.raises(ValidationError, match="重复"):
|
|
_spec(factors=[FactorSpec(name="momentum_20"), FactorSpec(name="momentum_20")])
|
|
|
|
def test_nonpositive_weight_rejected(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
_spec(factors=[FactorSpec(name="momentum_20", weight=0)])
|
|
|
|
def test_cost_bounds(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
CostSpec(commission_rate=0.5) # 超过 1%
|
|
with pytest.raises(ValidationError):
|
|
CostSpec(slippage_rate=-0.01)
|
|
|
|
|
|
class TestEvaluation:
|
|
def _panels(self):
|
|
# 强趋势 + 弱噪声，确保截面排序稳定（drift 差异远大于噪声）
|
|
drifts = {f"S{i:02d}": v for i, v in enumerate(np.linspace(0.006, -0.006, 12), start=1)}
|
|
daily = synthetic_daily(drifts, n=260)
|
|
_d, f20 = compute_factor("momentum_20", daily)
|
|
_d, f60 = compute_factor("momentum_60", daily)
|
|
close = daily.pivot(index="trade_date", columns="symbol", values="close").sort_index()
|
|
close.index = pd.to_datetime(close.index)
|
|
return daily, f20, f60, close
|
|
|
|
def test_cross_sectional_zscore_standardized(self) -> None:
|
|
_daily, f20, _f60, _close = self._panels()
|
|
z = cross_sectional_zscore(f20).dropna(how="all")
|
|
row = z.iloc[60].dropna()
|
|
assert abs(float(row.mean())) < 1e-9
|
|
assert abs(float(row.std()) - 1.0) < 1e-6
|
|
|
|
def test_composite_score_respects_direction(self) -> None:
|
|
_daily, f20, _f60, _close = self._panels()
|
|
pos = composite_score([("m", f20, 1.0, "higher_is_better")])
|
|
neg = composite_score([("m", f20, 1.0, "lower_is_better")])
|
|
row_date = f20.dropna(how="all").iloc[100].name
|
|
sym = f20.loc[row_date].dropna().index[0]
|
|
assert float(pos.loc[row_date, sym]) == pytest.approx(-float(neg.loc[row_date, sym]))
|
|
|
|
def test_momentum_ic_positive_on_trend_data(self) -> None:
|
|
daily, f20, _f60, close = self._panels()
|
|
forward = close.shift(-21) / close - 1.0
|
|
report = run_factor_test(f20, forward, factor_name="momentum_20")
|
|
assert report.sample_days > 10
|
|
assert report.ic_mean > 0
|
|
assert report.rank_ic_mean > 0
|
|
assert report.positive_ratio_pct > 50
|
|
|
|
def test_quantile_monotonic_on_trend(self) -> None:
|
|
daily, f20, _f60, close = self._panels()
|
|
forward = close.shift(-21) / close - 1.0
|
|
report = run_factor_test(f20, forward, factor_name="momentum_20", quantiles=5)
|
|
qs = {q.quantile: q.return_pct for q in report.quantile_returns}
|
|
assert qs[4] > qs[0] # 高动量层未来收益高于低动量层
|
|
assert report.spread_quantile is not None
|
|
|
|
def test_rebalance_dates_monthly_first(self) -> None:
|
|
idx = pd.bdate_range("2024-03-01", "2024-05-31")
|
|
out = rebalance_dates(idx, "monthly", date(2024, 3, 1))
|
|
assert [d.strftime("%Y-%m-%d") for d in out][:3] == [
|
|
"2024-03-01",
|
|
"2024-04-01",
|
|
"2024-05-01",
|
|
]
|
|
|
|
def test_rebalance_dates_respects_start(self) -> None:
|
|
idx = pd.bdate_range("2024-03-01", "2024-05-31")
|
|
out = rebalance_dates(idx, "monthly", date(2024, 4, 10))
|
|
assert out and out[0] >= pd.Timestamp("2024-04-10")
|
|
|
|
|
|
class TestSingleStockDegradation:
|
|
def test_zscore_single_stock_keeps_candidate(self) -> None:
|
|
from app.quant.composite import cross_sectional_zscore
|
|
|
|
daily = synthetic_daily({"ONLY": 0.001}, n=80)
|
|
_d, panel = compute_factor("momentum_20", daily)
|
|
z = cross_sectional_zscore(panel)
|
|
valid = z.dropna(how="all")
|
|
assert not valid.empty
|
|
assert (valid == 0.0).all().all() # 单股退化为 0，而非 NaN
|