From f93621e3fccd0645d5a8a644aef4a5d231403236 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Mon, 6 Jul 2026 08:06:10 +0800 Subject: [PATCH] =?UTF-8?q?feat(factor):=20=E5=9B=A0=E5=AD=90=E8=A1=A8?= =?UTF-8?q?=E8=BE=BE=E5=BC=8F=E6=B3=A8=E5=86=8C=20+=20Alpha158=20=E5=AD=90?= =?UTF-8?q?=E9=9B=86=E5=86=85=E7=BD=AE=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 实现因子表达式注册表（registry.py） - 实现 Alpha158 子集内置因子库（library.py） - 使用 vnpy.alpha ts_ 算子语法（ts_mean, ts_delta, ts_max, ts_min） - 修正列名：close/open/high/low/volume（非 _price 后缀） - TDD 实现：8 测试全部通过 - 支持因子分类过滤（custom/builtin） - 防止重复注册 Co-Authored-By: Claude --- sanguo_factor/library.py | 58 +++++++++++++++++++++++++++++++++++ sanguo_factor/registry.py | 50 ++++++++++++++++++++++++++++++ tests/factor/test_library.py | 29 ++++++++++++++++++ tests/factor/test_registry.py | 43 ++++++++++++++++++++++++++ 4 files changed, 180 insertions(+) create mode 100644 sanguo_factor/library.py create mode 100644 sanguo_factor/registry.py create mode 100644 tests/factor/test_library.py create mode 100644 tests/factor/test_registry.py diff --git a/sanguo_factor/library.py b/sanguo_factor/library.py new file mode 100644 index 0000000..589b250 --- /dev/null +++ b/sanguo_factor/library.py @@ -0,0 +1,58 @@ +"""内置因子表达式库（Alpha158/101 子集，使用 vnpy.alpha ts_ 算子）.""" +from .registry import register_factor + +# 内置因子列表（使用 ts_ 算子 + 列名 close/open/high/low/volume） +# 修正：使用 vnpy.alpha 表达式语法（ts_mean 等），而非 polars 语法 +BUILTIN_FACTORS = [ + { + "name": "ma5", + "expression": "ts_mean(close, 5)", + "category": "builtin" + }, + { + "name": "ma10", + "expression": "ts_mean(close, 10)", + "category": "builtin" + }, + { + "name": "ma20", + "expression": "ts_mean(close, 20)", + "category": "builtin" + }, + { + "name": "vol_ma5", + "expression": "ts_mean(volume, 5)", + "category": "builtin" + }, + { + "name": "return_1d", + "expression": "ts_delta(close, 1)", + "category": "builtin" + }, + { + "name": "return_5d", + "expression": "ts_delta(close, 5)", + "category": "builtin" + }, + { + "name": "high_low_5", + "expression": "(ts_max(high, 5) - ts_min(low, 5)) / close", + "category": "builtin" + }, +] + + +def _register_all() -> None: + """注册所有内置因子到注册表.""" + from .registry import _REGISTRY + for factor in BUILTIN_FACTORS: + if factor["name"] not in _REGISTRY: + register_factor( + factor["name"], + factor["expression"], + factor["category"] + ) + + +# 模块导入时自动注册内置因子 +_register_all() diff --git a/sanguo_factor/registry.py b/sanguo_factor/registry.py new file mode 100644 index 0000000..cddea69 --- /dev/null +++ b/sanguo_factor/registry.py @@ -0,0 +1,50 @@ +"""因子表达式注册表（包装 vnpy.alpha AlphaDataset 范式）.""" +from typing import Any + +_REGISTRY: dict[str, dict[str, Any]] = {} + + +def register_factor(name: str, expression: str, category: str = "custom") -> None: + """注册因子表达式. + + Args: + name: 因子名称 + expression: 因子表达式字符串（使用 ts_/cs_/math 算子 + 列名） + category: 因子分类（默认 "custom"） + + Raises: + ValueError: 当因子名称已存在时 + """ + if name in _REGISTRY: + raise ValueError(f"因子已存在: {name}") + _REGISTRY[name] = { + "name": name, + "expression": expression, + "category": category + } + + +def get_factor(name: str) -> dict[str, Any] | None: + """获取因子表达式. + + Args: + name: 因子名称 + + Returns: + 因子信息字典，不存在时返回 None + """ + return _REGISTRY.get(name) + + +def list_factors(category: str | None = None) -> list[dict[str, Any]]: + """列出因子表达式. + + Args: + category: 分类过滤器，None 表示返回所有因子 + + Returns: + 因子信息字典列表 + """ + if category is None: + return list(_REGISTRY.values()) + return [f for f in _REGISTRY.values() if f["category"] == category] diff --git a/tests/factor/test_library.py b/tests/factor/test_library.py new file mode 100644 index 0000000..c236c60 --- /dev/null +++ b/tests/factor/test_library.py @@ -0,0 +1,29 @@ +"""内置因子库测试.""" +from sanguo_factor.registry import list_factors, get_factor, _REGISTRY +import importlib +import sanguo_factor.library + + +def test_library_registers_on_import(): + """测试导入时自动注册内置因子.""" + _REGISTRY.clear() + importlib.reload(sanguo_factor.library) + assert len(list_factors(category="builtin")) > 0 + + +def test_builtin_factors_count(): + """测试内置因子数量正确.""" + _REGISTRY.clear() + importlib.reload(sanguo_factor.library) + builtin_factors = list_factors(category="builtin") + assert len(builtin_factors) == 7 + + +def test_builtin_factor_expressions(): + """测试内置因子表达式格式正确.""" + _REGISTRY.clear() + importlib.reload(sanguo_factor.library) + ma5 = get_factor("ma5") + assert ma5 is not None + assert ma5["category"] == "builtin" + assert "ts_mean(close, 5)" in ma5["expression"] diff --git a/tests/factor/test_registry.py b/tests/factor/test_registry.py new file mode 100644 index 0000000..70d91da --- /dev/null +++ b/tests/factor/test_registry.py @@ -0,0 +1,43 @@ +"""因子表达式注册表测试.""" +import pytest +from sanguo_factor.registry import register_factor, get_factor, list_factors, _REGISTRY + + +def test_register_and_get_factor(): + """测试注册和获取因子.""" + _REGISTRY.clear() + register_factor("ma5", "ts_mean(close, 5)", category="trend") + f = get_factor("ma5") + assert f["name"] == "ma5" + assert f["category"] == "trend" + assert f["expression"] == "ts_mean(close, 5)" + + +def test_list_factors_by_category(): + """测试按分类列出因子.""" + _REGISTRY.clear() + register_factor("ma5", "ts_mean(close, 5)", category="trend") + register_factor("rsi", "ts_rank(close, 14)", category="momentum") + assert len(list_factors(category="trend")) == 1 + assert len(list_factors()) == 2 + + +def test_register_duplicate_raises(): + """测试重复注册抛出异常.""" + _REGISTRY.clear() + register_factor("ma5", "ts_mean(close, 5)", category="trend") + with pytest.raises(ValueError, match="因子已存在"): + register_factor("ma5", "ts_mean(close, 10)", category="trend") + + +def test_get_nonexistent_factor_returns_none(): + """测试获取不存在的因子返回 None.""" + _REGISTRY.clear() + assert get_factor("nonexistent") is None + + +def test_list_factors_empty_registry(): + """测试空注册表返回空列表.""" + _REGISTRY.clear() + assert list_factors() == [] + assert list_factors(category="trend") == []