# -*- coding: utf-8 -*- """ 命令服务: 下达 → 校验 → 冲突识别 → 生效/规划 → 进度推进 ========================================================== 设计对应: §3 命令系统、§3.2 降仓全流程、§13 命令冲突由用户裁决。 铁律落点: * 命令至上 —— 命令类动作只过规则闸 (上限/一手/行业), 不送研判闸。 * 先记账后动作 —— 命令与方案先落表, 指令下发由择时执行器另行负责 (下一批)。 * 故障即守成 —— 任何一步失败都返回 ok=False 并保留已落表内容, 不产生新指令。 参数命令的事实源: 全局参数 → pms_runtime_param (ParamStore 读取, 页面即时生效) 个股参数 → pms_command 最新 EFFECTIVE 记录 (设计 §11 原话), 同时投影到 pms_position """ from __future__ import annotations import logging from datetime import datetime from app.core import command_spec as cs from app.core import planner as pl from app.core import tradedays as td from app.repo import pms_repo from app.services import industry, param_store, portfolio logger = logging.getLogger("pms.command") # 在途 (可被撤销/冲突判定) 的指令状态 LIVE_INSTR = ("PROPOSED", "RULE_PASSED", "JUDGE_PASSED", "DISPATCHED") STOCK_PARAM_TYPES = ("FREEZE_STOCK", "UNFREEZE_STOCK", "T0_ENABLE", "T0_DISABLE", "SET_STOP_PRICE", "SET_TARGET_PRICE", "BLACKLIST_ADD", "BLACKLIST_REMOVE", "WHITELIST_ADD", "WHITELIST_REMOVE") # ================================================================ 查询 def active_commands() -> list: """在途任务命令 + 生效中参数命令 (冲突判定与页面「在途」列表共用)。""" return pms_repo.list_commands( statuses=[cs.ST_PENDING, cs.ST_PLANNING, cs.ST_EXECUTING, cs.ST_PARTIAL, cs.ST_EFFECTIVE], limit=300) def effective_stock_params() -> dict: """个股参数命令当前值: {ts_code: {frozen, t0, stop_price, target_price, black, white}}""" rows = pms_repo.list_effective_stock_params(STOCK_PARAM_TYPES) out = {} for r in sorted(rows, key=lambda x: x["id"]): # 旧 → 新, 后者覆盖前者 code = r.get("ts_code") if not code: continue d = out.setdefault(code, {}) t, p = r["cmd_type"], r.get("params") or {} if t == "FREEZE_STOCK": d["frozen"] = True elif t == "UNFREEZE_STOCK": d["frozen"] = False elif t == "T0_ENABLE": d["t0"] = True d["t_ratio"] = p.get("t_ratio") elif t == "T0_DISABLE": d["t0"] = False elif t == "SET_STOP_PRICE": d["stop_price"] = p.get("price") elif t == "SET_TARGET_PRICE": d["target_price"] = p.get("price") elif t == "BLACKLIST_ADD": d["black"] = True elif t == "BLACKLIST_REMOVE": d["black"] = False elif t == "WHITELIST_ADD": d["white"] = True elif t == "WHITELIST_REMOVE": d["white"] = False return out def blacklist() -> set: return {c for c, d in effective_stock_params().items() if d.get("black")} def whitelist() -> set: return {c for c, d in effective_stock_params().items() if d.get("white")} # ================================================================ 下达 def issue(cmd_type: str, params: dict, *, note=None, issued_by="user", force_conflict: bool = False) -> dict: """下达一条命令。返回 {ok, command_id, errors, conflicts, plan}。""" spec = cs.SPECS.get(cmd_type) if not spec: return {"ok": False, "errors": [f"UNKNOWN_CMD: {cmd_type}"]} norm, errors = cs.validate(cmd_type, params or {}) if errors: return {"ok": False, "errors": errors} if spec.get("needs_sector_source") and not industry.ready(): return {"ok": False, "errors": [ "SECTOR_SOURCE_OFF: 行业划分数据源未配置 (PMS_SECTOR_SOURCE 为空), 行业类命令不可用"]} ts_code = norm.get("ts_code") try: conflicts = cs.detect_conflicts({"cmd_type": cmd_type, "ts_code": ts_code}, active_commands()) except Exception as e: logger.warning("冲突检测失败(按无冲突继续): %s", e) conflicts = [] if conflicts and not force_conflict: return {"ok": False, "errors": ["CONFLICT: 与在途命令冲突, 请裁决后重试 " "(确认要并行可带 force=true 重下)"], "conflicts": conflicts} ymd = td.ymd() try: seq = pms_repo.next_command_seq(ymd) command_id = cs.make_command_id(ymd, seq) pms_repo.insert_command(command_id=command_id, cmd_class=spec["cls"], cmd_type=cmd_type, ts_code=ts_code, params=norm, status=cs.ST_PENDING, issued_by=issued_by, note=note) except Exception as e: logger.exception("命令落表失败") return {"ok": False, "errors": [f"DB_ERROR: 命令落表失败 {type(e).__name__}: {e}"]} if spec["cls"] == cs.CLS_PARAM: res = _apply_param(command_id, cmd_type, spec, norm) else: res = plan_command(pms_repo.get_command(command_id)) res.setdefault("command_id", command_id) res["conflicts"] = conflicts return res def _apply_param(command_id: str, cmd_type: str, spec: dict, norm: dict) -> dict: """参数命令: 立即生效 + 旧记录置 SUPERSEDED + (全局)写运行参数 / (个股)投影到账本。""" ts_code = norm.get("ts_code") applied = {} try: if spec["scope"] == "global": key, val = spec["param_key"], norm[spec["value_field"]] r = param_store.set_param(key, val, updated_by="command") if not r.get("ok"): pms_repo.update_command(command_id, status=cs.ST_CANCELLED, note=f"参数写入失败: {r.get('error')}") return {"ok": False, "errors": [r.get("error")]} applied[key] = val else: _project_stock_param(cmd_type, spec, norm, ts_code) applied = dict(norm) pms_repo.supersede_param_commands(cmd_type, ts_code, keep_command_id=command_id) _supersede_opposites(cmd_type, ts_code) pms_repo.update_command(command_id, status=cs.ST_EFFECTIVE, progress={"applied": applied}, done_at=datetime.now()) return {"ok": True, "status": cs.ST_EFFECTIVE, "applied": applied, "plan": None} except Exception as e: logger.exception("参数命令生效失败") pms_repo.update_command(command_id, status=cs.ST_CANCELLED, note=f"生效失败: {type(e).__name__}: {e}") return {"ok": False, "errors": [f"APPLY_ERROR: {type(e).__name__}: {e}"]} _OPPOSITE = {"FREEZE_STOCK": "UNFREEZE_STOCK", "UNFREEZE_STOCK": "FREEZE_STOCK", "T0_ENABLE": "T0_DISABLE", "T0_DISABLE": "T0_ENABLE", "BLACKLIST_ADD": "BLACKLIST_REMOVE", "BLACKLIST_REMOVE": "BLACKLIST_ADD", "WHITELIST_ADD": "WHITELIST_REMOVE", "WHITELIST_REMOVE": "WHITELIST_ADD"} def _supersede_opposites(cmd_type: str, ts_code): opp = _OPPOSITE.get(cmd_type) if opp and ts_code: pms_repo.supersede_param_commands(opp, ts_code) def _project_stock_param(cmd_type: str, spec: dict, norm: dict, ts_code: str): """把个股参数命令投影到 pms_position, 供规则闸速读 (事实源仍是命令表)。""" proj = spec.get("projection") if not proj or not ts_code: return pms_repo.ensure_position(ts_code) fields = {} for k, v in proj.items(): fields[k] = norm.get(str(v)[1:]) if isinstance(v, str) and v.startswith("@") else v if cmd_type == "T0_ENABLE": fields["t0_ratio"] = norm.get("t_ratio") pms_repo.update_position(ts_code, **fields) # ================================================================ 规划 def plan_command(cmd: dict) -> dict: """任务命令 → 方案落表。命令状态推进到 EXECUTING (即时任务直接 DONE)。""" if not cmd: return {"ok": False, "errors": ["命令不存在"]} cmd_type, command_id = cmd["cmd_type"], cmd["command_id"] spec = cs.SPECS.get(cmd_type, {}) p = cmd.get("params") or {} try: pms_repo.update_command(command_id, status=cs.ST_PLANNING) except Exception as e: return {"ok": False, "errors": [f"DB_ERROR: {e}"]} try: result = _dispatch_planner(cmd_type, p, cmd) except Exception as e: logger.exception("方案生成失败 %s", command_id) pms_repo.update_command(command_id, status=cs.ST_CANCELLED, note=f"方案生成失败: {type(e).__name__}: {e}") return {"ok": False, "command_id": command_id, "errors": [f"PLAN_ERROR: {type(e).__name__}: {e}"]} items = result.get("items") or [] window = int(p.get("window_tdays") or param_store.get_int("PMS_EXEC_WINDOW_TDAYS", 3)) deadline = td.window_deadline(datetime.now().date(), window) rows = [] for i, it in enumerate(items, start=1): rows.append({"plan_id": cs.make_plan_id(command_id, i), "command_id": command_id, "ts_code": it["ts_code"], "action": it["action"], "qty": it.get("qty"), "amount": it.get("amount"), "priority": it.get("priority", 100), "deadline": deadline, # GATED = 建仓的回踩补足/盈利加仓批, 不随命令立即执行, # 等动作引擎按条件解锁后才转 PENDING (设计 §6) "status": "GATED" if it.get("gated") else "PENDING", "reason": it.get("reason")}) try: if rows: pms_repo.insert_plans(rows) except Exception as e: logger.exception("方案落表失败 %s", command_id) pms_repo.update_command(command_id, status=cs.ST_CANCELLED, note=f"方案落表失败: {type(e).__name__}: {e}") return {"ok": False, "command_id": command_id, "errors": [f"DB_ERROR: {e}"]} # 撤单类动作立刻执行 (撤销在途买入/全部在途指令) cancel_r = _cancel_marked_instructions(items) cancelled = cancel_r["cancelled"] cancel_failed = cancel_r["failed"] # 命令覆盖冲突 (设计铁律「命令至上」): 全额清仓类命令一下达, 立刻停掉这些票的**买入侧** # —— 撤活跃策略 (网格/做T/跟踪) 与待你确认的买入提议, 别再边卖边买。在途买单已由上面 # _cancel_marked_instructions 撤掉, 这里补策略与提议这两层。 # 由来 (2026-08-19 实机): 一键清仓时 002128 挂着网格, 而"清仓完成清场"撤策略要等持仓 # 归零才触发; 于是那天网格整个上午都在逐档买入、清仓命令同时在卖, 买进来的又是 T+1 # 锁着当天卖不掉 —— 清仓命令在追一个自己还在被买进的持仓。改成命令下达即撤, 不等归零。 stop_r = {} if cmd_type in ("LIQUIDATE_ALL", "EXIT_STOCK", "SECTOR_EXIT"): exit_codes = sorted({it["ts_code"] for it in items if it.get("action") == "EXIT" and it.get("ts_code")}) stop_r = _stop_buyside_for_exit(exit_codes, command_id) # 开关类命令写运行参数。 # **这行返回值绝对不能丢**: HALT_BUY / HALT_ALL 的"刹车"就是这一次写入 —— 规则闸 # 每一跳都去 param_store 读 PMS_GLOBAL_BUY_HALT / PMS_GLOBAL_EXEC_HALT, 参数没写进去 # 就等于刹车没踩。而这类命令 spec.instant=True, 原来无论写没写成都直接置 DONE 并回 # ok=True —— 页面上"全局暂停买入 已完成", 买单照下。宁可让命令报失败, 也不能让用户 # 以为已经停了。(2026-07-31 静默失败专项) switch_err = None if spec.get("switch_key") is not None: w = param_store.set_param(spec["switch_key"], spec.get("switch_value"), "command") or {} if not w.get("ok"): switch_err = f"{spec['switch_key']} 写入失败: {w.get('error')}" logger.error("[命令] %s %s 开关参数没写进去 —— **该命令没有生效**: %s", command_id, cmd_type, switch_err) rejects = result.get("rejects") or [] notes = list(result.get("notes", [])) if switch_err: notes.insert(0, f"⚠ 开关未生效: {switch_err}") # 撤单没撤干净必须顶到 notes 最前面 —— planner 写 notes 的时候还不知道撤单结果, # 它数的是"点名了几条", 真撤掉几条只有这里知道。埋在 cancel_failed 里没人看。 if cancel_failed: notes.insert(0, f"⚠ {len(cancel_failed)} 条在途指令**没撤掉, 仍在下游挂着**: " + "; ".join(f"{x['instruction_id']}({x['why']})" for x in cancel_failed[:5])) if stop_r.get("strategies") or stop_r.get("proposals") or stop_r.get("instructions"): notes.insert(0, f"清仓覆盖冲突: 已撤策略 {len(stop_r['strategies'])} 条、" f"驳回买入提议 {len(stop_r['proposals'])} 条、" f"撤在途买单 {len(stop_r.get('instructions') or [])} 条 (停止边卖边买)") if stop_r.get("errors"): notes.insert(0, f"⚠ 停买入侧有 {len(stop_r['errors'])} 处未成: " + "; ".join(str(x) for x in stop_r["errors"][:5])) progress = {"target_amount": result.get("target_amount", 0.0), "planned_amount": result.get("planned_amount", 0.0), "done_amount": 0.0, "gap": result.get("gap", 0.0), "plan_count": len(rows), "deadline": str(deadline), "notes": notes, "rejects": rejects, # 一行话说清"为什么只有这么少 / 一条都没有"。planner 的 notes 只会说 # "候选与补仓空间不足, 缺口 X 元" —— 那读起来像"没票可买", 而真相往往是 # 有一堆候选、全被同一道闸拒了。不聚合出来就得去翻 rejects 原始清单。 "reject_summary": pl.summarize_rejects(rejects), "cancelled_instructions": cancelled, "cancel_failed": cancel_failed, "stopped_buyside": stop_r or {}} # 零方案的两种情形必须分开 (2026-07-31 修) # ------------------------------------------------------------------ # 开关类命令 (HALT_BUY / RESUME_ALL …) 本来就不产出方案, 那是 DONE。 # 但一条**该**产出方案的任务命令一条都没产出, 它不是"完成", 是"没发生" —— 原来两者 # 都走 DONE, 于是页面显示已完成、`cancel()` 又因为 DONE 不在 ACTIVE_TASK_STATES 里而 # 拒绝撤销, 用户既看不出没执行成、也退不回来。日报统计也会把它算成完成的命令。 # 次序要紧: `instant` 是**命令规格**的属性 (下达即完成), 优先于有没有方案 —— # HALT_BUY 这类开关命令照样会产出撤单动作 (rows 非空), 但它下达完就该是 DONE。 # instant 命令自己失败了也不许算 DONE。`_adjust_window` / `_cancel_target` 靠返回 # result["ok"]=False 报"目标命令不存在 / 该命令不可撤销", 原来这个 ok 全库没人读 —— # 用户看到"撤销命令 已完成", 而目标命令还在跑。开关写不进去同理。 note, errors = None, [] instant_failed = switch_err or (spec.get("instant") and result.get("ok") is False) if instant_failed: status = cs.ST_CANCELLED note = ("命令未生效: " + (switch_err or "; ".join( str(x) for x in (result.get("notes") or ["方案生成器报告失败"]))))[:280] errors = [note] logger.error("[命令] %s %s 未生效 → 置 CANCELLED。%s", command_id, cmd_type, note) elif spec.get("instant"): status = cs.ST_DONE elif rows: status = cs.ST_EXECUTING else: status = cs.ST_CANCELLED note = ("未产出任何方案: " + (progress["reject_summary"] or "候选池为空"))[:280] logger.warning("[命令] %s %s 未产出任何方案 → 置 CANCELLED。%s", command_id, cmd_type, note) pms_repo.update_command(command_id, status=status, progress=progress, note=note, done_at=(datetime.now() if status in (cs.ST_DONE, cs.ST_CANCELLED) else None)) _ledger_rejects(rejects, command_id) return {"ok": not errors, "command_id": command_id, "status": status, "plan": progress, "items": items, "errors": errors} def _dispatch_planner(cmd_type: str, p: dict, cmd: dict) -> dict: """按命令类型调用对应的方案生成器 (纯逻辑在 core/planner.py)。""" view = portfolio.positions_view() sp = view["params"] positions = view["held"] scale = float(sp["scale"] or 0) exclude = _codes_with_live_plans() if cmd_type == "REDUCE_EXPOSURE": # 减仓不撤策略腿在途买单 (exclude_strategy=True): 整体降仓 ≠ 停网格/做T (设计: 清仓 # 才最高优先级、覆盖一切; 减仓看情况、默认不动策略)。非策略的在途买入 (自主/提议/命令) # 照撤 —— 正在减仓就别再往里投新钱。 return pl.plan_reduce_exposure( release_amount=scale * float(p["pct"]), positions=positions, pending_buys=_pending_buys(exclude_strategy=True), params={"weak_neg_days": sp["weak_neg_days"]}, exclude_codes=exclude) if cmd_type == "INCREASE_EXPOSURE": return pl.plan_increase_exposure( add_amount=scale * float(p["pct"]), positions=positions, candidates=_candidates(view), ctx=portfolio.caps_ctx(view), params=sp) if cmd_type in ("HALT_BUY",): return pl.plan_halt_buy(pending_buys=_pending_buys()) if cmd_type in ("HALT_ALL",): return pl.plan_halt_all(pending_instructions=_pending_instructions()) if cmd_type in ("RESUME_BUY", "RESUME_ALL"): return {"ok": True, "items": [], "target_amount": 0.0, "planned_amount": 0.0, "gap": 0.0, "notes": ["开关已恢复"], "rejects": []} if cmd_type == "LIQUIDATE_ALL": return pl.plan_liquidate_all(positions=positions, pending_buys=_pending_buys()) if cmd_type == "SECTOR_EXIT": return pl.plan_sector_exit(sector=p["sector"], positions=positions) if cmd_type == "SECTOR_CAP": r = pl.plan_sector_cap(sector=p["sector"], cap=float(p["cap"]), positions=positions) # 这条命令有两半: 减到线内 (方案) + 把线记下来 (参数)。参数没写上就只减了这一次, # 上限并没有立住 —— 必须说出来, 不能默默只做一半。 w = param_store.set_param(f"PMS_SECTOR_CAP_{p['sector']}", float(p["cap"]), "command") or {} if not w.get("ok"): logger.error("[命令] 行业上限参数没写进去 %s: %s", p["sector"], w.get("error")) r = {**r, "notes": [f"⚠ 本次已减到线内, 但行业上限**没能记下来** " f"({w.get('error')}), 下一轮不会自动守这条线"] + list(r.get("notes") or [])} return r if cmd_type == "OPEN_TARGET": code = p["ts_code"] px = _price_of(view, code) if not px: return {"ok": False, "items": [], "target_amount": 0.0, "planned_amount": 0.0, "gap": 0.0, "notes": [], "rejects": [ {"ts_code": code, "reasons": ["PRICE_MISSING: 取不到现价, 本轮不建仓"]}]} return pl.plan_open_target(ts_code=code, target_pct=float(p["target_pct"]), price=px, ctx=portfolio.caps_ctx(view, ts_code=code), params=sp) if cmd_type == "EXIT_STOCK": return pl.plan_exit_stock(ts_code=p["ts_code"], position=_pos_of(view, p["ts_code"])) if cmd_type == "REDUCE_STOCK": return pl.plan_reduce_stock(ts_code=p["ts_code"], target_pct=float(p["target_pct"]), position=_pos_of(view, p["ts_code"]), scale=scale) if cmd_type == "ADJUST_WINDOW": return _adjust_window(p) if cmd_type == "CANCEL_COMMAND": return _cancel_target(p) return {"ok": False, "items": [], "target_amount": 0.0, "planned_amount": 0.0, "gap": 0.0, "notes": [f"命令 {cmd_type} 暂无对应方案生成器"], "rejects": []} def _adjust_window(p: dict) -> dict: target = pms_repo.get_command(p["target_command_id"]) if not target: return {"ok": False, "items": [], "notes": ["目标命令不存在"], "rejects": [], "target_amount": 0.0, "planned_amount": 0.0, "gap": 0.0} dl = td.window_deadline(datetime.now().date(), int(p["window_tdays"])) pms_repo.set_plans_deadline(target["command_id"], dl) prog = dict(target.get("progress") or {}) prog["deadline"] = str(dl) pms_repo.update_command(target["command_id"], progress=prog) return {"ok": True, "items": [], "target_amount": 0.0, "planned_amount": 0.0, "gap": 0.0, "notes": [f"命令 {target['command_id']} 窗口调整为 {p['window_tdays']} 交易日 " f"(截止 {dl})"], "rejects": []} def _cancel_target(p: dict) -> dict: r = cancel(p["target_command_id"]) return {"ok": r.get("ok", False), "items": [], "target_amount": 0.0, "planned_amount": 0.0, "gap": 0.0, "notes": [r.get("message", "")], "rejects": []} # ================================================================ 撤销与进度 def cancel(command_id: str) -> dict: """撤销在途任务命令: 命令置 CANCELLED, 方案作废, 在途指令撤回 (设计: 在途自主指令自动撤销)。""" cmd = pms_repo.get_command(command_id) if not cmd: return {"ok": False, "message": f"命令 {command_id} 不存在"} if cmd["cmd_class"] != cs.CLS_TASK: return {"ok": False, "message": "参数命令不可撤销, 请下达新的参数命令覆盖"} if cmd["status"] not in cs.ACTIVE_TASK_STATES: return {"ok": False, "message": f"命令处于 {cmd['status']}, 不可撤销"} plans = pms_repo.list_plans(command_id=command_id, statuses=["PENDING", "EXECUTING"]) n_plan = pms_repo.cancel_plans_of_command(command_id) n_ins = 0 plan_ids = {p["plan_id"] for p in plans} for ins in pms_repo.list_instructions(statuses=list(LIVE_INSTR), limit=500): if ins.get("origin_type") == "plan" and ins.get("origin_id") in plan_ids: pms_repo.update_instruction(ins["instruction_id"], status="CANCELLED") n_ins += 1 pms_repo.update_command(command_id, status=cs.ST_CANCELLED, done_at=datetime.now()) return {"ok": True, "message": f"命令 {command_id} 已撤销 (作废方案 {n_plan} 条, " f"撤回指令 {n_ins} 条)"} def plan_pending(limit: int = 20) -> dict: """调度器每分钟一跳: 把 PENDING 的任务命令推进到 EXECUTING (幂等)。 **两个列表都空时要说清是哪一种空** (2026-08-03): 本函数只吃 `PENDING` 的**任务**命令, 它是「把已下达的命令排成方案」的泵, **自己不产生命令**。而 `{"planned": [], "failed": []}` 读起来像"排方案失败了", 实际绝大多数时候是"根本没有命令可排" —— 命令得先由人在页面 命令台下达 (或 POST /api/commands)。参数命令 (总规模、上限这类) 也不产出方案, 它们 是 CLS_PARAM, 压根不进这个查询。同 executor.materialize_plans 的毛病, 一并治。 """ done, errs = [], [] rows = pms_repo.list_commands(statuses=[cs.ST_PENDING], cmd_class=cs.CLS_TASK, limit=limit) for c in rows: r = plan_command(c) (done if r.get("ok") else errs).append(c["command_id"]) out = {"planned": done, "failed": errs, "scanned": len(rows), "note": ""} if not rows: by = {} for st in (cs.ST_PLANNING, cs.ST_EXECUTING, cs.ST_PARTIAL, cs.ST_DONE, cs.ST_CANCELLED): try: by[st] = len(pms_repo.list_commands(statuses=[st], cmd_class=cs.CLS_TASK, limit=200)) except Exception: pass out["commands_by_status"] = by live = sum(v for k, v in by.items() if k in (cs.ST_PLANNING, cs.ST_EXECUTING, cs.ST_PARTIAL)) out["note"] = ( "没有 PENDING 的任务命令可排 —— 这**不是**失败。本步只是把**你已经下达的**命令" "排成方案, 它自己不产生命令。" + (f"当前在途任务命令 {live} 条 (已排过方案, 看 make t-plans)。" if live else "一条在途任务命令都没有 —— 先去页面「命令台」下一条 (升仓/建仓/降仓…), " "或 POST /api/commands。参数命令 (总规模、上限这类) 不产出方案, 不算在内。")) return out def refresh_progress(command_id=None) -> dict: """按方案成交进度刷新命令进度与状态 (窗口末未达标 → PARTIAL)。""" cmds = ([pms_repo.get_command(command_id)] if command_id else pms_repo.list_commands(statuses=[cs.ST_EXECUTING, cs.ST_PARTIAL], cmd_class=cs.CLS_TASK, limit=100)) out = [] today = datetime.now().date() for c in [x for x in cmds if x]: plans = pms_repo.list_plans(command_id=c["command_id"]) done_amt = 0.0 for p in plans: qty, amt, filled = int(p.get("qty") or 0), float(p.get("amount") or 0), \ int(p.get("filled_qty") or 0) if qty > 0 and filled > 0: done_amt += amt * min(1.0, filled / qty) prog = dict(c.get("progress") or {}) prog["done_amount"] = round(done_amt, 2) dl = prog.get("deadline") over = bool(dl) and str(today) > str(dl)[:10] st = cs.settle_task_status(float(prog.get("target_amount") or 0), done_amt, over) if st != c["status"] and cs.can_transition(cs.CLS_TASK, c["status"], st): pms_repo.update_command(c["command_id"], status=st, progress=prog, done_at=datetime.now() if st == cs.ST_DONE else None) else: pms_repo.update_command(c["command_id"], progress=prog) out.append({"command_id": c["command_id"], "status": st, "done_amount": prog["done_amount"], "target_amount": prog.get("target_amount")}) return {"commands": out} # ================================================================ 内部助手 def _pending_buys(exclude_strategy: bool = False) -> list: """在途买入指令 (供 planner 生成撤单项)。 exclude_strategy=True 时**剔除策略腿** (origin_type='strategy') —— 减仓命令用: 整体降仓是把仓位压下来, 不该顺手把你为个股特意设的网格 / 做T 一并掐掉。设计铁律里 清仓 (LIQUIDATE_ALL / EXIT_STOCK / SECTOR_EXIT) 才是最高优先级、覆盖一切 (那几条仍 传默认 False, 策略连同在途买单一律撤); 减仓看情况、默认不动策略。 另一层原因: 减仓只发一次 (命令物化一次), 就算这里撤了策略的在途买单, 策略下一跳 又照发一张 —— 撤了等于没撤, 只是白撤一笔、还在页面上报"撤了 N 条策略买单"误导人。""" rows = pms_repo.list_instructions(statuses=list(LIVE_INSTR), side="buy", limit=500) out = [] for r in rows: if exclude_strategy and r.get("origin_type") == "strategy": continue out.append({"instruction_id": r["instruction_id"], "ts_code": r["ts_code"], "qty": int(r.get("qty") or 0), "amount": float(r.get("qty") or 0) * float(r.get("limit_price") or 0), "side": "buy", "origin_type": r.get("origin_type")}) return out def _pending_instructions() -> list: rows = pms_repo.list_instructions(statuses=list(LIVE_INSTR), limit=500) return [{"instruction_id": r["instruction_id"], "ts_code": r["ts_code"], "qty": int(r.get("qty") or 0), "amount": float(r.get("qty") or 0) * float(r.get("limit_price") or 0), "side": r.get("side")} for r in rows] def _cancel_marked_instructions(items: list) -> dict: """撤销命令点名的在途指令。返回 {"cancelled": [...], "failed": [{id, why}]}。 **必须走 executor.cancel_instruction, 不能只把本端的行标成 CANCELLED。** 2026-07-31 查出来: 原来这里只 `update_instruction(status="CANCELLED")`, 全仓库 `dispatcher.cancel` 在命令这条路上**一次都没被调用过**。切到 ws 之后, 「全局暂停买入」 会回一句「已撤销 N 条」, 而 `pms_qmt_order` 里的子单原封不动继续挂着、继续成交 —— 用户以为踩了刹车, 实际只是本端账面上把它划掉了。HALT_BUY / HALT_ALL / LIQUIDATE_ALL / REDUCE_EXPOSURE / 撤销命令 五条路全中。 失败的要**单独列出来**, 不能吞成 warning 后照样报"已撤销 N 条" —— 那个 N 原来数的是 命令点名的条数, 不是真撤掉的条数。 """ from app.services import executor ok, failed = [], [] for it in items or []: iid = it.get("cancel_instruction_id") if not iid: continue try: r = executor.cancel_instruction(iid, reason="命令撤销在途指令") or {} if r.get("ok"): ok.append(iid) else: failed.append({"instruction_id": iid, "why": r.get("message") or r.get("error") or "撤销未成功"}) except Exception as e: logger.exception("撤销指令失败 %s", iid) failed.append({"instruction_id": iid, "why": f"{type(e).__name__}: {e}"}) if failed: logger.error("[命令] %s 条在途指令没撤掉, **它们仍在下游挂着**: %s", len(failed), failed) return {"cancelled": ok, "failed": failed} def _codes_with_live_plans() -> list: rows = pms_repo.list_plans(statuses=["PENDING", "EXECUTING"], limit=500) return sorted({r["ts_code"] for r in rows}) 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, "price": 0.0} def _price_of(view: dict, ts_code: str): p = _pos_of(view, ts_code) if p.get("price"): return float(p["price"]) from app.services import market return market.get_price(ts_code) def _candidates(view: dict) -> list: """升仓候选池 = 上游选股计划 (∪ 旧买入计划表) ∪ 白名单, 剔除黑名单/已持有。 来源由 `PMS_CANDIDATE_SOURCE` 决定: plan_api (默认, 上游 /plan 接口) / buy_plan (旧 trading_buy_plan 表) / both (并集)。三条口径: 1. **计划不带价格** —— /plan 只回答「买什么、排第几」, 价格由 market.plan_price 现取: 实时价优先, 盘前回落昨收 (db13 存的是当日分钟线, 盘前那张 key 不存在, 2026-07-31 盘前实测前 30 只全部无实时价)。两者都没有才剔除并记 warning。 旧表的 target_price 若有则直接用。 2. **上游拿不到不回退旧表** —— 候选池宁可为空。那张表在目标架构下没有明确写入方, 拿它当事实源比没有候选更危险 (沿用「拿不到 ≠ 通过」的纪律)。 3. **白名单必须压过计划票** —— planner 只认 score 一把尺子, 而计划的 score 是 200+ 量级 (旧表的 prob_thresh 是 0~1)。白名单是用户点名的票, 故给 max(池内分)+1, 写死 1.0 会让点名票沉到池底。 """ from app.services import market, plan_feed held = {x["ts_code"] for x in view["held"]} sp = effective_stock_params() black = {c for c, d in sp.items() if d.get("black")} src = (param_store.get("PMS_CANDIDATE_SOURCE", plan_feed.SRC_PLAN_API) or plan_feed.SRC_PLAN_API).strip() out, seen, noprice, fallback = [], set(), [], [] def _push(c, score, tag, *, price=None, theme=None): if not c or c in held or c in black or c in seen: return px, psrc = float(price or 0), "given" if px <= 0: pp = market.plan_price(c) px, psrc = float(pp["price"] or 0), pp["source"] if px <= 0: noprice.append(c) return if psrc == "prev_close": fallback.append(c) seen.add(c) # sector 取已配置的行业源 (与持仓侧的 sector_*_map 同一套词表); theme 只作展示, # 不参与约束 —— theme 是"传导主题"(事件驱动、天天变、覆盖率约三成), 拿它当行业标签 # 会让集中度约束跟着漂, 已于 2026-07-31 明确不再灌行业表 (见 UPSTREAM_PLAN_API.md §6)。 out.append({"ts_code": c, "price": px, "price_source": psrc, "score": float(score or 0), "sector": industry.get(c), "theme": theme, "src": tag}) if src in (plan_feed.SRC_PLAN_API, plan_feed.SRC_BOTH): try: sel = plan_feed.candidates(held=held, black=black) for r in sel["items"]: _push(r["ts_code"], r["score"], "plan_api", theme=r.get("theme")) logger.info("[候选池] 上游计划 %s: 排序池 %s → 合格 %s → 取 %d 只 (丢弃 %s)", sel.get("date"), sel.get("considered"), sel.get("eligible"), len(sel["items"]), sel.get("dropped")) except plan_feed.PlanFeedError as e: logger.error("[候选池] 上游计划不可用, 计划票一只不进池 (按纪律不回退旧表): %s", e) except Exception as e: logger.exception("[候选池] 上游计划处理异常: %s", e) if src in (plan_feed.SRC_BUY_PLAN, plan_feed.SRC_BOTH): from app.repo import downstream_repo try: for p in downstream_repo.fetch_buy_plans(is_active=7, limit=100): _push(p["ts_code"], p.get("score"), "buy_plan", price=p.get("price")) except Exception as e: logger.warning("[候选池] 读旧买入计划表失败: %s", e) white_score = max([c["score"] for c in out], default=0.0) + 1.0 for c, d in sp.items(): if d.get("white"): _push(c, white_score, "whitelist") if fallback: logger.info("[候选池] %d 只候选无实时价, 已用昨收定量 (盘前正常): %s", len(fallback), fallback[:10]) if noprice: logger.warning("[候选池] %d 只候选实时价与昨收都取不到, 已剔除 (前 10): %s", len(noprice), noprice[:10]) return out def _ledger_rejects(rejects: list, command_id: str): """被上限/行业拦下的候选也要留痕 —— 「拒了的后来涨了多少」是调参核心数据 (设计 §7)。""" for r in rejects or []: try: pms_repo.insert_ledger(ts_code=r.get("ts_code") or "-", action="OPEN", arbiter="rule", verdict="REJECT", price_at=0, failed_checks=r.get("reasons"), ref_id=command_id, reason="命令规划期规则闸拦截") except Exception as e: logger.warning("拒绝留痕失败: %s", e) # ================================================================ 命令覆盖冲突: 清仓即停买入侧 def _stop_buyside_for_exit(codes: list, command_id: str) -> dict: """全额清仓类命令 (一键清仓 / 清仓某股 / 清仓某行业) 下达时, 立刻停掉这些票的买入侧。 三件事: 撤活跃策略 (网格/做T/跟踪) + 驳回待确认的买入提议 + **撤这些票当前在途的买入指令**。 在途买单这一层原来只有一键清仓兜得住 —— 它的 planner 会给所有在途买入排 HALT 撤单项, 经 _cancel_marked_instructions 撤掉。但清仓某股 (EXIT_STOCK) / 清仓某行业 (SECTOR_EXIT) 的 planner **不传 pending_buys**, 只排一条卖出, 于是「撤策略」只改了策略状态、**已经下发到 券商的那张网格/做T 买单还挂着继续成交** —— 你正清仓这只票, 它却又买进来一笔 T+1 锁死的货。 所以这里对被清仓的票再补一道: 直接撤掉当前在途买入指令。一键清仓时这些单多半已被 HALT 撤过、不在在途集里, 补这道也只是空跑, 幂等无害。 幂等、不抛异常: 单只失败记进 errors, 不拖垮命令本体 (故障即守成)。撤策略只改策略状态、 撤在途买单走 executor.cancel_instruction (下游没撤成不硬标终态), 都不动持仓, 安全。""" out = {"strategies": [], "proposals": [], "instructions": [], "errors": []} codeset = {c for c in (codes or []) if c} if not codeset: return out from app.services import strategy_service try: strategies = [s for s in pms_repo.list_strategies(statuses=["ACTIVE", "PAUSED"], limit=500) if s.get("ts_code") in codeset] except Exception as e: strategies = [] out["errors"].append(f"读策略失败: {type(e).__name__}: {e}") for s in strategies: try: r = strategy_service.set_status(s["strategy_id"], "CANCELLED", by="command") or {} if r.get("ok"): out["strategies"].append(s["strategy_id"]) try: pms_repo.insert_ledger( ts_code=s["ts_code"], action="CLEANUP", arbiter="rule", verdict="PASS", price_at=0, ref_id=command_id, hard_numbers={"strategy_id": s["strategy_id"], "command_id": command_id}, reason="清仓命令下达: 撤该票策略, 停止边卖边买 (命令覆盖冲突)") except Exception as e: logger.warning("[命令] 撤策略留痕失败 %s: %s", s.get("ts_code"), e) else: out["errors"].append(f"{s.get('ts_code')} 撤策略未成: {r.get('error')}") except Exception as e: out["errors"].append(f"{s.get('ts_code')} 撤策略失败: {type(e).__name__}: {e}") try: props = [p for p in pms_repo.list_proposals(statuses=("WAIT_USER",), limit=500) if p.get("ts_code") in codeset and p.get("action") in BUILD_ACTIONS] except Exception as e: props = [] out["errors"].append(f"读提议失败: {type(e).__name__}: {e}") for p in props: try: if pms_repo.decide_proposal(p["proposal_id"], "DECLINED"): out["proposals"].append(p["proposal_id"]) except Exception as e: out["errors"].append(f"{p.get('ts_code')} 驳回买入提议失败: {type(e).__name__}: {e}") # 第三层: 撤这些票当前在途的买入指令 (EXIT_STOCK/SECTOR_EXIT 的 planner 不排 HALT, 全靠这里)。 # 只查当前在途买入, 撤过的/已终态的不会回到这个集里, 所以一键清仓已 HALT 掉的这里空跑、不重撤。 from app.services import executor try: live_buys = [i for i in pms_repo.list_instructions(statuses=list(LIVE_INSTR), side="buy", limit=500) if i.get("ts_code") in codeset] except Exception as e: live_buys = [] out["errors"].append(f"读在途买单失败: {type(e).__name__}: {e}") for i in live_buys: iid = i.get("instruction_id") try: r = executor.cancel_instruction(iid, reason="清仓命令下达: 撤该票在途买入 (命令覆盖冲突)") or {} if r.get("ok"): out["instructions"].append(iid) else: # 下游没撤成: 指令仍在途, 单独列出来别吞 (与 _cancel_marked_instructions 同纪律) out["errors"].append(f"{i.get('ts_code')} 撤在途买单 {iid} 未成: {r.get('error') or r.get('message')}") except Exception as e: out["errors"].append(f"{i.get('ts_code')} 撤在途买单 {iid} 失败: {type(e).__name__}: {e}") if out["strategies"] or out["proposals"] or out["instructions"]: logger.warning("[命令] %s 清仓覆盖冲突: 撤策略 %s, 驳回买入提议 %s, 撤在途买单 %s", command_id, out["strategies"], out["proposals"], out["instructions"]) return out # ================================================================ 清仓完成后清场 CLEANUP_MARK_KEY = "PMS_EXIT_CLEANUP_DONE" # 已闭仓且已清场的代码集 (JSON) BUILD_ACTIONS = ("OPEN", "FILL", "ADD", "DCA") # 建仓类方案/指令 (买入) def _cleanup_todo(closed_codes, done_codes) -> tuple: """纯逻辑: 算出本轮要清场的代码与新的标记集 (2026-08-18, 便于单测)。 每只票在**一次闭仓周期里只清一次**: todo = 现在闭仓的 − 已清过的。 新标记 = 现在仍闭仓的那部分 —— 重新建仓的票会掉出闭仓集、从而掉出标记, 下次它再闭仓时会被重新清一遍。这条"只清一次"是防呆的关键: 闭仓后你若又下建仓命令, 持仓状态短暂还是 CLOSED, 标记挡住清场、不会把你刚下的建仓计划误撤。 """ closed = set(closed_codes or ()) done = set(done_codes or ()) todo = sorted(closed - done) new_done = closed # 只保留仍闭仓的; 重开的自然移除 return todo, new_done def _load_cleanup_done() -> set: try: raw = pms_repo.get_param(CLEANUP_MARK_KEY) import json return set(json.loads(raw)) if raw else set() except Exception: return set() def _save_cleanup_done(codes: set): try: import json pms_repo.set_param(CLEANUP_MARK_KEY, json.dumps(sorted(codes)), "system") except Exception as e: logger.warning("[清场] 标记写入失败: %s", e) def cleanup_exited_positions(limit: int = 500) -> dict: """持仓归零(清仓完成)后清场: 撤该股策略 / 在途建仓计划与买单 / 相关提议 (设计 2026-08-18)。 每分钟一跳, 幂等且防呆: * 只对 status=CLOSED 且 total_qty<=0 的持仓动手; * 每次闭仓只清一次 (靠 _cleanup_todo 的标记), 避免把闭仓后新下的建仓命令误撤; * 四类残留一次性清: 策略(网格/做T/跟踪) 撤下、在途建仓方案作废、在途买单撤回、 挂着的相关提议驳回。清场动作在账本留一行 (action=CLEANUP)。 你清仓离场后, 网格不再空转挂着、也不会因持仓被任何原因重建而复活逢跌买入。 """ from app.services import executor, strategy_service out = {"scanned": 0, "cleaned": [], "errors": [], "note": ""} try: positions = pms_repo.list_positions() except Exception as e: return {**out, "errors": [f"读持仓失败: {type(e).__name__}: {e}"]} closed = {p["ts_code"] for p in positions if str(p.get("status")) == "CLOSED" and int(p.get("total_qty") or 0) <= 0} done = _load_cleanup_done() todo, new_done = _cleanup_todo(closed, done) out["scanned"] = len(closed) if not todo: _save_cleanup_done(new_done) # 重开的票及时移出标记 out["note"] = "无新闭仓待清场" if closed else "当前无闭仓持仓" return out todo_set = set(todo) # 各类残留一次性拉取, 再按待清代码过滤 (省得逐票查库) try: strategies = [s for s in pms_repo.list_strategies(statuses=["ACTIVE", "PAUSED"], limit=limit) if s.get("ts_code") in todo_set] except Exception as e: strategies = [] out["errors"].append(f"读策略失败: {e}") try: plans = [pl for pl in pms_repo.list_plans(statuses=["PENDING", "GATED", "EXECUTING"], limit=limit) if pl.get("ts_code") in todo_set and pl.get("action") in BUILD_ACTIONS] except Exception as e: plans = [] out["errors"].append(f"读方案失败: {e}") try: buys = [i for i in pms_repo.list_instructions(statuses=list(LIVE_INSTR), side="buy", limit=limit) if i.get("ts_code") in todo_set] except Exception as e: buys = [] out["errors"].append(f"读指令失败: {e}") try: props = [pr for pr in pms_repo.list_proposals(statuses=("WAIT_USER",), limit=limit) if pr.get("ts_code") in todo_set] except Exception as e: props = [] out["errors"].append(f"读提议失败: {e}") per_code = {c: {"ts_code": c, "strategies": [], "plans": [], "instructions": [], "proposals": []} for c in todo} for s in strategies: try: r = strategy_service.set_status(s["strategy_id"], "CANCELLED", by="system") if r.get("ok") or r.get("status"): per_code[s["ts_code"]]["strategies"].append(s["strategy_id"]) except Exception as e: out["errors"].append(f"{s.get('ts_code')} 撤策略失败: {e}") for pl in plans: try: pms_repo.update_plan(pl["plan_id"], status="CANCELLED") per_code[pl["ts_code"]]["plans"].append(pl["plan_id"]) except Exception as e: out["errors"].append(f"{pl.get('ts_code')} 撤建仓方案失败: {e}") for ins in buys: try: r = executor.cancel_instruction(ins["instruction_id"], reason="清仓完成清场: 撤在途建仓") or {} if r.get("ok"): per_code[ins["ts_code"]]["instructions"].append(ins["instruction_id"]) else: out["errors"].append(f"{ins.get('ts_code')} 撤买单未成: {r.get('message') or r.get('error')}") except Exception as e: out["errors"].append(f"{ins.get('ts_code')} 撤买单失败: {e}") for pr in props: try: if pms_repo.decide_proposal(pr["proposal_id"], "DECLINED"): per_code[pr["ts_code"]]["proposals"].append(pr["proposal_id"]) except Exception as e: out["errors"].append(f"{pr.get('ts_code')} 撤提议失败: {e}") for c, acted in per_code.items(): if any(acted[k] for k in ("strategies", "plans", "instructions", "proposals")): out["cleaned"].append(acted) try: pms_repo.insert_ledger(ts_code=c, action="CLEANUP", arbiter="system", verdict="PASS", price_at=0, hard_numbers=acted, reason="清仓完成清场: 撤策略/在途建仓/相关提议") except Exception as e: logger.warning("[清场] %s 留痕失败: %s", c, e) logger.warning("[清场] %s 清仓完成, 已撤 策略%d/方案%d/买单%d/提议%d", c, len(acted["strategies"]), len(acted["plans"]), len(acted["instructions"]), len(acted["proposals"])) _save_cleanup_done(new_done) # 标记本轮闭仓集 (含刚清过的), 只清一次 return out