189 lines
8.2 KiB
Python
189 lines
8.2 KiB
Python
# 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_candidate, check_name, complexity_violations,
|
|
derive_source, dup_violations, 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"
|
|
|
|
|
|
def test_bare_operator_reference_blocked():
|
|
# 裸算子引用(非函数位)不是列名——仍须 unknown_column 拦住(T1 修复的回归钉)
|
|
assert _codes(precheck_expression("cs_rank + 1")) == ["unknown_column"]
|
|
|
|
|
|
def test_complexity_five_conditions():
|
|
cols7 = " + ".join(["close", "open", "volume", "vwap", "turnover",
|
|
"roe_ttm", "tacc"])
|
|
codes = [v.code for v in complexity_violations(f"cs_rank({cols7})")]
|
|
assert "complexity" in codes # 引用列 7>6
|
|
long_expr = "cs_rank(close + " + " + ".join(["close"] * 90) + ")"
|
|
assert any("符号长度" in v.detail for v in complexity_violations(long_expr))
|
|
deep = "close"
|
|
for _ in range(14): # 深度 15>12(_DEPTH_MAX=12:6 列合法和式结构深度≈9 留余量)
|
|
deep = f"ts_mean({deep}, 5)"
|
|
assert any("嵌套深度" in v.detail for v in complexity_violations(deep))
|
|
|
|
|
|
def test_complexity_names_count_columns_not_operators():
|
|
# 引用列数只数数据列(≤6),函数位算子名不计(T2 复审拍板钉:func_ids 同款)
|
|
six_cols_multi_op = ("cs_rank(cs_mean(close + open + volume)"
|
|
" + cs_std(vwap + turnover + roe_ttm))")
|
|
assert complexity_violations(six_cols_multi_op) == [] # 6 列+3 算子不误触
|
|
seven_cols = ("cs_rank(cs_mean(close + open + volume + tacc)"
|
|
" + cs_std(vwap + turnover + roe_ttm))")
|
|
assert _codes(complexity_violations(seven_cols)) == ["complexity"] # 7 列仍拦
|
|
|
|
|
|
def test_dup_exact_whole_tree():
|
|
lib = {"fund_gross_margin": "cs_rank(gross_margin)"}
|
|
vs = dup_violations("cs_rank(gross_margin)", lib)
|
|
assert [v.code for v in vs] == ["dup_exact"]
|
|
|
|
|
|
def test_dup_common_subtree_ge8():
|
|
shared = "ts_mean(close, 20) + ts_std(close, 20)" # 9 节点
|
|
lib = {"f1": f"cs_rank({shared})"}
|
|
vs = dup_violations(f"cs_rank(-({shared}))", lib)
|
|
assert [v.code for v in vs] == ["dup_subtree"]
|
|
|
|
|
|
def test_dup_commutative_matched():
|
|
shared = "ts_mean(close, 20) + ts_std(close, 20)"
|
|
lib = {"f1": f"cs_rank({shared})"}
|
|
swapped = " + ".join(reversed(shared.split(" + "))) # 加法交换
|
|
vs = dup_violations(f"cs_rank({swapped})", lib)
|
|
assert [v.code for v in vs] == ["dup_subtree"]
|
|
|
|
|
|
def test_p3_9_small_factor_commutative_swap_caught():
|
|
"""P3-9: 整树 <8 节点的交换律重排躲过 dup_exact(严格序键)也躲过
|
|
dup_subtree(<8 不触发)——close+open 在库 vs open+close 候选须拒."""
|
|
lib = {"f_small": "close + open"}
|
|
vs = dup_violations("open + close", lib)
|
|
assert [v.code for v in vs] == ["dup_subtree"]
|
|
|
|
|
|
def test_p3_9_non_commutative_not_flagged():
|
|
"""P3-9 不误伤:减法不可交换,close-open 与 open+close 键不同,放行."""
|
|
lib = {"f_small": "open + close"}
|
|
assert dup_violations("close - open", lib) == []
|
|
|
|
|
|
def test_p3_13_name_re_trailing_newline_rejected():
|
|
"""P3-13: NAME_RE $ 允许尾换行,'abc\\n' 曾过闸——改 \\Z 须拒."""
|
|
assert _codes(check_name("abc\n", set())) == ["bad_name"]
|
|
|
|
|
|
def test_dup_clean_when_different():
|
|
lib = {"f1": "cs_rank(gross_margin)"}
|
|
assert dup_violations("cs_rank(ts_mean(close, 20))", lib) == []
|
|
|
|
|
|
def test_check_candidate_composition():
|
|
lib = {"f1": "cs_rank(gross_margin)"}
|
|
vs = check_candidate("llm_x", "cs_rank(gross_margin)",
|
|
existing_names=set(), library_exprs=lib)
|
|
assert "dup_exact" in [v.code for v in vs]
|
|
assert check_candidate("llm_x", "cs_rank(ts_mean(close, 20))",
|
|
existing_names=set(), library_exprs=lib) == []
|
|
|
|
|
|
def test_precheck_bad_arity_blocked():
|
|
assert _codes(precheck_expression("cs_rank(ts_mean(close))")) == ["bad_arity"]
|
|
assert "bad_arity" in _codes(
|
|
precheck_expression("cs_rank(ts_delay(close, 20, 99))"))
|
|
assert "bad_arity" in _codes(
|
|
precheck_expression("cs_rank(cs_neutralize(roe_ttm))")) # 2 参只给 1
|
|
# 合法式不误拦(含一元/二元/TA 四参边界)
|
|
assert precheck_expression("cs_rank(ts_mean(close, 20))") == []
|
|
assert precheck_expression("cs_rank(ts_corr(close, open, 10))") == []
|
|
assert precheck_expression("cs_rank(ts_quantile(close, 20, 0.5))") == []
|
|
assert precheck_expression("ta_atr(high, low, close, 14)") == []
|
|
|
|
|
|
def test_precheck_blocked_containers():
|
|
assert _codes(precheck_expression("cs_rank((close, open))")) == ["syntax_not_allowed"]
|
|
assert _codes(precheck_expression("cs_rank([close])")) == ["syntax_not_allowed"]
|
|
|
|
|
|
def test_precheck_blocks_fstring_await_yield_family():
|
|
"""P2-8: 白名单曾漏 JoinedStr/FormattedValue/Await/Yield/YieldFrom——
|
|
f-string/await/yield 零违规过门拿「合法候选」身份,护栏契约被击穿."""
|
|
cases = {
|
|
'f"{close}"': "JoinedStr",
|
|
"await cs_rank(close)": "Await",
|
|
"cs_rank((yield close))": "Yield",
|
|
"cs_rank((yield from [close]))": "YieldFrom",
|
|
}
|
|
for expr, node in cases.items():
|
|
vs = precheck_expression(expr)
|
|
assert any(v.code == "syntax_not_allowed" and node in v.detail
|
|
for v in vs), (expr, [(v.code, v.detail) for v in vs])
|
|
|
|
|
|
def test_precheck_no_data_column():
|
|
vs = precheck_expression("cs_rank(1.0)") # 纯常量=零数据列
|
|
assert "no_data_column" in _codes(vs)
|
|
assert precheck_expression("cs_rank(close + 1)") == [] # 有数据列不误拦
|