2160 lines
69 KiB
Python
2160 lines
69 KiB
Python
"""src/shared/alpha_factors.py 单元测试（v1.2.0 Phase 0 第一批）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
import quant_engine.alpha_factors as alpha_factors_module
|
||
from quant_engine.alpha_factors import (
|
||
ALPHA158_REGISTRY,
|
||
ALPHA158_PHASE1_OPERATOR_SPECS,
|
||
ALPHA158_PHASE2_OPERATOR_SPECS,
|
||
ALPHA158_PHASE3_FORMULA_CATALOG_SHA256,
|
||
ALPHA158_PHASE3_FORMULA_CONTRACT_VERSION,
|
||
ALPHA158_PHASE3_FORMULA_SPECS,
|
||
ALPHA158_PHASE4_FORMULA_CATALOG_SHA256,
|
||
ALPHA158_PHASE4_FORMULA_CONTRACT_VERSION,
|
||
ALPHA158_PHASE4_FORMULA_SPECS,
|
||
ALPHA158_PHASE5_FORMULA_CATALOG_SHA256,
|
||
ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION,
|
||
ALPHA158_PHASE5_FORMULA_SPECS,
|
||
ALPHA158_PHASE6_FORMULA_CATALOG_SHA256,
|
||
ALPHA158_PHASE6_FORMULA_CONTRACT_VERSION,
|
||
ALPHA158_PHASE6_FORMULA_SPECS,
|
||
alpha_001,
|
||
alpha_002,
|
||
alpha_003,
|
||
alpha_004,
|
||
alpha_005,
|
||
alpha_006,
|
||
alpha_007,
|
||
alpha_008,
|
||
alpha_009,
|
||
alpha_010,
|
||
alpha_011,
|
||
alpha_012,
|
||
alpha_013,
|
||
alpha_014,
|
||
alpha_015,
|
||
alpha_016,
|
||
alpha_017,
|
||
alpha_018,
|
||
alpha_019,
|
||
alpha_020,
|
||
alpha_021,
|
||
alpha_022,
|
||
alpha_023,
|
||
alpha_024,
|
||
alpha_025,
|
||
alpha_026,
|
||
alpha_027,
|
||
alpha_028,
|
||
alpha_029,
|
||
alpha_030,
|
||
alpha_031,
|
||
alpha_032,
|
||
alpha_033,
|
||
alpha_034,
|
||
alpha_035,
|
||
alpha_036,
|
||
alpha_037,
|
||
alpha_038,
|
||
alpha_039,
|
||
alpha_040,
|
||
alpha_041,
|
||
alpha_042,
|
||
alpha_043,
|
||
alpha_044,
|
||
alpha_045,
|
||
alpha_046,
|
||
alpha_047,
|
||
alpha_048,
|
||
alpha_049,
|
||
alpha_050,
|
||
alpha_051,
|
||
alpha_052,
|
||
alpha_053,
|
||
alpha_054,
|
||
alpha_055,
|
||
alpha_056,
|
||
alpha_057,
|
||
alpha_058,
|
||
alpha_059,
|
||
alpha_060,
|
||
alpha_061,
|
||
alpha_062,
|
||
alpha_063,
|
||
alpha_064,
|
||
alpha_065,
|
||
alpha_066,
|
||
alpha_067,
|
||
alpha_068,
|
||
alpha_069,
|
||
alpha_070,
|
||
alpha_071,
|
||
alpha_072,
|
||
alpha_073,
|
||
alpha_074,
|
||
alpha_075,
|
||
alpha_076,
|
||
alpha_077,
|
||
alpha_078,
|
||
alpha_079,
|
||
alpha_080,
|
||
alpha_081,
|
||
alpha_082,
|
||
alpha_083,
|
||
alpha_084,
|
||
alpha_085,
|
||
alpha_086,
|
||
alpha_087,
|
||
alpha_088,
|
||
alpha_089,
|
||
alpha_090,
|
||
alpha_091,
|
||
alpha_092,
|
||
alpha_093,
|
||
alpha_094,
|
||
alpha_095,
|
||
alpha_096,
|
||
alpha_097,
|
||
alpha_098,
|
||
alpha_099,
|
||
alpha_100,
|
||
alpha_101,
|
||
alpha_102,
|
||
alpha_103,
|
||
alpha_104,
|
||
alpha_105,
|
||
alpha_106,
|
||
alpha_107,
|
||
alpha_108,
|
||
alpha_109,
|
||
alpha_110,
|
||
alpha_111,
|
||
alpha_112,
|
||
alpha_113,
|
||
alpha_114,
|
||
alpha_115,
|
||
alpha_116,
|
||
alpha_117,
|
||
alpha_118,
|
||
alpha_119,
|
||
alpha_120,
|
||
alpha_121,
|
||
alpha_122,
|
||
alpha_123,
|
||
alpha_124,
|
||
alpha_125,
|
||
alpha_126,
|
||
alpha_127,
|
||
alpha_128,
|
||
alpha_129,
|
||
alpha_130,
|
||
alpha_131,
|
||
alpha_132,
|
||
alpha_133,
|
||
alpha_134,
|
||
alpha_135,
|
||
alpha_136,
|
||
alpha_137,
|
||
alpha_138,
|
||
alpha_139,
|
||
alpha_140,
|
||
alpha_141,
|
||
alpha_142,
|
||
alpha_143,
|
||
alpha_144,
|
||
alpha_145,
|
||
alpha_146,
|
||
alpha_147,
|
||
alpha_148,
|
||
alpha_149,
|
||
alpha_150,
|
||
alpha_151,
|
||
alpha_152,
|
||
alpha_153,
|
||
alpha_154,
|
||
alpha_155,
|
||
alpha_156,
|
||
alpha_157,
|
||
alpha_158,
|
||
evaluate_phase1_operator,
|
||
evaluate_phase2_operator,
|
||
evaluate_phase3_formula,
|
||
evaluate_phase4_formula,
|
||
evaluate_phase5_formula,
|
||
evaluate_phase6_formula,
|
||
list_phase1_operators,
|
||
list_phase2_operators,
|
||
list_phase3_formulas,
|
||
list_phase4_formulas,
|
||
list_phase5_formulas,
|
||
list_phase6_formulas,
|
||
correlation,
|
||
covariance,
|
||
decay_linear,
|
||
delta,
|
||
get_alpha_meta,
|
||
indneutralize,
|
||
log,
|
||
max_pair,
|
||
min_pair,
|
||
product,
|
||
rank,
|
||
returns,
|
||
scale,
|
||
sign,
|
||
signed_power,
|
||
stddev,
|
||
ts_argmax,
|
||
ts_argmin,
|
||
ts_max,
|
||
ts_mean,
|
||
ts_min,
|
||
ts_rank,
|
||
ts_std,
|
||
ts_sum,
|
||
)
|
||
|
||
|
||
# ── rank 基础算子 ──────────────────────────────
|
||
|
||
|
||
def test_rank_uniform_input():
|
||
"""全相同输入 → 所有位置相同（pandas average tie-breaking）。"""
|
||
s = pd.Series([1.0, 1.0, 1.0, 1.0])
|
||
result = rank(s)
|
||
# pandas rank default method='average': ties get average rank
|
||
# 4 ties at positions 1,2,3,4 → average 2.5 → 2.5/4 = 0.625
|
||
expected = pd.Series([0.625, 0.625, 0.625, 0.625])
|
||
pd.testing.assert_series_equal(result, expected)
|
||
|
||
|
||
def test_rank_basic():
|
||
"""5 个升序值 → 0.2, 0.4, 0.6, 0.8, 1.0（pandas rank pct=True）。"""
|
||
s = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0])
|
||
result = rank(s)
|
||
expected = pd.Series([0.2, 0.4, 0.6, 0.8, 1.0])
|
||
pd.testing.assert_series_equal(result, expected)
|
||
|
||
|
||
def test_rank_empty():
|
||
"""空 series 返回空 series。"""
|
||
s = pd.Series([], dtype=float)
|
||
result = rank(s)
|
||
assert len(result) == 0
|
||
|
||
|
||
# ── delta 基础算子 ──────────────────────────────
|
||
|
||
|
||
def test_delta_basic():
|
||
"""5 期 delta。"""
|
||
s = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0, 6.0])
|
||
result = delta(s, 5)
|
||
expected = pd.Series([np.nan, np.nan, np.nan, np.nan, np.nan, 5.0])
|
||
pd.testing.assert_series_equal(result, expected)
|
||
|
||
|
||
def test_delta_zero():
|
||
"""delta n=0 → 全部 NaN（无前移）。"""
|
||
s = pd.Series([1.0, 2.0, 3.0])
|
||
result = delta(s, 0)
|
||
# shift(0) is identity, so result is all zeros except first NaN
|
||
assert result.iloc[0] == 0.0 or pd.isna(result.iloc[0])
|
||
|
||
|
||
def test_delta_negative_raises():
|
||
"""delta n<0 应报错。"""
|
||
s = pd.Series([1.0, 2.0, 3.0])
|
||
with pytest.raises(ValueError, match="non-negative"):
|
||
delta(s, -1)
|
||
|
||
|
||
# ── ts_mean 基础算子 ──────────────────────────────
|
||
|
||
|
||
def test_ts_mean_basic():
|
||
"""20 期均值。"""
|
||
s = pd.Series(range(20), dtype=float)
|
||
result = ts_mean(s, 5)
|
||
# 最后 5 个值 15,16,17,18,19 均值 17
|
||
assert result.iloc[-1] == 17.0
|
||
|
||
|
||
def test_ts_mean_n_1_raises():
|
||
"""ts_mean n<=0 应报错。"""
|
||
s = pd.Series([1.0, 2.0, 3.0])
|
||
with pytest.raises(ValueError, match="positive"):
|
||
ts_mean(s, 0)
|
||
|
||
|
||
# ── ts_std 基础算子 ──────────────────────────────
|
||
|
||
|
||
def test_ts_std_basic():
|
||
"""常量 series → std = 0。"""
|
||
s = pd.Series([5.0] * 20)
|
||
result = ts_std(s, 5)
|
||
assert result.iloc[-1] == 0.0
|
||
|
||
|
||
def test_ts_std_n_1_raises():
|
||
"""ts_std n<=0 应报错。"""
|
||
s = pd.Series([1.0, 2.0, 3.0])
|
||
with pytest.raises(ValueError, match="positive"):
|
||
ts_std(s, -1)
|
||
|
||
|
||
# ── ts_rank 基础算子 ──────────────────────────────
|
||
|
||
|
||
def test_ts_rank_basic():
|
||
"""时序 rank：最后一个值在窗口内应排 1.0。"""
|
||
s = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0])
|
||
result = ts_rank(s, 5)
|
||
# 最后一个值 5 在 [1,2,3,4,5] 内 rank = 1.0
|
||
assert result.iloc[-1] == 1.0
|
||
|
||
|
||
def test_ts_rank_n_1_raises():
|
||
"""ts_rank n<=0 应报错。"""
|
||
s = pd.Series([1.0, 2.0, 3.0])
|
||
with pytest.raises(ValueError, match="positive"):
|
||
ts_rank(s, 0)
|
||
|
||
|
||
# ── correlation 基础算子 ──────────────────────────────
|
||
|
||
|
||
def test_correlation_perfect_positive():
|
||
"""完全正相关 → 1.0。"""
|
||
s1 = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0])
|
||
s2 = s1 * 2 + 1
|
||
result = correlation(s1, s2, 5)
|
||
assert result.iloc[-1] == pytest.approx(1.0)
|
||
|
||
|
||
def test_correlation_perfect_negative():
|
||
"""完全负相关 → -1.0。"""
|
||
s1 = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0])
|
||
s2 = -s1
|
||
result = correlation(s1, s2, 5)
|
||
assert result.iloc[-1] == pytest.approx(-1.0)
|
||
|
||
|
||
def test_correlation_n_1_raises():
|
||
"""correlation n<=0 应报错。"""
|
||
s1 = pd.Series([1.0, 2.0, 3.0])
|
||
s2 = pd.Series([3.0, 2.0, 1.0])
|
||
with pytest.raises(ValueError, match="positive"):
|
||
correlation(s1, s2, 0)
|
||
|
||
|
||
# ── alpha_001 公式 ──────────────────────────────
|
||
|
||
|
||
def test_alpha_001_smoke():
|
||
"""alpha_001 端到端 smoke test。"""
|
||
idx = pd.MultiIndex.from_product(
|
||
[["d1", "d2", "d3", "d4", "d5"], ["s1", "s2", "s3"]],
|
||
names=["date", "stock"],
|
||
)
|
||
close = pd.Series(np.random.rand(15), index=idx)
|
||
result = alpha_001(close)
|
||
assert isinstance(result, pd.Series)
|
||
assert result.shape == close.shape
|
||
|
||
|
||
def test_alpha_001_constant_input():
|
||
"""常量输入 → ts_rank 在窗口内全 0.5 → rank 后仍全相同。"""
|
||
s = pd.Series([5.0] * 10)
|
||
result = alpha_001(s)
|
||
# 全 0.5 输入 → ts_rank = 0.5 → rank 在 6 个 0.5 之间取平均
|
||
# 6 个 ties at positions 1..6 → average 3.5 → 3.5/6 ≈ 0.5833
|
||
expected_value = 3.5 / 6.0
|
||
valid = result.dropna()
|
||
for v in valid:
|
||
assert v == pytest.approx(expected_value)
|
||
|
||
|
||
# ── alpha_002 公式 ──────────────────────────────
|
||
|
||
|
||
def test_alpha_002_matches_delta():
|
||
"""alpha_002 应等于 delta(close, 5)。"""
|
||
close = pd.Series(range(20), dtype=float)
|
||
assert alpha_002(close).equals(delta(close, 5))
|
||
|
||
|
||
# ── alpha_003 公式 ──────────────────────────────
|
||
|
||
|
||
def test_alpha_003_matches_ts_mean():
|
||
"""alpha_003 应等于 ts_mean(close, 20)。"""
|
||
close = pd.Series(range(30), dtype=float)
|
||
assert alpha_003(close).equals(ts_mean(close, 20))
|
||
|
||
|
||
# ── alpha_004 公式 ──────────────────────────────
|
||
|
||
|
||
def test_alpha_004_matches_ts_std():
|
||
"""alpha_004 应等于 ts_std(close, 20)。"""
|
||
close = pd.Series(range(30), dtype=float)
|
||
assert alpha_004(close).equals(ts_std(close, 20))
|
||
|
||
|
||
# ── alpha_005 公式 ──────────────────────────────
|
||
|
||
|
||
def test_alpha_005_matches_correlation():
|
||
"""alpha_005 应等于 correlation(close, volume, 10)。"""
|
||
close = pd.Series(range(20), dtype=float)
|
||
volume = pd.Series([1.0] * 10 + [2.0] * 10)
|
||
assert alpha_005(close, volume).equals(correlation(close, volume, 10))
|
||
|
||
|
||
# ── 元数据 ──────────────────────────────
|
||
|
||
|
||
def test_alpha_registry_complete():
|
||
"""158 个 alpha 算子全部在 registry 中（v1.2.0 第八批后——达成完整 alpha158）。"""
|
||
expected = {f"alpha_{i:03d}" for i in range(1, 159)}
|
||
assert set(ALPHA158_REGISTRY.keys()) == expected
|
||
|
||
|
||
def test_alpha_registry_required_fields():
|
||
"""每个 registry 条目都包含必需字段。"""
|
||
required = {"formula", "category", "complexity", "params", "description", "inputs"}
|
||
for alpha_id, meta in ALPHA158_REGISTRY.items():
|
||
assert required <= set(meta.keys()), f"{alpha_id} missing fields"
|
||
|
||
|
||
def test_get_alpha_meta_success():
|
||
"""已知 alpha_id 返回完整 meta。"""
|
||
meta = get_alpha_meta("alpha_001")
|
||
assert meta["formula"] == "rank(ts_rank(close, 5))"
|
||
assert meta["category"] == "momentum_rank"
|
||
|
||
|
||
def test_get_alpha_meta_unknown():
|
||
"""未知 alpha_id 抛 KeyError。"""
|
||
with pytest.raises(KeyError, match="alpha_999"):
|
||
get_alpha_meta("alpha_999")
|
||
|
||
|
||
def test_alpha_meta_jsonb_serializable():
|
||
"""元数据可以 JSON 序列化（PostgreSQL JSONB 落库前置验证）。"""
|
||
import json
|
||
|
||
for alpha_id, meta in ALPHA158_REGISTRY.items():
|
||
# 抛异常即失败
|
||
json.dumps(meta)
|
||
|
||
|
||
# ── v1.2.0 第二批基础算子（ts_min / ts_max / ts_sum / ts_argmin / ts_argmax / decay_linear / product / returns / scale / signed_power） ──
|
||
|
||
|
||
def test_ts_min_basic():
|
||
"""5 期滚动最小值。"""
|
||
s = pd.Series([5.0, 3.0, 8.0, 1.0, 6.0, 2.0])
|
||
assert ts_min(s, 5).iloc[-1] == 1.0
|
||
|
||
|
||
def test_ts_max_basic():
|
||
"""5 期滚动最大值。"""
|
||
s = pd.Series([5.0, 3.0, 8.0, 1.0, 6.0, 2.0])
|
||
assert ts_max(s, 5).iloc[-1] == 8.0
|
||
|
||
|
||
def test_ts_sum_basic():
|
||
"""5 期滚动求和。"""
|
||
s = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0, 6.0])
|
||
assert ts_sum(s, 5).iloc[-1] == 20.0 # 2+3+4+5+6
|
||
|
||
|
||
def test_ts_argmin_first_position():
|
||
"""最小值在窗口起点 → argmin = 0。"""
|
||
s = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0])
|
||
assert ts_argmin(s, 5).iloc[-1] == 0.0
|
||
|
||
|
||
def test_ts_argmax_last_position():
|
||
"""最大值在窗口终点 → argmax = 4。"""
|
||
s = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0])
|
||
assert ts_argmax(s, 5).iloc[-1] == 4.0
|
||
|
||
|
||
def test_decay_linear_recent_weighted():
|
||
"""线性衰减：近期权重大。"""
|
||
s = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0])
|
||
weights = np.array([1.0, 2.0, 3.0, 4.0, 5.0])
|
||
weights /= weights.sum() # [0.0667, 0.1333, 0.2, 0.2667, 0.3333]
|
||
expected = float(np.dot(s.values, weights))
|
||
assert decay_linear(s, 5).iloc[-1] == pytest.approx(expected)
|
||
|
||
|
||
def test_product_basic():
|
||
"""5 期滚动乘积。"""
|
||
s = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0])
|
||
assert product(s, 5).iloc[-1] == 120.0
|
||
|
||
|
||
def test_returns_basic():
|
||
"""简单收益率。"""
|
||
s = pd.Series([100.0, 110.0, 121.0])
|
||
assert returns(s).iloc[1] == pytest.approx(0.1)
|
||
assert returns(s).iloc[2] == pytest.approx(0.1)
|
||
|
||
|
||
def test_scale_sum_to_one():
|
||
"""scale 让 |x| 之和 = 1。"""
|
||
s = pd.Series([1.0, -2.0, 3.0, -4.0])
|
||
result = scale(s)
|
||
assert result.abs().sum() == pytest.approx(1.0)
|
||
|
||
|
||
def test_scale_zero_input():
|
||
"""scale 全 0 输入 → 原样返回（避免除零）。"""
|
||
s = pd.Series([0.0, 0.0, 0.0])
|
||
result = scale(s)
|
||
assert (result == 0.0).all()
|
||
|
||
|
||
def test_signed_power_positive():
|
||
"""正数 signed_power(2) → x^2。"""
|
||
s = pd.Series([3.0, 4.0])
|
||
result = signed_power(s, 2)
|
||
assert result.iloc[0] == pytest.approx(9.0)
|
||
assert result.iloc[1] == pytest.approx(16.0)
|
||
|
||
|
||
def test_signed_power_negative():
|
||
"""负数 signed_power(2) → -|x|^2。"""
|
||
s = pd.Series([-3.0])
|
||
result = signed_power(s, 2)
|
||
assert result.iloc[0] == pytest.approx(-9.0)
|
||
|
||
|
||
def test_ts_min_n_1_raises():
|
||
"""ts_min n<=0 应报错。"""
|
||
with pytest.raises(ValueError, match="positive"):
|
||
ts_min(pd.Series([1.0]), 0)
|
||
|
||
|
||
def test_ts_max_n_1_raises():
|
||
with pytest.raises(ValueError, match="positive"):
|
||
ts_max(pd.Series([1.0]), -1)
|
||
|
||
|
||
def test_decay_linear_n_1_raises():
|
||
with pytest.raises(ValueError, match="positive"):
|
||
decay_linear(pd.Series([1.0]), 0)
|
||
|
||
|
||
# ── v1.2.0 第二批 alpha 公式（alpha_006 – alpha_015） ─────
|
||
|
||
|
||
def test_alpha_006_smoke():
|
||
"""alpha_006 = rank(close - open) smoke test。"""
|
||
open_ = pd.Series([10.0, 11.0, 12.0])
|
||
close = pd.Series([11.0, 10.0, 13.0]) # diff: +1, -1, +1
|
||
result = alpha_006(open_, close)
|
||
# diff 序列 [1, -1, 1] → pandas rank pct=True：ranks [2.5, 1, 2.5] → pct [0.833, 0.333, 0.833]
|
||
assert result.iloc[0] == pytest.approx(5.0 / 6)
|
||
assert result.iloc[1] == pytest.approx(1.0 / 3)
|
||
assert result.iloc[2] == pytest.approx(5.0 / 6)
|
||
|
||
|
||
def test_alpha_007_volume_trend():
|
||
"""alpha_007 = mean(volume, 7) - mean(volume, 14)。"""
|
||
volume = pd.Series(range(20), dtype=float)
|
||
result = alpha_007(volume)
|
||
expected_last = volume.iloc[-7:].mean() - volume.iloc[-14:].mean()
|
||
assert result.iloc[-1] == pytest.approx(expected_last)
|
||
|
||
|
||
def test_alpha_008_smoke():
|
||
"""alpha_008 = rank(delta(vwap, 5))。"""
|
||
vwap = pd.Series(range(20), dtype=float)
|
||
result = alpha_008(vwap)
|
||
assert isinstance(result, pd.Series)
|
||
|
||
|
||
def test_alpha_009_min_rank():
|
||
"""alpha_009 = rank(ts_min(low, 5))。"""
|
||
low = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0])
|
||
result = alpha_009(low)
|
||
# 5 期 ts_min 都 = 1.0 → rank 全相同
|
||
valid = result.dropna()
|
||
assert valid.nunique() == 1
|
||
|
||
|
||
def test_alpha_010_max_rank():
|
||
"""alpha_010 = rank(ts_max(high, 5))。"""
|
||
high = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0])
|
||
result = alpha_010(high)
|
||
# 5 期 ts_max 都 = 5.0 → rank 全相同
|
||
valid = result.dropna()
|
||
assert valid.nunique() == 1
|
||
|
||
|
||
def test_alpha_011_constant_range():
|
||
"""alpha_011 高=低时（无振幅）→ 返回 0（safe_diff 处理）。"""
|
||
open_ = pd.Series([10.0, 11.0])
|
||
close = pd.Series([10.0, 11.0])
|
||
high = pd.Series([10.0, 11.0])
|
||
low = pd.Series([10.0, 11.0])
|
||
result = alpha_011(open_, close, high, low)
|
||
assert (result == 0.0).all()
|
||
|
||
|
||
def test_alpha_011_bullish():
|
||
"""alpha_011 阳线 close > open → 正值（多头强度）。"""
|
||
open_ = pd.Series([10.0] * 5)
|
||
close = pd.Series([10.5] * 5)
|
||
high = pd.Series([11.0] * 5)
|
||
low = pd.Series([9.5] * 5)
|
||
result = alpha_011(open_, close, high, low)
|
||
# close - low = 1.0; high - close = 0.5; high - low = 1.5
|
||
# (1.0 - 0.5) / 1.5 = 0.333
|
||
for v in result:
|
||
assert v == pytest.approx(1.0 / 3.0)
|
||
|
||
|
||
def test_alpha_012_volume_rank_change():
|
||
"""alpha_012 = rank(volume) - rank(volume.shift(5))。"""
|
||
volume = pd.Series(range(20), dtype=float)
|
||
result = alpha_012(volume)
|
||
# 前 5 个应为 NaN
|
||
assert result.iloc[:5].isna().all()
|
||
|
||
|
||
def test_alpha_013_returns_rank_change():
|
||
"""alpha_013 = rank(returns) - rank(returns.shift(3))。"""
|
||
close = pd.Series(range(20), dtype=float)
|
||
result = alpha_013(close)
|
||
# 前 4 个 NaN（returns 第一行 + shift(3) 3 行）
|
||
assert result.iloc[:4].isna().all()
|
||
|
||
|
||
def test_alpha_014_combined_momentum():
|
||
"""alpha_014 = -rank(delta(returns(open_), 3)) * correlation(open_, volume, 10)。"""
|
||
open_ = pd.Series(range(20), dtype=float)
|
||
volume = pd.Series([1.0] * 20)
|
||
result = alpha_014(open_, volume)
|
||
assert isinstance(result, pd.Series)
|
||
|
||
|
||
def test_alpha_015_high_volume_combo():
|
||
"""alpha_015 = -1 * ts_sum(rank(correlation(rank(high), rank(volume), 5)), 5)。
|
||
|
||
完全相关输入（high 和 volume 同方向）→ correlation = 1.0；
|
||
但 ts_sum 5 期窗口内 rank(1.0) 因 tie-breaking 在 [0, 1] 之间分布，
|
||
所以最后 5 期总和约为 -3.6875（实测值）。
|
||
"""
|
||
high = pd.Series(range(20), dtype=float)
|
||
volume = pd.Series(range(20), dtype=float)
|
||
result = alpha_015(high, volume)
|
||
# 取最后一个完全有效值（rolling warm-up 后）
|
||
valid_last = result.dropna().iloc[-1]
|
||
# 实测：-3.6875（具体数值取决于 pandas rank tie-breaking）
|
||
assert valid_last < 0 # 公式有 -1 系数
|
||
assert valid_last > -5.5 # 不会低于完全负相关
|
||
assert valid_last == pytest.approx(-3.6875, abs=1e-4)
|
||
|
||
|
||
def test_get_alpha_meta_alpha_006_to_015():
|
||
"""alpha_006 到 alpha_015 全部在 registry 中。"""
|
||
for i in range(6, 16):
|
||
alpha_id = f"alpha_{i:03d}"
|
||
meta = get_alpha_meta(alpha_id)
|
||
assert "formula" in meta
|
||
assert "inputs" in meta
|
||
assert len(meta["inputs"]) >= 1
|
||
|
||
|
||
# ── v1.2.0 第三批基础算子（stddev / covariance / log / abs / sign / max_pair / min_pair / indneutralize） ──
|
||
|
||
|
||
def test_stddev_alias():
|
||
"""stddev 与 ts_std 同义。"""
|
||
s = pd.Series(range(20), dtype=float)
|
||
assert stddev(s, 5).equals(ts_std(s, 5))
|
||
|
||
|
||
def test_covariance_perfect():
|
||
"""完全正相关 → 协方差 ≈ var(x)。"""
|
||
s1 = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0])
|
||
s2 = s1 * 2 + 1
|
||
cov = covariance(s1, s2, 5).iloc[-1]
|
||
var = s1.var()
|
||
# cov = corr * std1 * std2 = 1 * std1 * (2*std1) = 2*var
|
||
assert cov == pytest.approx(2 * var)
|
||
|
||
|
||
def test_log_basic():
|
||
"""log(e) ≈ 1。"""
|
||
s = pd.Series([1.0, np.e, np.e**2])
|
||
assert log(s).iloc[0] == pytest.approx(0.0)
|
||
assert log(s).iloc[1] == pytest.approx(1.0)
|
||
assert log(s).iloc[2] == pytest.approx(2.0)
|
||
|
||
|
||
def test_log_negative_input():
|
||
"""log(<=0) → NaN。"""
|
||
s = pd.Series([-1.0, 0.0, 1.0])
|
||
assert pd.isna(log(s).iloc[0])
|
||
assert pd.isna(log(s).iloc[1])
|
||
assert log(s).iloc[2] == pytest.approx(0.0)
|
||
|
||
|
||
def test_abs_basic():
|
||
"""abs_series: |−3| = 3。"""
|
||
from quant_engine.alpha_factors import abs_series
|
||
|
||
assert abs_series(pd.Series([-3.0, 0.0, 5.0])).tolist() == [3.0, 0.0, 5.0]
|
||
|
||
|
||
def test_sign_basic():
|
||
"""sign 函数。"""
|
||
assert sign(pd.Series([-5.0, 0.0, 5.0])).tolist() == [-1.0, 0.0, 1.0]
|
||
|
||
|
||
def test_max_pair():
|
||
"""逐元素 max。"""
|
||
s1 = pd.Series([1.0, 5.0, 3.0])
|
||
s2 = pd.Series([4.0, 2.0, 6.0])
|
||
assert max_pair(s1, s2).tolist() == [4.0, 5.0, 6.0]
|
||
|
||
|
||
def test_min_pair():
|
||
"""逐元素 min。"""
|
||
s1 = pd.Series([1.0, 5.0, 3.0])
|
||
s2 = pd.Series([4.0, 2.0, 6.0])
|
||
assert min_pair(s1, s2).tolist() == [1.0, 2.0, 3.0]
|
||
|
||
|
||
def test_indneutralize():
|
||
"""组内中性化：减去组均值。"""
|
||
s = pd.Series([1.0, 2.0, 3.0, 4.0], index=["a", "a", "b", "b"])
|
||
groups = pd.Series(["g1", "g1", "g2", "g2"], index=s.index)
|
||
result = indneutralize(s, groups)
|
||
# g1 均值 = 1.5；g2 均值 = 3.5
|
||
assert result.iloc[0] == pytest.approx(-0.5)
|
||
assert result.iloc[1] == pytest.approx(0.5)
|
||
assert result.iloc[2] == pytest.approx(-0.5)
|
||
assert result.iloc[3] == pytest.approx(0.5)
|
||
|
||
|
||
# ── v1.2.0 第三批 alpha 公式（alpha_016 – alpha_035） ─────
|
||
|
||
|
||
def test_alpha_016_smoke():
|
||
"""alpha_016 = rank(decay_linear(correlation(rank(high), rank(volume), 5), 5))。"""
|
||
high = pd.Series(np.random.rand(30).cumsum())
|
||
volume = pd.Series(np.random.rand(30) * 100 + 1.0)
|
||
result = alpha_016(high, volume)
|
||
assert isinstance(result, pd.Series)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_alpha_017_smoke():
|
||
"""alpha_017 smoke test。"""
|
||
close = pd.Series(np.random.rand(30).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(30) * 100 + 1.0)
|
||
result = alpha_017(close, volume)
|
||
assert isinstance(result, pd.Series)
|
||
|
||
|
||
def test_alpha_018_smoke():
|
||
"""alpha_018 smoke test。"""
|
||
open_ = pd.Series(np.random.rand(30).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(30) * 100 + 1.0)
|
||
result = alpha_018(open_, volume)
|
||
assert isinstance(result, pd.Series)
|
||
|
||
|
||
def test_alpha_019_smoke():
|
||
"""alpha_019 smoke test。"""
|
||
close = pd.Series(np.random.rand(30).cumsum() + 10)
|
||
open_ = pd.Series(np.random.rand(30).cumsum() + 10)
|
||
result = alpha_019(close, open_)
|
||
assert isinstance(result, pd.Series)
|
||
|
||
|
||
def test_alpha_020_smoke():
|
||
"""alpha_020 smoke test。"""
|
||
open_ = pd.Series(np.random.rand(30).cumsum() + 10)
|
||
close = pd.Series(np.random.rand(30).cumsum() + 10)
|
||
result = alpha_020(open_, close)
|
||
assert isinstance(result, pd.Series)
|
||
|
||
|
||
def test_alpha_021_volume_ratio():
|
||
"""alpha_021 = ts_mean(volume, 20) / ts_mean(volume, 60)。"""
|
||
volume = pd.Series([1.0] * 60 + [3.0] * 20)
|
||
result = alpha_021(volume).dropna()
|
||
# 后 20 个 volume=3, 前 40 个=1, ts_mean 20 = 3, ts_mean 60 = (40+60)/60 = 100/60
|
||
expected = 3.0 / (100.0 / 60.0)
|
||
assert result.iloc[-1] == pytest.approx(expected)
|
||
|
||
|
||
def test_alpha_021_zero_long_avg():
|
||
"""alpha_021 分母为 0 → NaN（safe 处理）。"""
|
||
volume = pd.Series([0.0] * 80)
|
||
result = alpha_021(volume)
|
||
assert result.dropna().empty
|
||
|
||
|
||
def test_alpha_022_smoke():
|
||
"""alpha_022 smoke test。"""
|
||
high = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(50) * 100 + 1.0)
|
||
close = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
result = alpha_022(high, volume, close)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_alpha_023_smoke():
|
||
"""alpha_023 = -1 * ts_mean(delta(close, 5), 20) * delta(close, 5) / close。"""
|
||
close = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
result = alpha_023(close)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_alpha_023_zero_close():
|
||
"""close 中含 0 → 跳过（safe_close）。"""
|
||
close = pd.Series([10.0, 11.0, 0.0, 12.0, 13.0] * 10)
|
||
result = alpha_023(close)
|
||
# 不抛异常；NaN 在 0 处
|
||
assert not result.isna().all()
|
||
|
||
|
||
def test_alpha_024_smoke():
|
||
"""alpha_024 smoke test。"""
|
||
close = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(50) * 100 + 1.0)
|
||
result = alpha_024(close, volume)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_alpha_025_smoke():
|
||
"""alpha_025 = rank(decay_linear(correlation(vwap, volume, 4), 8))。"""
|
||
vwap = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(50) * 100 + 1.0)
|
||
result = alpha_025(vwap, volume)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_alpha_026_smoke():
|
||
"""alpha_026 smoke test。"""
|
||
open_ = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(50) * 100 + 1.0)
|
||
close = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
result = alpha_026(open_, volume, close)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_alpha_027_smoke():
|
||
"""alpha_027 smoke test。"""
|
||
close = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(50) * 100 + 1.0)
|
||
result = alpha_027(close, volume)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_alpha_028_scale():
|
||
"""alpha_028 = scale(...) 应满足 |sum| ≈ 1。"""
|
||
close = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
open_ = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
result = alpha_028(close, open_)
|
||
valid = result.dropna()
|
||
assert valid.abs().sum() == pytest.approx(1.0, abs=1e-6)
|
||
|
||
|
||
def test_alpha_029_smoke():
|
||
"""alpha_029 嵌套组合 smoke。"""
|
||
close = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
result = alpha_029(close)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_alpha_030_smoke():
|
||
"""alpha_030 = -1 * rank(decay_linear(correlation(rank(low), rank(volume), 5), 5))。"""
|
||
low = pd.Series(np.random.rand(50).cumsum() + 5)
|
||
volume = pd.Series(np.random.rand(50) * 100 + 1.0)
|
||
result = alpha_030(low, volume)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_alpha_031_smoke():
|
||
"""alpha_031 smoke test。"""
|
||
close = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(50) * 100 + 1.0)
|
||
result = alpha_031(close, volume)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_alpha_032_scale():
|
||
"""alpha_032 = scale(...) → |sum| ≈ 1。"""
|
||
close = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(50) * 100 + 1.0)
|
||
result = alpha_032(close, volume)
|
||
valid = result.dropna()
|
||
assert valid.abs().sum() == pytest.approx(1.0, abs=1e-6)
|
||
|
||
|
||
def test_alpha_033_scale():
|
||
"""alpha_033 = scale(...) → |sum| ≈ 1。"""
|
||
close = pd.Series(np.random.rand(50).cumsum() + 10)
|
||
result = alpha_033(close)
|
||
valid = result.dropna()
|
||
assert valid.abs().sum() == pytest.approx(1.0, abs=1e-6)
|
||
|
||
|
||
def test_alpha_034_volume_ratio():
|
||
"""alpha_034 = ts_mean(volume, 12) / ts_mean(volume, 26)。"""
|
||
volume = pd.Series([1.0] * 50)
|
||
result = alpha_034(volume).dropna()
|
||
# 全部 = 1，ratio = 1.0
|
||
assert result.iloc[-1] == pytest.approx(1.0)
|
||
|
||
|
||
def test_alpha_035_volume_ratio():
|
||
"""alpha_035 = ts_mean(volume, 6) / ts_mean(volume, 12)。"""
|
||
close = pd.Series([1.0] * 30)
|
||
volume = pd.Series([1.0] * 30)
|
||
result = alpha_035(close, volume).dropna()
|
||
assert result.iloc[-1] == pytest.approx(1.0)
|
||
|
||
|
||
# ── v1.2.0 第四批 alpha 公式（alpha_036 – alpha_060） ─────
|
||
|
||
# 共 25 个 smoke test（大多数是组合公式，行为正确即可）
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"alpha_fn,args_fn",
|
||
[
|
||
("alpha_036", lambda c, v, o: (o, c)),
|
||
("alpha_037", lambda c, v, o: (o, c)),
|
||
("alpha_038", lambda c, v, o: (c,)),
|
||
("alpha_039", lambda c, v, o: (v,)),
|
||
("alpha_040", lambda c, v, o: (o * 1.1, c * 0.9, v)),
|
||
("alpha_041", lambda c, v, o: (o * 1.1, c * 0.9)),
|
||
("alpha_042", lambda c, v, o: (c, o * 1.1, c * 0.9)),
|
||
("alpha_043", lambda c, v, o: (c, v)),
|
||
("alpha_044", lambda c, v, o: (o, v)),
|
||
("alpha_045", lambda c, v, o: (c, v)),
|
||
("alpha_046", lambda c, v, o: (c,)),
|
||
("alpha_047", lambda c, v, o: (v, c)),
|
||
("alpha_048", lambda c, v, o: (c,)),
|
||
("alpha_049", lambda c, v, o: (c, v)),
|
||
("alpha_050", lambda c, v, o: (c, v)),
|
||
("alpha_051", lambda c, v, o: (o * 1.1, c * 0.9)),
|
||
("alpha_052", lambda c, v, o: (c,)),
|
||
("alpha_053", lambda c, v, o: (c, o * 1.1, c * 0.9)),
|
||
("alpha_054", lambda c, v, o: (o, c, c * 0.9)),
|
||
("alpha_055", lambda c, v, o: (o, o * 1.1, c * 0.9, v, c)),
|
||
("alpha_056", lambda c, v, o: (c, v)),
|
||
("alpha_057", lambda c, v, o: (c, v)),
|
||
("alpha_058", lambda c, v, o: (v, c)),
|
||
("alpha_059", lambda c, v, o: (c, v)),
|
||
("alpha_060", lambda c, v, o: (c, v)),
|
||
],
|
||
)
|
||
def test_alpha_036_to_060_smoke(alpha_fn, args_fn):
|
||
"""alpha_036 到 alpha_060 全部能跑通（smoke test）。"""
|
||
import sys
|
||
|
||
module = sys.modules["quant_engine.alpha_factors"]
|
||
fn = getattr(module, alpha_fn)
|
||
close = pd.Series(np.random.rand(60).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(60) * 100 + 1.0)
|
||
open_ = close.shift(1).fillna(close.iloc[0])
|
||
args = args_fn(close, volume, open_)
|
||
result = fn(*args)
|
||
assert isinstance(result, pd.Series)
|
||
# 大多数公式应该至少有少量有效值
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
# ── v1.2.0 第五批 alpha 公式（alpha_061 – alpha_085） ─────
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"alpha_fn,args_fn",
|
||
[
|
||
("alpha_061", lambda c, v, o, h, lo: (h, lo, v)),
|
||
("alpha_062", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_063", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_064", lambda c, v, o, h, lo: (o, c, v)),
|
||
("alpha_065", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_066", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_067", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
("alpha_068", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_069", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_070", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
("alpha_071", lambda c, v, o, h, lo: (o, c, lo, v)),
|
||
("alpha_072", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
("alpha_073", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_074", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
("alpha_075", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_076", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_077", lambda c, v, o, h, lo: (h, lo, v)),
|
||
("alpha_078", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
("alpha_079", lambda c, v, o, h, lo: (c, h, lo)),
|
||
("alpha_080", lambda c, v, o, h, lo: (o, c, v)),
|
||
("alpha_081", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_082", lambda c, v, o, h, lo: (o, v)),
|
||
("alpha_083", lambda c, v, o, h, lo: (h, lo, v)),
|
||
("alpha_084", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_085", lambda c, v, o, h, lo: (c, v)),
|
||
],
|
||
)
|
||
def test_alpha_061_to_085_smoke(alpha_fn, args_fn):
|
||
"""alpha_061 到 alpha_085 smoke test。"""
|
||
import sys
|
||
|
||
module = sys.modules["quant_engine.alpha_factors"]
|
||
fn = getattr(module, alpha_fn)
|
||
close = pd.Series(np.random.rand(60).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(60) * 100 + 1.0)
|
||
high = close * 1.05
|
||
low = close * 0.95
|
||
open_ = close.shift(1).fillna(close.iloc[0])
|
||
args = args_fn(close, volume, open_, high, low)
|
||
result = fn(*args)
|
||
assert isinstance(result, pd.Series)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_get_alpha_meta_alpha_061_to_085():
|
||
"""alpha_061 到 alpha_085 全部在 registry 中。"""
|
||
for i in range(61, 86):
|
||
alpha_id = f"alpha_{i:03d}"
|
||
meta = get_alpha_meta(alpha_id)
|
||
assert "formula" in meta
|
||
assert "inputs" in meta
|
||
assert len(meta["inputs"]) >= 1
|
||
|
||
|
||
# ── v1.2.0 第六批 alpha 公式（alpha_086 – alpha_110） ─────
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"alpha_fn,args_fn",
|
||
[
|
||
("alpha_086", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_087", lambda c, v, o, h, lo: (o, c, v)),
|
||
("alpha_088", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_089", lambda c, v, o, h, lo: (h, lo, v)),
|
||
("alpha_090", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_091", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
("alpha_092", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
("alpha_093", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_094", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_095", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_096", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_097", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
("alpha_098", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_099", lambda c, v, o, h, lo: (h, lo, v)),
|
||
("alpha_100", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
("alpha_101", lambda c, v, o, h, lo: (c, h, lo)),
|
||
("alpha_102", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_103", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_104", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
("alpha_105", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_106", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_107", lambda c, v, o, h, lo: (c, v)),
|
||
("alpha_108", lambda c, v, o, h, lo: (h, lo, v)),
|
||
("alpha_109", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
("alpha_110", lambda c, v, o, h, lo: (c, h, lo, v)),
|
||
],
|
||
)
|
||
def test_alpha_086_to_110_smoke(alpha_fn, args_fn):
|
||
"""alpha_086 到 alpha_110 smoke test。"""
|
||
import sys
|
||
|
||
module = sys.modules["quant_engine.alpha_factors"]
|
||
fn = getattr(module, alpha_fn)
|
||
close = pd.Series(np.random.rand(60).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(60) * 100 + 1.0)
|
||
high = close * 1.05
|
||
low = close * 0.95
|
||
open_ = close.shift(1).fillna(close.iloc[0])
|
||
args = args_fn(close, volume, open_, high, low)
|
||
result = fn(*args)
|
||
assert isinstance(result, pd.Series)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
# ── v1.2.0 第七批 alpha 公式（alpha_111 – alpha_135） ─────
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"alpha_id,args_fn",
|
||
[
|
||
(
|
||
f"alpha_{i:03d}",
|
||
lambda c, v, o, h, lo, _i=i: (
|
||
(c, v) if _i % 3 == 0 else (c, v) if _i % 3 == 1 else (h, lo, v)
|
||
),
|
||
)
|
||
for i in range(111, 136)
|
||
],
|
||
)
|
||
def test_alpha_111_to_135_smoke(alpha_id, args_fn):
|
||
"""alpha_111 到 alpha_135 smoke test（模板化 25 个）。"""
|
||
import sys
|
||
|
||
fn = getattr(sys.modules["quant_engine.alpha_factors"], alpha_id)
|
||
close = pd.Series(np.random.rand(60).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(60) * 100 + 1.0)
|
||
high = close * 1.05
|
||
low = close * 0.95
|
||
open_ = close.shift(1).fillna(close.iloc[0])
|
||
args = args_fn(close, volume, open_, high, low)
|
||
result = fn(*args)
|
||
assert isinstance(result, pd.Series)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
# ── v1.2.0 第八批 alpha 公式（alpha_136 – alpha_158） ─────
|
||
# 这是 alpha158 完整版的最后一批——达成 158 个公式
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"alpha_id",
|
||
[f"alpha_{i:03d}" for i in range(136, 159)],
|
||
)
|
||
def test_alpha_136_to_158_smoke(alpha_id):
|
||
"""alpha_136 到 alpha_158 smoke test（达成 158 完整版）。"""
|
||
import sys
|
||
|
||
fn = getattr(sys.modules["quant_engine.alpha_factors"], alpha_id)
|
||
close = pd.Series(np.random.rand(60).cumsum() + 10)
|
||
volume = pd.Series(np.random.rand(60) * 100 + 1.0)
|
||
high = close * 1.05
|
||
low = close * 0.95
|
||
args = fn.__code__.co_varnames[: fn.__code__.co_argcount]
|
||
args_series = {
|
||
"close": close,
|
||
"volume": volume,
|
||
"high": high,
|
||
"low": low,
|
||
"open": close.shift(1).fillna(close.iloc[0]),
|
||
}
|
||
fn_args = tuple(args_series[a] for a in args if a in args_series)
|
||
result = fn(*fn_args)
|
||
assert isinstance(result, pd.Series)
|
||
assert result.dropna().shape[0] > 0
|
||
|
||
|
||
def test_alpha_158_complete():
|
||
"""确认 alpha158 完整版：158 个算子全部可用 + 全部元数据。"""
|
||
assert len(ALPHA158_REGISTRY) == 158
|
||
for i in range(1, 159):
|
||
alpha_id = f"alpha_{i:03d}"
|
||
meta = get_alpha_meta(alpha_id)
|
||
assert "formula" in meta
|
||
assert "inputs" in meta
|
||
assert len(meta["inputs"]) >= 1
|
||
|
||
|
||
# ── v1.2.0 Phase 2: dump/load / JSONB 兼容工具 ─────
|
||
|
||
|
||
def test_dump_alpha_registry_jsonl(tmp_path):
|
||
"""dump_alpha_registry_jsonl 应产生 JSONL 文件。"""
|
||
from quant_engine.alpha_factors import dump_alpha_registry_jsonl, load_alpha_registry_jsonl
|
||
|
||
output_path = tmp_path / "alpha_registry.jsonl"
|
||
count = dump_alpha_registry_jsonl(str(output_path))
|
||
assert count == 158
|
||
assert output_path.exists()
|
||
# 加载回来验证
|
||
loaded = load_alpha_registry_jsonl(str(output_path))
|
||
assert len(loaded) == 158
|
||
assert "alpha_001" in loaded
|
||
assert "alpha_158" in loaded
|
||
|
||
|
||
def test_dump_load_roundtrip(tmp_path):
|
||
"""dump + load 应保持数据完整性。"""
|
||
from quant_engine.alpha_factors import (
|
||
ALPHA158_REGISTRY,
|
||
dump_alpha_registry_jsonl,
|
||
load_alpha_registry_jsonl,
|
||
)
|
||
|
||
output_path = tmp_path / "roundtrip.jsonl"
|
||
dump_alpha_registry_jsonl(str(output_path))
|
||
loaded = load_alpha_registry_jsonl(str(output_path))
|
||
# 抽样验证
|
||
for alpha_id in ["alpha_001", "alpha_050", "alpha_158"]:
|
||
assert alpha_id in loaded
|
||
original = ALPHA158_REGISTRY[alpha_id]
|
||
# 注意：load 后 id 字段已剥离，其余字段应完全一致
|
||
for key in original:
|
||
assert loaded[alpha_id].get(key) == original[key], f"Mismatch in {alpha_id}.{key}"
|
||
|
||
|
||
def test_alpha_registry_to_pg_rows():
|
||
"""alpha_registry_to_pg_rows 应输出 PG INSERT 行结构。"""
|
||
from quant_engine.alpha_factors import alpha_registry_to_pg_rows
|
||
|
||
rows = alpha_registry_to_pg_rows()
|
||
assert len(rows) == 158
|
||
# 验证第一行结构
|
||
row = rows[0]
|
||
assert "name" in row
|
||
assert "formula" in row
|
||
assert "metadata" in row
|
||
assert "version" in row
|
||
assert row["version"] == "1.2.0"
|
||
assert isinstance(row["metadata"], dict)
|
||
assert "category" in row["metadata"]
|
||
|
||
|
||
def test_parse_alpha_formula_basic():
|
||
"""parse_alpha_formula 应提取函数名与变量。"""
|
||
from quant_engine.alpha_factors import parse_alpha_formula
|
||
|
||
parsed = parse_alpha_formula("rank(ts_rank(close, 5))")
|
||
assert "raw" in parsed
|
||
assert "operators" in parsed
|
||
assert "functions" in parsed
|
||
assert "variables" in parsed
|
||
assert "close" in parsed["variables"]
|
||
assert "rank" in parsed["operators"]
|
||
|
||
|
||
def test_parse_alpha_formula_complex():
|
||
"""复杂公式解析应识别多函数 + 多变量。"""
|
||
from quant_engine.alpha_factors import parse_alpha_formula
|
||
|
||
parsed = parse_alpha_formula("-1 * rank(decay_linear(correlation(close, volume, 10), 5))")
|
||
# 应识别 correlation, decay_linear 函数调用
|
||
func_names = [f["name"] for f in parsed["functions"]]
|
||
assert "correlation" in func_names
|
||
assert "decay_linear" in func_names
|
||
# 应识别 close, volume 变量
|
||
assert "close" in parsed["variables"]
|
||
assert "volume" in parsed["variables"]
|
||
|
||
|
||
def test_parse_alpha_formula_round_trip_jsonb():
|
||
"""parse_alpha_formula 输出可 json.dumps → JSONB 落库。"""
|
||
import json
|
||
from quant_engine.alpha_factors import parse_alpha_formula
|
||
|
||
parsed = parse_alpha_formula("rank(decay_linear(rank(ts_rank(close, 5)), 5))")
|
||
# 应能 JSON 序列化
|
||
serialized = json.dumps(parsed)
|
||
assert isinstance(serialized, str)
|
||
assert "ts_rank" in serialized
|
||
|
||
|
||
# ── v1.2.0 Phase 1: deterministic operator dispatch contract ──────────────
|
||
|
||
|
||
def test_phase1_operator_catalog_is_explicit_and_serializable():
|
||
"""Phase 1 exposes a stable, JSON-friendly catalog for downstream callers."""
|
||
import json
|
||
|
||
expected = {
|
||
"rank",
|
||
"delta",
|
||
"ts_mean",
|
||
"ts_std",
|
||
"ts_rank",
|
||
"correlation",
|
||
"ts_min",
|
||
"ts_max",
|
||
"ts_sum",
|
||
"decay_linear",
|
||
}
|
||
assert set(list_phase1_operators()) == expected
|
||
assert set(ALPHA158_PHASE1_OPERATOR_SPECS) == expected
|
||
json.dumps(ALPHA158_PHASE1_OPERATOR_SPECS)
|
||
for name, spec in ALPHA158_PHASE1_OPERATOR_SPECS.items():
|
||
assert spec["name"] == name
|
||
assert isinstance(spec["inputs"], list)
|
||
assert isinstance(spec["formula"], str)
|
||
|
||
|
||
def test_phase1_unary_operators_preserve_index_and_are_deterministic():
|
||
values = pd.Series([1.0, 2.0, 3.0, 4.0], index=["a", "b", "c", "d"])
|
||
|
||
first = evaluate_phase1_operator("rank", values)
|
||
second = evaluate_phase1_operator("rank", values)
|
||
|
||
pd.testing.assert_series_equal(first, second)
|
||
assert first.index.equals(values.index)
|
||
assert first.iloc[-1] == pytest.approx(1.0)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("name", "window"),
|
||
[
|
||
("delta", 2),
|
||
("ts_mean", 2),
|
||
("ts_std", 2),
|
||
("ts_rank", 2),
|
||
("ts_min", 2),
|
||
("ts_max", 2),
|
||
("ts_sum", 2),
|
||
("decay_linear", 2),
|
||
],
|
||
)
|
||
def test_phase1_windowed_operators_require_explicit_window(name: str, window: int):
|
||
values = pd.Series([1.0, 2.0, 3.0, 4.0])
|
||
|
||
result = evaluate_phase1_operator(name, values, window=window)
|
||
|
||
assert result.index.equals(values.index)
|
||
with pytest.raises(ValueError, match="window"):
|
||
evaluate_phase1_operator(name, values)
|
||
with pytest.raises(ValueError, match="positive integer"):
|
||
evaluate_phase1_operator(name, values, window=1.5) # type: ignore[arg-type]
|
||
|
||
|
||
def test_phase1_binary_correlation_requires_aligned_secondary_input():
|
||
values = pd.Series([1.0, 2.0, 3.0, 4.0])
|
||
other = pd.Series([4.0, 3.0, 2.0, 1.0])
|
||
|
||
result = evaluate_phase1_operator("correlation", values, other, window=2)
|
||
|
||
assert result.iloc[-1] == pytest.approx(-1.0)
|
||
with pytest.raises(ValueError, match="secondary"):
|
||
evaluate_phase1_operator("correlation", values, window=2)
|
||
|
||
|
||
def test_phase1_dispatch_rejects_unknown_or_unused_arguments():
|
||
values = pd.Series([1.0, 2.0, 3.0])
|
||
|
||
with pytest.raises(KeyError, match="not registered"):
|
||
evaluate_phase1_operator("unknown", values)
|
||
with pytest.raises(ValueError, match="window"):
|
||
evaluate_phase1_operator("rank", values, window=2)
|
||
with pytest.raises(ValueError, match="secondary"):
|
||
evaluate_phase1_operator("rank", values, values)
|
||
|
||
|
||
def test_phase1_dispatch_rejects_window_above_supported_limit():
|
||
values = pd.Series([1.0, 2.0, 3.0])
|
||
|
||
with pytest.raises(ValueError, match="maximum"):
|
||
evaluate_phase1_operator("ts_mean", values, window=2**63)
|
||
|
||
|
||
# ── v1.2.0 Phase 2: cumulative deterministic operator contract ─────────────
|
||
|
||
|
||
def test_phase2_operator_catalog_is_cumulative_stable_and_serializable():
|
||
"""Phase 2 exposes all existing building blocks without changing Phase 1."""
|
||
import json
|
||
|
||
phase1 = list_phase1_operators()
|
||
expected_phase2 = (
|
||
*phase1,
|
||
"ts_argmin",
|
||
"ts_argmax",
|
||
"product",
|
||
"returns",
|
||
"scale",
|
||
"signed_power",
|
||
"stddev",
|
||
"covariance",
|
||
"log",
|
||
"abs_series",
|
||
"sign",
|
||
"max_pair",
|
||
"min_pair",
|
||
"indneutralize",
|
||
)
|
||
|
||
assert list_phase2_operators() == expected_phase2
|
||
assert tuple(ALPHA158_PHASE2_OPERATOR_SPECS) == expected_phase2
|
||
assert tuple(ALPHA158_PHASE1_OPERATOR_SPECS) == phase1
|
||
json.dumps(ALPHA158_PHASE2_OPERATOR_SPECS)
|
||
for name, spec in ALPHA158_PHASE2_OPERATOR_SPECS.items():
|
||
assert spec["name"] == name
|
||
assert isinstance(spec["inputs"], list)
|
||
assert isinstance(spec["parameters"], list)
|
||
assert isinstance(spec["formula"], str)
|
||
|
||
|
||
@pytest.mark.parametrize("name", ["ts_argmin", "ts_argmax", "product", "stddev"])
|
||
def test_phase2_windowed_unary_dispatch_is_deterministic(name: str):
|
||
values = pd.Series([3.0, 1.0, 4.0, 2.0], index=["a", "b", "c", "d"])
|
||
|
||
first = evaluate_phase2_operator(name, values, window=3)
|
||
second = evaluate_phase2_operator(name, values, window=3)
|
||
|
||
pd.testing.assert_series_equal(first, second)
|
||
assert first.index.equals(values.index)
|
||
with pytest.raises(ValueError, match="window"):
|
||
evaluate_phase2_operator(name, values)
|
||
|
||
|
||
@pytest.mark.parametrize("name", ["returns", "scale", "log", "abs_series", "sign"])
|
||
def test_phase2_unary_dispatch_rejects_unused_arguments(name: str):
|
||
values = pd.Series([1.0, 2.0, 4.0], index=["a", "b", "c"])
|
||
|
||
result = evaluate_phase2_operator(name, values)
|
||
|
||
assert result.index.equals(values.index)
|
||
with pytest.raises(ValueError, match="window"):
|
||
evaluate_phase2_operator(name, values, window=2)
|
||
with pytest.raises(ValueError, match="secondary"):
|
||
evaluate_phase2_operator(name, values, secondary=values)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("name", "window"),
|
||
[("correlation", 2), ("covariance", 2), ("max_pair", None), ("min_pair", None)],
|
||
)
|
||
def test_phase2_binary_dispatch_requires_aligned_secondary(name: str, window: int | None):
|
||
values = pd.Series([1.0, 2.0, 3.0], index=["a", "b", "c"])
|
||
secondary = pd.Series([3.0, 2.0, 1.0], index=values.index)
|
||
|
||
result = evaluate_phase2_operator(name, values, secondary=secondary, window=window)
|
||
|
||
assert result.index.equals(values.index)
|
||
with pytest.raises(ValueError, match="secondary is required"):
|
||
evaluate_phase2_operator(name, values, window=window)
|
||
with pytest.raises(ValueError, match="secondary index"):
|
||
evaluate_phase2_operator(
|
||
name,
|
||
values,
|
||
secondary=secondary.rename(index={"c": "z"}),
|
||
window=window,
|
||
)
|
||
|
||
|
||
def test_phase2_signed_power_requires_finite_numeric_exponent():
|
||
values = pd.Series([-4.0, 0.0, 9.0])
|
||
|
||
result = evaluate_phase2_operator("signed_power", values, exponent=0.5)
|
||
|
||
pd.testing.assert_series_equal(result, pd.Series([-2.0, 0.0, 3.0]))
|
||
for exponent in (None, True, float("inf"), float("nan"), "2"):
|
||
with pytest.raises(ValueError, match="exponent"):
|
||
evaluate_phase2_operator( # type: ignore[arg-type]
|
||
"signed_power",
|
||
values,
|
||
exponent=exponent,
|
||
)
|
||
|
||
|
||
def test_phase2_indneutralize_requires_aligned_groups():
|
||
values = pd.Series([1.0, 3.0, 10.0, 14.0], index=["a", "b", "c", "d"])
|
||
groups = pd.Series(["x", "x", "y", "y"], index=values.index)
|
||
|
||
result = evaluate_phase2_operator("indneutralize", values, groups=groups)
|
||
|
||
pd.testing.assert_series_equal(result, pd.Series([-1.0, 1.0, -2.0, 2.0], index=values.index))
|
||
with pytest.raises(ValueError, match="groups is required"):
|
||
evaluate_phase2_operator("indneutralize", values)
|
||
with pytest.raises(ValueError, match="groups index"):
|
||
evaluate_phase2_operator(
|
||
"indneutralize",
|
||
values,
|
||
groups=groups.rename(index={"d": "z"}),
|
||
)
|
||
|
||
|
||
def test_phase2_dispatch_validates_primary_series_and_unused_parameters():
|
||
values = pd.Series([1.0, 2.0, 3.0])
|
||
|
||
with pytest.raises(TypeError, match="series must be a pandas Series"):
|
||
evaluate_phase2_operator("rank", [1.0, 2.0, 3.0]) # type: ignore[arg-type]
|
||
with pytest.raises(KeyError, match="not registered"):
|
||
evaluate_phase2_operator("unknown", values)
|
||
with pytest.raises(ValueError, match="exponent"):
|
||
evaluate_phase2_operator("rank", values, exponent=2.0)
|
||
with pytest.raises(ValueError, match="groups"):
|
||
evaluate_phase2_operator("rank", values, groups=pd.Series(["x", "x", "x"]))
|
||
with pytest.raises(ValueError, match="maximum"):
|
||
evaluate_phase2_operator("product", values, window=253)
|
||
|
||
|
||
# ── Alpha158 Phase 3: versioned alpha001-alpha050 formula contract ──────────
|
||
|
||
|
||
def _phase3_market_inputs() -> dict[str, pd.Series]:
|
||
positions = np.arange(80, dtype=float)
|
||
index = pd.RangeIndex(len(positions), name="row")
|
||
open_ = pd.Series(100.0 + positions * 0.2 + np.sin(positions / 4.0), index=index)
|
||
close = pd.Series(100.5 + positions * 0.18 + np.cos(positions / 5.0), index=index)
|
||
high = pd.Series(np.maximum(open_, close) + 1.0, index=index)
|
||
low = pd.Series(np.minimum(open_, close) - 1.0, index=index)
|
||
volume = pd.Series(1_000.0 + positions**1.3 + 20.0 * np.sin(positions / 3.0), index=index)
|
||
vwap = (open_ + close + high + low) / 4.0
|
||
return {
|
||
"open": open_,
|
||
"close": close,
|
||
"high": high,
|
||
"low": low,
|
||
"volume": volume,
|
||
"vwap": vwap,
|
||
}
|
||
|
||
|
||
def test_phase3_formula_catalog_is_versioned_exact_and_content_addressed():
|
||
import hashlib
|
||
import json
|
||
from collections import Counter
|
||
|
||
expected_ids = tuple(f"alpha_{number:03d}" for number in range(1, 51))
|
||
expected_fields = {
|
||
"name",
|
||
"contract_version",
|
||
"formula",
|
||
"category",
|
||
"complexity",
|
||
"parameters",
|
||
"description",
|
||
"references",
|
||
"call_inputs",
|
||
"formula_inputs",
|
||
"input_category",
|
||
}
|
||
|
||
assert ALPHA158_PHASE3_FORMULA_CONTRACT_VERSION == "1.0.0"
|
||
assert list_phase3_formulas() == expected_ids
|
||
assert tuple(ALPHA158_PHASE3_FORMULA_SPECS) == expected_ids
|
||
assert Counter(
|
||
spec["input_category"] for spec in ALPHA158_PHASE3_FORMULA_SPECS.values()
|
||
) == {"single": 19, "pair": 26, "triple": 4, "quadruple": 1}
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE3_FORMULA_SPECS.items():
|
||
assert set(spec) == expected_fields
|
||
assert spec["name"] == alpha_id
|
||
assert spec["contract_version"] == ALPHA158_PHASE3_FORMULA_CONTRACT_VERSION
|
||
assert spec["formula"] == ALPHA158_REGISTRY[alpha_id]["formula"]
|
||
|
||
serializable_specs = {
|
||
alpha_id: {
|
||
field: list(value) if isinstance(value, tuple) else value
|
||
for field, value in spec.items()
|
||
}
|
||
for alpha_id, spec in ALPHA158_PHASE3_FORMULA_SPECS.items()
|
||
}
|
||
encoded = json.dumps(
|
||
serializable_specs,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
ensure_ascii=False,
|
||
).encode()
|
||
assert hashlib.sha256(encoded).hexdigest() == ALPHA158_PHASE3_FORMULA_CATALOG_SHA256
|
||
assert ALPHA158_PHASE3_FORMULA_CATALOG_SHA256 == (
|
||
"9a3360e5ee77a85a35d3c2fdab1efaa531fa0c263a2cb2bb5b965a1fb96fe1bd"
|
||
)
|
||
|
||
|
||
def test_phase3_formula_catalog_is_recursively_immutable():
|
||
import operator
|
||
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(ALPHA158_PHASE3_FORMULA_SPECS, "alpha_001", {})
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(
|
||
ALPHA158_PHASE3_FORMULA_SPECS["alpha_001"],
|
||
"formula",
|
||
"changed",
|
||
)
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(
|
||
ALPHA158_PHASE3_FORMULA_SPECS["alpha_001"]["call_inputs"],
|
||
0,
|
||
"volume",
|
||
)
|
||
|
||
|
||
def test_phase3_catalog_freezes_callable_signatures_without_rewriting_formulas():
|
||
import inspect
|
||
|
||
legacy_formula_input_differences = {
|
||
"alpha_011": ("close", "high", "low"),
|
||
"alpha_035": ("volume",),
|
||
"alpha_036": ("close",),
|
||
"alpha_040": ("high", "low"),
|
||
"alpha_042": ("close",),
|
||
"alpha_043": ("volume",),
|
||
}
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE3_FORMULA_SPECS.items():
|
||
function = getattr(alpha_factors_module, alpha_id)
|
||
signature_inputs = tuple(
|
||
"open" if name == "open_" else name
|
||
for name in inspect.signature(function).parameters
|
||
)
|
||
assert spec["call_inputs"] == signature_inputs
|
||
assert spec["formula_inputs"] == legacy_formula_input_differences.get(
|
||
alpha_id,
|
||
signature_inputs,
|
||
)
|
||
|
||
assert ALPHA158_PHASE3_FORMULA_SPECS["alpha_035"]["call_inputs"] == (
|
||
"close",
|
||
"volume",
|
||
)
|
||
assert ALPHA158_REGISTRY["alpha_035"]["inputs"] == ["volume"]
|
||
|
||
|
||
def test_phase3_dispatch_matches_all_existing_alpha001_alpha050_functions():
|
||
inputs = _phase3_market_inputs()
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE3_FORMULA_SPECS.items():
|
||
call_inputs = spec["call_inputs"]
|
||
function = getattr(alpha_factors_module, alpha_id)
|
||
expected = function(*(inputs[name] for name in call_inputs))
|
||
actual = evaluate_phase3_formula(
|
||
alpha_id,
|
||
**{name: inputs[name] for name in reversed(call_inputs)},
|
||
)
|
||
pd.testing.assert_series_equal(actual, expected)
|
||
|
||
|
||
def test_phase3_dispatch_rejects_unknown_missing_extra_and_non_series_inputs():
|
||
inputs = _phase3_market_inputs()
|
||
|
||
with pytest.raises(KeyError, match="not registered"):
|
||
evaluate_phase3_formula("alpha_051", close=inputs["close"])
|
||
with pytest.raises(ValueError, match=r"missing inputs.*volume"):
|
||
evaluate_phase3_formula("alpha_005", close=inputs["close"])
|
||
with pytest.raises(ValueError, match=r"unexpected inputs.*vwap"):
|
||
evaluate_phase3_formula(
|
||
"alpha_005",
|
||
close=inputs["close"],
|
||
volume=inputs["volume"],
|
||
vwap=inputs["vwap"],
|
||
)
|
||
with pytest.raises(TypeError, match="close must be a pandas Series"):
|
||
evaluate_phase3_formula( # type: ignore[arg-type]
|
||
"alpha_005",
|
||
close=[1.0, 2.0],
|
||
volume=inputs["volume"],
|
||
)
|
||
|
||
|
||
def test_phase3_dispatch_rejects_implicit_series_alignment():
|
||
inputs = _phase3_market_inputs()
|
||
misaligned_volume = inputs["volume"].rename(index={79: 80})
|
||
|
||
with pytest.raises(ValueError, match="volume index must align with close"):
|
||
evaluate_phase3_formula(
|
||
"alpha_005",
|
||
close=inputs["close"],
|
||
volume=misaligned_volume,
|
||
)
|
||
|
||
|
||
# ── Alpha158 Phase 4: versioned alpha051-alpha100 formula contract ──────────
|
||
|
||
|
||
def test_phase4_formula_catalog_is_versioned_exact_and_content_addressed():
|
||
import hashlib
|
||
import json
|
||
from collections import Counter
|
||
|
||
expected_ids = tuple(f"alpha_{number:03d}" for number in range(51, 101))
|
||
expected_fields = {
|
||
"name",
|
||
"contract_version",
|
||
"formula",
|
||
"category",
|
||
"complexity",
|
||
"parameters",
|
||
"description",
|
||
"references",
|
||
"call_inputs",
|
||
"formula_inputs",
|
||
"input_category",
|
||
}
|
||
|
||
assert ALPHA158_PHASE4_FORMULA_CONTRACT_VERSION == "1.0.0"
|
||
assert list_phase4_formulas() == expected_ids
|
||
assert tuple(ALPHA158_PHASE4_FORMULA_SPECS) == expected_ids
|
||
assert Counter(
|
||
spec["input_category"] for spec in ALPHA158_PHASE4_FORMULA_SPECS.values()
|
||
) == {"single": 1, "pair": 27, "triple": 11, "quadruple": 10, "quintuple": 1}
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE4_FORMULA_SPECS.items():
|
||
assert set(spec) == expected_fields
|
||
assert spec["name"] == alpha_id
|
||
assert spec["contract_version"] == ALPHA158_PHASE4_FORMULA_CONTRACT_VERSION
|
||
assert spec["formula"] == ALPHA158_REGISTRY[alpha_id]["formula"]
|
||
|
||
serializable_specs = {
|
||
alpha_id: {
|
||
field: list(value) if isinstance(value, tuple) else value
|
||
for field, value in spec.items()
|
||
}
|
||
for alpha_id, spec in ALPHA158_PHASE4_FORMULA_SPECS.items()
|
||
}
|
||
encoded = json.dumps(
|
||
serializable_specs,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
ensure_ascii=False,
|
||
).encode()
|
||
assert hashlib.sha256(encoded).hexdigest() == ALPHA158_PHASE4_FORMULA_CATALOG_SHA256
|
||
assert ALPHA158_PHASE4_FORMULA_CATALOG_SHA256 == (
|
||
"858daf5e2abab5063fc28fcf7c79936096e2c76dde582fc7bab78b3458f17054"
|
||
)
|
||
|
||
|
||
def test_phase4_formula_catalog_is_recursively_immutable():
|
||
import operator
|
||
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(ALPHA158_PHASE4_FORMULA_SPECS, "alpha_051", {})
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(
|
||
ALPHA158_PHASE4_FORMULA_SPECS["alpha_051"],
|
||
"formula",
|
||
"changed",
|
||
)
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(
|
||
ALPHA158_PHASE4_FORMULA_SPECS["alpha_051"]["call_inputs"],
|
||
0,
|
||
"volume",
|
||
)
|
||
|
||
|
||
def test_phase4_catalog_freezes_callable_and_formula_inputs():
|
||
import inspect
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE4_FORMULA_SPECS.items():
|
||
function = getattr(alpha_factors_module, alpha_id)
|
||
signature_inputs = tuple(
|
||
"open" if name == "open_" else name
|
||
for name in inspect.signature(function).parameters
|
||
)
|
||
assert spec["call_inputs"] == signature_inputs
|
||
assert spec["formula_inputs"] == tuple(ALPHA158_REGISTRY[alpha_id]["inputs"])
|
||
|
||
assert ALPHA158_PHASE4_FORMULA_SPECS["alpha_055"]["call_inputs"] == (
|
||
"open",
|
||
"high",
|
||
"low",
|
||
"volume",
|
||
"close",
|
||
)
|
||
|
||
|
||
def test_phase4_dispatch_matches_all_existing_alpha051_alpha100_functions():
|
||
inputs = _phase3_market_inputs()
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE4_FORMULA_SPECS.items():
|
||
call_inputs = spec["call_inputs"]
|
||
function = getattr(alpha_factors_module, alpha_id)
|
||
expected = function(*(inputs[name] for name in call_inputs))
|
||
actual = evaluate_phase4_formula(
|
||
alpha_id,
|
||
**{name: inputs[name] for name in reversed(call_inputs)},
|
||
)
|
||
pd.testing.assert_series_equal(actual, expected)
|
||
|
||
|
||
def test_phase4_dispatch_rejects_unknown_missing_extra_and_non_series_inputs():
|
||
inputs = _phase3_market_inputs()
|
||
|
||
with pytest.raises(KeyError, match="not registered"):
|
||
evaluate_phase4_formula("alpha_050", close=inputs["close"])
|
||
with pytest.raises(ValueError, match=r"missing inputs.*low"):
|
||
evaluate_phase4_formula("alpha_051", high=inputs["high"])
|
||
with pytest.raises(ValueError, match=r"unexpected inputs.*vwap"):
|
||
evaluate_phase4_formula(
|
||
"alpha_051",
|
||
high=inputs["high"],
|
||
low=inputs["low"],
|
||
vwap=inputs["vwap"],
|
||
)
|
||
with pytest.raises(TypeError, match="high must be a pandas Series"):
|
||
evaluate_phase4_formula( # type: ignore[arg-type]
|
||
"alpha_051",
|
||
high=[1.0, 2.0],
|
||
low=inputs["low"],
|
||
)
|
||
|
||
|
||
def test_phase4_dispatch_rejects_implicit_series_alignment():
|
||
inputs = _phase3_market_inputs()
|
||
misaligned_low = inputs["low"].rename(index={79: 80})
|
||
|
||
with pytest.raises(ValueError, match="low index must align with high"):
|
||
evaluate_phase4_formula(
|
||
"alpha_051",
|
||
high=inputs["high"],
|
||
low=misaligned_low,
|
||
)
|
||
|
||
|
||
# ── Alpha158 Phase 5: versioned alpha101-alpha150 formula contract ──────────
|
||
|
||
|
||
def test_phase5_formula_catalog_is_versioned_exact_and_content_addressed():
|
||
import hashlib
|
||
import json
|
||
from collections import Counter
|
||
|
||
expected_ids = tuple(f"alpha_{number:03d}" for number in range(101, 151))
|
||
expected_fields = {
|
||
"name",
|
||
"contract_version",
|
||
"formula",
|
||
"category",
|
||
"complexity",
|
||
"parameters",
|
||
"description",
|
||
"references",
|
||
"call_inputs",
|
||
"formula_inputs",
|
||
"input_category",
|
||
}
|
||
|
||
assert ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION == "1.0.0"
|
||
assert list_phase5_formulas() == expected_ids
|
||
assert tuple(ALPHA158_PHASE5_FORMULA_SPECS) == expected_ids
|
||
assert Counter(
|
||
spec["input_category"] for spec in ALPHA158_PHASE5_FORMULA_SPECS.values()
|
||
) == {"pair": 33, "triple": 14, "quadruple": 3}
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE5_FORMULA_SPECS.items():
|
||
assert set(spec) == expected_fields
|
||
assert spec["name"] == alpha_id
|
||
assert spec["contract_version"] == ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION
|
||
assert spec["formula"] == ALPHA158_REGISTRY[alpha_id]["formula"]
|
||
|
||
serializable_specs = {
|
||
alpha_id: {
|
||
field: list(value) if isinstance(value, tuple) else value
|
||
for field, value in spec.items()
|
||
}
|
||
for alpha_id, spec in ALPHA158_PHASE5_FORMULA_SPECS.items()
|
||
}
|
||
encoded = json.dumps(
|
||
serializable_specs,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
ensure_ascii=False,
|
||
).encode()
|
||
assert hashlib.sha256(encoded).hexdigest() == ALPHA158_PHASE5_FORMULA_CATALOG_SHA256
|
||
assert ALPHA158_PHASE5_FORMULA_CATALOG_SHA256 == (
|
||
"3368796169c9fbd39c4a34ea137e569964b15882fbf4de25124790d548db6533"
|
||
)
|
||
|
||
|
||
def test_phase5_formula_catalog_is_recursively_immutable():
|
||
import operator
|
||
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(ALPHA158_PHASE5_FORMULA_SPECS, "alpha_101", {})
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(
|
||
ALPHA158_PHASE5_FORMULA_SPECS["alpha_101"],
|
||
"formula",
|
||
"changed",
|
||
)
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(
|
||
ALPHA158_PHASE5_FORMULA_SPECS["alpha_101"]["call_inputs"],
|
||
0,
|
||
"volume",
|
||
)
|
||
|
||
|
||
def test_phase5_catalog_freezes_callable_and_formula_inputs():
|
||
import inspect
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE5_FORMULA_SPECS.items():
|
||
function = getattr(alpha_factors_module, alpha_id)
|
||
signature_inputs = tuple(
|
||
"open" if name == "open_" else name
|
||
for name in inspect.signature(function).parameters
|
||
)
|
||
assert spec["call_inputs"] == signature_inputs
|
||
assert spec["formula_inputs"] == tuple(ALPHA158_REGISTRY[alpha_id]["inputs"])
|
||
|
||
assert ALPHA158_PHASE5_FORMULA_SPECS["alpha_101"]["call_inputs"] == (
|
||
"close",
|
||
"high",
|
||
"low",
|
||
)
|
||
|
||
|
||
def test_phase5_dispatch_matches_all_existing_alpha101_alpha150_functions():
|
||
inputs = _phase3_market_inputs()
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE5_FORMULA_SPECS.items():
|
||
call_inputs = spec["call_inputs"]
|
||
function = getattr(alpha_factors_module, alpha_id)
|
||
expected = function(*(inputs[name] for name in call_inputs))
|
||
actual = evaluate_phase5_formula(
|
||
alpha_id,
|
||
**{name: inputs[name] for name in reversed(call_inputs)},
|
||
)
|
||
pd.testing.assert_series_equal(actual, expected)
|
||
|
||
|
||
def test_phase5_dispatch_rejects_unknown_missing_extra_and_non_series_inputs():
|
||
inputs = _phase3_market_inputs()
|
||
|
||
with pytest.raises(KeyError, match="not registered"):
|
||
evaluate_phase5_formula("alpha_100", close=inputs["close"])
|
||
with pytest.raises(KeyError, match="not registered"):
|
||
evaluate_phase5_formula("alpha_151", close=inputs["close"])
|
||
with pytest.raises(ValueError, match=r"missing inputs.*low"):
|
||
evaluate_phase5_formula(
|
||
"alpha_101",
|
||
close=inputs["close"],
|
||
high=inputs["high"],
|
||
)
|
||
with pytest.raises(ValueError, match=r"unexpected inputs.*vwap"):
|
||
evaluate_phase5_formula(
|
||
"alpha_101",
|
||
close=inputs["close"],
|
||
high=inputs["high"],
|
||
low=inputs["low"],
|
||
vwap=inputs["vwap"],
|
||
)
|
||
with pytest.raises(TypeError, match="high must be a pandas Series"):
|
||
evaluate_phase5_formula( # type: ignore[arg-type]
|
||
"alpha_101",
|
||
close=inputs["close"],
|
||
high=[1.0, 2.0],
|
||
low=inputs["low"],
|
||
)
|
||
|
||
|
||
def test_phase5_dispatch_rejects_length_and_index_alignment_errors():
|
||
inputs = _phase3_market_inputs()
|
||
shorter_low = inputs["low"].iloc[:-1]
|
||
misaligned_high = inputs["high"].rename(index={79: 80})
|
||
|
||
with pytest.raises(ValueError, match="low length must match close"):
|
||
evaluate_phase5_formula(
|
||
"alpha_101",
|
||
close=inputs["close"],
|
||
high=inputs["high"],
|
||
low=shorter_low,
|
||
)
|
||
with pytest.raises(ValueError, match="high index must align with close"):
|
||
evaluate_phase5_formula(
|
||
"alpha_101",
|
||
close=inputs["close"],
|
||
high=misaligned_high,
|
||
low=inputs["low"],
|
||
)
|
||
|
||
|
||
def test_phase5_contract_is_publicly_exported():
|
||
assert {
|
||
"ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION",
|
||
"ALPHA158_PHASE5_FORMULA_CATALOG_SHA256",
|
||
"ALPHA158_PHASE5_FORMULA_SPECS",
|
||
"list_phase5_formulas",
|
||
"evaluate_phase5_formula",
|
||
} <= set(alpha_factors_module.__all__)
|
||
|
||
|
||
# ── Alpha158 Phase 6: versioned alpha151-alpha158 formula contract ──────────
|
||
|
||
|
||
def test_phase6_formula_catalog_is_versioned_exact_and_content_addressed():
|
||
import hashlib
|
||
import json
|
||
from collections import Counter
|
||
|
||
expected_ids = tuple(f"alpha_{number:03d}" for number in range(151, 159))
|
||
expected_fields = {
|
||
"name",
|
||
"contract_version",
|
||
"formula",
|
||
"category",
|
||
"complexity",
|
||
"parameters",
|
||
"description",
|
||
"references",
|
||
"call_inputs",
|
||
"formula_inputs",
|
||
"input_category",
|
||
}
|
||
|
||
assert ALPHA158_PHASE6_FORMULA_CONTRACT_VERSION == "1.0.0"
|
||
assert list_phase6_formulas() == expected_ids
|
||
assert tuple(ALPHA158_PHASE6_FORMULA_SPECS) == expected_ids
|
||
assert Counter(
|
||
spec["input_category"] for spec in ALPHA158_PHASE6_FORMULA_SPECS.values()
|
||
) == {"pair": 6, "triple": 2}
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE6_FORMULA_SPECS.items():
|
||
assert set(spec) == expected_fields
|
||
assert spec["name"] == alpha_id
|
||
assert spec["contract_version"] == ALPHA158_PHASE6_FORMULA_CONTRACT_VERSION
|
||
assert spec["formula"] == ALPHA158_REGISTRY[alpha_id]["formula"]
|
||
|
||
serializable_specs = {
|
||
alpha_id: {
|
||
field: list(value) if isinstance(value, tuple) else value
|
||
for field, value in spec.items()
|
||
}
|
||
for alpha_id, spec in ALPHA158_PHASE6_FORMULA_SPECS.items()
|
||
}
|
||
encoded = json.dumps(
|
||
serializable_specs,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
ensure_ascii=False,
|
||
).encode()
|
||
assert hashlib.sha256(encoded).hexdigest() == ALPHA158_PHASE6_FORMULA_CATALOG_SHA256
|
||
assert ALPHA158_PHASE6_FORMULA_CATALOG_SHA256 == (
|
||
"70ffae16ca6cbb59a1e8644f5cdeec1d291fa75ff8b5647fb534af0083408219"
|
||
)
|
||
|
||
|
||
def test_phase6_formula_catalog_is_recursively_immutable():
|
||
import operator
|
||
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(ALPHA158_PHASE6_FORMULA_SPECS, "alpha_151", {})
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(
|
||
ALPHA158_PHASE6_FORMULA_SPECS["alpha_151"],
|
||
"formula",
|
||
"changed",
|
||
)
|
||
with pytest.raises(TypeError):
|
||
operator.setitem(
|
||
ALPHA158_PHASE6_FORMULA_SPECS["alpha_151"]["call_inputs"],
|
||
0,
|
||
"volume",
|
||
)
|
||
|
||
|
||
def test_phase6_catalog_freezes_callable_and_formula_inputs():
|
||
import inspect
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE6_FORMULA_SPECS.items():
|
||
function = getattr(alpha_factors_module, alpha_id)
|
||
signature_inputs = tuple(
|
||
"open" if name == "open_" else name
|
||
for name in inspect.signature(function).parameters
|
||
)
|
||
assert spec["call_inputs"] == signature_inputs
|
||
assert spec["formula_inputs"] == tuple(ALPHA158_REGISTRY[alpha_id]["inputs"])
|
||
|
||
assert ALPHA158_PHASE6_FORMULA_SPECS["alpha_158"]["call_inputs"] == (
|
||
"high",
|
||
"low",
|
||
"volume",
|
||
)
|
||
|
||
|
||
def test_phase6_dispatch_matches_all_existing_alpha151_alpha158_functions():
|
||
inputs = _phase3_market_inputs()
|
||
|
||
for alpha_id, spec in ALPHA158_PHASE6_FORMULA_SPECS.items():
|
||
call_inputs = spec["call_inputs"]
|
||
function = getattr(alpha_factors_module, alpha_id)
|
||
expected = function(*(inputs[name] for name in call_inputs))
|
||
actual = evaluate_phase6_formula(
|
||
alpha_id,
|
||
**{name: inputs[name] for name in reversed(call_inputs)},
|
||
)
|
||
pd.testing.assert_series_equal(actual, expected)
|
||
|
||
|
||
def test_phase6_dispatch_rejects_unknown_missing_extra_and_non_series_inputs():
|
||
inputs = _phase3_market_inputs()
|
||
|
||
with pytest.raises(KeyError, match="not registered"):
|
||
evaluate_phase6_formula("alpha_150", close=inputs["close"])
|
||
with pytest.raises(KeyError, match="not registered"):
|
||
evaluate_phase6_formula("alpha_159", close=inputs["close"])
|
||
with pytest.raises(ValueError, match=r"missing inputs.*volume"):
|
||
evaluate_phase6_formula("alpha_151", close=inputs["close"])
|
||
with pytest.raises(ValueError, match=r"unexpected inputs.*vwap"):
|
||
evaluate_phase6_formula(
|
||
"alpha_151",
|
||
close=inputs["close"],
|
||
volume=inputs["volume"],
|
||
vwap=inputs["vwap"],
|
||
)
|
||
with pytest.raises(TypeError, match="volume must be a pandas Series"):
|
||
evaluate_phase6_formula( # type: ignore[arg-type]
|
||
"alpha_151",
|
||
close=inputs["close"],
|
||
volume=[1.0, 2.0],
|
||
)
|
||
|
||
|
||
def test_phase6_dispatch_rejects_length_and_index_alignment_errors():
|
||
inputs = _phase3_market_inputs()
|
||
shorter_volume = inputs["volume"].iloc[:-1]
|
||
misaligned_low = inputs["low"].rename(index={79: 80})
|
||
|
||
with pytest.raises(ValueError, match="volume length must match close"):
|
||
evaluate_phase6_formula(
|
||
"alpha_151",
|
||
close=inputs["close"],
|
||
volume=shorter_volume,
|
||
)
|
||
with pytest.raises(ValueError, match="low index must align with high"):
|
||
evaluate_phase6_formula(
|
||
"alpha_158",
|
||
high=inputs["high"],
|
||
low=misaligned_low,
|
||
volume=inputs["volume"],
|
||
)
|
||
|
||
|
||
def test_phase6_preserves_complete_alpha001_alpha158_formula_body_fingerprint():
|
||
import ast
|
||
import hashlib
|
||
import inspect
|
||
import json
|
||
import textwrap
|
||
|
||
fingerprints = {}
|
||
for number in range(1, 159):
|
||
alpha_id = f"alpha_{number:03d}"
|
||
source = textwrap.dedent(inspect.getsource(getattr(alpha_factors_module, alpha_id)))
|
||
node = ast.parse(source).body[0]
|
||
assert isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||
body = ast.dump(
|
||
ast.Module(body=node.body, type_ignores=[]),
|
||
include_attributes=False,
|
||
)
|
||
fingerprints[alpha_id] = hashlib.sha256(body.encode()).hexdigest()
|
||
|
||
encoded = json.dumps(
|
||
fingerprints,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
).encode()
|
||
assert hashlib.sha256(encoded).hexdigest() == (
|
||
"f3ae807983cf8ca083d0b924c3807ffd84a62a5bd354f9efb1a1a16ca40c7da6"
|
||
)
|
||
|
||
|
||
def test_phase6_contract_is_publicly_exported():
|
||
assert {
|
||
"ALPHA158_PHASE6_FORMULA_CONTRACT_VERSION",
|
||
"ALPHA158_PHASE6_FORMULA_CATALOG_SHA256",
|
||
"ALPHA158_PHASE6_FORMULA_SPECS",
|
||
"list_phase6_formulas",
|
||
"evaluate_phase6_formula",
|
||
} <= set(alpha_factors_module.__all__)
|