# -*- coding: utf-8 -*- """ 执行编排: 方案 → 指令 → 分日出手 → 回执 (设计 §7 / §8 / §9 的落地) ===================================================================== 纯逻辑在 core/rule_gate.py (规则闸) 与 core/exec_timing.py (择时实现B), 本模块负责 取数、落表与状态推进。三个入口: materialize_plans() 把命令方案里「可执行」的条目转成 pms_instruction (先记账后动作)。 gated 方案 (建仓的 FILL/ADD 批) 状态为 GATED, 由动作引擎解锁, 此处不动。 run_tick() 盘中每分钟一跳: 配额 → 择时判定 → 规则闸终检 → 下发 → 记子单。 sweep_windows() 窗口耗尽收口: 命令类置部分完成并告警, 自主类作废。 指令与子单的关系: 一条 pms_instruction 承载一个方案条目的**总量**, 每日出手记一条子单到 progress_json.children[]; exec_qty 由回放认领回填 (账本以下游成交为准, 不拿下发量当成交量)。 """ from __future__ import annotations import logging from datetime import datetime from app.core import command_spec as cs from app.core import exec_timing as et from app.core import rule_gate from app.core import tradedays as td from app.repo import pms_repo from app.services import (command_service, dispatcher, industry, market, param_store, portfolio) logger = logging.getLogger("pms.exec") # 指令状态 ST_PROPOSED, ST_RULE_PASSED, ST_DISPATCHED = "PROPOSED", "RULE_PASSED", "DISPATCHED" ST_CONFIRMED, ST_REJECTED, ST_EXPIRED, ST_CANCELLED = ("CONFIRMED", "REJECTED", "EXPIRED", "CANCELLED") LIVE = (ST_PROPOSED, ST_RULE_PASSED, ST_DISPATCHED) # 方案状态: GATED = 建仓的补足/加仓批, 等动作引擎按条件解锁 (DDL 注释已同步) PLAN_PENDING, PLAN_GATED, PLAN_EXEC, PLAN_DONE = "PENDING", "GATED", "EXECUTING", "DONE" SELL_ACTIONS = {"EXIT", "TRIM"} BUY_ACTIONS = {"OPEN", "FILL", "ADD", "DCA"} # ================================================================ 方案 → 指令 def materialize_plans(limit: int = 100) -> dict: """把在途命令的可执行方案条目转成指令。幂等: 已生成过指令的方案置 EXECUTING 不再重复。""" out = {"created": [], "skipped": [], "errors": []} plans = pms_repo.list_plans(statuses=[PLAN_PENDING], limit=limit) if not plans: return out cmd_cache = {} ymd = td.ymd() seq = 0 for p in plans: act = p.get("action") if act == "HALT": # 撤单类在规划期已执行完毕 pms_repo.update_plan(p["plan_id"], status=PLAN_DONE) out["skipped"].append({"plan_id": p["plan_id"], "why": "HALT 已在规划期执行"}) continue side = "sell" if act in SELL_ACTIONS else ("buy" if act in BUY_ACTIONS else None) if not side: out["skipped"].append({"plan_id": p["plan_id"], "why": f"动作 {act} 无对应指令"}) continue cid = p["command_id"] if cid not in cmd_cache: cmd_cache[cid] = pms_repo.get_command(cid) cmd = cmd_cache[cid] if not cmd or cmd["status"] not in (cs.ST_EXECUTING, cs.ST_PARTIAL): out["skipped"].append({"plan_id": p["plan_id"], "why": f"命令状态 {cmd['status'] if cmd else '缺失'}"}) continue qty = int(p.get("qty") or 0) if qty <= 0: out["skipped"].append({"plan_id": p["plan_id"], "why": "数量为 0"}) continue seq += 1 iid = cs.make_instruction_id(ymd, p["ts_code"], act, seq) window = int((cmd.get("progress") or {}).get("window_tdays") or param_store.get_int("PMS_EXEC_WINDOW_TDAYS", 3)) try: pms_repo.insert_instruction( instruction_id=iid, origin_type="plan", origin_id=p["plan_id"], ts_code=p["ts_code"], action=act, side=side, qty=qty, limit_price=None, window_tdays=window, status=ST_PROPOSED, progress={"deadline": str(p.get("deadline") or ""), "command_id": cid, "is_command": True, "children": []}) pms_repo.update_plan(p["plan_id"], status=PLAN_EXEC) out["created"].append(iid) except Exception as e: logger.exception("方案转指令失败 %s", p["plan_id"]) out["errors"].append(f"{p['plan_id']}: {type(e).__name__}: {e}") return out # ================================================================ 出手 tick def run_tick(*, now=None, dry_run: bool = False) -> dict: """盘中每分钟一跳。dry_run=True 时只算不发不落库 (页面「试算」用)。""" now = now or datetime.now() out = {"ok": True, "checked": 0, "fired": [], "waited": [], "rejected": [], "errors": [], "mode": dispatcher.mode(), "dry_run": dry_run} instrs = [i for i in pms_repo.list_instructions(statuses=list(LIVE), limit=200) if int(i.get("qty") or 0) > int(i.get("exec_qty") or 0)] if not instrs: return out view = portfolio.positions_view() stock_params = command_service.effective_stock_params() prm = view["params"] exec_prm = { "sell_avoid_open_min": param_store.get_int("PMS_SELL_AVOID_OPEN_MIN", 30), "buy_halt_dayup": param_store.get_float("PMS_BUY_HALT_DAYUP", 0.05), "eod_force_time": param_store.get("PMS_EOD_FORCE_TIME", "14:45"), "eod_force_discount": param_store.get_float("PMS_EOD_FORCE_DISCOUNT", 0.998), "no_chase_ma5": param_store.get_float("PMS_NO_CHASE_MA5", 0.06), } slices = param_store.get_int("PMS_EXEC_SLICES", 1) brake_active = td.ymd() < param_store.get_int("PMS_BRAKE_UNTIL", 0) ymd_today = td.ymd() for ins in instrs: out["checked"] += 1 code, side = ins["ts_code"], str(ins.get("side") or "").lower() prog = dict(ins.get("progress") or {}) children = list(prog.get("children") or []) remaining = int(ins["qty"]) - int(ins.get("exec_qty") or 0) deadline = prog.get("deadline") or "" tdays_left = td.trade_days_left(deadline, now) if deadline else 1 is_last = tdays_left <= 1 try: day = market.day_snapshot(code) pos = _pos_of(view, code) quota = et.daily_quota(remaining, tdays_left, allow_odd_tail=(ins.get("action") == "EXIT")) fired_today = sum(int(c.get("qty") or 0) for c in children if int(c.get("ymd") or 0) == ymd_today) day_ctx = {**day, "support": pos.get("support_ref"), "limit_up": _limit_up(day), "limit_down": _limit_down(day), "halted": not day or not day.get("price")} d = et.decide(side=side, now=now, day=day_ctx, params=exec_prm, is_last_day=is_last, fired_today=fired_today, quota=quota) if d["action"] != et.ACT_FIRE: prog["last_decision"] = {"at": now.strftime("%H:%M"), **d} if not dry_run: pms_repo.update_instruction(ins["instruction_id"], progress=prog) out["waited"].append({"instruction_id": ins["instruction_id"], "code": code, "action": d["action"], "reason": d["reason"]}) continue slice_list = et.slice_qty(min(d["qty_hint"], remaining), slices) qty = slice_list[0] if slice_list else 0 if side == "sell": qty = min(qty, int(pos.get("avail_qty") or 0)) if qty <= 0: prog["last_decision"] = {"at": now.strftime("%H:%M"), "action": et.ACT_WAIT, "reason": "可卖量不足, 顺延"} if not dry_run: pms_repo.update_instruction(ins["instruction_id"], progress=prog) out["waited"].append({"instruction_id": ins["instruction_id"], "code": code, "action": et.ACT_WAIT, "reason": "可卖量不足, 顺延"}) continue ma5 = market.get_ma5(code) if side == "buy" else None gate = rule_gate.check( side=side, action=ins.get("action"), qty=qty, price=day_ctx.get("price"), ctx={"ts_code": code, "position": pos, "day": {**day_ctx, "ma5": ma5}, "params": {**exec_prm, "sector_source_ready": view["sector_ready"]}, "caps": portfolio.caps_ctx(view, ts_code=code) if side == "buy" else None, "flags": {"buy_halt": prm["buy_halt"], "exec_halt": prm["exec_halt"], "brake_active": brake_active, "blacklisted": bool(stock_params.get(code, {}).get("black")), "is_command": bool(prog.get("is_command"))}}) if not gate["passed"]: out["rejected"].append({"instruction_id": ins["instruction_id"], "code": code, "failed": gate["failed"]}) if not dry_run: pms_repo.insert_ledger( ts_code=code, action=ins.get("action"), arbiter="rule", verdict="REJECT", price_at=day_ctx.get("price") or 0, hard_numbers=gate["hard_numbers"], failed_checks=gate["failed"], ref_id=ins["instruction_id"], reason="规则闸终检未通过 (宁可不动)") prog["last_decision"] = {"at": now.strftime("%H:%M"), "action": "REJECT", "reason": "; ".join(gate["failed"])} pms_repo.update_instruction(ins["instruction_id"], progress=prog) continue if dry_run: out["fired"].append({"instruction_id": ins["instruction_id"], "code": code, "qty": qty, "limit": d["limit_price"], "reason": d["reason"], "dry_run": True}) continue res = dispatcher.dispatch( instruction_id=_child_id(ins["instruction_id"], len(children) + 1), ts_code=code, side=side, qty=qty, limit_price=d["limit_price"], valid_until=now) if not res.get("ok"): out["errors"].append(f"{ins['instruction_id']} 下发失败: {res.get('error')}") continue children.append({"ymd": ymd_today, "at": now.strftime("%H:%M:%S"), "qty": qty, "limit": d["limit_price"], "mode": res["mode"], "ref": res["ref"], "forced": d.get("forced", False), "reason": d["reason"]}) prog["children"] = children prog.setdefault("dispatched_at", now.strftime("%Y-%m-%d %H:%M:%S")) prog["last_decision"] = {"at": now.strftime("%H:%M"), "action": et.ACT_FIRE, "reason": d["reason"]} pms_repo.update_instruction(ins["instruction_id"], status=ST_DISPATCHED, limit_price=None, dispatch_ref=res["ref"], progress=prog) pms_repo.insert_ledger(ts_code=code, action=ins.get("action"), arbiter="rule", verdict="PASS", price_at=day_ctx.get("price") or 0, hard_numbers={**gate["hard_numbers"], "limit": d["limit_price"], "mode": res["mode"]}, ref_id=ins["instruction_id"], reason=d["reason"]) out["fired"].append({"instruction_id": ins["instruction_id"], "code": code, "qty": qty, "limit": d["limit_price"], "mode": res["mode"], "note": res.get("note"), "reason": d["reason"]}) except Exception as e: logger.exception("出手处理失败 %s", ins.get("instruction_id")) out["errors"].append(f"{ins.get('instruction_id')}: {type(e).__name__}: {e}") out["ok"] = not out["errors"] return out # ================================================================ 窗口收口 def sweep_windows(*, now=None) -> dict: """窗口耗尽的指令收口, 并把成交量回写方案 (命令进度据此结算)。""" now = now or datetime.now() out = {"expired": [], "partial": [], "synced": 0, "errors": []} for ins in pms_repo.list_instructions(statuses=list(LIVE), limit=300): try: prog = dict(ins.get("progress") or {}) exec_qty = int(ins.get("exec_qty") or 0) remaining = int(ins["qty"]) - exec_qty if ins.get("origin_type") == "plan" and ins.get("origin_id"): pms_repo.update_plan(ins["origin_id"], filled_qty=exec_qty, status=PLAN_DONE if remaining <= 0 else None) out["synced"] += 1 if remaining <= 0: pms_repo.update_instruction(ins["instruction_id"], status=ST_CONFIRMED) continue deadline = prog.get("deadline") or "" if not deadline: continue left = td.trade_days_left(deadline, now) v = et.window_verdict(remaining_qty=remaining, tdays_left=left, is_command=bool(prog.get("is_command"))) if v["verdict"] == "RUNNING": continue prog["window_verdict"] = v if v["verdict"] == "EXPIRED": pms_repo.update_instruction(ins["instruction_id"], status=ST_EXPIRED, progress=prog) out["expired"].append(ins["instruction_id"]) else: pms_repo.update_instruction(ins["instruction_id"], progress=prog) out["partial"].append({"instruction_id": ins["instruction_id"], "remaining": remaining, "note": v["note"]}) logger.warning("[窗口耗尽] %s 仍剩 %s 股 —— %s", ins["instruction_id"], remaining, v["note"]) except Exception as e: out["errors"].append(f"{ins.get('instruction_id')}: {type(e).__name__}: {e}") try: command_service.refresh_progress() except Exception as e: out["errors"].append(f"命令进度刷新失败: {e}") return out def cancel_instruction(instruction_id: str, reason: str = "页面人工撤销") -> dict: ins = pms_repo.get_instruction(instruction_id) if not ins: return {"ok": False, "error": "指令不存在"} if ins["status"] not in LIVE: return {"ok": False, "error": f"指令处于 {ins['status']}, 不可撤销"} r = dispatcher.cancel(instruction_id=instruction_id, dispatch_ref=ins.get("dispatch_ref")) pms_repo.update_instruction(instruction_id, status=ST_CANCELLED) pms_repo.insert_ledger(ts_code=ins["ts_code"], action=ins.get("action"), arbiter="user", verdict="REJECT", price_at=0, ref_id=instruction_id, reason=reason) return {"ok": True, "downstream": r, "message": f"指令 {instruction_id} 已撤销"} # ================================================================ 内部 def _pos_of(view: dict, ts_code: str) -> dict: for x in view["positions"]: if x["ts_code"] == ts_code: return x return {"ts_code": ts_code, "total_qty": 0, "avail_qty": 0, "frozen_reason": "NONE"} def _child_id(instruction_id: str, n: int) -> str: return f"{instruction_id}_D{n:02d}" def _limit_up(day: dict) -> bool: """一字板粗判: 当日最高=最低=现价 且 涨幅为正 (无涨跌停价字段时的兜底口径)。""" if not day: return False hi, lo, px = day.get("high"), day.get("low"), day.get("price") chg = day.get("day_chg_from_open") return bool(hi and lo and px and hi == lo == px and (chg or 0) >= 0 and day.get("bars", 0) > 3) def _limit_down(day: dict) -> bool: if not day: return False hi, lo, px = day.get("high"), day.get("low"), day.get("price") chg = day.get("day_chg_from_open") return bool(hi and lo and px and hi == lo == px and (chg or 0) <= 0 and day.get("bars", 0) > 3)