feat(factor): 因子表达式注册 + Alpha158 子集内置库

- 实现因子表达式注册表（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 <noreply@anthropic.com>

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()

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]

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"]

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