476 lines
26 KiB
Python
476 lines
26 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
第二十二批模块单测 (逻辑状态四态接入 PMS, 零外部依赖, 不连库不触网)
|
||
==================================================================
|
||
运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch22_units.py
|
||
|
||
背景 (2026-09-07 下一阶段方案第三件): 选股系统每早给每只票算一个「支撑它的研究证据还在不在」的
|
||
状态 (逻辑强化 / 逻辑成立 / 无法判断 / 逻辑存疑), 随计划每行下发, 另有按代码查询的接口给持仓票用。
|
||
此前 PMS 收到了却没人读; 持仓不绑入场论点; 「逻辑不证伪不退出」没有判据。本批钉住:
|
||
|
||
* 解析层收逻辑状态, 硬数字带它, 但送研判的白名单不收 (一致性检查表第九行: 择时层不判产业逻辑);
|
||
* 判决候选而逻辑存疑的新建仓强制人工确认 (复用强制确认字段, 与关注判决同一条队列);
|
||
* 逻辑存疑的持仓停增持侧自主动作, 减持侧照评; 开关关着一行都不执行; 没有读数不拦;
|
||
* 研究走弱的减持: 默认关, 开了也必定交人 (research_weak 来源一票否决), 数量按可卖量夹紧;
|
||
* 同轮只发一条减持的优先级: 到价清仓 > 研究走弱 > 保垫减仓;
|
||
* 早上取回: 取不到写空映射带原因、绝不折成存疑、不动已有暂停; 映射超龄按没有读数;
|
||
* 策略买入腿按来源暂停与恢复, 不动风控与定性停的;
|
||
* 持仓视图两栏: 有指令链 / 外部成交并入 / 账本无行 三种情形都说得出话, 仓库读失败页面不塌。
|
||
约定同前: 全过输出 "ALL PASS (n cases)" 退出码 0。
|
||
"""
|
||
import os
|
||
import sys
|
||
import traceback
|
||
from datetime import date, datetime, timedelta
|
||
|
||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, os.path.dirname(_HERE))
|
||
sys.path.insert(0, _HERE)
|
||
|
||
from app.core import action_engine as ae # noqa: E402
|
||
from app.services import judge as jd # noqa: E402
|
||
from app.services import plan_feed as pf # noqa: E402
|
||
from app.services import logic_state_service as lss # noqa: E402
|
||
|
||
RESULTS = []
|
||
|
||
|
||
def case(name):
|
||
def deco(fn):
|
||
RESULTS.append((name, fn))
|
||
return fn
|
||
return deco
|
||
|
||
|
||
NOW = datetime(2026, 9, 8, 10, 30)
|
||
CODE = "600000.SH"
|
||
TODAY = date.today().isoformat()
|
||
|
||
|
||
# ================================================================ 夹具
|
||
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, mod, values):
|
||
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)
|
||
|
||
|
||
def doubt(as_of="2026-09-04"):
|
||
return {"state": "逻辑存疑", "raw_state": "逻辑存疑", "why": None, "settle_note": "进入逻辑存疑即刻成立",
|
||
"as_of": as_of, "usable": ["券商行动"], "missing": ["研报论断", "产业研判", "公司事件"],
|
||
"reasons": ["券商行动的转弱达到进入条件：每股收益预测下修且覆盖机构收缩",
|
||
"券商行动（2026-09-04）：每股收益预测中位数下修 40%，同时覆盖机构从 5 家收缩到 3 家"]}
|
||
|
||
|
||
def hold():
|
||
return {"state": "逻辑成立", "raw_state": "逻辑成立", "why": None, "settle_note": "维持",
|
||
"as_of": "2026-09-04", "usable": ["研报论断"], "missing": ["产业研判", "券商行动", "公司事件"],
|
||
"reasons": ["研报论断（2026-08-25）：最近一条利好"]}
|
||
|
||
|
||
def pos(code=CODE, **kw):
|
||
p = {"ts_code": code, "total_qty": 3000, "avail_qty": 3000, "price": 10.0, "price_ok": True,
|
||
"frozen_reason": "NONE", "cushion_pct": 0.02, "cushion_peak": 0.03, "avg_cost": 9.8}
|
||
p.update(kw)
|
||
return p
|
||
|
||
|
||
SCAN_PARAMS = {"trim_peak": 0.06, "trim_giveback": 0.5, "logic_state_route": True,
|
||
"logic_doubt_trim": False, "logic_doubt_trim_ratio": 1.0 / 3, "scale": 1_000_000}
|
||
|
||
PARAMS_BASE = {"PMS_EXEC_WINDOW_TDAYS": 3, "PMS_PROPOSAL_TTL_HOURS": 24,
|
||
"PMS_JUDGE_TICK_BUDGET_SEC": 150}
|
||
JUDGE_PASS = {"verdict": "PASS", "reason": "理由仍成立", "degraded": False,
|
||
"raw": {"verdict": "PASS"}, "confidence": 70.0}
|
||
|
||
|
||
def _route(c, *, autonomy="propose_only", judge_resp=JUDGE_PASS, dry_run=False, values=None,
|
||
price=10.0):
|
||
"""把一条候选送进 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": [], "judge_calls": []}
|
||
with _Patch() as p:
|
||
p(rule_gate, "check", lambda **kw: {"passed": True, "failed": [], "warnings": []})
|
||
p(judge, "request", lambda cand_, context=None, **kw: (
|
||
got["judge_calls"].append(cand_.get("action")) or 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: [])
|
||
p(pms_repo, "update_position", lambda code, **kw: 1)
|
||
_patch_params(p, param_store, {**PARAMS_BASE, **(values or {})})
|
||
out = {"autonomy": autonomy, "open_autonomy": autonomy, "executed": [], "queued": [],
|
||
"rejected": [], "skipped": [], "errors": [], "degraded": False}
|
||
held = [{"ts_code": c["ts_code"], "total_qty": 3000, "avail_qty": 3000,
|
||
"price": price, "frozen_reason": "NONE"}]
|
||
view = {"positions": held, "held": held, "sector_ready": False,
|
||
"params": {}, "totals": {}}
|
||
psvc._route_one(c, view, {"_mkt": {}}, {}, False, NOW, dry_run, out)
|
||
return out, got
|
||
|
||
|
||
# ================================================================ 一, 解析层与硬数字
|
||
@case("解析层·逻辑状态按约定的键归一, 出处最多三条; 缺键 / 非字典 / 没有 state 一律 None")
|
||
def _():
|
||
raw = {"state": "逻辑存疑", "raw_state": "逻辑存疑", "why": None, "settle_note": "进入逻辑存疑即刻成立",
|
||
"as_of": "2026-09-04", "usable": ["券商行动"], "missing": ["研报论断", "公司事件"],
|
||
"reasons": ["a", "b", "c", "d"], "paths": [{"path": "x"}], "prev_state": "逻辑成立"}
|
||
st = pf._logic_state_or_none(raw)
|
||
assert st["state"] == "逻辑存疑" and st["as_of"] == "2026-09-04", st
|
||
assert st["reasons"] == ["a", "b", "c"] and st["usable"] == ["券商行动"], st
|
||
assert "paths" not in st and "prev_state" not in st, st # 只收约定的键
|
||
assert pf._logic_state_or_none(None) is None and pf._logic_state_or_none("逻辑存疑") is None
|
||
assert pf._logic_state_or_none({"why": "x"}) is None
|
||
# 没有原始态的旧形状: 原始态退回落定态
|
||
assert pf._logic_state_or_none({"state": "逻辑成立"})["raw_state"] == "逻辑成立"
|
||
row = pf._rows([{"code": "600000.SH", "rank": 1, "score": 210, "logic_state": raw}], "main")[0]
|
||
assert row["logic_state"]["state"] == "逻辑存疑", row
|
||
row2 = pf._rows([{"code": "600000.SH", "rank": 1, "score": 210}], "main")[0]
|
||
assert row2["logic_state"] is None, row2
|
||
|
||
|
||
@case("解析层·按代码查询接口的应答映射回 PMS 的代码形态, 带 error 的行跳过, source 与日期带上")
|
||
def _():
|
||
payload = {"date": "2026-09-04", "count": 3, "states": [
|
||
{"input": "600000.SH", "code": "SH600000", "source": "daily", **doubt()},
|
||
{"input": "300750", "code": "SZ300750", "source": "computed", **hold()},
|
||
{"input": "abc", "error": "认不出的代码形态"}]}
|
||
m = pf.parse_logic_states(payload)
|
||
assert set(m) == {"600000.SH", "300750.SZ"}, m
|
||
assert m["600000.SH"]["state"] == "逻辑存疑" and m["600000.SH"]["source"] == "daily"
|
||
assert m["300750.SZ"]["source"] == "computed" and m["300750.SZ"]["date"] == "2026-09-04"
|
||
assert pf.parse_logic_states(None) == {} and pf.parse_logic_states({"states": "x"}) == {}
|
||
assert pf.fetch_logic_states([]) == {} # 没票不触网
|
||
|
||
|
||
@case("候选筛选·items 带 logic_state (缺就是 None), 不改资格也不改次序")
|
||
def _():
|
||
plan = {"date": "2026-09-04", "main": pf._rows([
|
||
{"code": "600000.SH", "rank": 1, "score": 220, "verdict": "候选", "logic_state": doubt()},
|
||
{"code": "300750.SZ", "rank": 2, "score": 210, "verdict": "候选"}], "main"), "observe": []}
|
||
sel = pf.select_candidates(plan, top_n=10, route_by_verdict=True)
|
||
items = sel["items"]
|
||
assert [x["ts_code"] for x in items] == ["600000.SH", "300750.SZ"], items
|
||
assert items[0]["logic_state"]["state"] == "逻辑存疑" and items[1]["logic_state"] is None
|
||
|
||
|
||
@case("硬数字带逻辑状态, 而送研判的白名单不收它 (择时层不判产业逻辑, 检查表第九行)")
|
||
def _():
|
||
c = {"ts_code": CODE, "price": 10.0, "score": 220, "rank": 1, "bucket": "main", "src": "plan_api",
|
||
"verdict": "候选", "basis": "三门槛全过", "logic": ["研报说好"], "logic_state": doubt()}
|
||
params = {"scale": 1_000_000, "stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25)}
|
||
with _Patch() as p:
|
||
p(ae, "check_all_caps", lambda **kw: [])
|
||
p(ae, "_new_name_ctx", lambda caps, c: caps)
|
||
cand, why = ae.eval_open(c, params, {"names_count": 0, "max_names": 10}, 500_000)
|
||
assert cand and not why, why
|
||
hn = cand["hard_numbers"]
|
||
assert hn["logic_state"]["state"] == "逻辑存疑" and hn["basis"] == "三门槛全过", hn
|
||
sent = jd._judge_hard_numbers("OPEN", hn)
|
||
assert "logic_state" not in sent and "basis" not in sent and "verdict" in sent, sent
|
||
assert "logic_state" not in jd.OPEN_JUDGE_KEYS
|
||
|
||
|
||
@case("安全边际整句·解析层收 valuation_text, 候选与硬数字带它, 研判白名单不收 (第四件, 只给人看)")
|
||
def _():
|
||
txt = "安全边际（2026 年度预测，9 家）：悲观 42.64 元（-15.0%）、中性 56.75 元（+13.1%）；赔率 0.87 比 1"
|
||
row = pf._rows([{"code": "600000.SH", "rank": 1, "score": 210, "valuation_text": txt}], "main")[0]
|
||
assert row["valuation_text"] == txt, row
|
||
assert pf._rows([{"code": "600000.SH", "rank": 1, "score": 210, "valuation_text": ""}], "main")[0]["valuation_text"] is None
|
||
plan = {"date": "2026-09-04", "main": [row], "observe": []}
|
||
item = pf.select_candidates(plan, top_n=10)["items"][0]
|
||
assert item["valuation_text"] == txt, item
|
||
c = {**item, "price": 10.0}
|
||
params = {"scale": 1_000_000, "stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25)}
|
||
with _Patch() as p:
|
||
p(ae, "check_all_caps", lambda **kw: [])
|
||
p(ae, "_new_name_ctx", lambda caps, c: caps)
|
||
cand, why = ae.eval_open(c, params, {"names_count": 0, "max_names": 10}, 500_000)
|
||
assert cand and cand["hard_numbers"]["valuation_text"] == txt, why
|
||
assert "valuation_text" not in jd._judge_hard_numbers("OPEN", cand["hard_numbers"])
|
||
assert "valuation_text" not in jd.OPEN_JUDGE_KEYS
|
||
|
||
|
||
# ================================================================ 二, 新建仓与持仓的分流
|
||
def _scan_open(cands, *, route_logic=True):
|
||
params = {"scale": 1_000_000, "stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25),
|
||
"open_route_by_verdict": True, "open_route_by_logic": route_logic,
|
||
"open_signal_priority": False}
|
||
with _Patch() as p:
|
||
p(ae, "check_all_caps", lambda **kw: [])
|
||
p(ae, "_new_name_ctx", lambda caps, c: caps)
|
||
p(ae, "_ctx_after", lambda ctx, *a, **kw: ctx)
|
||
return ae.scan_open(candidates=cands, params=params,
|
||
caps={"names_count": 0, "max_names": 10}, room_amt=900_000, slots=5)
|
||
|
||
|
||
@case("新建仓·判决候选而逻辑存疑 → 强制人工确认并写明出处; 关注保留自己的原因; 成立不强制; 开关关不强制")
|
||
def _():
|
||
base = {"price": 10.0, "score": 220, "rank": 1, "bucket": "main", "src": "plan_api"}
|
||
cands = [{**base, "ts_code": "600000.SH", "verdict": "候选", "logic_state": doubt()},
|
||
{**base, "ts_code": "600001.SH", "verdict": "关注", "logic_state": doubt()},
|
||
{**base, "ts_code": "600002.SH", "verdict": "候选", "logic_state": hold()},
|
||
{**base, "ts_code": "600003.SH", "verdict": "候选"}]
|
||
r = _scan_open(cands)
|
||
by = {c["ts_code"]: c for c in r["candidates"]}
|
||
assert set(by) == {"600000.SH", "600001.SH", "600002.SH", "600003.SH"}, r["skipped"]
|
||
a = by["600000.SH"]
|
||
assert a["needs_user_confirm"] and "逻辑存疑" in a["confirm_why"] and "证据截止 2026-09-04" in a["confirm_why"], a
|
||
assert "每股收益预测下修" in a["confirm_why"], a["confirm_why"]
|
||
b = by["600001.SH"]
|
||
assert b["needs_user_confirm"] and b["confirm_why"] == ae.WHY_WATCH_CONFIRM, b
|
||
assert not by["600002.SH"].get("needs_user_confirm") and not by["600003.SH"].get("needs_user_confirm")
|
||
r2 = _scan_open(cands, route_logic=False)
|
||
by2 = {c["ts_code"]: c for c in r2["candidates"]}
|
||
assert not by2["600000.SH"].get("needs_user_confirm"), by2["600000.SH"]
|
||
assert ae.logic_confirm_why(hold()) is None and ae.logic_confirm_why(None) is None
|
||
|
||
|
||
@case("持仓·逻辑存疑停增持侧 (补足/加仓/补仓写明原因), 减持侧照评; 开关关不停; 没有读数不停")
|
||
def _():
|
||
p = pos(logic_state=doubt(), cushion_peak=0.10, cushion_pct=0.04) # 保垫回吐过半 → TRIM 照评
|
||
r = ae.scan(positions=[p], params=SCAN_PARAMS, market={CODE: {}}, skip={})
|
||
acts = {c["action"] for c in r["candidates"]}
|
||
assert acts == {"TRIM"}, r["candidates"]
|
||
stopped = {s["action"]: s["why"] for s in r["skipped"] if "逻辑存疑" in s["why"]}
|
||
assert set(stopped) == {"ADD", "FILL", "DCA"}, r["skipped"]
|
||
assert all("停掉增持侧" in w and "证据截止 2026-09-04" in w for w in stopped.values()), stopped
|
||
r2 = ae.scan(positions=[pos(logic_state=doubt())], params={**SCAN_PARAMS, "logic_state_route": False},
|
||
market={CODE: {}}, skip={})
|
||
assert not [s for s in r2["skipped"] if "逻辑存疑" in s["why"]], r2["skipped"]
|
||
r3 = ae.scan(positions=[pos()], params=SCAN_PARAMS, market={CODE: {}}, skip={})
|
||
assert not [s for s in r3["skipped"] if "逻辑存疑" in s["why"]], r3["skipped"]
|
||
|
||
|
||
@case("研究走弱的减持·默认关不产出; 开了按三分之一取整并按可卖量夹紧, 来源 research_weak 且强制确认; 不足一手不产出")
|
||
def _():
|
||
p = pos(logic_state=doubt())
|
||
assert ae.eval_weak_research(p, SCAN_PARAMS) is None # 开关关
|
||
on = {**SCAN_PARAMS, "logic_doubt_trim": True}
|
||
c = ae.eval_weak_research(p, on)
|
||
assert c and c["action"] == "TRIM" and c["side"] == "sell" and c["qty"] == 1000, c
|
||
assert c["source"] == ae.SRC_RESEARCH_WEAK and c["needs_user_confirm"] and c["confirm_why"], c
|
||
assert "拟减 1000 股" in c["reason"] and "证据截止 2026-09-04" in c["reason"], c["reason"]
|
||
assert c["hard_numbers"]["logic_state"]["state"] == "逻辑存疑" and c["hard_numbers"]["trim_ratio"] == 0.3333
|
||
c2 = ae.eval_weak_research(pos(logic_state=doubt(), avail_qty=400), on)
|
||
assert c2["qty"] == 400, c2 # 夹到可卖量
|
||
assert ae.eval_weak_research(pos(logic_state=doubt(), total_qty=200, avail_qty=200), on) is None
|
||
assert ae.eval_weak_research(pos(logic_state=hold()), on) is None
|
||
assert ae.eval_weak_research(pos(), on) is None # 没有读数
|
||
c3 = ae.eval_weak_research(p, {**on, "logic_doubt_trim_ratio": 0.5})
|
||
assert c3["qty"] == 1500, c3
|
||
|
||
|
||
@case("同轮只发一条减持·优先级 到价清仓 > 研究走弱 > 保垫减仓, 让路的记进跳过原因")
|
||
def _():
|
||
on = {**SCAN_PARAMS, "logic_doubt_trim": True}
|
||
p = pos(logic_state=doubt(), cushion_peak=0.10, cushion_pct=0.04)
|
||
r = ae.scan(positions=[p], params=on, market={CODE: {}}, skip={},
|
||
stock_params={CODE: {"target_price": 9.5}})
|
||
assert [c["action"] for c in r["candidates"]] == ["EXIT"], r["candidates"]
|
||
yielded = [s for s in r["skipped"] if "一轮只发一条减持" in s["why"]]
|
||
assert len(yielded) == 2 and all(s["action"] == "TRIM" for s in yielded), yielded
|
||
r2 = ae.scan(positions=[p], params=on, market={CODE: {}}, skip={})
|
||
assert len(r2["candidates"]) == 1 and r2["candidates"][0]["source"] == ae.SRC_RESEARCH_WEAK, r2
|
||
assert ae._sell_priority({"action": "EXIT"}) < ae._sell_priority({"action": "TRIM", "source": "research_weak"}) \
|
||
< ae._sell_priority({"action": "TRIM", "source": "engine"})
|
||
|
||
|
||
@case("分流·研究走弱的减持在 full 档也入人工队列, 不落指令, 提议带来源与逻辑状态")
|
||
def _():
|
||
c = ae.eval_weak_research(pos(logic_state=doubt()), {**SCAN_PARAMS, "logic_doubt_trim": True})
|
||
for autonomy in ("propose_only", "full"):
|
||
out, got = _route(c, autonomy=autonomy)
|
||
assert not out["executed"] and len(out["queued"]) == 1, (autonomy, out)
|
||
assert "研究证据走弱" in out["queued"][0]["why"], out["queued"]
|
||
assert not got["instructions"] and len(got["proposals"]) == 1, (autonomy, got)
|
||
hn = got["proposals"][0]["hard_numbers"]
|
||
assert hn["source"] == ae.SRC_RESEARCH_WEAK and hn["logic_state"]["state"] == "逻辑存疑", hn
|
||
|
||
|
||
# ================================================================ 三, 早上取回与策略买入腿
|
||
@case("映射·超过三个自然日没刷新按没有读数; 时刻缺失或坏了也按没有读数; attach 只给有读数的行挂键")
|
||
def _():
|
||
fresh = {"at": f"{TODAY} 08:41:00", "states": {CODE: doubt()}}
|
||
assert lss.state_map(fresh) == {CODE: doubt()}
|
||
old = {"at": (date.today() - timedelta(days=5)).isoformat() + " 08:41:00", "states": {CODE: doubt()}}
|
||
assert lss.state_map(old) == {}
|
||
assert lss.state_map({"states": {CODE: doubt()}}) == {} and lss.state_map({}) == {}
|
||
rows = [pos(), pos("300750.SZ", logic_state="陈旧的")]
|
||
lss.attach(rows, {CODE: doubt()})
|
||
assert rows[0]["logic_state"]["state"] == "逻辑存疑" and "logic_state" not in rows[1], rows
|
||
|
||
|
||
@case("策略买入腿·存疑按来源 logic 暂停; 明确不存疑只清本来源; 没读数不动; 开关关整段跳过")
|
||
def _():
|
||
from app.services import param_store, strategy_service
|
||
calls = {"pause": [], "clear": []}
|
||
with _Patch() as p:
|
||
p(strategy_service, "pause_buy", lambda code, *, reason="", source="signal": (
|
||
calls["pause"].append((code, source, reason)) or ["S1"]))
|
||
p(strategy_service, "clear_buypause", lambda code, only_source=None: (
|
||
calls["clear"].append((code, only_source)) or {"ok": True, "cleared": code == "600001.SH"}))
|
||
_patch_params(p, param_store, {"PMS_LOGIC_STATE_ROUTE": True})
|
||
r = lss.apply_pauses({CODE: doubt(), "600001.SH": hold(), "600002.SH": hold()},
|
||
[CODE, "600001.SH", "600002.SH", "600003.SH"])
|
||
assert r["paused"] == [CODE] and r["resumed"] == ["600001.SH"] and not r["errors"], r
|
||
assert calls["pause"] == [(CODE, "logic", calls["pause"][0][2])] and "停掉增持侧" in calls["pause"][0][2]
|
||
assert [c for c, _ in calls["clear"]] == ["600001.SH", "600002.SH"], calls # 没读数的 600003 不动
|
||
assert all(s == "logic" for _, s in calls["clear"]), calls
|
||
with _Patch() as p:
|
||
p(strategy_service, "pause_buy", lambda *a, **k: (_ for _ in ()).throw(AssertionError("不该调")))
|
||
p(strategy_service, "clear_buypause", lambda *a, **k: (_ for _ in ()).throw(AssertionError("不该调")))
|
||
_patch_params(p, param_store, {"PMS_LOGIC_STATE_ROUTE": False})
|
||
r2 = lss.apply_pauses({CODE: doubt()}, [CODE])
|
||
assert r2["paused"] == [] and r2.get("skipped"), r2
|
||
|
||
|
||
@case("早上取回·查不到写空映射带原因且不动暂停 (绝不折成存疑); 查到写映射并按结果暂停")
|
||
def _():
|
||
from app.repo import pms_repo
|
||
from app.services import param_store, strategy_service
|
||
saved, calls = {}, {"pause": [], "clear": []}
|
||
with _Patch() as p:
|
||
p(pms_repo, "list_positions", lambda *, only_open=False: [{"ts_code": CODE}, {"ts_code": "600001.SH"}])
|
||
p(param_store, "set_param", lambda k, v, by="user": saved.__setitem__(k, v) or {"ok": True})
|
||
_patch_params(p, param_store, {"PMS_LOGIC_STATE_ROUTE": True})
|
||
p(strategy_service, "pause_buy", lambda code, *, reason="", source="signal": calls["pause"].append(code) or [])
|
||
p(strategy_service, "clear_buypause", lambda code, only_source=None: calls["clear"].append(code) or {"ok": True, "cleared": False})
|
||
|
||
def _boom(codes):
|
||
raise pf.PlanFeedError("选股系统连不上")
|
||
r = lss.pull_for_held(now=NOW, fetch=_boom)
|
||
import json
|
||
m = json.loads(saved[lss.MAP_KEY])
|
||
assert r["ok"] is False and m["states"] == {} and "连不上" in m["error"], (r, m)
|
||
assert not calls["pause"] and not calls["clear"], calls
|
||
|
||
r2 = lss.pull_for_held(now=NOW, fetch=lambda codes: {CODE: {**doubt(), "date": "2026-09-04"}})
|
||
m2 = json.loads(saved[lss.MAP_KEY])
|
||
assert r2["ok"] and r2["got"] == 1 and r2["by_state"] == {"逻辑存疑": 1}, r2
|
||
assert m2["date"] == "2026-09-04" and m2["states"][CODE]["state"] == "逻辑存疑", m2
|
||
assert r2["paused"] == [CODE] and r2["missing"] == ["600001.SH"], r2 # 没查到的票记明, 不动
|
||
|
||
|
||
# ================================================================ 四, 持仓视图两栏
|
||
def _fake_repo():
|
||
from test_wiring import FakeRepo
|
||
f = FakeRepo()
|
||
f.insert_instruction(instruction_id="I1", origin_type="proposal", origin_id="P1", ts_code=CODE,
|
||
action="OPEN", side="buy", qty=1000)
|
||
f.insert_ledger(ts_code=CODE, action="OPEN", arbiter="user", verdict="PASS", price_at=10.0,
|
||
hard_numbers={"basis": "三门槛全过、无硬风险", "logic": ["研报说好 —— 出处 A", "B", "C", "D"],
|
||
"verdict": "候选", "logic_state": hold()},
|
||
ref_id="P1", reason="人工采纳: 看好")
|
||
f.insert_lot(ts_code=CODE, lot_type="BASE", qty=1000, open_price=10.0, open_date="2026-09-01",
|
||
instruction_id="I1")
|
||
f.insert_lot(ts_code="600001.SH", lot_type="BASE", qty=1000, open_price=10.0, open_date="2026-09-01")
|
||
f.insert_instruction(instruction_id="I2", origin_type="command", origin_id="C1", ts_code="600002.SH",
|
||
action="OPEN", side="buy", qty=1000)
|
||
f.insert_lot(ts_code="600002.SH", lot_type="BASE", qty=1000, open_price=10.0, open_date="2026-09-02",
|
||
instruction_id="I2")
|
||
return f
|
||
|
||
|
||
@case("入场论点·有指令链 (批次→指令→提议号→账本) / 外部成交并入 / 账本无行 三种情形都说得出话")
|
||
def _():
|
||
from app.repo import pms_repo
|
||
f = _fake_repo()
|
||
with _Patch() as p:
|
||
p(pms_repo, "list_lots", f.list_lots)
|
||
p(pms_repo, "get_instruction", f.get_instruction)
|
||
p(pms_repo, "ledger_by_ref", f.ledger_by_ref)
|
||
a = lss.entry_view(CODE)
|
||
b = lss.entry_view("600001.SH")
|
||
c = lss.entry_view("600002.SH")
|
||
d = lss.entry_view("600009.SH")
|
||
assert a["basis"] == "三门槛全过、无硬风险" and a["logic"] == ["研报说好 —— 出处 A", "B", "C"], a
|
||
assert a["verdict"] == "候选" and a["logic_state_at_entry"] == "逻辑成立" and a["open_date"] == "2026-09-01", a
|
||
assert a["instruction_id"] == "I1" and a["arbiter"] == "user", a
|
||
assert "外部成交并入" in b["why"] and b["open_date"] == "2026-09-01", b
|
||
assert "账本里没有" in c["why"] and c["instruction_id"] == "I2", c
|
||
assert "没有未平的批次" in d["why"], d
|
||
assert f.ledger_by_ref(["P1", "X"]) and not f.ledger_by_ref([]), "假仓库的按引用取账本"
|
||
|
||
|
||
@case("持仓接口两栏·仓库读失败也不抛, 每行都有 entry 与 logic_now; 没有读数写明; 空仓行 entry 为空")
|
||
def _():
|
||
from app.repo import pms_repo
|
||
rows = [pos(), pos("600001.SH"), pos("600002.SH", total_qty=0)]
|
||
with _Patch() as p:
|
||
p(pms_repo, "list_lots", lambda *a, **k: (_ for _ in ()).throw(OSError("db down")))
|
||
lss.decorate_positions(rows, {CODE: doubt()})
|
||
assert rows[0]["logic_now"]["state"] == "逻辑存疑" and "每股收益预测下修" in rows[0]["logic_now"]["text"]
|
||
assert rows[0]["logic_now"]["as_of"] == "2026-09-04" and "批次读取失败" in rows[0]["entry"]["why"], rows[0]
|
||
assert rows[1]["logic_now"]["state"] is None and "没有读数" in rows[1]["logic_now"]["text"], rows[1]
|
||
assert rows[2]["entry"] is None, rows[2]
|
||
nv = lss.now_view(hold())
|
||
assert nv["state"] == "逻辑成立" and nv["settle_note"] == "维持" and "研报论断" in nv["text"], nv
|
||
|
||
|
||
@case("假仓库·ledger_by_ref 与真 repo 同名同签名 (第十批 [M1] 也会扫, 这里先钉一次)")
|
||
def _():
|
||
import ast
|
||
import inspect
|
||
from test_wiring import FakeRepo
|
||
path = os.path.join(os.path.dirname(_HERE), "app", "repo", "pms_repo.py")
|
||
with open(path, encoding="utf-8") as fh:
|
||
names = {n.name: [a.arg for a in n.args.args] for n in ast.parse(fh.read()).body
|
||
if isinstance(n, ast.FunctionDef)}
|
||
assert names["ledger_by_ref"] == ["ref_ids", "limit"], names["ledger_by_ref"]
|
||
fp = list(inspect.signature(FakeRepo.ledger_by_ref).parameters)
|
||
assert fp == ["self", "ref_ids", "limit"], fp
|
||
|
||
|
||
# ================================================================ 入口
|
||
def main():
|
||
ok = 0
|
||
for name, fn in RESULTS:
|
||
try:
|
||
fn()
|
||
ok += 1
|
||
print(f" ok {name}")
|
||
except Exception:
|
||
print(f" FAIL {name}")
|
||
traceback.print_exc()
|
||
print("-" * 60)
|
||
if ok == len(RESULTS):
|
||
print(f"ALL PASS ({ok} cases)")
|
||
return 0
|
||
print(f"FAILED {len(RESULTS) - ok}/{len(RESULTS)}")
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|