diff --git a/src/quant_engine/alpha_factors.py b/src/quant_engine/alpha_factors.py index 4ed8773..f71fcc8 100644 --- a/src/quant_engine/alpha_factors.py +++ b/src/quant_engine/alpha_factors.py @@ -15,8 +15,9 @@ v1.2.0 Phase 0：5 个基础算子 + 5 个 alpha 公式（alpha001–alpha005） from __future__ import annotations -from collections.abc import Callable -from typing import Any +from collections.abc import Callable, Mapping +from types import MappingProxyType +from typing import Any, cast import numpy as np import pandas as pd @@ -3092,6 +3093,184 @@ def parse_alpha_formula(formula_str: str) -> dict[str, Any]: return parsed +# ── Phase 3 formula contract: frozen alpha001-alpha050 surface ────────────── + +# Formula functions remain the implementation source of truth. This contract +# freezes their callable surface separately from formula dependencies so that +# historical compatibility-only arguments remain explicit without rewriting +# formulas or changing direct-call APIs. +ALPHA158_PHASE3_FORMULA_CONTRACT_VERSION = "1.0.0" +ALPHA158_PHASE3_FORMULA_CATALOG_SHA256 = ( + "9a3360e5ee77a85a35d3c2fdab1efaa531fa0c263a2cb2bb5b965a1fb96fe1bd" +) + +_PHASE3_FORMULA_FUNCTIONS: dict[str, Callable[..., pd.Series]] = { + "alpha_001": alpha_001, + "alpha_002": alpha_002, + "alpha_003": alpha_003, + "alpha_004": alpha_004, + "alpha_005": alpha_005, + "alpha_006": alpha_006, + "alpha_007": alpha_007, + "alpha_008": alpha_008, + "alpha_009": alpha_009, + "alpha_010": alpha_010, + "alpha_011": alpha_011, + "alpha_012": alpha_012, + "alpha_013": alpha_013, + "alpha_014": alpha_014, + "alpha_015": alpha_015, + "alpha_016": alpha_016, + "alpha_017": alpha_017, + "alpha_018": alpha_018, + "alpha_019": alpha_019, + "alpha_020": alpha_020, + "alpha_021": alpha_021, + "alpha_022": alpha_022, + "alpha_023": alpha_023, + "alpha_024": alpha_024, + "alpha_025": alpha_025, + "alpha_026": alpha_026, + "alpha_027": alpha_027, + "alpha_028": alpha_028, + "alpha_029": alpha_029, + "alpha_030": alpha_030, + "alpha_031": alpha_031, + "alpha_032": alpha_032, + "alpha_033": alpha_033, + "alpha_034": alpha_034, + "alpha_035": alpha_035, + "alpha_036": alpha_036, + "alpha_037": alpha_037, + "alpha_038": alpha_038, + "alpha_039": alpha_039, + "alpha_040": alpha_040, + "alpha_041": alpha_041, + "alpha_042": alpha_042, + "alpha_043": alpha_043, + "alpha_044": alpha_044, + "alpha_045": alpha_045, + "alpha_046": alpha_046, + "alpha_047": alpha_047, + "alpha_048": alpha_048, + "alpha_049": alpha_049, + "alpha_050": alpha_050, +} + +_PHASE3_FORMULA_INPUT_OVERRIDES: dict[str, list[str]] = { + "alpha_011": ["close", "high", "low"], + "alpha_035": ["volume"], + "alpha_036": ["close"], + "alpha_040": ["high", "low"], + "alpha_042": ["close"], + "alpha_043": ["volume"], +} + +_PHASE3_INPUT_CATEGORIES = { + 1: "single", + 2: "pair", + 3: "triple", + 4: "quadruple", +} + + +def _phase3_call_inputs(function: Callable[..., pd.Series]) -> list[str]: + import inspect + + parameters = list(inspect.signature(function).parameters.values()) + if any( + parameter.kind is not inspect.Parameter.POSITIONAL_OR_KEYWORD + or parameter.default is not inspect.Parameter.empty + for parameter in parameters + ): + raise RuntimeError(f"unsupported formula signature for {function.__name__}") + return ["open" if parameter.name == "open_" else parameter.name for parameter in parameters] + + +def _phase3_string_list(meta: dict[str, Any], field: str, alpha_id: str) -> list[str]: + value = meta[field] + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + raise RuntimeError(f"{field} must be a list of strings for {alpha_id}") + return list(value) + + +def _build_phase3_formula_specs() -> dict[str, dict[str, Any]]: + specs: dict[str, dict[str, Any]] = {} + for alpha_id, function in _PHASE3_FORMULA_FUNCTIONS.items(): + meta = ALPHA158_REGISTRY[alpha_id] + call_inputs = _phase3_call_inputs(function) + formula_inputs = _PHASE3_FORMULA_INPUT_OVERRIDES.get(alpha_id, call_inputs) + input_category = _PHASE3_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_PHASE3_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": list(formula_inputs), + "input_category": input_category, + } + return specs + + +def _freeze_phase3_formula_specs( + specs: dict[str, dict[str, Any]], +) -> Mapping[str, Mapping[str, Any]]: + frozen_specs: dict[str, Mapping[str, Any]] = {} + for alpha_id, spec in specs.items(): + frozen_specs[alpha_id] = MappingProxyType( + {field: tuple(value) if isinstance(value, list) else value for field, value in spec.items()} + ) + return MappingProxyType(frozen_specs) + + +ALPHA158_PHASE3_FORMULA_SPECS: Mapping[str, Mapping[str, Any]] = ( + _freeze_phase3_formula_specs(_build_phase3_formula_specs()) +) + + +def list_phase3_formulas() -> tuple[str, ...]: + """Return the frozen alpha001-alpha050 formula IDs in stable order.""" + return tuple(ALPHA158_PHASE3_FORMULA_SPECS) + + +def evaluate_phase3_formula(name: str, **inputs: pd.Series) -> pd.Series: + """Evaluate a Phase 3 formula with an exact, alignment-safe input contract.""" + if name not in ALPHA158_PHASE3_FORMULA_SPECS: + raise KeyError(f"formula {name!r} not registered") + + spec = ALPHA158_PHASE3_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 not primary.index.equals(inputs[field].index): + raise ValueError(f"{field} index must align with {primary_field}") + + function = _PHASE3_FORMULA_FUNCTIONS[name] + return function(*(inputs[field] for field in required_inputs)) + + __all__ = [ "rank", "delta", @@ -3125,6 +3304,11 @@ __all__ = [ "ALPHA158_PHASE2_OPERATOR_SPECS", "list_phase2_operators", "evaluate_phase2_operator", + "ALPHA158_PHASE3_FORMULA_CONTRACT_VERSION", + "ALPHA158_PHASE3_FORMULA_CATALOG_SHA256", + "ALPHA158_PHASE3_FORMULA_SPECS", + "list_phase3_formulas", + "evaluate_phase3_formula", "alpha_001", "alpha_002", "alpha_003", diff --git a/tests/test_alpha_factors.py b/tests/test_alpha_factors.py index 4174d3d..de3044f 100644 --- a/tests/test_alpha_factors.py +++ b/tests/test_alpha_factors.py @@ -6,10 +6,14 @@ 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, alpha_001, alpha_002, alpha_003, @@ -170,8 +174,10 @@ from quant_engine.alpha_factors import ( alpha_158, evaluate_phase1_operator, evaluate_phase2_operator, + evaluate_phase3_formula, list_phase1_operators, list_phase2_operators, + list_phase3_formulas, correlation, covariance, decay_linear, @@ -1463,3 +1469,175 @@ def test_phase2_dispatch_validates_primary_series_and_unused_parameters(): 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, + )