From fad1bb1fb5de42406503bc9442afb5704f68ec8b Mon Sep 17 00:00:00 2001 From: baobao Date: Mon, 19 Oct 2026 14:16:20 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BA=A4=E6=98=93=E6=89=A7=E8=A1=8C=EF=BC=9A?= =?UTF-8?q?=E6=B8=85=E4=BB=93=E5=81=9C=E6=AD=A2=E6=97=A7=E4=B9=B0=E5=85=A5?= =?UTF-8?q?=EF=BC=8C=E8=A1=A5=E9=BD=90=E8=B7=A8=E8=BD=AE=E4=BA=92=E6=96=A5?= =?UTF-8?q?=E4=B8=8E=E7=9C=9F=E5=AE=9E=E8=BF=9B=E5=BA=A6=EF=BC=88=E5=8F=B0?= =?UTF-8?q?=E8=B4=A6=20076=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一、科士达旧建仓600股已成400股，风控清仓400股已成200股，旧买入仍活跃。完整清仓统一停止买入指令、提议及未解锁批次，停止依据追加账本，旧进度不能复活终态。普通减仓仅暂停买入，做T配对不误作清仓。 二、出口先撤后发，认领排除待撤，发送前复核来源与方向。页面按子委托事实显示等待，补回上游原始卖出理由。22项专项及45批隔离禁网回归通过。 三、不修改成交、成本、盈亏、交易时段、仓位和策略阈值。仅拟部署155模拟实例，188不动；部署与自然调度判收另记。 --- DEVLOG.md | 13 + app/core/signal_rules.py | 2 +- app/repo/pms_repo.py | 11 +- app/repo/qmt_repo.py | 6 +- app/services/dispatcher.py | 11 + app/services/execution_conflicts.py | 108 ++++++++ app/services/executor.py | 27 +- app/services/proposal_service.py | 2 + app/services/strategy_service.py | 9 +- app/web/main.py | 15 +- app/web/static/index.html | 10 +- app/ws/runner.py | 18 +- docs/开发规格_买卖方向冲突修复_2026-10-19.md | 48 ++++ scripts/run_tests.py | 3 +- scripts/test_batch41_units.py | 256 +++++++++++++++++++ scripts/test_wiring.py | 6 + 16 files changed, 528 insertions(+), 17 deletions(-) create mode 100644 app/services/execution_conflicts.py create mode 100644 docs/开发规格_买卖方向冲突修复_2026-10-19.md create mode 100644 scripts/test_batch41_units.py diff --git a/DEVLOG.md b/DEVLOG.md index 843bf9a..0b6cfc7 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -2016,3 +2016,16 @@ bafecb7 经 45 策略用例、44 批本地及候选镜像禁网回归通过，11 四个新增反例在旧实现失败，修复后30专项及44批本地隔离禁网回归通过，Node交互替身通过。部署与真机状态待本节后续补记。只部署155模拟仓，188不动。未改交易时段、准入参数、数量算法或研判结论。 - 14:45部署补记：60de23e在14:39:51于155统一重建，镜像befb6246、指纹3777e125a88a。候选镜像44批禁网回归通过。14:40正常扫描撤下已保存明确拒绝，账本6858唯一、零目标指令，14:45复查无重复。浏览器待确认0条、无错误；未代点采纳。15参数及账务保持，对账正常。24条历史指令仅一条正常WAIT时刻前进一分钟，数量和修复审计不变。188未操作。完整备份与回退见提议采纳失败排查14:45节及四仓交接。 + + +## 2026-10-19 买卖方向冲突修复（台账076） + +**做了什么。** 科士达旧建仓与风控清仓同时保持活跃。统一在执行前停止清仓票的旧买入，普通减仓期间暂停买入。停止事实留在追加账本，旧进度回写不能复活指令。出口先撤后发且发送前核对。原始audit_reason补回，页面按实际委托状态显示。 + +**动了哪些文件。** execution_conflicts服务、executor、dispatcher、proposal_service、strategy_service、signal_rules、两仓储、ws runner、web与页面。增加batch41并登记测试与替身。 + +**部署方式。** 本地22项专项与45批禁网回归通过。155需候选镜像断网验证后统一重建web、worker、beat、ws。当前此节尚未部署；后续追加真实读数。 + +**真机判收。** 修复前14:10仅科士达存在买卖重叠。已保存指令、成交、账务与参数原值，详见规格。未改成交，不恢复七条历史暂停指令，不操作188。 + +**还欠什么。** 候选镜像测试、部署、生效与页面核验，按真实证据追加。 diff --git a/app/core/signal_rules.py b/app/core/signal_rules.py index 1896b2a..dc19864 100644 --- a/app/core/signal_rules.py +++ b/app/core/signal_rules.py @@ -113,7 +113,7 @@ def parse_intraday(fields: dict, *, msg_id: str = None) -> dict: "confidence": _norm_conf(f.get("confidence")), "minute_qrs": _num(scores.get("minute_qrs")), "suggested_price": _num(f.get("suggested_price")) or None, - "reason": f.get("reason") or f.get("llm_reason") or "", + "reason": f.get("reason") or f.get("llm_reason") or f.get("audit_reason") or scores.get("reason") or "", "dominant_signal": f.get("dominant_signal") or "", # 发送方标识 (2026-09-03 起保留): 缺就是 unknown, 不猜是谁发的 "producer_id": str(f.get("producer_id") or "").strip() or PRODUCER_UNKNOWN} diff --git a/app/repo/pms_repo.py b/app/repo/pms_repo.py index 7533e18..ebf07be 100644 --- a/app/repo/pms_repo.py +++ b/app/repo/pms_repo.py @@ -439,6 +439,13 @@ def list_instructions(*, statuses=None, side=None, ts_code=None, limit: int = 30 return rows +def instruction_buy_stop(instruction_id: str): + # 停止依据留在只追加的动作账本，避免并发的进度更新覆盖 JSON 标记。 + return fetch_one("SELECT id, reason, ref_id FROM pms_action_ledger " + "WHERE ref_id = :iid AND action = 'BUY_STOP' AND verdict = 'PASS' " + "ORDER BY id DESC LIMIT 1", {'iid': instruction_id}) + + def get_instruction(instruction_id: str): r = fetch_one("SELECT * FROM pms_instruction WHERE instruction_id = :iid", {"iid": instruction_id}) @@ -465,7 +472,9 @@ def update_instruction(instruction_id: str, *, status=None, exec_qty=None, exec_ if progress is not None: sets.append("progress_json = :pg") p["pg"] = _dumps(progress) - return execute(f"UPDATE pms_instruction SET {', '.join(sets)} WHERE instruction_id = :iid", p) + guard = " AND status IN ('PROPOSED','RULE_PASSED','JUDGE_PASSED','DISPATCHED')" if status == 'DISPATCHED' else '' + # 下发后的旧进度回写不能把已撤销指令复活。真实迟到成交仍由成交入口记账。 + return execute(f"UPDATE pms_instruction SET {', '.join(sets)} WHERE instruction_id = :iid" + guard, p) def add_instruction_exec(instruction_id: str, qty: int) -> int: diff --git a/app/repo/qmt_repo.py b/app/repo/qmt_repo.py index 3ee71a6..bd6be52 100644 --- a/app/repo/qmt_repo.py +++ b/app/repo/qmt_repo.py @@ -389,6 +389,10 @@ def list_orders(*, statuses=None, parent_id=None, ts_code=None, limit: int = 200 return fetch_all(sql, p) +def orders_for_parent(parent_id: str) -> list: + return fetch_all('SELECT * FROM pms_qmt_order WHERE parent_id = :pid ORDER BY id', {'pid': parent_id}) + + def next_queued(limit: int = 20, side=None) -> list: """取待发委托。按 id 升序 = 先进先发, 保证同一只票的分笔不乱序 (§1「不开第二条连接」 是为了避免乱序, 出口这一端也得守住)。""" @@ -408,7 +412,7 @@ def claim_order(instruction_id: str) -> bool: 而不是靠"我们约定只起一个"。 """ n = execute("UPDATE pms_qmt_order SET status = 'SENDING', updated_at = :ts " - "WHERE instruction_id = :iid AND status = 'QUEUED'", + "WHERE instruction_id = :iid AND status = 'QUEUED' AND cancel_state = 'NONE'", {"iid": instruction_id, "ts": _NOW()}) return bool(n) diff --git a/app/services/dispatcher.py b/app/services/dispatcher.py index 744f9b6..42a0651 100644 --- a/app/services/dispatcher.py +++ b/app/services/dispatcher.py @@ -140,6 +140,17 @@ def dispatch(*, instruction_id: str, ts_code: str, side: str, qty: int, limit_pr """递一手给下游。返回 {ok, ref, mode, note, error}; ok=False 时 executor 不改指令状态。""" m = mode() try: + if str(side).lower() == 'buy': + from app.services.execution_conflicts import buy_blocker + from app.repo import pms_repo + pid = _parent_of(instruction_id, parent_id) + parent = pms_repo.get_instruction(pid) + stopped = pms_repo.instruction_buy_stop(pid) + blocked = ('清仓已终止原买入安排' if stopped or ((parent or {}).get('progress') or {}).get('stopped_by_exit') else buy_blocker(ts_code)) + if parent and parent.get('status') not in ('PROPOSED', 'RULE_PASSED', 'JUDGE_PASSED', 'DISPATCHED'): + blocked = '来源买入指令已结束，不再发单' + if blocked: + return {'ok': False, 'ref': None, 'mode': m, 'note': '', 'error': blocked} if m == MODE_WS: return _ws(instruction_id=instruction_id, ts_code=ts_code, side=side, qty=qty, limit_price=limit_price, valid_until=valid_until, diff --git a/app/services/execution_conflicts.py b/app/services/execution_conflicts.py new file mode 100644 index 0000000..f094671 --- /dev/null +++ b/app/services/execution_conflicts.py @@ -0,0 +1,108 @@ +"""统一执行方向检查。只消解既有指令冲突，不重做研究或择时。""" +from app.repo import pms_repo, qmt_repo + +LIVE = ('PROPOSED', 'RULE_PASSED', 'JUDGE_PASSED', 'DISPATCHED') +BUY_ACTIONS = ('OPEN', 'FILL', 'ADD', 'DCA') + + +def buy_blocker(ts_code): + # 查询失败必须抛给调用者，不能当作无冲突。按票查，不受全局列表裁剪影响。 + rows = pms_repo.list_instructions(statuses=list(LIVE), side='sell', ts_code=ts_code, limit=500) + for row in rows: + if row.get('action') in ('EXIT', 'TRIM') and int(row.get('qty') or 0) > int(row.get('exec_qty') or 0): + verb = '清仓' if row['action'] == 'EXIT' else '减仓' + return f'该股正在{verb}，暂停买入' + return None + + +def queued_buy_blocker(row): + """发出前再次检查。异常由通道保留队列等待，不按无冲突继续。""" + if row.get('side') != 'buy': + return None + parent = pms_repo.get_instruction(row.get('parent_id')) + if not parent: + raise RuntimeError('买入来源指令无法核实，暂不发送') + if pms_repo.instruction_buy_stop(row.get('parent_id')) or (parent.get('progress') or {}).get('stopped_by_exit'): + return '清仓已终止原买入安排' + if parent.get('status') not in LIVE: + return '来源买入指令已结束，未发出的委托作废' + return buy_blocker(row['ts_code']) + + +def enforce_exit(ins): + """完整清仓停止旧买入。每跳重试，撤单失败不会阻断风控卖出。""" + out = {'ok': True, 'instructions': [], 'plans': [], 'errors': []} + if ins.get('action') != 'EXIT' or ins.get('side') != 'sell': + return out + from app.services import strategy_service + why = '清仓指令已生效，停止该股原有买入安排' + # 先持久化停止依据。即使下游撤单暂时失败、清仓随后成交，也不能恢复原买入。 + buys = pms_repo.list_instructions(statuses=list(LIVE), side='buy', ts_code=ins['ts_code'], limit=500) + for buy in buys: + if not pms_repo.instruction_buy_stop(buy['instruction_id']): + if not pms_repo.insert_ledger(ts_code=ins['ts_code'], action='BUY_STOP', arbiter='system', + verdict='PASS', price_at=0, ref_id=buy['instruction_id'], reason=why, + hard_numbers={'exit_instruction_id': ins['instruction_id'], + 'qty': buy.get('qty'), 'exec_qty': buy.get('exec_qty')}): + raise RuntimeError('清仓停止依据未写入动作账本') + progress = dict(buy.get('progress') or {}) + if not progress.get('stopped_by_exit'): + progress['stopped_by_exit'] = ins['instruction_id'] + progress['last_decision'] = {'action': 'WAIT', 'reason': why} + if not pms_repo.update_instruction(buy['instruction_id'], progress=progress): + raise RuntimeError('清仓停止买入依据未写入') + result = strategy_service.cancel_by_code(ins['ts_code'], reason=why) + out.update(result) + # 未解锁与尚未转指令的命令批也要收尾，否则下一跳又生出买单。 + try: + plans = pms_repo.list_plans(ts_code=ins['ts_code'], statuses=['PENDING', 'GATED', 'EXECUTING']) + for plan in plans: + if plan.get('action') not in BUY_ACTIONS: + continue + pms_repo.insert_ledger(ts_code=ins['ts_code'], action='CLEANUP', arbiter='system', + verdict='PASS', price_at=0, ref_id=plan['plan_id'], + reason=why, hard_numbers={'exit_instruction_id': ins['instruction_id']}) + if not pms_repo.update_plan(plan['plan_id'], status='CANCELLED'): + raise RuntimeError('停止买入批次未写入') + out['plans'].append(plan['plan_id']) + except Exception as exc: + out['ok'] = False + out['errors'].append(str(exc)) + return out + + +def instruction_progress(rows): + """只读展示。父指令发过单，不代表当前还有未成交委托。""" + from app.services import dispatcher + ws = dispatcher.mode() == 'ws' + for row in rows: + if row.get('status') not in LIVE and row.get('status') != 'CANCELLED': + continue + if not ws: + row['execution_label'] = '影子执行待核对' + continue + try: + orders = qmt_repo.orders_for_parent(row['instruction_id']) + active = [o for o in orders if o.get('status') in qmt_repo.LIVE] + if any(o.get('cancel_state') in ('REQUESTED', 'SENT') for o in active): + label = '撤单处理中' + elif any(o.get('status') not in ('QUEUED', 'SENDING') for o in active): + label = '委托待成交' + elif active: + label = '委托待发送' + else: + reason = ((row.get('progress') or {}).get('last_decision') or {}).get('reason') or '' + if row.get('status') == 'CANCELLED': + label = '已撤销' + elif '可卖量不足' in reason: + label = '等待可卖' + elif '当日配额' in reason: + label = '等待下一批' + elif '暂停买入' in reason: + label = '买入已暂停' + else: + label = '等待执行' + row['execution_label'] = label + except Exception: + row['execution_label'] = '委托状态待核实' + return rows diff --git a/app/services/executor.py b/app/services/executor.py index 6f5da5f..162419d 100644 --- a/app/services/executor.py +++ b/app/services/executor.py @@ -188,6 +188,29 @@ def run_tick(*, now=None, dry_run: bool = False) -> dict: is_last = tdays_left <= 1 try: + from app.services import execution_conflicts + if ins.get('action') == 'EXIT' and side == 'sell' and not dry_run: + cleanup = execution_conflicts.enforce_exit(ins) + out['errors'].extend(cleanup.get('errors') or []) + if side == 'buy': + # 上面的清仓可能已撤销本轮列表中的旧买入，不能拿旧列表继续发。 + fresh = pms_repo.get_instruction(ins['instruction_id']) + if not fresh or fresh.get('status') not in LIVE: + continue + prog = dict(fresh.get('progress') or {}) + if pms_repo.instruction_buy_stop(ins['instruction_id']) or prog.get('stopped_by_exit'): + if not dry_run: + cancelled = cancel_instruction(ins['instruction_id'], reason='清仓已终止原买入安排', arbiter='system') + if not cancelled.get('ok'): + out['errors'].append(cancelled.get('message') or '清仓停止买入尚未完成') + continue + blocked = execution_conflicts.buy_blocker(code) + if blocked: + prog['last_decision'] = {'at': now.strftime('%H:%M'), 'action': 'WAIT', 'reason': blocked} + if not dry_run: + pms_repo.update_instruction(ins['instruction_id'], progress=prog) + out['waited'].append({'instruction_id': ins['instruction_id'], 'code': code, 'reason': blocked}) + continue day = market.day_snapshot(code) pos = _pos_of(view, code) quota = et.daily_quota(remaining, tdays_left, lot=lot_of(code), @@ -427,7 +450,7 @@ def sweep_windows(*, now=None) -> dict: return out -def cancel_instruction(instruction_id: str, reason: str = "页面人工撤销") -> dict: +def cancel_instruction(instruction_id: str, reason: str = "页面人工撤销", *, arbiter="user") -> dict: ins = pms_repo.get_instruction(instruction_id) if not ins: return {"ok": False, "error": "指令不存在"} @@ -453,7 +476,7 @@ def cancel_instruction(instruction_id: str, reason: str = "页面人工撤销") except Exception as e: logger.warning("撤销指令进度回写失败 %s: %s", instruction_id, e) pms_repo.update_instruction(instruction_id, status=ST_CANCELLED) - pms_repo.insert_ledger(ts_code=ins["ts_code"], action=ins.get("action"), arbiter="user", + pms_repo.insert_ledger(ts_code=ins["ts_code"], action=ins.get("action"), arbiter=arbiter, verdict="REJECT", price_at=0, ref_id=instruction_id, reason=reason, hard_numbers=(ins.get('progress') or {}).get('plan') or {}) return {"ok": True, "downstream": r, "message": f"指令 {instruction_id} 已撤销"} diff --git a/app/services/proposal_service.py b/app/services/proposal_service.py index 009c039..644bb17 100644 --- a/app/services/proposal_service.py +++ b/app/services/proposal_service.py @@ -1637,6 +1637,8 @@ def _inflight_keys() -> dict: if action in ae.SELL_SIDE_ACTIONS: for other in ae.SELL_SIDE_ACTIONS: keys.setdefault((code, other), why + " (同票另一种减持一并让路)") + for buy in ('OPEN', 'FILL', 'ADD', 'DCA'): + keys.setdefault((code, buy), '该股有减持安排待处理，暂不新增买入提议') try: for p in pms_repo.list_proposals(statuses=("WAIT_USER",), limit=200): _mark(p["ts_code"], p["action"], f"已有在途提议 {p.get('proposal_id')} 在等人确认") diff --git a/app/services/strategy_service.py b/app/services/strategy_service.py index e911f46..d1382fc 100644 --- a/app/services/strategy_service.py +++ b/app/services/strategy_service.py @@ -180,7 +180,8 @@ def cancel_by_code(ts_code: str, reason: str = "") -> dict: from app.services import command_service, executor try: props = [p for p in pms_repo.list_proposals(statuses=("WAIT_USER",), limit=500) - if p.get("ts_code") == ts_code and p.get("action") in command_service.BUILD_ACTIONS] + if p.get("ts_code") == ts_code and (p.get("action") in command_service.BUILD_ACTIONS + or (p.get('hard_numbers') or {}).get('side') == 'buy')] except Exception as e: # noqa: BLE001 props = [] out["ok"] = False @@ -189,13 +190,15 @@ def cancel_by_code(ts_code: str, reason: str = "") -> dict: try: if pms_repo.decide_proposal(p["proposal_id"], "DECLINED"): out["proposals"].append(p["proposal_id"]) + pms_repo.insert_ledger(ts_code=ts_code, action=p.get('action'), arbiter='system', + verdict='REJECT', price_at=0, ref_id=p['proposal_id'], reason=why) except Exception as e: # noqa: BLE001 out["ok"] = False out["errors"].append(f"驳回买入提议 {p.get('proposal_id')} 失败: {type(e).__name__}: {e}") # 三, 撤该票当前在途的买入指令 (已下发到券商的网格/补仓买单, 撤了方案它还挂着) try: live_buys = [i for i in pms_repo.list_instructions( - statuses=list(command_service.LIVE_INSTR), side="buy", limit=500) + statuses=list(command_service.LIVE_INSTR), side="buy", ts_code=ts_code, limit=500) if i.get("ts_code") == ts_code] except Exception as e: # noqa: BLE001 live_buys = [] @@ -204,7 +207,7 @@ def cancel_by_code(ts_code: str, reason: str = "") -> dict: for i in live_buys: iid = i.get("instruction_id") try: - r = executor.cancel_instruction(iid, reason=why) or {} + r = executor.cancel_instruction(iid, reason=why, arbiter="system") or {} if r.get("ok"): out["instructions"].append(iid) else: diff --git a/app/web/main.py b/app/web/main.py index 584fef4..50e5493 100644 --- a/app/web/main.py +++ b/app/web/main.py @@ -419,8 +419,11 @@ def api_lots(ts_code: str, status: str = Query("OPEN")): def api_instructions(status: str = Query(None), limit: int = 200, include_archived: bool = Query(False)): statuses = [s for s in (status or "").split(",") if s] or None - return ok(pms_repo.list_instructions, statuses=statuses, limit=limit, - include_archived=include_archived) + def _list(): + from app.services.execution_conflicts import instruction_progress + return instruction_progress(pms_repo.list_instructions(statuses=statuses, limit=limit, + include_archived=include_archived)) + return ok(_list) @app.get("/api/ledger") @@ -523,6 +526,14 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default= except Exception as e: # noqa: BLE001 logger.warning("[提议] 过期清扫失败: %s", e) return {"ok": False, "error": "提议已过失效时刻, 不能再裁决; 该票由系统重新评估"} + if decision == 'ACCEPTED': + from app.services.execution_conflicts import buy_blocker + hn0 = p.get('hard_numbers') or {} + side0 = hn0.get('side') if hn0.get('kind') == 'strategy' else ('sell' if p['action'] in ('TRIM', 'EXIT') else 'buy') + if side0 == 'buy': + blocked = buy_blocker(p['ts_code']) + if blocked: + return {'ok': False, 'error': blocked, 'proposal_id': proposal_id} expected_hard = None if (decision == 'ACCEPTED' and p['action'] == ae.A_OPEN and param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False)): diff --git a/app/web/static/index.html b/app/web/static/index.html index 2c9a925..76e7f91 100644 --- a/app/web/static/index.html +++ b/app/web/static/index.html @@ -1144,7 +1144,7 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;} - + @@ -1154,7 +1154,7 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;} - + @@ -1162,7 +1162,7 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;} - + @@ -2865,7 +2865,7 @@ createApp({ const T = { // 账本的动作有十六个：八个真动作（planner.py:31 那批常量）加八个伪动作 // （账本里直写的字面量，例如对账修正、信号留痕）。少一个就会在页面上露原码。 - action: { OPEN:'建新仓', OPEN_REASK:'重问研判', FILL:'回落补足', ADD:'盈利加仓', DCA:'浮亏补仓', TRIM:'保盈减仓', + action: { BUY_STOP:'停止买入', OPEN:'建新仓', OPEN_REASK:'重问研判', FILL:'回落补足', ADD:'盈利加仓', DCA:'浮亏补仓', TRIM:'保盈减仓', EXIT:'清仓', T0:'做T', T0_ROUND:'做T', SWITCH:'换仓', HALT:'停手', RECON:'对账修正', CLEANUP:'收尾清理', MACRO:'大盘冷热', SIGNAL:'信号留痕', SIGNAL_BUY:'买入信号', @@ -2879,7 +2879,7 @@ createApp({ lot: { BASE:'首批', FILL:'回落补足', ADD:'盈利加仓', DCA:'浮亏补仓', T0:'做T仓', RECON:'对账修正' }, frozen: { NONE:'', COMMAND_HALT:'命令要求只卖不买', BRAKE:'组合刹车，只卖不买', MANUAL:'人工冻结，只卖不买' }, posStatus: { PLANNED:'计划建仓', OPENING:'建仓中', HOLDING:'持有中', EXITING:'清仓中', CLOSED:'已清仓' }, - insStatus: { PROPOSED:'待校验', RULE_PASSED:'合规检查已过', JUDGE_PASSED:'决策系统复核已过', DISPATCHED:'已挂单待成交', + insStatus: { PROPOSED:'待校验', RULE_PASSED:'合规检查已过', JUDGE_PASSED:'决策系统复核已过', DISPATCHED:'执行中', CONFIRMED:'已成交', REJECTED:'已拦截', EXPIRED:'到期作废', CANCELLED:'已撤销' }, arbiter: { rule:'合规检查', judge:'决策系统', user:'人工', system:'系统自动' }, // 这一组是**评审账本**的结论，只有四个值。研判闸的结论是另一组 judgeVerdict， diff --git a/app/ws/runner.py b/app/ws/runner.py index cc61e32..206c95d 100644 --- a/app/ws/runner.py +++ b/app/ws/runner.py @@ -843,8 +843,8 @@ class WsRunner: """ while not self._stop.is_set(): try: - await self._drain_outbox(seed) await self._drain_cancels(seed) + await self._drain_outbox(seed) except asyncio.CancelledError: raise except Exception as e: @@ -871,6 +871,22 @@ class WsRunner: if not await _db(qmt_repo.claim_order, iid): continue # 被别的进程抢走了 (滚动重启的瞬间) try: + # 排队后可能发生清仓或撤单。认领后再核对，未知状态保留重试。 + fresh = await _db(qmt_repo.get_order, iid) + if not fresh or fresh.get('cancel_state') in ('REQUESTED', 'SENT'): + await _db(qmt_repo.abort_order, iid, '发出前撤销') + continue + from app.services.execution_conflicts import queued_buy_blocker + try: + blocked = await _db(queued_buy_blocker, fresh) + except Exception as exc: + await _db(qmt_repo.requeue_order, iid, f'买入依据暂不可核实: {exc}', + self._p('max_attempts', 3), False) + logger.warning('[outbox] %s 买入依据暂不可核实，保留队列', iid) + continue + if blocked: + await _db(qmt_repo.abort_order, iid, blocked) + continue payload = wsc.place_order_payload( instruction_id=iid, ts_code=r["ts_code"], side=r["side"], qty=int(r["qty"]), limit_price=r["limit_price"], diff --git a/docs/开发规格_买卖方向冲突修复_2026-10-19.md b/docs/开发规格_买卖方向冲突修复_2026-10-19.md new file mode 100644 index 0000000..559893e --- /dev/null +++ b/docs/开发规格_买卖方向冲突修复_2026-10-19.md @@ -0,0 +1,48 @@ +# 买卖方向冲突修复开发规格 + +2026-10-19，北京时间。用户已授权修复并检查其他流程。 + +## 目标与已核实反例 + +科士达旧建仓目标600股，累计买入400股。今天09:51风控清仓400股，卖出200股。 +旧建仓仍活跃，剩余200股可能隔日再买。实际子委托均已成交，页面却写已挂单。 +原始卖出理由在audit_reason及component_scores.reason，PMS解析遗漏。 +证据保存在155 ~/pms_logs/kstar-conflict-20261019-1402.json及kstar-signal-20261019.json。 + +## 实现与边界 + +1. 停止依据追加到动作账本，不依赖可被并发覆盖的进度JSON。下发回写不得复活终态指令。 + 全部完整清仓指令在执行前统一停买入侧。复用既有策略、提议、指令撤销路径。 + 覆盖信号自动清仓、命令清仓、人工采纳、技术面离场及跟踪止盈。 + 每跳重试未完成收尾，也覆盖上线前遗留的清仓指令。不修改已成交数量、成本或盈亏。 +2. 同票有活跃EXIT或TRIM时，买入发出前再检查。查询失败停止买入。 + 普通减仓只暂缓买入，不永久撤销建仓或策略。做T自身T0_ROUND不当清仓。 + 完整清仓期间，做T买回也不能抵消清仓；没有清仓时原有配对流程不变。 +3. 待确认卖出仅阻止新的自主增持提议，不自动撤销已有买单。 + 用户采纳买入前核对已生效的卖出指令，冲突时保留提议和说明，不写已采纳。 +4. 出口队列先处理撤单，认领时排除待撤单。发送前重查买入冲突及父指令状态。 + 尚未发出且已失效的买单本地作废，已经发出的买单走撤单回执。 + 撤单受理不冒充成交撤回，迟到成交继续按已有逐笔路径记账。 +5. 页面读取实际子委托状态。没有活跃子单时显示等待下一批、等待可卖或执行等待。 + 技术单号只在详情出现。卖出理由采用原文，不新调用模型。 + +## 预判与验证 + +先建立科士达跨日反例，再验证各入口共用的执行终检。 +覆盖部分成交、T+1、普通减仓、T0配对、撤单失败重试、排队后清仓、查询失败、重复执行。 +页面验证已发出且终态的子单不能显示为挂单，取数失败明确为状态待核实。 +全部测试使用配置隔离与禁网入口。运行三页面守卫。 +155部署前保存参数、账务、指令与镜像证据。只修模拟实例，188不动。 +上线后先查自然调度清理旧建仓，保留真实400股买入与200股卖出。 +不采纳提议、不制造成交，不改交易时段或业务阈值。 + +## 回退 + +保留成交与撤销审计。155挂载/app，须反向提交、git pull后重建四容器。 +回退不恢复已经撤销的买入指令。撤销前保存旧镜像与证据。 + +## 本地验证记录 + +22项专项回归通过，覆盖五类清仓入口共用执行器、撤单失败与旧进度覆盖、T0不误挡、人工确认被拦时保持待确认。 +45批配置隔离、禁止联网的回归通过，包含三页面守卫。生产状态另记部署判收。 +原临时虚拟环境被系统清理，本轮另建 /private/tmp/pms-conflict-20261019-venv。 diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 0647472..56b24d6 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -5,6 +5,7 @@ 运行: docker compose run --rm pms-web python scripts/run_tests.py 包含 (例数按 2026-10-12 隔离回归实跑校正): + test_batch41_units.py 跨流程买卖互斥、清仓收尾、出口撤单与真实进展 (22 例) test_batch40_units.py 逐笔历史修复、进度暂停、并发与事务回滚与均价失效 (22 例) test_batch39_units.py 成交双来源归属、迟到、部分成交与失败守成 (12 例) test_batch38_units.py 每日计划顺序、版本核验、重新确认、并发采纳、历史保留及采纳失败说明、明确否决撤下 (30 例) @@ -191,7 +192,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py", "test_batch35_units.py", "test_batch36_units.py", "test_batch37_units.py", - "test_batch38_units.py", "test_batch39_units.py", "test_batch40_units.py", + "test_batch38_units.py", "test_batch39_units.py", "test_batch40_units.py", "test_batch41_units.py", "test_page_copy_guard.py", "test_page_enum_guard.py", "test_page_wiring_guard.py", diff --git a/scripts/test_batch41_units.py b/scripts/test_batch41_units.py new file mode 100644 index 0000000..cbc6d7c --- /dev/null +++ b/scripts/test_batch41_units.py @@ -0,0 +1,256 @@ +"""本地：python scripts/unit_test_guard.py scripts/test_batch41_units.py。配置隔离、禁止联网。""" +import unit_test_guard +import os +import sys +import unittest +from datetime import datetime, timedelta +from unittest.mock import patch +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from test_wiring import install_fakes +from app.repo import pms_repo, qmt_repo +from app.services import executor, dispatcher, proposal_service, signal_service +from app.core import signal_rules + +CODE = '002518.SZ' + +def ins(fake, iid, action, side, qty=600, done=0): + fake.insert_instruction(instruction_id=iid, origin_type='proposal', origin_id=None, + ts_code=CODE, action=action, side=side, qty=qty, status='DISPATCHED', + progress={'children': [], 'deadline': '2026-10-21', 'is_command': False}) + fake.instructions[iid]['exec_qty'] = done + return fake.instructions[iid] + +class ConflictsTest(unittest.TestCase): + def setUp(self): + self.fake=install_fakes(params={'PMS_DISPATCH_MODE':'ws'}) + self.fake.qmt.set_channel() + + def test_old_partial_buy_stopped_before_exit(self): + from app.services import execution_conflicts as c + buy=ins(self.fake,'old','OPEN','buy',600,400) + sell=ins(self.fake,'exit','EXIT','sell',400,200) + result=c.enforce_exit(sell) + self.assertTrue(result['ok']) + self.assertEqual(buy['status'],'CANCELLED') + self.assertEqual((buy['qty'],buy['exec_qty']),(600,400)) + self.assertEqual(sell['exec_qty'],200) + n=len(self.fake.ledger) + c.enforce_exit(sell) + self.assertEqual(len(self.fake.ledger),n) + + def test_trim_pauses_without_cancelling_buy(self): + from app.services import execution_conflicts as c + buy=ins(self.fake,'old','OPEN','buy') + sell=ins(self.fake,'trim','TRIM','sell') + self.assertTrue(c.buy_blocker(CODE)) + c.enforce_exit(sell) + self.assertEqual(buy['status'],'DISPATCHED') + sell['status']='CONFIRMED' + self.assertFalse(c.buy_blocker(CODE)) + + def test_t0_pair_is_not_exit(self): + from app.services import execution_conflicts as c + ins(self.fake,'t0','T0_ROUND','sell') + self.assertFalse(c.buy_blocker(CODE)) + + def test_dispatch_rejects_buy_with_active_sell(self): + ins(self.fake,'exit','EXIT','sell',400) + result=dispatcher.dispatch(instruction_id='buy_D01',parent_id='buy',ts_code=CODE, + side='buy',qty=200,limit_price=28,valid_until=datetime.now()+timedelta(minutes=10)) + self.assertFalse(result['ok']) + self.assertEqual(self.fake.qmt.orders,{}) + + def test_lookup_failure_cannot_dispatch_buy(self): + with patch.object(pms_repo,'list_instructions',side_effect=RuntimeError('source down')): + r=dispatcher.dispatch(instruction_id='buy_D01',ts_code=CODE,side='buy',qty=200,limit_price=28) + self.assertFalse(r['ok']) + self.assertEqual(self.fake.qmt.orders,{}) + + def test_pending_sell_blocks_new_buy_proposals(self): + self.fake.insert_proposal(proposal_id='sell',ts_code=CODE,action='EXIT',qty=200, + hard_numbers={},expire_at=datetime.now()+timedelta(hours=1)) + keys=proposal_service._inflight_keys() + for action in ('OPEN','ADD','FILL','DCA'): + self.assertIn((CODE,action),keys) + + def test_signal_reason_preserved(self): + parsed=signal_rules.parse_intraday({'ts_code':CODE,'action':'SELL','confidence':'0.95', + 'audit_reason':'支撑失守，资金流出','component_scores':'{"reason":"另一份理由"}'}) + self.assertEqual(parsed['reason'],'支撑失守，资金流出') + parsed=signal_rules.parse_intraday({'component_scores':'{"reason":"资金流出"}'}) + self.assertEqual(parsed['reason'],'资金流出') + + def test_cancel_failure_keeps_stop_after_exit_completes(self): + from app.services import execution_conflicts as c + buy=ins(self.fake,'old','OPEN','buy',600,400) + sell=ins(self.fake,'exit','EXIT','sell',400,200) + with patch.object(dispatcher,'cancel',return_value={'ok':False,'error':'offline'}): + result=c.enforce_exit(sell) + self.assertFalse(result['ok']) + self.assertEqual(buy['status'],'DISPATCHED') + self.assertEqual(buy['progress']['stopped_by_exit'],'exit') + sell['status']='CONFIRMED' + r=dispatcher.dispatch(instruction_id='old_D03',parent_id='old',ts_code=CODE, + side='buy',qty=200,limit_price=28) + self.assertFalse(r['ok']) + self.assertEqual(self.fake.qmt.orders,{}) + + def test_cleanup_cancels_gated_plans_and_strategy_buy_proposal(self): + from app.services import execution_conflicts as c + self.fake.plans=[{'plan_id':'p1','ts_code':CODE,'action':'ADD','status':'GATED','filled_qty':0}] + self.fake.insert_proposal(proposal_id='t0buy',ts_code=CODE,action='T0_ROUND',qty=200, + hard_numbers={'kind':'strategy','side':'buy'},expire_at=datetime.now()+timedelta(hours=1)) + result=c.enforce_exit(ins(self.fake,'exit','EXIT','sell',400)) + self.assertTrue(result['ok'],result) + self.assertEqual(self.fake.plans[0]['status'],'CANCELLED') + self.assertEqual(self.fake.proposals['t0buy']['status'],'DECLINED') + self.assertTrue(any(x['arbiter']=='system' and x['ref_id']=='t0buy' for x in self.fake.ledger)) + + def test_all_exit_origins_share_executor_cleanup(self): + from app.services import execution_conflicts as c, portfolio, exec_advisor + for origin in ('system','proposal','plan','strategy'): + with self.subTest(origin=origin): + self.fake.instructions.clear() + buy=ins(self.fake,'old','OPEN','buy',600,400) + sell=ins(self.fake,'exit','EXIT','sell',400,200) + sell['origin_type']=origin + with patch.object(portfolio,'positions_view',return_value={'positions':[],'params':{}}), \ + patch.object(executor.td,'trade_days_left',return_value=2), \ + patch.object(exec_advisor,'decide',return_value={'action':'WAIT','reason':'等待可卖'}): + result=executor.run_tick() + self.assertEqual(buy['status'],'CANCELLED',result) + self.assertFalse(result['fired']) + self.assertEqual(sell['exec_qty'],200) + + def test_dry_run_does_not_cancel(self): + from app.services import portfolio, exec_advisor + buy=ins(self.fake,'old','OPEN','buy',600,400) + ins(self.fake,'exit','EXIT','sell',400,200) + with patch.object(portfolio,'positions_view',return_value={'positions':[],'params':{}}), \ + patch.object(executor.td,'trade_days_left',return_value=2), \ + patch.object(exec_advisor,'decide',return_value={'action':'WAIT','reason':'等待可卖'}): + executor.run_tick(dry_run=True) + self.assertEqual(buy['status'],'DISPATCHED') + self.assertFalse(buy['progress'].get('stopped_by_exit')) + self.assertFalse(self.fake.ledger) + + def test_queued_buy_rechecks_parent_and_exit(self): + from app.services import execution_conflicts as c + buy=ins(self.fake,'old','OPEN','buy') + row={'parent_id':'old','ts_code':CODE,'side':'buy'} + self.assertIsNone(c.queued_buy_blocker(row)) + ins(self.fake,'exit','EXIT','sell') + self.assertTrue(c.queued_buy_blocker(row)) + self.fake.instructions['exit']['status']='CONFIRMED' + buy['status']='CANCELLED' + self.assertTrue(c.queued_buy_blocker(row)) + self.assertIsNone(c.queued_buy_blocker(dict(row,side='sell'))) + + def test_missing_parent_keeps_queue_unverified(self): + from app.services import execution_conflicts as c + with self.assertRaises(RuntimeError): + c.queued_buy_blocker({'parent_id':'missing','ts_code':CODE,'side':'buy'}) + + def test_finished_children_show_wait_instead_of_pending_order(self): + from app.services import execution_conflicts as c + row=ins(self.fake,'old','OPEN','buy',600,400) + row['progress']['last_decision']={'reason':'当日配额已出完'} + with patch.object(qmt_repo,'orders_for_parent',return_value=[{'status':'FILLED'}]): + self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'等待下一批') + row['progress']['last_decision']={'reason':'可卖量不足, 顺延'} + with patch.object(qmt_repo,'orders_for_parent',return_value=[]): + self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'等待可卖') + + def test_pending_cancel_never_shown_as_finished(self): + from app.services import execution_conflicts as c + row=ins(self.fake,'old','OPEN','buy');row['status']='CANCELLED' + with patch.object(qmt_repo,'orders_for_parent',return_value=[{'status':'PARTIAL','cancel_state':'REQUESTED'}]): + self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'撤单处理中') + with patch.object(qmt_repo,'orders_for_parent',side_effect=RuntimeError('unavailable')): + self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'委托状态待核实') + + def test_other_stock_not_blocked(self): + from app.services import execution_conflicts as c + ins(self.fake,'exit','EXIT','sell') + self.assertIsNone(c.buy_blocker('600000.SH')) + + def test_claim_rejects_cancel_requested_atomically(self): + with patch.object(qmt_repo,'execute',return_value=0) as write: + self.assertFalse(qmt_repo.claim_order('child')) + self.assertIn("cancel_state = 'NONE'",write.call_args.args[0]) + + def test_outbox_drops_new_conflict_without_sending(self): + import asyncio + from unittest.mock import AsyncMock + from app.ws.runner import WsRunner + buy=ins(self.fake,'old','OPEN','buy') + ins(self.fake,'exit','EXIT','sell') + row={'instruction_id':'old_D03','parent_id':'old','ts_code':CODE,'side':'buy', + 'qty':200,'limit_price':28,'valid_until':int((datetime.now()+timedelta(minutes=10)).timestamp()*1000), + 'cancel_state':'NONE','status':'SENDING'} + loop=asyncio.new_event_loop() + asyncio.set_event_loop(loop) + self.addCleanup(loop.close) + runner=WsRunner() + with patch.object(qmt_repo,'next_queued',return_value=[row]), \ + patch.object(qmt_repo,'claim_order',return_value=True), \ + patch.object(qmt_repo,'get_order',return_value=row), \ + patch.object(qmt_repo,'abort_order') as abort, \ + patch.object(qmt_repo,'mark_sent'), \ + patch.object(runner,'_send',new_callable=AsyncMock) as send: + asyncio.run(runner._drain_outbox('unused')) + abort.assert_called_once() + send.assert_not_called() + + def test_outbox_source_failure_preserves_queue(self): + import asyncio + from unittest.mock import AsyncMock + from app.ws.runner import WsRunner + row={'instruction_id':'old_D03','parent_id':'old','ts_code':CODE,'side':'buy', + 'valid_until':int((datetime.now()+timedelta(minutes=10)).timestamp()*1000), 'cancel_state':'NONE'} + loop=asyncio.new_event_loop() + asyncio.set_event_loop(loop) + self.addCleanup(loop.close) + runner=WsRunner() + with patch.object(qmt_repo,'next_queued',return_value=[row]), \ + patch.object(qmt_repo,'claim_order',return_value=True), \ + patch.object(qmt_repo,'get_order',return_value=row), \ + patch.object(qmt_repo,'requeue_order') as retry, \ + patch.object(qmt_repo,'mark_sent'), \ + patch.object(runner,'_send',new_callable=AsyncMock) as send: + asyncio.run(runner._drain_outbox('unused')) + retry.assert_called_once() + self.assertIs(retry.call_args.args[-1],False) + send.assert_not_called() + + def test_stale_progress_cannot_restore_stopped_buy(self): + from app.services import execution_conflicts as c + buy=ins(self.fake,'old','OPEN','buy',600,400) + sell=ins(self.fake,'exit','EXIT','sell',400,200) + with patch.object(dispatcher,'cancel',return_value={'ok':False,'error':'offline'}): + c.enforce_exit(sell) + buy['progress']={'children':[]} # 模拟并发旧进度覆盖，停止事实仍在追加账本中。 + sell['status']='CONFIRMED' + result=dispatcher.dispatch(instruction_id='old_D03',parent_id='old',ts_code=CODE,side='buy',qty=200,limit_price=28) + self.assertFalse(result['ok']) + + def test_terminal_parent_cannot_be_revived_by_old_dispatch_write(self): + buy=ins(self.fake,'old','OPEN','buy');buy['status']='CANCELLED' + self.assertEqual(pms_repo.update_instruction('old',status='DISPATCHED'),0) + self.assertEqual(buy['status'],'CANCELLED') + + def test_human_accept_is_blocked_before_changing_proposal(self): + from app.web import main + ins(self.fake,'exit','EXIT','sell') + self.fake.insert_proposal(proposal_id='p',ts_code=CODE,action='OPEN',qty=200, + hard_numbers={},expire_at=datetime.now()+timedelta(hours=1)) + with patch.object(main,'ok_logged',side_effect=lambda name,fn,**kw:fn()), \ + patch.object(main,'_actor',return_value='test'): + result=main.api_decide(None,'p',{'decision':'ACCEPTED','reason':'用户确认'}) + self.assertFalse(result['ok']) + self.assertEqual(self.fake.proposals['p']['status'],'WAIT_USER') + self.assertEqual(len(self.fake.instructions),1) + +if __name__=='__main__':unittest.main() + + diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index bd50e98..c5bbc88 100644 --- a/scripts/test_wiring.py +++ b/scripts/test_wiring.py @@ -257,6 +257,10 @@ class FakeRepo: and (not ts_code or i["ts_code"] == ts_code)] return out[:limit] + def instruction_buy_stop(self, instruction_id): + return next((r for r in reversed(self.ledger) if r.get('ref_id') == instruction_id + and r.get('action') == 'BUY_STOP' and r.get('verdict') == 'PASS'), None) + def get_instruction(self, iid): return self.instructions.get(iid) @@ -269,6 +273,8 @@ class FakeRepo: i = self.instructions.get(iid) if not i: return 0 + if status == 'DISPATCHED' and i.get('status') not in ('PROPOSED', 'RULE_PASSED', 'JUDGE_PASSED', 'DISPATCHED'): + return 0 for k, v in (("status", status), ("exec_qty", exec_qty), ("exec_avg_price", exec_avg_price), ("dispatch_ref", dispatch_ref), ("progress", progress)):