diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index 4bf05aa..130c49f 100644 --- a/scripts/test_wiring.py +++ b/scripts/test_wiring.py @@ -298,12 +298,67 @@ class FakeRepo: return len(rows) +class FakeQmtRepo: + """ws 通道三表的内存替身 (pms_qmt_order / pms_qmt_inbox / pms_ws_state)。 + + 只实现 dispatcher 用得到的那几个 —— 出栈、落库、seq 水位是常驻进程 (app/ws/runner.py) + 的事, 由 test_batch6_units.py 的纯逻辑用例覆盖; 这里只管「服务层接线对不对」。 + """ + LIVE = ("QUEUED", "SENDING", "SENT", "ACCEPTED", "SUBMITTED", "PARTIAL") + + def __init__(self): + self.state = {"id": 1, "last_seq": 0, "acked_seq": 0, "server_seq": 0, + "conn_state": "INIT", "heartbeat_at": None, "connected_at": None, + "resync_flag": 0, "last_error": None, "stat": {}} + self.orders = {} + + def set_channel(self, conn_state="ONLINE", alive=True): + """摆一个通道状态出来 —— 「进程活没活」和「连接通没通」是两件事。""" + self.state["conn_state"] = conn_state + self.state["heartbeat_at"] = datetime.now() if alive else None + + # ---- 以下与真 repo 同名同义 ---- + def get_state(self): + return dict(self.state) + + def inbox_pending_count(self): + return 0 + + def queue_depth(self): + d = {} + for o in self.orders.values(): + d[o["status"]] = d.get(o["status"], 0) + 1 + return d + + def enqueue_order(self, *, instruction_id, parent_id, ts_code, side, qty, limit_price, + valid_until, intent="OPEN", note=None): + self.orders[instruction_id] = { + "instruction_id": instruction_id, "parent_id": parent_id, "ts_code": ts_code, + "side": side, "qty": int(qty), "limit_price": float(limit_price), + "valid_until": int(valid_until), "intent": intent, "note": note, + "status": "QUEUED", "cancel_state": "NONE", "cancel_id": None} + return 1 + + def request_cancel(self, *, parent_id, cancel_id): + n = 0 + for o in self.orders.values(): + if (o["parent_id"] == parent_id and o["cancel_state"] == "NONE" + and o["status"] in self.LIVE): + o.update({"cancel_state": "REQUESTED", "cancel_id": cancel_id}) + n += 1 + return n + + def install_fakes(prices=None, positions=None, params=None, high5=None): - """把内存桩装到各模块上, 返回 FakeRepo 实例。""" - from app.repo import downstream_repo, pms_repo + """把内存桩装到各模块上, 返回 FakeRepo 实例 (ws 通道桩挂在 .qmt 上)。""" + from app.repo import downstream_repo, pms_repo, qmt_repo from app.services import industry, market, param_store, portfolio fake = FakeRepo() + fake.qmt = FakeQmtRepo() + for _n in ("get_state", "inbox_pending_count", "queue_depth", "enqueue_order", + "request_cancel"): + setattr(qmt_repo, _n, getattr(fake.qmt, _n)) fake.params.update(params or {}) for p in (positions or []): base = {"ts_code": p["ts_code"], "status": "HOLDING", "frozen_reason": "NONE", @@ -849,8 +904,9 @@ def _(): assert fake.instructions["INS_A2"]["status"] == "EXPIRED" -@case("下发通道·两模式描述与影子回执; ws 未实现时明确拒发; 撤销走本地置状态") +@case("下发通道·影子回执 / ws 出口队列 / 进程与连接两级降级 / 撤销走本地置状态") def _(): + from datetime import datetime as _dtm from app.services import dispatcher, executor, param_store fake = install_fakes() assert dispatcher.mode() == "shadow" @@ -861,12 +917,47 @@ def _(): # 已作废的通道名不能再被设进来 for dead in ("bad_mode", "plan_x", "channel_y"): assert param_store.set_param("PMS_DISPATCH_MODE", dead)["ok"] is False, dead - # ws 通道未实现: 必须 ok=False 而不是静默成功 —— executor 据此不改指令状态 assert param_store.set_param("PMS_DISPATCH_MODE", "ws")["ok"] is True - d2 = dispatcher.dispatch(instruction_id="INS_2", ts_code="600000.SH", side="sell", - qty=1000, limit_price=9.98) - assert d2["ok"] is False and d2["mode"] == "ws" and "尚未实现" in d2["error"], d2 - assert dispatcher.cancel(instruction_id="INS_2")["ok"] is False + vu = _dtm.now().replace(hour=14, minute=45, second=0, microsecond=0) + + def _send(iid, side="sell", qty=1000, px=9.98): + return dispatcher.dispatch(instruction_id=iid, parent_id=iid.rsplit("_D", 1)[0], + ts_code="600000.SH", side=side, qty=qty, + limit_price=px, valid_until=vu, intent="TRIM") + + # ① ws 进程没在跑 (心跳陈旧): 买卖一律拒发 —— 排进队列也没人发, 装作成功更危险 + fake.qmt.set_channel("OFFLINE", alive=False) + for side in ("sell", "buy"): + r = _send(f"INS_2{side}_D01", side=side, qty=1000) + assert r["ok"] is False and r["mode"] == "ws" and "未在线" in r["error"], r + assert not fake.qmt.orders, "进程不在时不该往出口表里塞东西" + + # ② 进程在、连接断: 协议 §6.3 —— 停发增持, 减持照常入队等重连 + fake.qmt.set_channel("OFFLINE", alive=True) + rb = _send("INS_3buy_D01", side="buy", qty=1000) + assert rb["ok"] is False and "§6.3" in rb["error"], rb + rs = _send("INS_3sell_D01", side="sell", qty=1000) + assert rs["ok"] and "重连后" in rs["note"], rs + assert fake.qmt.orders["INS_3sell_D01"]["status"] == "QUEUED" + assert fake.qmt.orders["INS_3sell_D01"]["parent_id"] == "INS_3sell" + + # ③ 连接正常: 买入也放行, 落出口表即返回 (不等 QMT 的 ack) + fake.qmt.set_channel("ONLINE", alive=True) + rb2 = _send("INS_4buy_D01", side="buy", qty=1000, px=9.876) + assert rb2["ok"] and rb2["ref"] == "INS_4buy_D01", rb2 + o = fake.qmt.orders["INS_4buy_D01"] + assert o["limit_price"] == 9.88 and o["valid_until"] > 10 ** 12, o # 2位小数 + 毫秒 + + # ④ 不合协议的参数在本地就拦下, 不换对端一个 BAD_PARAM + bad = _send("INS_5buy_D01", side="buy", qty=150) # 买入非整百 + assert bad["ok"] is False and "不合协议" in bad["error"], bad + assert "INS_5buy_D01" not in fake.qmt.orders + + # ⑤ 撤单: 标记父指令名下所有在途子单, 由 ws 进程逐张发 cancel_order + c = dispatcher.cancel(instruction_id="INS_3sell") + assert c["ok"] and c["cancelled"] == 1 and c["cancel_id"].startswith("CXL-"), c + assert fake.qmt.orders["INS_3sell_D01"]["cancel_state"] == "REQUESTED" + assert dispatcher.cancel(instruction_id="INS_NOBODY")["cancelled"] == 0 param_store.set_param("PMS_DISPATCH_MODE", "shadow") fake.insert_instruction(instruction_id="INS_C", origin_type="plan", origin_id="P1",