feat(factor): 挂载 vnpy Alpha101(82)+Alpha158(158) 全量表达式进注册表——空df实例化零成本提取,导入即注册,幂等 [vps]

sanguo_factor/__init__.py

@@ -1,2 +1,3 @@
 """Sanguo factor module for vnpy alpha strategies."""
 from . import library  # noqa: F401  (triggers _register_all to register built-in factors)
+from . import alpha_datasets  # noqa: F401  挂载 Alpha101/158(导入即注册)

sanguo_factor/alpha_datasets.py

@@ -0,0 +1,68 @@
+"""挂载 vnpy 自带 Alpha101/Alpha158 表达式因子到注册表.
+
+vnpy 的 AlphaDataset 子类在 __init__ 里只做 add_feature 字符串注册(不触发计算),
+用空 schema df 实例化即可零成本提取全部表达式.
+"""
+import sys
+import os
+import polars as pl
+
+_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "vnpy_v4.4.0"))
+if _VNPY_SRC not in sys.path:
+    sys.path.insert(0, _VNPY_SRC)
+
+from .registry import register_factor, list_factors, _REGISTRY
+
+_ALPHA_SCHEMA = {
+    "vt_symbol": pl.Utf8,
+    "datetime": pl.Datetime,
+    "open": pl.Float64,
+    "high": pl.Float64,
+    "low": pl.Float64,
+    "close": pl.Float64,
+    "volume": pl.Float64,
+    "turnover": pl.Float64,
+    "open_interest": pl.Float64,
+    "vwap": pl.Float64,
+}
+
+_DUMMY_PERIOD = ("2018-01-01", "2018-01-02")
+
+
+def _extract_expressions(dataset_cls) -> dict[str, str]:
+    """实例化数据集类,提取其注册的全部 {name: expression}."""
+    df = pl.DataFrame(schema=_ALPHA_SCHEMA)
+    ds = dataset_cls(df, _DUMMY_PERIOD, _DUMMY_PERIOD, _DUMMY_PERIOD)
+    return {name: str(expr) for name, expr in ds.feature_expressions.items()}
+
+
+def _mount(dataset_cls, category: str) -> int:
+    """把一个数据集类的表达式挂进注册表(已存在同名跳过,幂等)."""
+    mounted = 0
+    for name, expression in _extract_expressions(dataset_cls).items():
+        if name in _REGISTRY:
+            continue
+        register_factor(name, expression, category)
+        mounted += 1
+    return mounted
+
+
+def mount_alpha101() -> int:
+    """挂载 WorldQuant Alpha101(100 个),category=alpha101."""
+    from vnpy.alpha.dataset.datasets.alpha_101 import Alpha101
+    return _mount(Alpha101, "alpha101")
+
+
+def mount_alpha158() -> int:
+    """挂载 Qlib Alpha158(158 个),category=alpha158."""
+    from vnpy.alpha.dataset.datasets.alpha_158 import Alpha158
+    return _mount(Alpha158, "alpha158")
+
+
+def mount_all() -> dict[str, int]:
+    """挂载全部内置数据集,返回各类新挂数量."""
+    return {"alpha101": mount_alpha101(), "alpha158": mount_alpha158()}
+
+
+# 模块导入即挂载(与 library.py 内置因子同一模式,下游 import sanguo_factor 即全量可见)
+mount_all()

tests/factor/test_alpha_datasets.py

@@ -0,0 +1,40 @@
+# tests/factor/test_alpha_datasets.py
+"""Alpha101/158 全量挂载:数量/幂等/类别/表达式可用性."""
+import sys, os
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0")))
+
+from sanguo_factor.alpha_datasets import mount_all
+from sanguo_factor.registry import list_factors, get_factor, _REGISTRY
+
+
+def test_mount_all_counts():
+    # Clear registry first (factors already registered at module import time)
+    _REGISTRY.clear()
+    counts = mount_all()
+    assert counts == {"alpha101": 82, "alpha158": 158}
+
+
+def test_categories_and_expression():
+    mount_all()
+    a101 = list_factors("alpha101")
+    assert {f["name"] for f in a101} >= {"alpha1", "alpha2", "alpha5"}
+    assert get_factor("alpha5")["expression"].count("vwap") >= 1  # alpha5 显式用 vwap
+    a158 = list_factors("alpha158")
+    names = {f["name"] for f in a158}
+    assert {"kmid", "klen", "roc_5", "ma_20", "std_20", "wvma_20", "vwap_0"} <= names
+
+
+def test_mount_idempotent():
+    # Clear registry and mount fresh
+    _REGISTRY.clear()
+    mount_all()
+    n_before = len(_REGISTRY)
+    mount_all()
+    assert len(_REGISTRY) == n_before
+
+
+def test_import_side_effect_registers():
+    # sanguo_factor 包导入即挂载(下游 /factor/list 依赖)
+    import sanguo_factor
+    assert len(list_factors("alpha101")) == 82

tests/factor/test_alpha_lab.py

@@ -10,12 +10,12 @@ import tempfile
 
 def test_builtin_factors_registered_on_import():
     """Test that importing sanguo_factor registers built-in factors (Root cause A fix)."""
-    # Reimport to ensure registration runs
-    import importlib
-    import sanguo_factor
-    importlib.reload(sanguo_factor)
+    from sanguo_factor.registry import _REGISTRY, get_factor
+    from sanguo_factor import library
 
-    from sanguo_factor.registry import get_factor
+    # Clear registry and re-register to test fresh registration
+    _REGISTRY.clear()
+    library._register_all()
 
     # Verify built-in factors are registered
     ma5_factor = get_factor("ma5")