zlt
/
tradingSystem
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
tradingSystem/scripts/test_batch20_units.py

557 lines
30 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# -*- coding: utf-8 -*-
"""
第二十批模块单测 (零外部依赖, 不连库不触网)
============================================
运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch20_units.py
覆盖: 选股系统候选卡判决接入 (2026-09-03, 方案《主观量化系统方案_2026-09-03》第三节 PMS 表):
* 按判决分流三态 (候选 / 关注 / 仅展示) 与没有判决时的回退, 判决值认不出一律交人;
* 仅展示在候选阶段就剔掉 (不白占 top_n 名额) 并单列;
* 自动执行开关的四条件 (判决候选 + 研判真通过 + 规则闸通过 + 风险列表为空), 任一缺失仍入队;
* 信号来源区分: producer_id 缺省 unknown、留痕文案按来源分写、新建仓插队只认择时决策系统;
* 密钥名单含会话密钥; 研判键放行五键、新建仓必答题、置信度保留、提议硬数字带研判结论与置信度;
* 页面静态守卫: 提议卡按 plan_rank 取名次、裁决理由必填并随请求体发。
约定同前: 全过输出 "ALL PASS (n cases)" 退出码 0。
"""
import os
import sys
import traceback
from datetime import datetime
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 signal_rules as sr # noqa: E402
from app.services import judge as jd # noqa: E402
from app.services import param_store as pstore # noqa: E402
from app.services import plan_feed as pf # noqa: E402
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
RESULTS = []
def case(name):
def deco(fn):
RESULTS.append((name, fn))
return fn
return deco
# ================================================================ 夹具 (口径同第十二批)
def params(**kw):
p = {"scale": 2_000_000.0, "stock_target_default": 0.06,
"batch_split": (0.5, 0.25, 0.25), "min_lot_merge": True}
p.update(kw)
return p
def caps(**kw):
c = {"scale": 2_000_000.0, "portfolio_cap": 0.70, "stock_cap": 0.08,
"max_names": 20, "portfolio_mv": 0.0, "names_count": 0,
"stock_mv": 0.0, "is_new_name": False, "sector": None,
"sector_names": 0, "sector_mv": 0.0,
"sector_names_map": {}, "sector_mv_map": {},
"sector_max_names": 4, "sector_max_ratio": 0.40,
"cash_reserve": 0.0, "sector_source_ready": True,
"cash_avail": None, "cash_source": "estimate"}
c.update(kw)
return c
def cand(code, price=10.0, score=100.0, sector=None, rank=1, **kw):
"""候选池一行 (plan_feed.select_candidates 的 items 形状 + 现价)。不传 verdict 就是旧版计划。"""
d = {"ts_code": code, "price": price, "score": score, "sector": sector,
"theme": kw.pop("theme", None), "tier": kw.pop("tier", "强传导"),
"upside": kw.pop("upside", 0.3), "heat": kw.pop("heat", 0.5),
"rank": rank, "bucket": "main", "src": "plan_api"}
d.update(kw)
return d
FIVE = ("verdict", "reasons", "missing", "risk", "card_rank")
def _scan(rows, *, route=True, slots=20, names_count=0, room=1_400_000, **pkw):
return ae.scan_open(candidates=rows, params=params(open_route_by_verdict=route, **pkw),
caps=caps(names_count=names_count), room_amt=room, slots=slots)
class _Patch:
"""临时替换若干模块属性, 退出时原样还回去 —— 单测不连库不触网。"""
def __init__(self):
self.saved = []
def __call__(self, mod, name, value):
self.saved.append((mod, name, getattr(mod, name)))
setattr(mod, name, value)
def __enter__(self):
return self
def __exit__(self, *a):
for mod, name, old in reversed(self.saved):
setattr(mod, name, old)
return False
def _patch_params(p: _Patch, mod, values: dict):
"""把某模块引用的 param_store 取值函数换成一份字典。"""
v = values
def _get_list(k, d=None, sep=","):
x = v.get(k)
if x in (None, ""):
return list(d or [])
return [s.strip() for s in str(x).split(sep) if s.strip()]
p(mod, "get", lambda k, d=None: v.get(k, d))
p(mod, "get_int", lambda k, d=0: int(v.get(k, d)))
p(mod, "get_float", lambda k, d=0.0: float(v.get(k, d)))
p(mod, "get_bool", lambda k, d=False: bool(v.get(k, d)))
p(mod, "get_list", _get_list)
# ================================================================ 判决分流 (动作引擎)
@case("判决分流·候选: 走现有自动流程, 不强制人工确认, 硬数字带候选卡五键")
def _():
row = cand("600000.SH", verdict="候选", reasons=["强传导链首环", "券商覆盖 7 家"],
missing=[], risk=[], card_rank=1)
r = _scan([row])
assert len(r["candidates"]) == 1 and not r["skipped"], r
c = r["candidates"][0]
assert c["needs_user_confirm"] is False and "confirm_why" not in c, c
assert c["judge_required"] is True, "候选照样送研判闸, 分流不绕闸"
hn = c["hard_numbers"]
assert hn["verdict"] == "候选" and hn["reasons"] == ["强传导链首环", "券商覆盖 7 家"], hn
assert hn["missing"] == [] and hn["risk"] == [] and hn["card_rank"] == 1, hn
@case("判决分流·关注: 照常产出候选、照常占名额, 但强制人工确认并写明「无法判断, 交人裁决」")
def _():
rows = [cand("600000.SH", score=300, verdict="关注", reasons=["强传导"],
missing=["确认线缺失"], risk=[], card_rank=3),
cand("600001.SH", score=200, verdict="候选", reasons=[], missing=[], risk=[])]
r = _scan(rows, slots=1, names_count=19)
assert [c["ts_code"] for c in r["candidates"]] == ["600000.SH"], r["candidates"]
c = r["candidates"][0]
assert c["needs_user_confirm"] is True, c
assert c["confirm_why"] == ae.WHY_WATCH_CONFIRM, c["confirm_why"]
assert "关注" in c["confirm_why"] and "交人裁决" in c["confirm_why"], c["confirm_why"]
assert c["hard_numbers"]["verdict"] == "关注" and c["hard_numbers"]["missing"] == ["确认线缺失"]
# 关注占掉了唯一的名额, 第二只按名额用完跳过 (关注不是免费的)
assert any(s["ts_code"] == "600001.SH" and "名额" in s["why"] for s in r["skipped"]), r["skipped"]
@case("判决分流·仅展示: 不产出、不占名额、不进任何闸, 跳过原因说清为什么不买")
def _():
rows = [cand("600000.SH", score=300, verdict="仅展示", reasons=["只差覆盖"], missing=["券商覆盖"],
risk=[]),
cand("600001.SH", score=200, verdict="候选", reasons=[], missing=[], risk=[])]
r = _scan(rows, slots=1, names_count=19)
assert [c["ts_code"] for c in r["candidates"]] == ["600001.SH"], r["candidates"]
sk = {s["ts_code"]: s["why"] for s in r["skipped"]}
# 这一栏回答的是「为什么今天没买」, 所以它只写原因, 不写判决名, 也不许写成
# 「名额用完」—— 仅展示的行排在名额判断之前, 跳过原因不能被名额那一刀盖掉。
assert "名额" not in sk["600000.SH"], sk
assert sk["600000.SH"] == "选股系统今天不建议买入", sk
# 名额没被仅展示占掉: 只有一个名额, 候选那只拿到了
assert r["candidates"][0]["needs_user_confirm"] is False
@case("判决分流·仅展示的跳过原因写出这一只自己的判决依据 (三种来路要分得开)")
def _():
# 三只都是仅展示, 但来路完全不同: 一只差当日涨幅、一只吸筹没到确认线、
# 一只被昨夜坏信号一票否决。写成同一句话的话, 人看不出哪只该继续盯。
rows = [cand("600000.SH", score=300, verdict="仅展示", basis="门槛未过：started"),
cand("600002.SH", score=290, verdict="仅展示",
basis="三门槛全过，但吸筹评分为「潜在吸筹」，未达确认线，不升格关注"),
cand("600003.SH", score=280, verdict="仅展示",
basis="硬风险否决：决策系统昨夜信号 SELL"),
cand("600004.SH", score=270, verdict="仅展示")] # 旧版计划: 没有判决依据
sk = {x["ts_code"]: x["why"] for x in _scan(rows, slots=3)["skipped"]}
assert sk["600000.SH"].endswith("：门槛未过：started"), sk["600000.SH"]
assert "潜在吸筹" in sk["600002.SH"], sk["600002.SH"]
assert "SELL" in sk["600003.SH"], sk["600003.SH"]
# 三句话必须互不相同 —— 这正是这条用例要守住的东西
assert len({sk["600000.SH"], sk["600002.SH"], sk["600003.SH"]}) == 3, sk
# 判决依据缺失时退回原来那句固定文案, 不许让这一栏空掉
assert sk["600004.SH"] == "选股系统今天不建议买入", sk["600004.SH"]
# 四句话里一个代码名、一句内部说法都不许有 —— 这一栏是给人读的
for c in ("600000.SH", "600002.SH", "600003.SH", "600004.SH"):
for w in ("started", "pointed", "covered", "SELL", "AVOID",
"仅展示", "不出提议", "判决"):
assert w not in sk[c], (c, w, sk[c])
@case("候选一只不剩时的说明: 只念不为零的那几项, 不把计数字典原样打印出来")
def _():
from app.services import proposal_service as psvc
z = {"held": 0, "black": 0, "tier": 0, "score": 0, "sources": 0, "upside": 0,
"st": 0, "theme": 0, "dup": 0, "capped": 0, "display": 83}
w = psvc._no_candidate_why({"considered": 83, "dropped": z})
assert w == "今天考察了 83 只票，一只都没留下：选股系统判定今天不买 83 只。", w
# 页面上出现过 "落选明细 {'held': 0, ...}" 这种东西, 这条就是钉住不许再出现
for bad in ("{", "}", "'", "held", "display", "capped"):
assert bad not in w, (bad, w)
# 多项不为零时按筛选实际发生的顺序念
w2 = psvc._no_candidate_why({"considered": 50,
"dropped": {"held": 3, "display": 40, "capped": 7}})
assert w2.index("已经持有") < w2.index("不买") < w2.index("名额"), w2
# 一只都没考察过时不硬拼一个空清单
assert psvc._no_candidate_why({"considered": 0, "dropped": {}}) == "今天没有票可考察。"
@case("判决分流·没有判决 (旧版计划) 时开关开着与关着产出逐字一致, 五键全 None")
def _():
rows = [cand("600000.SH", score=300), cand("600001.SH", score=200, rank=2)]
on, off = _scan(rows, route=True), _scan(rows, route=False)
assert on == off, (on, off)
assert [c["ts_code"] for c in on["candidates"]] == ["600000.SH", "600001.SH"]
for c in on["candidates"]:
assert c["needs_user_confirm"] is False and "confirm_why" not in c, c
for k in FIVE:
assert k in c["hard_numbers"] and c["hard_numbers"][k] is None, (k, c["hard_numbers"])
# 计划层同样: 没有 verdict 的行永远不会被 display 那一刀碰到
p = pf.parse_plan({"date": "2026-09-02", "main": [
{"rank": 1, "code": "SH600000", "name": "甲", "score": 300},
{"rank": 2, "code": "SH600001", "name": "乙", "score": 200}]})
s = pf.select_candidates(p, top_n=5, route_by_verdict=True)
assert len(s["items"]) == 2 and s["dropped"]["display"] == 0 and s["display_only"] == []
@case("判决分流·开关关着时判决不起作用: 关注不强制确认、仅展示照常产出 (按档位, 即旧行为)")
def _():
rows = [cand("600000.SH", score=300, verdict="关注", reasons=["a"], missing=["b"], risk=[]),
cand("600001.SH", score=200, rank=2, verdict="仅展示", reasons=[], missing=[], risk=[])]
r = _scan(rows, route=False)
assert [c["ts_code"] for c in r["candidates"]] == ["600000.SH", "600001.SH"], r
assert not r["skipped"], r["skipped"]
for c in r["candidates"]:
assert c["needs_user_confirm"] is False and "confirm_why" not in c, c
# 判决仍然原样进硬数字 (开关只管分流, 不管记录)
assert r["candidates"][0]["hard_numbers"]["verdict"] == "关注"
assert r["candidates"][1]["hard_numbers"]["verdict"] == "仅展示"
@case("判决分流·判决值认不出一律交人 (上游改了词表不许被静默当成候选)")
def _():
r = _scan([cand("600000.SH", verdict="推荐", reasons=[], missing=[], risk=[])])
c = r["candidates"][0]
assert c["needs_user_confirm"] is True and "无法识别" in c["confirm_why"], c
assert "推荐" in c["confirm_why"] and "交人裁决" in c["confirm_why"], c["confirm_why"]
# 三个合法值与空值的判定钉死
assert ae.verdict_confirm_why(None) is None and ae.verdict_confirm_why("") is None
assert ae.verdict_confirm_why(ae.VERDICT_CANDIDATE) is None
assert ae.verdict_confirm_why(ae.VERDICT_WATCH) == ae.WHY_WATCH_CONFIRM
assert ae.verdict_confirm_why(ae.VERDICT_DISPLAY).startswith(ae.WHY_DISPLAY_ONLY)
assert (ae.VERDICT_CANDIDATE, ae.VERDICT_WATCH, ae.VERDICT_DISPLAY) == ("候选", "关注", "仅展示")
@case("候选筛选·仅展示在 top_n 截断之前剔掉并单列 (否则前 N 名被仅展示占满, 候选永远轮不上)")
def _():
def row(i, code, score, verdict=None):
d = {"rank": i, "code": code, "name": f"票{i}", "score": score, "tier": "强传导",
"evidence": {"theme": "整车"}}
if verdict:
d["verdict"] = verdict
return d
p = pf.parse_plan({"date": "2026-09-02", "main": [
row(1, "SH600001", 300, "仅展示"), row(2, "SH600002", 299, "仅展示"),
row(3, "SH600003", 298, "仅展示"), row(4, "SH600004", 250, "候选"),
row(5, "SH600005", 240, "关注"), row(6, "SH600006", 230), # 没有判决 (旧行)
row(7, "SH600007", 220, "候选")]})
on = pf.select_candidates(p, top_n=3, route_by_verdict=True)
assert [x["ts_code"] for x in on["items"]] == ["600004.SH", "600005.SH", "600006.SH"], on["items"]
assert on["dropped"]["display"] == 3 and on["dropped"]["capped"] == 1, on["dropped"]
# 2026-09-04 起这里是带判决依据的行, 不再是一串代码。
assert [x["ts_code"] for x in on["display_only"]] == \
["600001.SH", "600002.SH", "600003.SH"], on["display_only"]
assert on["eligible"] == 4 and on["considered"] == 7
# 不分流: 前三名全是仅展示 —— 这正是要在候选阶段剔掉的理由
off = pf.select_candidates(p, top_n=3)
assert {x["verdict"] for x in off["items"]} == {"仅展示"}, off["items"]
assert off["dropped"]["display"] == 0 and off["display_only"] == []
# ================================================================ 自动执行开关 (提议分流)
def _route(c, *, values, judge_resp, open_autonomy="propose_only"):
"""把一条新建仓候选送进 proposal_service._route_one, 规则闸/研判/落表全部换成桩。"""
from app.core import rule_gate
from app.repo import pms_repo
from app.services import judge, param_store, portfolio, proposal_service as psvc
got = {"ledger": [], "instructions": [], "proposals": []}
with _Patch() as p:
p(rule_gate, "check", lambda **kw: {"passed": True, "failed": [], "warnings": []})
p(judge, "request", lambda cand_, context=None, **kw: dict(judge_resp))
p(portfolio, "caps_ctx", lambda *a, **kw: {})
p(pms_repo, "insert_ledger", lambda **kw: got["ledger"].append(kw) or 1)
p(pms_repo, "insert_instruction", lambda **kw: got["instructions"].append(kw) or 1)
p(pms_repo, "insert_proposal", lambda **kw: got["proposals"].append(kw) or 1)
p(pms_repo, "list_ledger", lambda **kw: [])
_patch_params(p, param_store, {"PMS_EXEC_WINDOW_TDAYS": 3, "PMS_PROPOSAL_TTL_HOURS": 24,
"PMS_JUDGE_TICK_BUDGET_SEC": 150, **values})
out = {"autonomy": "propose_only", "open_autonomy": open_autonomy, "executed": [],
"queued": [], "rejected": [], "skipped": [], "errors": [], "degraded": False}
view = {"positions": [], "held": [], "sector_ready": False, "params": {}, "totals": {}}
psvc._route_one(c, view, {"_mkt": {}}, {}, False, datetime(2026, 9, 3, 10, 0), False, out)
return out, got
PASS_REAL = {"verdict": "PASS", "reason": "理由仍成立", "degraded": False,
"raw": {"verdict": "PASS", "confidence": 72}, "confidence": 72.0}
def _open_cand(verdict="候选", risk=None):
r = _scan([cand("600000.SH", verdict=verdict, reasons=["强传导"], missing=[],
risk=([] if risk is None else risk), card_rank=1)])
assert len(r["candidates"]) == 1, r
return r["candidates"][0]
@case("自动执行开关·四条件齐: propose_only 档位下判决候选直接落指令, 账本写「判决候选自动执行」")
def _():
out, got = _route(_open_cand(), values={"PMS_OPEN_AUTO_EXEC_ON_VERDICT": True},
judge_resp=PASS_REAL)
assert len(out["executed"]) == 1 and not out["queued"], out
assert "判决候选自动执行" in out["executed"][0]["why"], out["executed"]
assert len(got["instructions"]) == 1 and not got["proposals"], got
ins = got["instructions"][0]
assert ins["action"] == "OPEN" and ins["side"] == "buy" and ins["qty"] == 6000, ins
assert ins["progress"]["auto"] is True and ins["progress"]["is_command"] is False
led = got["ledger"][0]
assert led["arbiter"] == "judge" and led["verdict"] == "PASS", led
assert led["reason"].startswith("判决候选自动执行"), led["reason"]
assert led["hard_numbers"]["verdict"] == "候选" and led["ref_id"] == ins["instruction_id"]
@case("自动执行开关·任一条件不满足仍入队 (开关关 / 关注 / 研判不可用 / 没问过决策系统的放行 / 有风险)")
def _():
# ① 开关关着 (默认): 四条其余全满足也入队, 硬数字带研判结论与置信度
out, got = _route(_open_cand(), values={}, judge_resp=PASS_REAL)
assert not out["executed"] and len(out["queued"]) == 1, out
assert out["queued"][0]["why"] == "新建仓档位 propose_only", out["queued"]
hn = got["proposals"][0]["hard_numbers"]
assert hn["judge_verdict"] == "PASS" and hn["judge_conf"] == 72.0, hn
assert hn["verdict"] == "候选" and hn["reasons"] == ["强传导"], hn
on = {"PMS_OPEN_AUTO_EXEC_ON_VERDICT": True}
# ② 判为关注: 强制入队, 原因是关注那句 (开关开着也不许自动)
out, got = _route(_open_cand("关注"), values=on, judge_resp=PASS_REAL)
assert not out["executed"] and out["queued"][0]["why"] == ae.WHY_WATCH_CONFIRM, out
assert got["proposals"][0]["hard_numbers"]["needs_user_confirm"] is True
# ②b 关注在 full 档位下同样入队 (needs_user_confirm 压过档位)
out, _ = _route(_open_cand("关注"), values=on, judge_resp=PASS_REAL, open_autonomy="full")
assert not out["executed"] and out["queued"][0]["why"] == ae.WHY_WATCH_CONFIRM, out
# ③ 研判不可用: 降级入队
out, _ = _route(_open_cand(), values=on, judge_resp={"verdict": "UNAVAILABLE", "reason": "未接通",
"degraded": True, "raw": None,
"confidence": None})
assert not out["executed"] and "研判不可用" in out["queued"][0]["why"], out
# ④ PASS 但没有应答体 (动作不在研判范围那种放行) 不算研判通过
out, _ = _route(_open_cand(), values=on, judge_resp={"verdict": "PASS", "reason": "不在研判范围",
"degraded": False, "raw": None,
"confidence": None})
assert not out["executed"] and len(out["queued"]) == 1, out
# ⑤ 上游标了风险: 入队
out, _ = _route(_open_cand(risk=["昨夜信号陈旧"]), values=on, judge_resp=PASS_REAL)
assert not out["executed"] and len(out["queued"]) == 1, out
# ⑥ 没有判决 (旧版计划): 入队 —— 自动执行只认明确的「候选」
out, _ = _route(_open_cand(None), values=on, judge_resp=PASS_REAL)
assert not out["executed"] and len(out["queued"]) == 1, out
# ⑦ 研判驳回: 既不执行也不入队, 留驳回痕 (分流不改这条老规矩)
out, got = _route(_open_cand(), values=on, judge_resp={"verdict": "REJECT", "reason": "形态走坏",
"degraded": False, "raw": {"verdict": "REJECT"},
"confidence": 80.0})
assert not out["executed"] and not out["queued"] and out["rejected"][0]["by"] == "judge", out
assert got["ledger"][0]["verdict"] == "REJECT" and not got["proposals"]
# ================================================================ 信号来源区分
@case("信号来源·producer_id 缺省 unknown; 留痕文案按来源分写, 硬数字带发送方")
def _():
base = {"ts_code": "600000.SH", "action": "BUY", "confidence": "0.9", "reason": "放量突破"}
s = sr.parse_intraday(dict(base))
assert s["producer_id"] == "unknown", s
d = sr.digest(s, None, {})
assert d["action"] == sr.ACT_NOTE_BUY and d["hard_numbers"]["producer_id"] == "unknown", d
assert d["reason"].startswith("盘中择时程序买入信号（来源 unknown）"), d["reason"]
# 择时决策系统 (大脑广播) 的写法: 以 bionic 开头 → 老文案原句
sb = sr.parse_intraday(dict(base, producer_id="bionic_brain_intraday_v2.0"))
db = sr.digest(sb, {"total_qty": 1000}, {})
assert db["reason"].startswith("决策系统盘中判该股转多"), db["reason"]
assert db["hard_numbers"]["producer_id"] == "bionic_brain_intraday_v2.0"
assert sr.is_bionic_buy_note(db["reason"]) and not sr.is_bionic_buy_note(d["reason"])
# 盘中择时程序的写法
st = sr.parse_intraday(dict(base, producer_id="intraday_timing_v0.1.0"))
dt = sr.digest(st, None, {})
assert dt["reason"].startswith("盘中择时程序买入信号（来源 intraday_timing_v0.1.0）"), dt["reason"]
assert not sr.is_bionic_producer("intraday_timing_v0.1.0") and sr.is_bionic_producer("Bionic_x")
# 两类文案的尾巴一个字没变: 只留痕、买不买归动作引擎、持仓状态
for x in (d, db, dt):
assert "只留痕" in x["reason"] and "买不买由动作引擎" in x["reason"], x["reason"]
assert "无持仓" in d["reason"] and "有持仓" in db["reason"]
# 卖出与 HOLD 的口径不受影响
assert sr.digest(sr.parse_intraday({"ts_code": "600000.SH", "action": "HOLD"}), None, {})["action"] == sr.ACT_RECORD
@case("信号来源·新建仓插队只认择时决策系统的留痕 (盘中择时程序的触发不插队; 读失败按空)")
def _():
from app.repo import pms_repo
from app.services import proposal_service as psvc
rows = {"600000.SH": {"reason": "决策系统盘中判该股转多 (置信度 90%) —— 只留痕", "at": "", "price": 10.0},
"600001.SH": {"reason": "盘中择时程序买入信号（来源 intraday_timing_v0.1.0） (置信度 80%)",
"at": "", "price": 11.0},
"600002.SH": {"reason": None, "at": "", "price": 0.0}}
with _Patch() as p:
p(pms_repo, "buy_signals_today", lambda since: dict(rows))
got = psvc._buy_signals_today()
assert set(got) == {"600000.SH"}, got
assert got["600000.SH"]["price"] == 10.0
def _boom(since):
raise RuntimeError("库挂了")
with _Patch() as p:
p(pms_repo, "buy_signals_today", _boom)
assert psvc._buy_signals_today() == {}
# ================================================================ 密钥名单
@case("密钥名单·会话密钥 PMS_SESSION_SECRET 不进参数中心 (页面读不到、改不了、快照里不出现)")
def _():
assert "PMS_SESSION_SECRET" in pstore.SECRET_KEYS, pstore.SECRET_KEYS
assert {"PMS_QMT_SIGN_SEED_HEX", "PMS_QMT_PEER_PUBKEY_B64"} <= set(pstore.SECRET_KEYS)
assert "PMS_SESSION_SECRET" not in pstore._editable_keys()
assert pstore.get("PMS_SESSION_SECRET") == ""
r = pstore.set_param("PMS_SESSION_SECRET", "deadbeef")
assert r["ok"] is False and "不可修改" in r["error"], r
# 两个新开关是普通业务参数: 可调、有说明、类型是布尔
ek = pstore._editable_keys()
for k in ("PMS_PLAN_ROUTE_BY_VERDICT", "PMS_OPEN_AUTO_EXEC_ON_VERDICT"):
assert k in ek and ek[k].annotation is bool and k in pstore.DESC, k
from config.settings import Settings
assert Settings.model_fields["PMS_PLAN_ROUTE_BY_VERDICT"].default is True
assert Settings.model_fields["PMS_OPEN_AUTO_EXEC_ON_VERDICT"].default is False
# ================================================================ 研判闸
@case("研判键放行·候选卡五键随硬数字送研判, 仓位数字照旧不送; 新建仓带必答「每条理由是否仍成立」")
def _():
hard = {"price": 10.0, "score": 300.0, "verdict": "候选", "reasons": ["强传导", "券商覆盖"],
"missing": [], "risk": [], "card_rank": 2,
"target_amount": 120000.0, "names_before": 3, "room_amt_before": 1_400_000.0}
got = jd._judge_hard_numbers("OPEN", hard)
for k in FIVE:
assert k in got and got[k] == hard[k], (k, got)
for k in ("target_amount", "names_before", "room_amt_before"):
assert k not in got, k
assert set(FIVE) <= set(jd.OPEN_JUDGE_KEYS)
assert jd.must_answer_for("OPEN") == ["上游候选卡的每条理由到今天是否仍成立"]
assert jd.must_answer_for("DCA") == ["下跌是杀逻辑还是杀情绪"]
assert jd.must_answer_for("ADD") == [] and jd.must_answer_for("TRIM") == []
# 请求体真的带上了 (假 requests 抓请求)
import types
sent = []
fake = types.ModuleType("requests")
class _Resp:
status_code = 200
def raise_for_status(self):
pass
def json(self):
return {"verdict": "PASS", "reason": "都成立", "confidence": 66}
fake.post = lambda url, json=None, timeout=None: sent.append({"url": url, "json": json}) or _Resp()
prev = sys.modules.get("requests")
sys.modules["requests"] = fake
try:
with _Patch() as p:
_patch_params(p, jd.param_store, {"PMS_JUDGE_ENABLED": True,
"PMS_JUDGE_API_BASE": "http://bionic:38000",
"PMS_JUDGE_ACTIONS": "FILL,ADD,DCA,SWITCH,OPEN",
"PMS_JUDGE_TIMEOUT": 5,
"PMS_JUDGE_PATH": "/api/intraday/pms_judge"})
r = jd.request({"action": "OPEN", "ts_code": "600000.SH", "qty": 6000,
"reason": "新建仓", "hard_numbers": hard})
finally:
if prev is None:
sys.modules.pop("requests", None)
else:
sys.modules["requests"] = prev
body = sent[0]["json"]
assert body["must_answer"] == ["上游候选卡的每条理由到今天是否仍成立"], body
assert body["hard_numbers"]["verdict"] == "候选" and body["hard_numbers"]["reasons"] == ["强传导", "券商覆盖"]
assert "target_amount" not in body["hard_numbers"]
assert r["verdict"] == jd.PASS and r["confidence"] == 66.0 and r["raw"]["verdict"] == "PASS", r
@case("研判应答·置信度保留 (缺失/非数字为 None), 提议硬数字带 judge_verdict 与 judge_conf")
def _():
r = jd._map_verdict({"verdict": "PASS", "reason": "ok", "confidence": 72})
assert r["verdict"] == jd.PASS and r["confidence"] == 72.0, r
assert jd._map_verdict({"verdict": "REJECT", "confidence": "abc"})["confidence"] is None
assert jd._map_verdict({"verdict": "REJECT", "confidence": "55.5"})["confidence"] == 55.5
u = jd._map_verdict({"verdict": "UNAVAILABLE", "reason": "无昨夜结论"})
assert u["degraded"] is True and u["confidence"] is None and u["reason"] == "无昨夜结论", u
assert jd._map_verdict({"verdict": "???"})["confidence"] is None
# 不在研判范围 / 未接通两条早退路径也带 confidence 键 (下游按键取)
with _Patch() as p:
_patch_params(p, jd.param_store, {"PMS_JUDGE_ENABLED": False, "PMS_JUDGE_ACTIONS": "OPEN"})
assert jd.request({"action": "TRIM"})["confidence"] is None
assert jd.request({"action": "OPEN"})["confidence"] is None
# _make_proposal 把结论与置信度写进硬数字
from app.repo import pms_repo
from app.services import param_store, proposal_service as psvc
got = []
c = _open_cand()
with _Patch() as p:
p(pms_repo, "insert_proposal", lambda **kw: got.append(kw) or 1)
_patch_params(p, param_store, {"PMS_PROPOSAL_TTL_HOURS": 24})
pid = psvc._make_proposal(c, 10.0, {"verdict": "PASS", "reason": "都成立", "degraded": False,
"raw": {}, "confidence": 72.0})
assert pid.startswith("PRP_") and pid.endswith("_600000SH_OPEN"), pid
hn = got[0]["hard_numbers"]
assert hn["judge_verdict"] == "PASS" and hn["judge_conf"] == 72.0, hn
assert hn["price"] == 10.0 and hn["verdict"] == "候选" and hn["plan_rank"] == 1, hn
assert got[0]["judge_verdict"] == "PASS" and got[0]["judge_reason"] == "都成立"
# ================================================================ 页面静态守卫
@case("页面守卫·提议卡按 plan_rank 取候选榜名次并显示判决与理由; 采纳/驳回都弹必填理由并随请求体发")
def _():
path = os.path.join(ROOT, "app", "web", "static", "index.html")
with open(path, encoding="utf-8") as f:
html = f.read()
i = html.index("function propWhy(")
seg = html[i:i + 800]
assert "h.plan_rank" in seg, "提议卡的候选榜名次要读 plan_rank (硬数字里从来没有 rank 这个键)"
assert "function propReasons(" in html and "propReasons," in html, "理由函数要定义并注册到模板"
assert "判为「{{ (p.hard_numbers||{}).verdict }}」" in html, "提议卡要显示选股系统的判决"
j = html.index("async function decide(")
dseg = html[j:html.index("async function loadStrategies(")]
assert "ElMessageBox.prompt" in dseg and "inputValidator" in dseg, "采纳与驳回都要弹必填理由"
assert "{ decision, reason }" in dseg, "理由要随请求体发 (后端 /decide 已读 payload.reason)"
assert "if (!reason)" in dseg, "空理由不许提交"
def main():
import logging
logging.disable(logging.CRITICAL)
passed, failed = 0, 0
for name, fn in RESULTS:
try:
fn()
print(f" PASS {name}")
passed += 1
except Exception:
print(f" FAIL {name}")
traceback.print_exc()
failed += 1
print("-" * 60)
if failed:
print(f"FAILED: {failed} / {passed + failed}")
sys.exit(1)
print(f"ALL PASS ({passed} cases)")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink