Simon
e9f59d3cf8
- domain：ResearchSpec（universe/factors/selection/rebalance/costs 校验）+ 标准化 BacktestResult / FactorTestReport - 因子引擎：注册表 + 元数据，内置 9 个行情因子（momentum/volatility/量比/乖离/反转），支持自定义注册；只用行情字段规避未来函数 - 评估：横截面 IC / RankIC（rank+pearson 免 scipy）/ ICIR / 分层收益 - 回测：TopK 等权低频，无未来函数记账（t 收盘成交、自 t+1 计收益），成本/涨跌停/停牌约束，未建模项显式写入 unimplemented（AGENT §24） - 引擎抽象 QuantEngine + LocalEngine（pandas 默认实现）；qlib_adapter 桥接占位 —— pyqlib 无 aarch64+cp312 wheel（ROADMAP 已备注） - 真实链路冒烟：600519 2024 月度动量回测闭环产出标准结果 - 测试 60 passed / ruff clean
90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
"""因子计算与注册表测试（合成数据、确定性断言方向与相对排序）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pandas as pd
|
||
from app.quant.factors import (
|
||
FactorDef,
|
||
FactorError,
|
||
compute_factor,
|
||
get_factor,
|
||
list_factors,
|
||
register,
|
||
)
|
||
|
||
from conftest_quant import synthetic_daily
|
||
|
||
|
||
def test_registry_builtins_present() -> None:
|
||
names = {f.name for f in list_factors()}
|
||
assert {"momentum_20", "momentum_60", "volatility_20", "ma_bias_20"} <= names
|
||
|
||
|
||
def test_unknown_factor_raises() -> None:
|
||
try:
|
||
compute_factor("not_a_factor", synthetic_daily({"A": 0.0}))
|
||
except FactorError:
|
||
return
|
||
raise AssertionError("应抛 FactorError")
|
||
|
||
|
||
def test_momentum_ordering_matches_drift() -> None:
|
||
daily = synthetic_daily({"AAA": 0.002, "BBB": 0.0, "CCC": -0.002}, n=160)
|
||
_defn, panel = compute_factor("momentum_20", daily)
|
||
tail = panel.iloc[-1]
|
||
assert tail["AAA"] > tail["BBB"] > tail["CCC"]
|
||
assert tail["AAA"] > 0 # 上涨股 20 日动量为正
|
||
|
||
|
||
def test_volatility_ranks_noise() -> None:
|
||
# 手写：SMOOTH 每日 +0.2%；WILD 在 ±5% 间摆动 → WILD 的 20 日波动率应显著更高
|
||
|
||
from datetime import date, timedelta
|
||
|
||
dates = [date(2024, 1, 1) + timedelta(days=i) for i in range(120)]
|
||
rows = []
|
||
smooth, wild = 100.0, 100.0
|
||
for j, d in enumerate(dates):
|
||
smooth *= 1.002
|
||
wild *= 1.05 if j % 2 == 0 else 0.95
|
||
for sym, px in (("SMOOTH", smooth), ("WILD", wild)):
|
||
rows.append(
|
||
{
|
||
"symbol": sym,
|
||
"trade_date": d,
|
||
"close": px,
|
||
"high": px,
|
||
"low": px,
|
||
"volume": 1e6,
|
||
"amount": 1e8,
|
||
}
|
||
)
|
||
daily = pd.DataFrame(rows)
|
||
_defn, panel = compute_factor("volatility_20", daily)
|
||
assert float(panel["WILD"].iloc[-1]) > float(panel["SMOOTH"].iloc[-1]) * 5
|
||
|
||
|
||
def test_custom_factor_registration() -> None:
|
||
@register(FactorDef("test_double_close", "close*2 测试因子", "close * 2", lookback=1))
|
||
def _fn(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
|
||
return fields["close"] * 2
|
||
|
||
try:
|
||
daily = synthetic_daily({"A": 0.001}, n=40)
|
||
defn, panel = compute_factor("test_double_close", daily)
|
||
assert defn.direction == "higher_is_better"
|
||
assert float(panel.iloc[-1, 0]) > 200.0
|
||
finally:
|
||
# 清理注册表，避免污染其他测试
|
||
from app.quant import factors as _factors
|
||
|
||
_factors._REGISTRY.pop("test_double_close", None) # noqa: SLF001
|
||
|
||
|
||
def test_factor_def_metadata_present() -> None:
|
||
defn, _fn = get_factor("momentum_60")
|
||
assert defn.description
|
||
assert defn.formula
|
||
assert defn.lookback == 60
|
||
assert defn.direction in {"higher_is_better", "lower_is_better"}
|