# -*- coding: utf-8 -*- """系统自决包 (2026-09-14 台账 015)。后端离线纯逻辑单测, 不连库。 把「该不该做」交回三源合议与研判闸。人只处理四类例外: 研判不可用、上游硬风险、合议装不上、 判决词认不出。这里测四块: A 合议路由的择时异议观察 (consensus.decide timing_dissent)。 B 裁决理由码表与拼句 (decide_reasons.compose)。 C 决策核心 _self_decide: 决定表逐格 (放行强/试探仓/逻辑存疑/深亏补仓/目标价/上限/四类交人)、 四档 off/record/decline/full 的降级。 D 动作引擎 scan_open 打 decide_kind (关掉不打, 逐字回旧) 与 mark_decide 优先级、参数与哨兵。 端到端的自动清仓/自动采纳落表见 test_wiring.py (目标价那两例)。 """ import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.core import action_engine as ae # noqa: E402 from app.core import consensus # noqa: E402 from app.core import decide_reasons as dr # noqa: E402 from app.services import judge # noqa: E402 from app.services import param_store # noqa: E402 from app.services.proposal_service import _self_decide # noqa: E402 RESULTS = [] def case(name): def deco(fn): RESULTS.append((name, fn)) return fn return deco def _v(verdict=judge.PASS, conf=78, degraded=False): return {"verdict": verdict, "confidence": conf, "degraded": degraded, "reason": ""} def _cand(action=ae.A_OPEN, decide_kind=None, consensus=True, risk=None): c = {"ts_code": "600000.SH", "action": action, "side": "buy", "qty": 1000, "reason": "x", "hard_numbers": {}} if action in (ae.A_EXIT, ae.A_TRIM): c["side"] = "sell" if decide_kind: c["decide_kind"] = decide_kind if consensus: c["consensus"] = {"direction": "看多", "route": "放行", "strength": "强"} if risk: c["hard_numbers"]["risk"] = risk return c # ================================================================ A 合议择时异议 @case("A 合议·方向看多但择时看空: 默认交人 (逐字回旧)") def _(): d = consensus.decide("看多", "看多", "看空") assert d["route"] == "交人" and "择时" in d["route_reason"], d assert "disp" not in d, d @case("A 合议·自决开 (timing_dissent=observe): 改观察, 等择时转多, disp=wait_timing") def _(): d = consensus.decide("看多", "看多", "看空", timing_dissent="observe") assert d["route"] == "观察" and d["route_reason"] == "等择时转多", d assert d.get("disp") == "wait_timing", d # 择时不看空时 observe 不改别的格 (三方看多仍放行) assert consensus.decide("看多", "看多", "看多", timing_dissent="observe")["route"] == "放行" # ================================================================ B 裁决理由拼句 @case("B decide_reasons·采纳预设码 → 句子与码") def _(): s, code = dr.compose("ACCEPTED", "ACCEPT_CONSENSUS", None, None) assert code == "ACCEPT_CONSENSUS" and "合议放行" in s, (s, code) @case("B decide_reasons·选其他不填补充 → 报错; 填了 → 带补充") def _(): try: dr.compose("DECLINED", "DECLINE_OTHER", "", None) assert False, "空补充应报错" except ValueError as e: assert "其他" in str(e) or "补充" in str(e), e s, code = dr.compose("DECLINED", "DECLINE_OTHER", "估值太贵", None) assert code == "DECLINE_OTHER" and "估值太贵" in s, (s, code) @case("B decide_reasons·旧自由文本兼容 / 全空报错 / 码串表报错") def _(): s, code = dr.compose("ACCEPTED", None, None, "我自己判断成立") assert code is None and s == "我自己判断成立", (s, code) try: dr.compose("ACCEPTED", None, None, None) assert False except ValueError: pass try: dr.compose("ACCEPTED", "DECLINE_TECH_WEAK", None, None) # 采纳用了驳回码 assert False except ValueError as e: assert "不适用" in str(e), e # ================================================================ C 决策核心 @case("C 不变式·off 一律不归自决管 (decision none)") def _(): d = _self_decide(_cand(), _v(), {"self_decide": "off"}, 0) assert d["decision"] == "none", d @case("C 放行强·PASS 把握度 78, full → 采纳标准仓") def _(): d = _self_decide(_cand(), _v(conf=78), {"self_decide": "full"}, 0) assert d["decision"] == "accept" and d["tier"] == "standard", d assert d["rule"] == "consensus_strong", d assert d["why"].startswith("系统采纳"), d @case("C 放行强·PASS 把握度 40: TRIAL 关转交人 / 开则采纳试探仓") def _(): off = _self_decide(_cand(), _v(conf=40), {"self_decide": "full", "self_decide_trial": False}, 0) assert off["decision"] == "confirm", off on = _self_decide(_cand(), _v(conf=40), {"self_decide": "full", "self_decide_trial": True}, 0) assert on["decision"] == "accept" and on["tier"] == "trial", on @case("C 只有一方表态 (weak_consensus): TRIAL 开采纳试探 / 关交人") def _(): c = _cand(decide_kind=ae.DK_WEAK_CONSENSUS) on = _self_decide(c, _v(conf=78), {"self_decide": "full", "self_decide_trial": True}, 0) assert on["decision"] == "accept" and on["tier"] == "trial", on off = _self_decide(c, _v(conf=78), {"self_decide": "full", "self_decide_trial": False}, 0) assert off["decision"] == "confirm", off @case("C 异议只在技术面 (tech_bear): 采纳试探仓") def _(): d = _self_decide(_cand(decide_kind=ae.DK_TECH_BEAR), _v(conf=78), {"self_decide": "full", "self_decide_trial": True}, 0) assert d["decision"] == "accept" and d["tier"] == "trial", d @case("C 四类交人·研判不可用 / 硬风险 / 合议装不上 / 判决认不出 → confirm") def _(): fp = {"self_decide": "full", "self_decide_trial": True} assert _self_decide(_cand(), _v(degraded=True), fp, 0)["decision"] == "confirm" assert _self_decide(_cand(risk=["高位派发"]), _v(), fp, 0)["decision"] == "confirm" assert _self_decide(_cand(consensus=False), _v(), fp, 0)["decision"] == "confirm" assert _self_decide(_cand(decide_kind=ae.DK_VERDICT_UNKNOWN), _v(), fp, 0)["decision"] == "confirm" @case("C 研判不可用带 judge_unavail 标记 (研判不可用占比复核用)") def _(): d = _self_decide(_cand(), _v(degraded=True), {"self_decide": "full"}, 0) assert d["why"].startswith("研判不可用"), d @case("C 逻辑存疑 (logic_doubt): 自动放弃 decline") def _(): d = _self_decide(_cand(decide_kind=ae.DK_LOGIC_DOUBT), _v(), {"self_decide": "full"}, 0) assert d["decision"] == "decline", d @case("C 深亏补仓: PASS≥60 自动补 / PASS<60 放弃 / 不可用交人") def _(): dca = _cand(action=ae.A_DCA, decide_kind=ae.DK_DCA_DEEP, consensus=False) dca["side"] = "buy" fp = {"self_decide": "full"} assert _self_decide(dca, _v(conf=70), fp, 0)["decision"] == "accept" assert _self_decide(dca, _v(conf=40), fp, 0)["decision"] == "decline" assert _self_decide(dca, _v(degraded=True), fp, 0)["decision"] == "confirm" @case("C 目标价到价: EXIT 直接采纳 (不进研判范围)") def _(): ex = _cand(action=ae.A_EXIT, decide_kind=ae.DK_TARGET_PRICE, consensus=False) d = _self_decide(ex, _v(verdict=judge.PASS, conf=None), {"self_decide": "full"}, 0) assert d["decision"] == "accept" and d["rule"] == ae.DK_TARGET_PRICE, d @case("C 每日上限: 达上限 3 时第四只转交人") def _(): fp = {"self_decide": "full", "self_decide_daily_max": 3} assert _self_decide(_cand(), _v(conf=78), fp, 2)["decision"] == "accept" # 第三只 over = _self_decide(_cand(), _v(conf=78), fp, 3) # 第四只 assert over["decision"] == "confirm" and "上限" in over["why"], over @case("C 档位降级·record 只记不做 (would 留原判); decline 只放弃不采纳") def _(): # record: 本会采纳但 decision 降为 none, would 留 accept rec = _self_decide(_cand(), _v(conf=78), {"self_decide": "record"}, 0) assert rec["decision"] == "none" and rec["would"] == "accept", rec # decline: 采纳降 none, 放弃仍放弃 dec_acc = _self_decide(_cand(), _v(conf=78), {"self_decide": "decline"}, 0) assert dec_acc["decision"] == "none" and dec_acc["would"] == "accept", dec_acc dec_dec = _self_decide(_cand(decide_kind=ae.DK_LOGIC_DOUBT), _v(), {"self_decide": "decline"}, 0) assert dec_dec["decision"] == "decline", dec_dec # ================================================================ D 动作引擎与参数 @case("D mark_decide·按优先级取最保守: 认不出 < 逻辑存疑 < 关注 < 试探仓") def _(): c = {} ae.mark_decide(c, ae.DK_ADVICE_TRIAL) assert c["decide_kind"] == ae.DK_ADVICE_TRIAL ae.mark_decide(c, ae.DK_VERDICT_WATCH) # 更高优先, 覆盖 assert c["decide_kind"] == ae.DK_VERDICT_WATCH ae.mark_decide(c, ae.DK_LOGIC_DOUBT) # 再高, 覆盖 assert c["decide_kind"] == ae.DK_LOGIC_DOUBT ae.mark_decide(c, ae.DK_ADVICE_TRIAL) # 更低, 不覆盖 assert c["decide_kind"] == ae.DK_LOGIC_DOUBT @case("D self_decide_on: off/空 → False; record/decline/full → True") def _(): assert ae.self_decide_on({"self_decide": "off"}) is False assert ae.self_decide_on({}) is False assert ae.self_decide_on({"self_decide": "record"}) is True assert ae.self_decide_on({"self_decide": "full"}) is True @case("D eval_dca 深档·自决关不打 decide_kind、confirm 真; 自决开打 dca_deep") def _(): p = {"ts_code": "688001.SH", "cushion_pct": -0.20, "dca_qty": 0, "dca_count": 0, "price": 10.0, "base_qty": 2000} pr = {"dca_triggers": (-0.08, -0.15), "dca_deep_confirm": -0.15, "dca_max_ratio": 0.5} off = ae.eval_dca(p, pr, {}) assert off and off["needs_user_confirm"] is True and "decide_kind" not in off, off on = ae.eval_dca(p, {**pr, "self_decide": "full"}, {}) assert on and on.get("decide_kind") == ae.DK_DCA_DEEP, on @case("D 参数登记·三键在 RUNTIME_EXTRA, 合法值与范围校验") def _(): for k in ("PMS_SELF_DECIDE", "PMS_SELF_DECIDE_TRIAL", "PMS_SELF_DECIDE_DAILY_MAX"): assert k in param_store.RUNTIME_EXTRA, k assert param_store.RUNTIME_EXTRA["PMS_SELF_DECIDE"][0] == "off" # 代码默认 off assert param_store.FAIL_CLOSED.get("PMS_SELF_DECIDE") == "off" # 读不到按 off assert param_store._range_check("PMS_SELF_DECIDE", "full") is None assert param_store._range_check("PMS_SELF_DECIDE", "wtf") is not None # 非法值被拒 assert param_store._range_check("PMS_SELF_DECIDE_DAILY_MAX", 3) is None assert param_store._range_check("PMS_SELF_DECIDE_DAILY_MAX", 99) is not None @case("D repo 两个自决查询函数存在且单表") def _(): from app.repo import pms_repo assert callable(pms_repo.system_decided_today) and callable(pms_repo.system_pass_count_today) assert pms_repo.SYSTEM_ACCEPT_REASON_PREFIX == "系统采纳" def main(): ok = 0 for name, fn in RESULTS: try: fn() ok += 1 print(" ok " + name) except Exception: import traceback print(" FAIL " + name) traceback.print_exc() print("-" * 60) if ok == len(RESULTS): print("ALL PASS (%d cases)" % ok) return 0 print("FAILED %d/%d" % (len(RESULTS) - ok, len(RESULTS))) return 1 if __name__ == "__main__": sys.exit(main())