128 lines
5.6 KiB
Python
128 lines
5.6 KiB
Python
"""plan_reconcile.verdict() 纯逻辑单测（无需 DB / 无需 pandas 实体）。
|
||
|
||
verdict() 只读 L 字典、不碰库，故可离线跑通。校验四种 decision 分支 + tier 三档
|
||
+ 前20外提示，且 verdict_text 与命令行 explain()『▶ 判决：』之后的文案逐字一致
|
||
（页面一行解释 == 命令行对账，同口径）。
|
||
|
||
隔离约定：仅对"当前尚未导入"的依赖装临时桩，导入后立即还原，绝不覆盖真实模块——
|
||
因此与 test_pool_logic.py / test_xxl_trigger.py 同进程 pytest 收集也不会互相污染。
|
||
|
||
跑法：python3 test_plan_verdict.py 或 pytest test_plan_verdict.py
|
||
"""
|
||
import importlib.util
|
||
import os
|
||
import sys
|
||
import types
|
||
|
||
_NAMES = ("pandas", "common", "db", "tracks", "psycopg", "pymysql")
|
||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||
_PR_PATH = os.path.join(_HERE, "plan_reconcile.py")
|
||
|
||
|
||
def _load_isolated():
|
||
"""装最小桩→加载 plan_reconcile→还原 import-only 桩。
|
||
|
||
config 留到 run() 结束再还原：verdict() 的 gate=0 分支在运行期才 `import config`，
|
||
需要 UPSIDE_NEG_TOLERANCE 可读（真实环境用真 config，离线用桩）。
|
||
"""
|
||
saved = {n: sys.modules.get(n) for n in _NAMES + ("config",)}
|
||
for n in _NAMES:
|
||
if sys.modules.get(n) is None:
|
||
sys.modules[n] = types.ModuleType(n) # 仅在缺席时装桩，不覆盖真实模块
|
||
if sys.modules.get("config") is None:
|
||
cfg = types.ModuleType("config")
|
||
cfg.UPSIDE_NEG_TOLERANCE = 0.0
|
||
sys.modules["config"] = cfg
|
||
spec = importlib.util.spec_from_file_location("plan_reconcile_uut", _PR_PATH)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
for n in _NAMES: # 还原 import-only 桩
|
||
if saved[n] is None:
|
||
sys.modules.pop(n, None)
|
||
else:
|
||
sys.modules[n] = saved[n]
|
||
return mod, saved
|
||
|
||
|
||
def _restore_config(saved):
|
||
if saved["config"] is None:
|
||
sys.modules.pop("config", None)
|
||
else:
|
||
sys.modules["config"] = saved["config"]
|
||
|
||
|
||
def _L():
|
||
"""合成档位数据（主榜 30 只 / 观察档 40 只），覆盖四种判决分支。"""
|
||
return {
|
||
"names": {
|
||
"SH600519": "贵州茅台", "SH600000": "浦发银行", "SZ000002": "万科A",
|
||
"SZ000111": "*ST锐电", "SZ000625": "长安汽车", "SH601111": "中国国航",
|
||
},
|
||
"gate": {
|
||
"SH600519": 2.0, "SH601111": 2.0, "SZ000625": 2.0,
|
||
"SH600000": 1.0, "SZ000002": 0.0, "SZ000111": 0.0,
|
||
# SZ300999 故意缺席 → absent
|
||
},
|
||
"score": {
|
||
"SH600519": 235.0, "SH601111": 232.0, "SZ000625": 215.0,
|
||
"SH600000": 105.0,
|
||
},
|
||
"upside": {"SH600519": 0.25, "SZ000002": -0.05},
|
||
"heat": {"SH600519": 0.9},
|
||
"tr": {"SH600519": 55.0},
|
||
"hits": {"SH600519": [("光伏", "CATL", "graph_segment")]},
|
||
"tmap": {"SH600519": ["宁德时代(源2×空间30%)"]},
|
||
"ev": {"SH600519": {"segs": {"电池"}, "chains": {"锂电链"}}},
|
||
"main": ["x"] * 30,
|
||
"obs": ["y"] * 40,
|
||
"rank_main": {"SH600519": 3, "SH601111": 25, "SZ000625": 5},
|
||
"rank_obs": {"SH600000": 7},
|
||
}
|
||
|
||
|
||
def test_verdict():
|
||
pr, saved = _load_isolated()
|
||
try:
|
||
L = _L()
|
||
|
||
v = pr.verdict("SH600519", L) # 主榜·强传导
|
||
assert v["decision"] == "main" and v["tier"] == "强传导", v
|
||
assert v["verdict_text"] == "主榜第 3/30 名（score=235.0，强传导档）", v["verdict_text"]
|
||
assert v["score"] == 235.0 and v["rank"] == 3 and v["rank_total"] == 30, v
|
||
assert v["upside"] == 0.25 and v["tracks"][0]["rule"] == "graph_segment", v
|
||
assert v["graph_segments"] == ["电池"] and v["transmission"] == 55.0, v
|
||
|
||
v = pr.verdict("SH601111", L) # 主榜但名次>20 → 带前20提示
|
||
assert v["decision"] == "main" and v["tier"] == "强传导", v
|
||
assert v["verdict_text"] == (
|
||
"主榜第 25/30 名（score=232.0，强传导档）"
|
||
"——计划默认只显示前 20，名次靠后不等于不在计划里"), v["verdict_text"]
|
||
|
||
v = pr.verdict("SZ000625", L) # score 215 → 弱传导
|
||
assert v["decision"] == "main" and v["tier"] == "弱传导", v
|
||
assert v["verdict_text"] == "主榜第 5/30 名（score=215.0，弱传导档）", v["verdict_text"]
|
||
|
||
v = pr.verdict("SH600000", L) # 观察档
|
||
assert v["decision"] == "observe", v
|
||
assert v["verdict_text"] == "观察档第 7/40 名（score=105.0，无券商覆盖、低置信）", v["verdict_text"]
|
||
|
||
v = pr.verdict("SZ000002", L) # gate=0，有覆盖 upside<0 → 贵了不买
|
||
assert v["decision"] == "reject", v
|
||
assert "贵了不买是绝对下限" in v["reason"], v
|
||
assert v["verdict_text"] == f"不采纳（gate=0）。原因：{v['reason']}", v["verdict_text"]
|
||
|
||
v = pr.verdict("SZ000111", L) # gate=0，名字 *ST → 风险闸
|
||
assert v["decision"] == "reject" and v["risk"] is True, v
|
||
assert "风险闸" in v["reason"], v
|
||
|
||
v = pr.verdict("SZ300999", L) # 不在 gate 表 → absent
|
||
assert v["decision"] == "absent", v
|
||
assert v["verdict_text"] == "当日档位表无此票（可能不在覆盖池 universe，或当日未出行）", v["verdict_text"]
|
||
finally:
|
||
_restore_config(saved)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
test_verdict()
|
||
print("ALL OK — verdict() 四分支 / tier 三档 / 前20提示 全部通过")
|