90 lines
3.9 KiB
Python
90 lines
3.9 KiB
Python
# tests/factor/test_expression_match.py
|
|
"""AST 防换皮(spec §4.2 2026-10-10):交换律归一化+最大公共子树占比."""
|
|
from sanguo_factor.expression_match import similarity, top_similar
|
|
|
|
|
|
def test_identical_and_commutative():
|
|
assert similarity("close/open", "close/open") == 1.0
|
|
# 交换律:a+b ≡ b+a(归一化后同构)
|
|
assert similarity("close + open", "open + close") == 1.0
|
|
assert similarity("ts_mean(close, 5) * volume",
|
|
"volume * ts_mean(close, 5)") == 1.0
|
|
|
|
|
|
def test_partial_common_subtree():
|
|
# 节点口径(Name/Constant/Call 各计 1,算子与 ctx 为元数据不计):
|
|
# 公共子树 ts_mean(close,5)=4 节点(Call+func+close+5);
|
|
# 整树=BinOp+Call+volume(turnover)=6 节点;ratio=4/6≈0.6667
|
|
r = similarity("ts_mean(close, 5) / volume",
|
|
"ts_mean(close, 5) * turnover")
|
|
assert r is not None and 0.6 < r < 1.0
|
|
|
|
|
|
def test_no_common_and_invalid():
|
|
assert similarity("close / open", "volume * turnover") == 0.0
|
|
assert similarity("close +++", "close/open") is None # 解析失败如实 None
|
|
assert similarity("", "close") is None
|
|
|
|
|
|
def test_top_similar_floor():
|
|
cands = {"fa_a": "close + open", "fa_b": "open + close",
|
|
"fa_c": "volume * turnover"}
|
|
hits = top_similar("close + open", cands, floor=0.6)
|
|
assert [h["name"] for h in hits] == ["fa_a", "fa_b"]
|
|
assert all(h["ratio"] >= 0.6 for h in hits)
|
|
# 空表达式/坏 candidates 不炸
|
|
assert top_similar("close", {"fa_x": ""}) == []
|
|
|
|
|
|
# —— codex review 修批(2026-10-10):UnaryOp/keywords/limit/深树防护 ——
|
|
def test_unaryop_distinct_operands_not_similar():
|
|
"""-close vs -open:UnaryOp 显式编码,不再落 ? 兜底桶互相撞签."""
|
|
assert similarity("-close", "-open") == 0.0
|
|
assert similarity("-ts_mean(close,5)", "-ts_mean(open,99)") == 0.0
|
|
# 同构 UnaryOp 仍 1.0
|
|
assert similarity("-close", "-close") == 1.0
|
|
# containment 语义:close 整个是 -close 的子树,较小表达式全包含→1.0
|
|
# (口径=公共子树/较小表达式节点数,非对称同构判定)
|
|
assert similarity("-close", "close") == 1.0
|
|
|
|
|
|
def test_call_keywords_encoded():
|
|
"""Call 关键字参数须进签名:kw 值不同→不再 1.0(共享 close 操作数
|
|
仅余 1/5 地板噪声,低于提示阈 0.6)."""
|
|
r = similarity("ts_mean(close, window=5)", "ts_mean(close, window=99)")
|
|
assert r is not None and r < 0.6
|
|
# 同 kwargs 同构
|
|
assert similarity("f(close, w=5)", "f(close, w=5)") == 1.0
|
|
# kwargs 顺序无关(按名排序后拼接)
|
|
assert similarity("f(a=1, b=2)", "f(b=2, a=1)") == 1.0
|
|
# kwargs 有无也是结构差异:无 kw vs 有 kw 不撞签(仅共享 close,1/3)
|
|
r2 = similarity("f(close)", "f(close, w=5)")
|
|
assert r2 is not None and r2 < 0.6
|
|
|
|
|
|
def test_nested_call_common_subtree():
|
|
"""嵌套 Call:全同→1;算子名也不同→0;仅共享外层算子名→占比 1/6 低于提示阈."""
|
|
assert similarity("cs_rank(ts_mean(close, 5))",
|
|
"cs_rank(ts_mean(close, 5))") == 1.0
|
|
assert similarity("cs_rank(ts_mean(close, 5))",
|
|
"zscore(ts_mean(open, 9))") == 0.0
|
|
r = similarity("cs_rank(ts_mean(close, 5))", "cs_rank(ts_mean(open, 9))")
|
|
assert r is not None and r < 0.6
|
|
|
|
|
|
def test_top_similar_limit_and_empty():
|
|
"""limit=5 上限(注册路径 registry 只存前 5)+空候选不炸."""
|
|
cands = {f"fa_{i}": "close + open" for i in range(7)}
|
|
hits = top_similar("close + open", cands)
|
|
assert len(hits) == 5
|
|
assert top_similar("close", {}) == []
|
|
assert top_similar("", {"fa_a": "close"}) == []
|
|
|
|
|
|
def test_deep_expression_no_crash():
|
|
"""深表达式递归爆栈不 raise,如实 None(factor_guard _DEPTH_MAX=12 挡真实
|
|
注册,此处为纯防护——RecursionError 捕获返回 None)."""
|
|
deep = "close" + " + close" * 3000
|
|
assert similarity(deep, deep) is None
|
|
assert top_similar(deep, {"fa_a": deep}) == []
|