276 lines
12 KiB
Python
276 lines
12 KiB
Python
|
# -*- 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())
|