feat(factor): 分解器护栏第一层——白名单三源预校验(ast.parse+Name)+列名空间三族同源+source 确定性推导 #84 [vps] [no-doc]

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-10-02 07:52:53 +08:00
parent 7ca6345cb8
commit 62c797534a
2 changed files with 195 additions and 0 deletions
@@ -0,0 +1,129 @@
+# sanguo_factor/factor_guard.py
+"""分解器确定性护栏(C1/D1,spec §4.8 P4-2 折入)——纯函数零 IO.
+
+白名单三源:算子(vnpy EXPRESSION_FUNCTIONS 干净子集,42 注册名裁掉
+DataProxy/quesval 等杂项)+列名(三族 adapter 落地列,与 adapter 常量同源
+=零漂移)+常量(数字字面量).全部违规结构化返回(中文 detail)供反馈循环
+渲染(C3).上游范式: qlib OpsWrapper 命名空间封闭,修正为 ast.parse+Name
+预校验(调研报告 §2).
+"""
+from __future__ import annotations
+
+import ast
+import re
+from dataclasses import dataclass
+
+from .fundamental_schema import FEATURE_COLUMNS
+from .sentiment_adapter import SENTIMENT_COLUMNS
+
+BAR_COLUMNS = ("open", "high", "low", "close", "volume", "turnover", "vwap")
+COLUMN_DOMAINS: dict[str, str] = {
+    **{c: "bars_daily" for c in BAR_COLUMNS},
+    **{c: "fundamentals_pit" for c in FEATURE_COLUMNS},
+    **{c: "corpus_sentiment" for c in SENTIMENT_COLUMNS},
+}
+WIRED_DOMAINS = frozenset(COLUMN_DOMAINS.values())
+
+ALLOWED_OPERATORS = frozenset({
+    "ts_delay", "ts_delta", "ts_sum", "ts_mean", "ts_std", "ts_min", "ts_max",
+    "ts_argmax", "ts_argmin", "ts_rank", "ts_quantile", "ts_slope",
+    "ts_rsquare", "ts_resi", "ts_corr", "ts_cov", "ts_decay_linear",
+    "ts_product", "ts_less", "ts_greater", "ts_log", "ts_abs",
+    "cs_rank", "cs_mean", "cs_std", "cs_sum", "cs_scale", "cs_neutralize",
+    "ta_atr", "ta_rsi", "log", "sign", "abs", "greater", "less",
+})
+
+SOURCE_CATEGORY = {"bars_daily": "technical", "fundamentals_pit": "fundamental",
+                   "corpus_sentiment": "sentiment"}
+
+NAME_RE = re.compile(r"^[a-z][a-z0-9_]{2,39}$")
+
+_ALLOWED_BINOPS = (ast.Add, ast.Sub, ast.Mult, ast.Div)
+_BLOCKED_NODES = (ast.Attribute, ast.Subscript, ast.Compare, ast.BoolOp,
+                  ast.IfExp, ast.Lambda, ast.ListComp, ast.SetComp, ast.DictComp,
+                  ast.GeneratorExp, ast.Starred, ast.NamedExpr)
+
+
+@dataclass(frozen=True)
+class Violation:
+    code: str   # syntax_error/unknown_operator/unknown_column/…
+    detail: str  # 中文,含具体名字,直接渲染进反馈 prompt
+
+
+def _column_hint(col: str, unwired: tuple[str, ...]) -> str:
+    msg = f"列 {col} 不在底座可用清单"
+    if unwired:
+        msg += (f"(卡片关联域 {', '.join(unwired)} 尚未接线表达式底座,"
+                "若该列属之需先由 data 域接入)")
+    return msg
+
+
+def precheck_expression(expr: str,
+                        unwired_domains: tuple[str, ...] = ()) -> list[Violation]:
+    """ast.parse+Name 白名单预校验(C1).语法错→单条 syntax_error,其余逐节点."""
+    try:
+        tree = ast.parse(expr, mode="eval")
+    except (SyntaxError, ValueError) as e:
+        return [Violation("syntax_error", f"表达式无法解析: {e}")]
+    out: list[Violation] = []
+    func_ids = {id(n.func) for n in ast.walk(tree)
+                if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)}
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Name):
+            if id(node) not in func_ids and node.id not in COLUMN_DOMAINS:
+                out.append(Violation(
+                    "unknown_column", _column_hint(node.id, unwired_domains)))
+        elif isinstance(node, ast.Call):
+            fname = getattr(node.func, "id", None)
+            if not isinstance(node.func, ast.Name) or fname not in ALLOWED_OPERATORS:
+                out.append(Violation("unknown_operator",
+                                     f"算子 {fname or '(复杂表达式)'} 不在白名单"))
+            if node.keywords:
+                out.append(Violation("syntax_not_allowed", "算子调用不允许关键字参数"))
+        elif isinstance(node, ast.BinOp) and not isinstance(node.op, _ALLOWED_BINOPS):
+            out.append(Violation(
+                "syntax_not_allowed",
+                f"二元运算 {type(node.op).__name__} 不允许(只支持 + - * /)"))
+        elif isinstance(node, ast.UnaryOp) and not isinstance(node.op, ast.USub):
+            out.append(Violation("syntax_not_allowed", "一元运算只允许取负 -"))
+        elif isinstance(node, _BLOCKED_NODES):
+            out.append(Violation("syntax_not_allowed",
+                                 f"语法 {type(node).__name__} 不允许"))
+        elif (isinstance(node, ast.Constant)
+              and not (isinstance(node.value, (int, float))
+                       and not isinstance(node.value, bool))):
+            out.append(Violation("syntax_not_allowed", "只允许数字字面量"))
+    return list({(v.code, v.detail): v for v in out}.values())
+
+
+def check_name(name: str, existing_names: set[str]) -> list[Violation]:
+    out: list[Violation] = []
+    if not NAME_RE.match(name):
+        out.append(Violation("bad_name", "因子名须 ^[a-z][a-z0-9_]{2,39}$"))
+    if name in existing_names or name in ALLOWED_OPERATORS or name in COLUMN_DOMAINS:
+        out.append(Violation("name_taken",
+                             f"名字 {name} 已被占用(在库因子/算子/列名)"))
+    return out
+
+
+def derive_source(expr: str) -> str | Violation:
+    """由表达式列族确定性推导 params.source(source 非LLM声明=防编数).
+
+    规则: 列至多 bars+单一特征族(跨 fund+sent 不支持——batch_eval 按单一
+    category join 特征,跨族列会全 NaN).
+    """
+    tree = ast.parse(expr, mode="eval")   # 调用方已过 precheck
+    func_ids = {id(n.func) for n in ast.walk(tree)
+                if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)}
+    domains = {COLUMN_DOMAINS[n.id] for n in ast.walk(tree)
+               if isinstance(n, ast.Name) and id(n) not in func_ids}
+    feature = domains - {"bars_daily"}
+    if len(feature) > 1:
+        return Violation("cross_domain",
+                         f"跨特征族 {sorted(feature)} 不支持(至多 bars+单一特征族)")
+    return next(iter(feature)) if feature else "bars_daily"
+
+
+def category_for_source(source: str) -> str:
+    """batch_eval 特征 join 的 category 映射(动态注册链消费,Task 6)."""
+    return SOURCE_CATEGORY.get(source, "technical")
@@ -0,0 +1,66 @@
+# tests/factor/test_factor_guard.py
+"""分解器确定性护栏:白名单三源+名字/来源推导(spec §4.8 D1/C1)."""
+from sanguo_factor.factor_guard import (
+    ALLOWED_OPERATORS, COLUMN_DOMAINS, WIRED_DOMAINS, Violation,
+    category_for_source, check_name, derive_source, precheck_expression,
+)
+
+
+def _codes(vs):
+    return [v.code for v in vs]
+
+
+def test_column_namespace_three_families():
+    assert COLUMN_DOMAINS["close"] == "bars_daily"
+    assert COLUMN_DOMAINS["roe_ttm"] == "fundamentals_pit"
+    assert COLUMN_DOMAINS["sent_mom_20"] == "corpus_sentiment"
+    assert WIRED_DOMAINS == {"bars_daily", "fundamentals_pit", "corpus_sentiment"}
+    assert len(COLUMN_DOMAINS) == 7 + 90 + 4
+
+
+def test_precheck_clean_expression_passes():
+    assert precheck_expression("cs_rank(ts_mean(close, 20))") == []
+    assert precheck_expression("cs_rank(-tacc)") == []
+
+
+def test_precheck_unknown_operator():
+    vs = precheck_expression("cs_rank(foo(close))")
+    assert _codes(vs) == ["unknown_operator"]
+
+
+def test_precheck_unknown_column_with_unwired_hint():
+    vs = precheck_expression("cs_rank(pledge_pct)", unwired_domains=("pledge_ratio",))
+    assert _codes(vs) == ["unknown_column"]
+    assert "尚未接线" in vs[0].detail and "pledge_ratio" in vs[0].detail
+
+
+def test_precheck_blocked_syntax():
+    assert "syntax_not_allowed" in _codes(precheck_expression("close[1]"))      # Subscript
+    assert "syntax_not_allowed" in _codes(precheck_expression("close > 1"))     # Compare
+    assert "syntax_not_allowed" in _codes(precheck_expression("close ** 2"))    # Pow
+    assert "syntax_not_allowed" in _codes(precheck_expression('cs_rank("x")'))  # 字符串字面量
+    assert _codes(precheck_expression("close +")) == ["syntax_error"]
+
+
+def test_check_name_rules():
+    assert _codes(check_name("close", set())) == ["name_taken"]          # 列名占用
+    assert _codes(check_name("cs_rank", set())) == ["name_taken"]        # 算子占用
+    assert _codes(check_name("Bad-Name", set())) == ["bad_name"]
+    assert _codes(check_name("ab", set())) == ["bad_name"]               # 太短
+    assert check_name("llm_mom20", {"fund_roe_ttm"}) == []               # existing_names 由调用方传在库全名(Task 5)
+    assert _codes(check_name("fund_roe_ttm", {"fund_roe_ttm"})) == ["name_taken"]
+
+
+def test_derive_source_families():
+    assert derive_source("cs_rank(ts_mean(close, 20))") == "bars_daily"
+    assert derive_source("cs_rank(-tacc)") == "fundamentals_pit"
+    assert derive_source("cs_rank(sent_mom_20)") == "corpus_sentiment"
+    mixed = derive_source("cs_rank(roe_ttm + sent_mom_20)")
+    assert isinstance(mixed, Violation) and mixed.code == "cross_domain"
+
+
+def test_category_for_source():
+    assert category_for_source("fundamentals_pit") == "fundamental"
+    assert category_for_source("corpus_sentiment") == "sentiment"
+    assert category_for_source("bars_daily") == "technical"
+    assert category_for_source("unknown_domain") == "technical"