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") == []