ageorge156/quant_engine
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into codex/research-alpha158-phase5-formula-con (#12)
CI / lite (push) Successful in 7s
Details

Browse Source
This commit was merged in pull request #12.
This commit is contained in:
ageorge156
2026-08-28 18:32:18 +08:00
parent 03e38d5123
commit 2bc8aea435
2 changed files with 309 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/quant_engine/alpha_factors.py
+136
View File
@@ -3408,6 +3408,137 @@ def evaluate_phase4_formula(name: str, **inputs: pd.Series) -> pd.Series:
return function(*(inputs[field] for field in required_inputs))
# ── Phase 5 formula contract: frozen alpha101-alpha150 surface ──────────────
# Phase 5 extends the versioned formula contract without mutating any earlier
# catalogue, digest, dispatch surface, or existing formula implementation.
ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION = "1.0.0"
ALPHA158_PHASE5_FORMULA_CATALOG_SHA256 = (
"3368796169c9fbd39c4a34ea137e569964b15882fbf4de25124790d548db6533"
)
_PHASE5_FORMULA_FUNCTIONS: dict[str, Callable[..., pd.Series]] = {
"alpha_101": alpha_101,
"alpha_102": alpha_102,
"alpha_103": alpha_103,
"alpha_104": alpha_104,
"alpha_105": alpha_105,
"alpha_106": alpha_106,
"alpha_107": alpha_107,
"alpha_108": alpha_108,
"alpha_109": alpha_109,
"alpha_110": alpha_110,
"alpha_111": alpha_111,
"alpha_112": alpha_112,
"alpha_113": alpha_113,
"alpha_114": alpha_114,
"alpha_115": alpha_115,
"alpha_116": alpha_116,
"alpha_117": alpha_117,
"alpha_118": alpha_118,
"alpha_119": alpha_119,
"alpha_120": alpha_120,
"alpha_121": alpha_121,
"alpha_122": alpha_122,
"alpha_123": alpha_123,
"alpha_124": alpha_124,
"alpha_125": alpha_125,
"alpha_126": alpha_126,
"alpha_127": alpha_127,
"alpha_128": alpha_128,
"alpha_129": alpha_129,
"alpha_130": alpha_130,
"alpha_131": alpha_131,
"alpha_132": alpha_132,
"alpha_133": alpha_133,
"alpha_134": alpha_134,
"alpha_135": alpha_135,
"alpha_136": alpha_136,
"alpha_137": alpha_137,
"alpha_138": alpha_138,
"alpha_139": alpha_139,
"alpha_140": alpha_140,
"alpha_141": alpha_141,
"alpha_142": alpha_142,
"alpha_143": alpha_143,
"alpha_144": alpha_144,
"alpha_145": alpha_145,
"alpha_146": alpha_146,
"alpha_147": alpha_147,
"alpha_148": alpha_148,
"alpha_149": alpha_149,
"alpha_150": alpha_150,
}
def _build_phase5_formula_specs() -> dict[str, dict[str, Any]]:
specs: dict[str, dict[str, Any]] = {}
for alpha_id, function in _PHASE5_FORMULA_FUNCTIONS.items():
meta = ALPHA158_REGISTRY[alpha_id]
call_inputs = _phase3_call_inputs(function)
formula_inputs = _phase3_string_list(meta, "inputs", alpha_id)
input_category = _PHASE4_INPUT_CATEGORIES.get(len(call_inputs))
if input_category is None:
raise RuntimeError(f"unsupported formula input count for {alpha_id}")
specs[alpha_id] = {
"name": alpha_id,
"contract_version": ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION,
"formula": meta["formula"],
"category": meta["category"],
"complexity": meta["complexity"],
"parameters": _phase3_string_list(meta, "params", alpha_id),
"description": meta["description"],
"references": _phase3_string_list(meta, "references", alpha_id),
"call_inputs": list(call_inputs),
"formula_inputs": formula_inputs,
"input_category": input_category,
}
return specs
ALPHA158_PHASE5_FORMULA_SPECS: Mapping[str, Mapping[str, Any]] = (
_freeze_phase3_formula_specs(_build_phase5_formula_specs())
)
def list_phase5_formulas() -> tuple[str, ...]:
"""Return the frozen alpha101-alpha150 formula IDs in stable order."""
return tuple(ALPHA158_PHASE5_FORMULA_SPECS)
def evaluate_phase5_formula(name: str, **inputs: pd.Series) -> pd.Series:
"""Evaluate a Phase 5 formula with an exact, alignment-safe input contract."""
if name not in ALPHA158_PHASE5_FORMULA_SPECS:
raise KeyError(f"formula {name!r} not registered")
spec = ALPHA158_PHASE5_FORMULA_SPECS[name]
required_inputs = cast(tuple[str, ...], spec["call_inputs"])
missing_inputs = [field for field in required_inputs if field not in inputs]
unexpected_inputs = sorted(field for field in inputs if field not in required_inputs)
if missing_inputs or unexpected_inputs:
details: list[str] = []
if missing_inputs:
details.append(f"missing inputs {missing_inputs}")
if unexpected_inputs:
details.append(f"unexpected inputs {unexpected_inputs}")
raise ValueError(f"invalid inputs for {name}: {'; '.join(details)}")
for field in required_inputs:
if not isinstance(inputs[field], pd.Series):
raise TypeError(f"{field} must be a pandas Series")
primary_field = required_inputs[0]
primary = inputs[primary_field]
for field in required_inputs[1:]:
if len(inputs[field]) != len(primary):
raise ValueError(f"{field} length must match {primary_field}")
if not primary.index.equals(inputs[field].index):
raise ValueError(f"{field} index must align with {primary_field}")
function = _PHASE5_FORMULA_FUNCTIONS[name]
return function(*(inputs[field] for field in required_inputs))
__all__ = [
"rank",
"delta",
@@ -3451,6 +3582,11 @@ __all__ = [
"ALPHA158_PHASE4_FORMULA_SPECS",
"list_phase4_formulas",
"evaluate_phase4_formula",
"ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION",
"ALPHA158_PHASE5_FORMULA_CATALOG_SHA256",
"ALPHA158_PHASE5_FORMULA_SPECS",
"list_phase5_formulas",
"evaluate_phase5_formula",
"alpha_001",
"alpha_002",
"alpha_003",
tests/test_alpha_factors.py
+173
View File
@@ -17,6 +17,9 @@ from quant_engine.alpha_factors import (
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,
alpha_001,
alpha_002,
alpha_003,
@@ -179,10 +182,12 @@ from quant_engine.alpha_factors import (
evaluate_phase2_operator,
evaluate_phase3_formula,
evaluate_phase4_formula,
evaluate_phase5_formula,
list_phase1_operators,
list_phase2_operators,
list_phase3_formulas,
list_phase4_formulas,
list_phase5_formulas,
correlation,
covariance,
decay_linear,
@@ -1789,3 +1794,171 @@ def test_phase4_dispatch_rejects_implicit_series_alignment():
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__)