diff --git a/README.md b/README.md
index 24d36b0..8c9fb0f 100644
--- a/README.md
+++ b/README.md
@@ -25,6 +25,8 @@ app/
command_spec.py 命令目录(A/B/C 全量 27 类) / 参数校验 / 双状态机 / 冲突识别
planner.py 方案生成器: 降仓凑额四档 / 升仓 / 建仓 / 清仓 / 行业 / 撤单
recon.py 成交认领与入账映射 / 对账差异与修正 / 除权检测 / T+1 可用量
+ rule_gate.py 规则闸终检: 上限/一手/可卖/冻结/刹车/行业/不追高 (减持只放行不阻拦)
+ exec_timing.py 择时实现B: 分日配额 / 分笔 / 买卖出手判定 / 14:45 兜底 / 窗口收口
tradedays.py 交易日历: 调度守卫与执行窗口计算
db/session.py 三库连接 + **严格单表访问守卫** (JOIN/逗号连表/跨表子查询一律拒绝)
repo/ 单表数据访问: pms_repo (自有 10 表) / downstream_repo (下游只读三表)
@@ -32,6 +34,8 @@ app/
param_store.py 运行参数中心 (表值优先于 settings 初值, 页面调参即时生效)
portfolio.py 组合快照 (账本+行情+行业 → 方案/规则闸/页面的统一输入)
command_service.py 命令下达→校验→冲突→生效/规划→进度推进→撤销
+ executor.py 方案→指令→分日出手→窗口收口 (规则闸与择时的编排落点)
+ dispatcher.py 下发通道三适配器: shadow(默认) / plan_x / channel_y
ledger_service.py 成交回放 / 对账 / 除权 / 盘前 / 日终结算 / 运营日报
market.py 行情 (Redis db13) 与参考位 (决策系统主口径 + 兜底自算)
industry.py 行业划分可插拔适配器 (custom_table / gp_stock_category / 停用)
@@ -41,7 +45,8 @@ scripts/
run_tests.py 一次跑完全部单测 (见下方「Docker 部署」)
test_core_units.py 仓位与安全垫核心逻辑 14 例
test_batch2_units.py 命令 / 方案 / 回放对账 纯逻辑 35 例
- test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 18 例
+ test_batch3_units.py 规则闸 / 择时执行器实现B 纯逻辑 18 例
+ test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 25 例
init_db.py 建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练)
check_db.py 实机连通性与表结构自检 (需真实 .env)
```
@@ -91,7 +96,7 @@ git pull && docker compose build && docker compose up -d
| 盘前准备 | 交易日 08:50 | T+1 可卖重置 / 参考位取数 / 刹车结算 | ✅ |
| 命令轮询 | 每 1 分钟（全天） | 新命令解析 → 方案生成 → 状态机推进 | ✅ |
| 成交回放 | 交易时段每 5 分钟 | `trading_order` 增量回放 + 盘中轻对账 | ✅ |
-| 盘中执行 | 交易时段每 1 分钟 | 择时出手 + 自主提议扫描 | 🔜 下一批（当前只做在途指令超时收口） |
+| 盘中执行 | 交易时段每 1 分钟 | 方案转指令 → 择时出手（规则闸终检 → 下发 → 记子单） | ✅（自主提议扫描待下一批） |
| 信号消化 | 交易时段每 1 分钟 | 订阅决策系统盘中信号 | 🔜 下一批 |
| T 仓平回 | 14:50 | 做T强制平回 | 🔜 二期（现只自证 T 仓为 0） |
| 日终结算 | 15:10 | 除权检测 / 全量对账 / 安全垫 / 命令进度日结 | ✅ |
@@ -99,11 +104,21 @@ git pull && docker compose build && docker compose up -d
调度器三条守卫：交易日守卫、故障即守成（任务内异常吞掉记 ERROR，绝不因调度异常产生新指令）、全局暂停执行（休假模式下除对账与日报外全部跳过）。
+## 指令下发通道（设计 §9，参数 `PMS_DISPATCH_MODE`，默认 `shadow`）
+
+| 模式 | 行为 | 什么时候用 |
+|---|---|---|
+| `shadow`（默认） | 指令照常过规则闸、照常置 DISPATCHED，但**不写下游**。你在 QMT 侧人工执行，成交由回放按 FIFO 认领回账本 | 通道协商完成前的一期口径（设计 §9：命令类降仓/清仓由用户人工执行、PMS 记账跟踪） |
+| `plan_x` | 买入写 `trading_buy_plan`（`is_active=6` 待挂单、署名 `approved_by='pms'`）；**卖出无对应通道，自动退回影子** | QMT 侧确认沿用旧通道过渡时 |
+| `channel_y` | 写统一指令表 `pms_order_request`（DDL 见需求清单 B1） | B1 协商落地、表建好之后 |
+
+影子模式下的完整闭环：页面下命令 → 方案落表 → 方案转指令 → 择时按日配额给出「今天该出多少、什么价」→ 你照着在 QMT 下单 → 5 分钟一次的回放把成交认领回批次账本 → 命令进度自动推进。整条链路除了「人手下单」这一步，其余与实盘接管后完全一致。
+
## 已实现 / 待开发
-**已实现**：建表 DDL；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；管理页面四块 + 运维/日报抽屉；调度器骨架；单测 67 例。
+**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；**规则闸终检**；**择时执行器实现 B**（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口）；**三模式下发通道**；管理页面四块 + 运维/日报抽屉；调度器八个调度位；单测 92 例。
-**待开发（下一批）**：择时执行器（实现 B 内置保守择时 + 分日配额 + 子指令）、规则闸终检与指令下发通道、动作引擎（FILL/ADD/DCA/TRIM 自主提议扫描）、决策系统信号订阅与研判闸对接、T0 做T（二期）。
+**待开发（下一批）**：动作引擎（FILL/ADD/DCA/TRIM 自主提议扫描 + 提议入队）、决策系统信号订阅（风控 SELL / 止盈 / 反转 → 卖出方案）、研判闸对接与择时实现 A、T0 做T（二期）。
**待外部协商**：`QMT_INTERFACE_REQUIREMENTS.md` 的 A/B/C/D 各项——尤其 A1（`trading_position` 完整 DDL 与可用数量列）、A2（`trading_order` 状态枚举与**来源标识**）、B1（统一指令通道）。在来源标识到位前，回放按「同股同向 + 下发早于成交 + FIFO」贪心认领，认领不上即判外部成交并告警；持仓数量列用候选名探测，探测结果可经页面「运维 → 导出下游表结构」查看，也是回填 D1 的现成材料。
diff --git a/app/core/exec_timing.py b/app/core/exec_timing.py
new file mode 100644
index 0000000..721d275
--- /dev/null
+++ b/app/core/exec_timing.py
@@ -0,0 +1,177 @@
+# -*- coding: utf-8 -*-
+"""
+择时执行器 · 实现B「内置保守择时」(纯逻辑, 无外部依赖, 可单测)
+================================================================
+设计 POSITION_MGMT_DESIGN.md §8。实现A (委托决策系统盘中择时研判) 走同一个
+`decide()` 接口, 研判接通后在 services 层换实现即可, 本模块是兜底也是一期主力。
+
+规则原文与落点:
+ 每日配额 = 剩余量 ÷ 剩余窗口天数, 向上取整到一手 → daily_quota()
+ 卖出: 避开开盘 30 分钟; 现价 ≥ 当日均价(VWAP) 时分笔卖 → decide() side=sell
+ 14:45 未完成 → 现价 × 0.998 限价兜底
+ 买入: 现价 ≤ 当日均价或进入回踩带时买; 当日涨幅 > 5% 停止买入 (不追高)
+ 窗口末日 14:45 强制限价完成或按命令属性作废
+ 停牌/一字板当日跳过顺延; 窗口耗尽未完成 → 命令置「部分完成」并告警
+
+时点一律用「分钟数」比较 (hm_to_min), 免去跨时区与字符串比较的坑。
+"""
+from __future__ import annotations
+
+from app.core.sizer import LOT
+
+OPEN_MIN = 9 * 60 + 30 # 09:30 开盘
+CLOSE_MIN = 15 * 60 # 15:00 收盘
+LUNCH_START = 11 * 60 + 30
+LUNCH_END = 13 * 60
+
+# decide() 的动作词
+ACT_FIRE = "FIRE" # 出手
+ACT_WAIT = "WAIT" # 条件未到, 本轮不动
+ACT_SKIP = "SKIP" # 当日跳过 (停牌/一字板), 顺延
+ACT_STOP = "STOP" # 本日不再出手 (如买入触发不追高)
+
+
+def hm_to_min(hm) -> int:
+ """"14:45" / datetime / (h, m) → 当日分钟数。"""
+ if hm is None:
+ return 0
+ if isinstance(hm, (tuple, list)):
+ return int(hm[0]) * 60 + int(hm[1])
+ if hasattr(hm, "hour"):
+ return hm.hour * 60 + hm.minute
+ s = str(hm).strip()
+ if ":" in s:
+ h, m = s.split(":")[:2]
+ return int(h) * 60 + int(m)
+ return int(s)
+
+
+def in_session(now_min: int) -> bool:
+ return (OPEN_MIN <= now_min <= LUNCH_START) or (LUNCH_END <= now_min <= CLOSE_MIN)
+
+
+def daily_quota(remaining_qty: int, tdays_left: int, *, total_qty: int = None,
+ lot: int = LOT, allow_odd_tail: bool = False) -> int:
+ """当日配额 = 剩余量 ÷ 剩余交易日, 向上取整到一手。
+
+ * 最后一日 (tdays_left ≤ 1) 或剩余不足一手 → 全部剩余 (含零股尾巴)。
+ * allow_odd_tail=True (整票清仓) 时不做整百取整, 零股一并出。
+ """
+ r = max(0, int(remaining_qty or 0))
+ if r <= 0:
+ return 0
+ d = max(1, int(tdays_left or 1))
+ if d <= 1 or r <= lot:
+ return r
+ raw = r / d
+ q = int(-(-raw // lot)) * lot # 向上取整到一手
+ q = min(q, r)
+ if not allow_odd_tail and q % lot and q != r:
+ q = (q // lot) * lot
+ # 若本次取整后剩下不足一手的尾巴, 并进本次一起出, 免得最后一天剩 30 股卡住
+ if 0 < r - q < lot:
+ q = r
+ return max(q, 0) if q > 0 else r
+
+
+def slice_qty(quota: int, slices: int = 1, lot: int = LOT) -> list:
+ """把当日配额切成 N 笔 (设计「分笔卖出配额」)。最后一笔兜底吃掉余数。"""
+ quota = max(0, int(quota or 0))
+ n = max(1, int(slices or 1))
+ if quota <= 0:
+ return []
+ if n == 1 or quota <= lot:
+ return [quota]
+ per = int(quota / n / lot) * lot
+ if per <= 0:
+ return [quota]
+ out = [per] * (n - 1)
+ out.append(quota - per * (n - 1))
+ return [q for q in out if q > 0]
+
+
+def decide(*, side: str, now, day: dict, params: dict, is_last_day: bool,
+ fired_today: int = 0, quota: int = 0) -> dict:
+ """单条指令在「此刻」该不该出手。
+
+ day: {price, vwap, halted, limit_up, limit_down, day_chg_from_open, support}
+ params: {sell_avoid_open_min, buy_halt_dayup, eod_force_time, eod_force_discount}
+ 返回 {"action", "qty_hint", "limit_price", "reason", "forced"}
+ """
+ now_min = hm_to_min(now)
+ price = float(day.get("price") or 0)
+ vwap = float(day.get("vwap") or 0)
+ eod_min = hm_to_min(params.get("eod_force_time") or "14:45")
+ disc = float(params.get("eod_force_discount") or 0.998)
+ left = max(0, int(quota) - int(fired_today))
+
+ def out(action, reason, limit=None, forced=False):
+ return {"action": action, "qty_hint": left, "limit_price": limit,
+ "reason": reason, "forced": forced}
+
+ if price <= 0:
+ return out(ACT_SKIP, "取不到现价, 当日跳过")
+ if day.get("halted"):
+ return out(ACT_SKIP, "停牌, 当日跳过顺延")
+ if left <= 0:
+ return out(ACT_WAIT, "当日配额已出完")
+ if not in_session(now_min):
+ return out(ACT_WAIT, "非交易时段")
+
+ if side == "sell":
+ if day.get("limit_down") and not is_last_day:
+ return out(ACT_SKIP, "跌停一字板, 当日跳过顺延")
+ avoid = int(params.get("sell_avoid_open_min") or 30)
+ if now_min < OPEN_MIN + avoid:
+ return out(ACT_WAIT, f"避开开盘 {avoid} 分钟 (至 {_fmt(OPEN_MIN + avoid)})")
+ if now_min >= eod_min:
+ return out(ACT_FIRE, f"{_fmt(eod_min)} 兜底: 限价 = 现价×{disc}",
+ limit=round(price * disc, 2), forced=True)
+ if vwap > 0 and price >= vwap:
+ return out(ACT_FIRE, f"现价 {price} ≥ 当日均价 {vwap}, 分笔卖出配额",
+ limit=round(price * disc, 2))
+ return out(ACT_WAIT, f"现价 {price} < 当日均价 {vwap or '—'}, 等更好的价")
+
+ if side == "buy":
+ if day.get("limit_up"):
+ return out(ACT_SKIP, "涨停一字板, 当日跳过顺延")
+ dayup = day.get("day_chg_from_open")
+ cap = float(params.get("buy_halt_dayup") or 0.05)
+ if dayup is not None and float(dayup) > cap:
+ return out(ACT_STOP, f"当日涨幅 {float(dayup):.2%} > {cap:.0%}, 停止买入 (不追高)")
+ premium = round(2 - disc, 4) # 买入兜底与卖出对称: 0.998 → 1.002
+ if now_min >= eod_min:
+ if is_last_day:
+ return out(ACT_FIRE, f"窗口末日 {_fmt(eod_min)} 强制完成: 限价 = 现价×{premium}",
+ limit=round(price * premium, 2), forced=True)
+ return out(ACT_WAIT, f"{_fmt(eod_min)} 后不新开买单, 顺延次日")
+ support = float(day.get("support") or 0)
+ if vwap > 0 and price <= vwap:
+ return out(ACT_FIRE, f"现价 {price} ≤ 当日均价 {vwap}, 买入配额",
+ limit=round(price * premium, 2))
+ if support > 0 and price <= support * 1.01:
+ return out(ACT_FIRE, f"现价 {price} 进入回踩带 (支撑 {support}), 买入配额",
+ limit=round(price * premium, 2))
+ return out(ACT_WAIT, f"现价 {price} > 当日均价 {vwap or '—'} 且未回踩, 等回调")
+
+ return out(ACT_SKIP, f"方向 {side!r} 非法")
+
+
+def _fmt(m: int) -> str:
+ return f"{m // 60:02d}:{m % 60:02d}"
+
+
+def window_verdict(*, remaining_qty: int, tdays_left: int, is_command: bool) -> dict:
+ """窗口耗尽时的收口 (设计 §8 末句)。
+
+ 还有剩余且窗口已尽 → 命令置「部分完成」并告警; 命令类保留人工兜底提示,
+ 自主类直接作废。
+ """
+ if int(remaining_qty or 0) <= 0:
+ return {"verdict": "DONE", "note": "已足额完成"}
+ if int(tdays_left or 0) > 0:
+ return {"verdict": "RUNNING", "note": f"窗口内剩余 {tdays_left} 交易日"}
+ return {"verdict": "PARTIAL" if is_command else "EXPIRED",
+ "note": f"窗口耗尽仍剩 {remaining_qty} 股 —— "
+ + ("命令置部分完成并告警, 请在页面决定顺延或人工完成"
+ if is_command else "自主指令作废")}
diff --git a/app/core/rule_gate.py b/app/core/rule_gate.py
new file mode 100644
index 0000000..07a92c5
--- /dev/null
+++ b/app/core/rule_gate.py
@@ -0,0 +1,148 @@
+# -*- coding: utf-8 -*-
+"""
+规则闸 · 一级关口 (纯逻辑, 无外部依赖, 可单测)
+================================================
+设计 POSITION_MGMT_DESIGN.md §7:
+ 「全指令必过, PMS 纯代码。下发前用最新数据终检 —— 命令参数上限、一手检查、
+ T+1 可卖、冻结状态、刹车状态、行业集中度、(买入)不追高。
+ 任一不过或必要输入缺失 → 拒绝 (宁可不动), 未通过项落账本。」
+
+两条容易搞反的口径, 在此钉死:
+ 1. **冻结与刹车只挡增持, 不挡减持** —— 卖出、止损任何时候都放行 (设计 §3.1/§5)。
+ 2. **命令驱动不受刹车限制** —— 刹车是给自主提议踩的, 用户命令优先, 只提示不拦
+ (设计 §5「命令类不受限, 执行前提示」)。上限/一手/可卖量这类硬约束则一视同仁。
+
+输出结构固定为 {"passed", "failed", "warnings", "hard_numbers"}, 直接喂 pms_action_ledger
+的 failed_checks_json / hard_numbers_json —— 「拒了的后来涨了多少」靠这份留痕做判分。
+"""
+from __future__ import annotations
+
+from app.core.sizer import LOT
+from app.core.planner import check_all_caps
+
+BUY, SELL = "buy", "sell"
+
+
+def _num(v, default=0.0):
+ try:
+ return float(v)
+ except (TypeError, ValueError):
+ return default
+
+
+def check(*, side: str, action: str, qty: int, price: float, ctx: dict) -> dict:
+ """指令下发前终检。
+
+ ctx 由调用方备齐 (services/executor.py):
+ position {total_qty, avail_qty, frozen_reason} 该股账本快照
+ caps check_all_caps 所需组合上下文 (仅买入用)
+ day {price, vwap, ma5, day_chg_from_open, halted, limit_up, limit_down}
+ params {no_chase_ma5, buy_halt_dayup, sector_source_ready}
+ flags {buy_halt, exec_halt, brake_active, blacklisted, is_command}
+ """
+ pos = ctx.get("position") or {}
+ day = ctx.get("day") or {}
+ prm = ctx.get("params") or {}
+ flg = ctx.get("flags") or {}
+ is_cmd = bool(flg.get("is_command"))
+ side = str(side or "").lower()
+
+ failed, warns = [], []
+ qty = int(qty or 0)
+ price = _num(price)
+ total_qty = int(pos.get("total_qty") or 0)
+ avail_qty = int(pos.get("avail_qty") or 0)
+
+ hard = {"side": side, "action": action, "qty": qty, "price": price,
+ "total_qty": total_qty, "avail_qty": avail_qty,
+ "frozen_reason": pos.get("frozen_reason") or "NONE",
+ "day_chg_from_open": day.get("day_chg_from_open"),
+ "vwap": day.get("vwap"), "ma5": day.get("ma5"), "is_command": is_cmd}
+
+ # ---- 通用: 必要输入缺失一律拒绝 (宁可不动) ----
+ if qty <= 0:
+ failed.append(f"QTY_INVALID: 数量 {qty} 非法")
+ if price <= 0:
+ failed.append("PRICE_MISSING: 取不到现价, 无法定限价与校验 (宁可不动)")
+ if day.get("halted"):
+ failed.append("HALTED: 停牌, 当日跳过顺延")
+ if flg.get("exec_halt"):
+ failed.append("EXEC_HALT: 全局暂停执行 (休假模式) 生效中")
+
+ if side == SELL:
+ _check_sell(failed, warns, qty, total_qty, avail_qty, day)
+ elif side == BUY:
+ _check_buy(failed, warns, qty, price, ctx, pos, day, prm, flg, is_cmd, hard)
+ else:
+ failed.append(f"SIDE_INVALID: 方向 {side!r} 非法")
+
+ return {"passed": not failed, "failed": failed, "warnings": warns, "hard_numbers": hard}
+
+
+def _check_sell(failed, warns, qty, total_qty, avail_qty, day):
+ """减持方向: 冻结/刹车/上限一概不拦, 只看「卖得出去吗」。"""
+ if qty > total_qty:
+ failed.append(f"OVER_SELL: 卖出 {qty} > 持仓 {total_qty}")
+ elif qty > avail_qty:
+ failed.append(f"T1_UNAVAILABLE: 卖出 {qty} > T+1 可卖 {avail_qty} (当日买入次日才可卖)")
+ # 清仓允许卖零股 (A股规则); 部分减持必须整百
+ if qty % LOT != 0 and qty != total_qty:
+ failed.append(f"LOT_INVALID: 部分减持 {qty} 股非整百 (零股只能在清仓时一次性卖出)")
+ if day.get("limit_down"):
+ warns.append("LIMIT_DOWN: 跌停封板, 大概率成交不了, 已按限价挂出")
+
+
+def _check_buy(failed, warns, qty, price, ctx, pos, day, prm, flg, is_cmd, hard):
+ """增持方向: 全部硬约束都要过。"""
+ if qty % LOT != 0:
+ failed.append(f"LOT_INVALID: 买入 {qty} 股非整百")
+ if (pos.get("frozen_reason") or "NONE") != "NONE":
+ failed.append(f"FROZEN: 该股处于 {pos['frozen_reason']}, 禁止增持")
+ if flg.get("buy_halt"):
+ failed.append("BUY_HALT: 全局暂停买入生效中")
+ if flg.get("blacklisted"):
+ failed.append("BLACKLIST: 该股在黑名单 (永不买入)")
+ if flg.get("brake_active"):
+ if is_cmd:
+ warns.append("BRAKE_ACTIVE: 组合刹车中 —— 命令驱动不受限, 仅提示")
+ else:
+ failed.append("BRAKE_ACTIVE: 组合刹车中, 自主增持暂停")
+
+ if day.get("limit_up"):
+ failed.append("LIMIT_UP: 涨停封板, 不追买")
+
+ # 不追高: 当日涨幅 与 距 MA5 两道
+ dayup = day.get("day_chg_from_open")
+ cap_dayup = _num(prm.get("buy_halt_dayup"), 0.05)
+ if dayup is not None and _num(dayup) > cap_dayup:
+ failed.append(f"NO_CHASE_DAYUP: 当日涨幅 {_num(dayup):.2%} > 上限 {cap_dayup:.0%}")
+ ma5 = _num(day.get("ma5"))
+ cap_ma5 = _num(prm.get("no_chase_ma5"), 0.06)
+ if ma5 > 0:
+ gap = price / ma5 - 1
+ hard["ma5_gap"] = round(gap, 4)
+ if gap > cap_ma5:
+ failed.append(f"NO_CHASE_MA5: 距 MA5 {gap:.2%} > 上限 {cap_ma5:.0%}")
+ else:
+ warns.append("MA5_MISSING: 取不到 MA5, 距均线一项未校验")
+
+ caps = ctx.get("caps")
+ if caps:
+ bad = check_all_caps(ts_code=ctx.get("ts_code") or "-", add_amount=qty * price, ctx=caps)
+ failed.extend(bad)
+ if not caps.get("sector_source_ready", True):
+ warns.append("SECTOR_OFF: 行业数据源未配置, 行业集中度未校验 (页面已提示)")
+ else:
+ failed.append("CAPS_MISSING: 缺组合上下文, 无法校验上限 (宁可不动)")
+
+
+def summarize(results: list) -> dict:
+ """一批指令的闸门统计 (日报关注区用)。"""
+ passed = [r for r in results if r.get("passed")]
+ reasons = {}
+ for r in results:
+ for f in r.get("failed") or []:
+ k = f.split(":")[0]
+ reasons[k] = reasons.get(k, 0) + 1
+ return {"total": len(results), "passed": len(passed),
+ "rejected": len(results) - len(passed), "by_reason": reasons}
diff --git a/app/scheduler.py b/app/scheduler.py
index b0e9288..e5035c3 100644
--- a/app/scheduler.py
+++ b/app/scheduler.py
@@ -132,10 +132,14 @@ def replay_fills():
@celery_app.task(name="pms.intraday_exec")
@guard(trade_day=True, session=True)
def intraday_exec():
- """盘中执行: 择时出手 + 自主提议扫描。**执行器与动作引擎为下一批交付**,
- 本轮只做在途指令的超时收口 (设计 §13: 不自动重发)。"""
- from app.services import ledger_service
- return {"expired": ledger_service.expire_stale_instructions(), "executor": "pending_batch3"}
+ """盘中执行: 方案转指令 → 择时出手 (规则闸终检 → 下发 → 记子单)。
+
+ 自主提议扫描 (FILL/ADD/DCA/TRIM 动作引擎) 仍为下一批交付。
+ """
+ from app.services import executor
+ r = {"materialized": executor.materialize_plans()}
+ r.update(executor.run_tick())
+ return r
@celery_app.task(name="pms.signal_digest")
@@ -160,8 +164,18 @@ def t0_close():
@celery_app.task(name="pms.daily_settle")
@guard(trade_day=True, respect_exec_halt=False) # 对账属守成动作, 休假模式下照跑
def daily_settle():
- from app.services import ledger_service
- return ledger_service.daily_settle()
+ from app.services import executor, ledger_service
+ out = ledger_service.daily_settle()
+ try:
+ out["steps"]["windows"] = executor.sweep_windows() # 窗口耗尽收口 + 方案成交回写
+ except Exception as e:
+ out.setdefault("errors", []).append(f"窗口收口失败: {e}")
+ out["ok"] = False
+ try:
+ out["steps"]["expired_instructions"] = ledger_service.expire_stale_instructions()
+ except Exception as e:
+ out.setdefault("errors", []).append(f"指令过期处理失败: {e}")
+ return out
@celery_app.task(name="pms.daily_report")
diff --git a/app/services/command_service.py b/app/services/command_service.py
index 0a96016..baf4e2d 100644
--- a/app/services/command_service.py
+++ b/app/services/command_service.py
@@ -218,7 +218,9 @@ def plan_command(cmd: dict) -> dict:
"ts_code": it["ts_code"], "action": it["action"], "qty": it.get("qty"),
"amount": it.get("amount"), "priority": it.get("priority", 100),
"deadline": deadline,
- "status": "PENDING" if not it.get("gated") else "PENDING",
+ # GATED = 建仓的回踩补足/盈利加仓批, 不随命令立即执行,
+ # 等动作引擎按条件解锁后才转 PENDING (设计 §6)
+ "status": "GATED" if it.get("gated") else "PENDING",
"reason": it.get("reason")})
try:
if rows:
diff --git a/app/services/dispatcher.py b/app/services/dispatcher.py
new file mode 100644
index 0000000..cdf26c1
--- /dev/null
+++ b/app/services/dispatcher.py
@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+"""
+指令下发通道 (设计 §9 权限移交的落点)
+======================================
+三个适配器, 由参数 `PMS_DISPATCH_MODE` 切换, **默认 shadow**:
+
+ shadow 影子运行 —— 只记账不下发。指令照常过规则闸、照常置 DISPATCHED,
+ 等用户人工在 QMT 侧执行, 成交由回放按 FIFO 认领回来。
+ 这是设计 §9 的一期口径:「通道未通前由用户人工执行、PMS 记账跟踪」。
+ plan_x 过渡兼容 —— 买入沿用 `trading_buy_plan` (置 is_active=6 待挂单,
+ 署名 approved_by='pms')。**卖出无对应通道**, 自动退回 shadow。
+ 列清单按下游现表推断, QMT 侧确认前请勿在实盘开启。
+ channel_y 推荐方案 —— 写统一指令表 `pms_order_request` (DDL 见需求清单 B1)。
+ 该表归属与形态仍在协商 (B1.1/B1.7), 表未建时会明确报错而非静默吞掉。
+
+无论哪种模式, **指令先落 pms_instruction 再下发** (先记账后动作), 本模块只负责
+「往下游递一手」, 不改指令状态 —— 状态由 executor 统一推进。
+"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime
+
+from app.db.session import execute
+from app.services import param_store
+
+logger = logging.getLogger("pms.dispatch")
+
+MODE_SHADOW, MODE_PLAN_X, MODE_CHANNEL_Y = "shadow", "plan_x", "channel_y"
+MODES = (MODE_SHADOW, MODE_PLAN_X, MODE_CHANNEL_Y)
+
+
+def mode() -> str:
+ m = (param_store.get("PMS_DISPATCH_MODE", MODE_SHADOW) or MODE_SHADOW).strip()
+ return m if m in MODES else MODE_SHADOW
+
+
+def describe() -> dict:
+ m = mode()
+ return {"mode": m, "shadow": m == MODE_SHADOW, "modes": list(MODES),
+ "hint": {
+ MODE_SHADOW: "影子运行: 指令只记账不下发, 请在 QMT 侧人工执行, "
+ "成交由回放自动认领回账本",
+ MODE_PLAN_X: "过渡通道: 买入写 trading_buy_plan(is_active=6), 卖出退回影子",
+ MODE_CHANNEL_Y: "统一通道: 写 pms_order_request, 由下游轮询执行",
+ }[m]}
+
+
+def dispatch(*, instruction_id: str, ts_code: str, side: str, qty: int, limit_price=None,
+ valid_until=None, stock_name=None) -> dict:
+ """递一手给下游。返回 {ok, ref, mode, note, error}; ok=False 时 executor 不改指令状态。"""
+ m = mode()
+ try:
+ if m == MODE_CHANNEL_Y:
+ return _channel_y(instruction_id, ts_code, side, qty, limit_price, valid_until)
+ if m == MODE_PLAN_X and str(side).lower() == "buy":
+ return _plan_x(instruction_id, ts_code, qty, limit_price, stock_name)
+ if m == MODE_PLAN_X:
+ r = _shadow(instruction_id, side)
+ r["note"] = "plan_x 无卖出通道, 本单退回影子运行 (待 QMT B1 落地)"
+ return r
+ return _shadow(instruction_id, side)
+ except Exception as e:
+ logger.exception("下发失败 %s", instruction_id)
+ return {"ok": False, "ref": None, "mode": m, "note": "",
+ "error": f"{type(e).__name__}: {e}"}
+
+
+def _shadow(instruction_id: str, side: str) -> dict:
+ return {"ok": True, "ref": f"manual:{instruction_id}", "mode": MODE_SHADOW,
+ "note": f"影子运行: 请在 QMT 侧人工{'买入' if side == 'buy' else '卖出'}, "
+ f"成交由回放认领", "error": None}
+
+
+def _channel_y(instruction_id, ts_code, side, qty, limit_price, valid_until) -> dict:
+ now = datetime.now()
+ execute(
+ "INSERT INTO pms_order_request (instruction_id, ts_code, side, qty, limit_price, "
+ "valid_until, source, status, cancel_flag, create_time, update_time) VALUES "
+ "(:iid, :code, :side, :qty, :lp, :vu, 'pms', 'NEW', 0, :ts, :ts)",
+ {"iid": instruction_id, "code": ts_code, "side": str(side).lower(), "qty": int(qty),
+ "lp": limit_price, "vu": valid_until or now, "ts": now})
+ return {"ok": True, "ref": instruction_id, "mode": MODE_CHANNEL_Y,
+ "note": "已写入 pms_order_request, 等下游轮询执行", "error": None}
+
+
+def _plan_x(instruction_id, ts_code, qty, limit_price, stock_name) -> dict:
+ """买入走上游既有计划表。amount 由 数量×限价 反算 (下游按 buy_amount 挂单)。"""
+ px = float(limit_price or 0)
+ if px <= 0:
+ return {"ok": False, "ref": None, "mode": MODE_PLAN_X, "note": "",
+ "error": "plan_x 通道要求限价 (下游按 target_price 挂单)"}
+ now = datetime.now()
+ execute(
+ "INSERT INTO trading_buy_plan (stock_code, stock_name, target_price, buy_amount, "
+ "is_active, trading_time, create_time, update_time, approved_by, change_reason) "
+ "VALUES (:code, :name, :px, :amt, 6, :ts, :ts, :ts, 'pms', :rsn)",
+ {"code": ts_code, "name": stock_name or ts_code, "px": px,
+ "amt": round(qty * px, 2), "ts": now,
+ "rsn": f"PMS 指令 {instruction_id}"})
+ return {"ok": True, "ref": f"buy_plan:{instruction_id}", "mode": MODE_PLAN_X,
+ "note": "已写 trading_buy_plan(is_active=6) 待下游挂单", "error": None}
+
+
+def cancel(*, instruction_id: str, dispatch_ref=None) -> dict:
+ """请求撤单。shadow/plan_x 无撤单语义, 只回执由 executor 置本地状态。"""
+ m = mode()
+ if m != MODE_CHANNEL_Y:
+ return {"ok": True, "mode": m, "note": "本模式无下游撤单动作, 仅本地置撤销"}
+ try:
+ n = execute("UPDATE pms_order_request SET cancel_flag = 1, update_time = :ts "
+ "WHERE instruction_id = :iid AND status IN ('NEW', 'ACCEPTED', 'EXECUTING')",
+ {"iid": instruction_id, "ts": datetime.now()})
+ return {"ok": True, "mode": m, "note": f"已置 cancel_flag, 影响 {n} 行"}
+ except Exception as e:
+ return {"ok": False, "mode": m, "error": f"{type(e).__name__}: {e}"}
diff --git a/app/services/executor.py b/app/services/executor.py
new file mode 100644
index 0000000..49a26c7
--- /dev/null
+++ b/app/services/executor.py
@@ -0,0 +1,314 @@
+# -*- 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)
diff --git a/app/services/market.py b/app/services/market.py
index 2bf35dc..6c1f0ed 100644
--- a/app/services/market.py
+++ b/app/services/market.py
@@ -167,6 +167,28 @@ def self_calc_refs(ts_code: str, base_cost=None) -> dict:
"source": "self_calc", "bars": len(rows)}
+_ma_cache = {"day": None, "data": {}}
+
+
+def get_ma5(ts_code: str):
+ """MA5 (规则闸「不追高」用)。按日缓存 —— 因子分表是日频数据, 盘中重复查没意义。"""
+ today = datetime.now().strftime("%Y%m%d")
+ if _ma_cache["day"] != today:
+ _ma_cache.update({"day": today, "data": {}})
+ if ts_code in _ma_cache["data"]:
+ return _ma_cache["data"][ts_code]
+ val = None
+ try:
+ rows = _factor_rows(ts_code, days=10)
+ closes = [float(r["close_qfq"]) for r in rows if r.get("close_qfq")]
+ if len(closes) >= 5:
+ val = round(sum(closes[-5:]) / 5, 3)
+ except Exception as e:
+ logger.warning("MA5 取数失败 [%s]: %s", ts_code, e)
+ _ma_cache["data"][ts_code] = val
+ return val
+
+
def get_refs(ts_code: str, *, base_cost=None) -> dict:
"""参考位: 决策系统主口径 → 日龄超期/缺失时兜底自算 → 都拿不到返回 source=none。"""
stale_days = param_store.get_int("PMS_REF_STALE_TDAYS", 3)
diff --git a/app/services/param_store.py b/app/services/param_store.py
index df7531a..d3efbfb 100644
--- a/app/services/param_store.py
+++ b/app/services/param_store.py
@@ -64,6 +64,9 @@ DESC = {
"PMS_EOD_FORCE_TIME": "当日配额兜底时点", "PMS_EOD_FORCE_DISCOUNT": "兜底限价系数 (卖出)",
"PMS_MIN_LOT_MERGE": "一手检查: 批次自动合并",
"PMS_DISPATCH_EXPIRE_MIN": "指令下发后未被接受的过期时间 (分钟)",
+ "PMS_DISPATCH_MODE": "下发通道: shadow=只记账待人工 / plan_x=买入走 trading_buy_plan / "
+ "channel_y=写 pms_order_request",
+ "PMS_EXEC_SLICES": "当日配额分几笔出手",
"PMS_RISK_WARN_ENTRY": "单笔敞口告警线 (占规模)", "PMS_RISK_WARN_PORTFOLIO": "组合敞口告警线",
"PMS_BRAKE_DRAWDOWN": "组合刹车: 自高水位回撤", "PMS_BRAKE_DAYS": "刹车持续交易日",
"PMS_STOP_ATR_MULT": "自算止损参考: 成本 − N×ATR",
@@ -223,6 +226,8 @@ def _range_check(key, v):
return "PMS_AUTONOMY 只能是 full / propose_only / off"
if key == "PMS_SECTOR_SOURCE" and v not in ("", "custom_table", "gp_stock_category"):
return "PMS_SECTOR_SOURCE 只能是 空 / custom_table / gp_stock_category"
+ if key == "PMS_DISPATCH_MODE" and v not in ("shadow", "plan_x", "channel_y"):
+ return "PMS_DISPATCH_MODE 只能是 shadow / plan_x / channel_y"
lo_hi = _RANGES.get(key)
if lo_hi and isinstance(v, (int, float)) and not isinstance(v, bool):
lo, hi = lo_hi
diff --git a/app/web/main.py b/app/web/main.py
index 10d15b4..ed614a5 100644
--- a/app/web/main.py
+++ b/app/web/main.py
@@ -282,6 +282,39 @@ def api_plan_pending():
return ok(command_service.plan_pending)
+@app.post("/api/ops/materialize")
+def api_materialize():
+ """方案 → 指令 (先记账后动作)。"""
+ from app.services import executor
+ return ok(executor.materialize_plans)
+
+
+@app.post("/api/ops/exec-tick")
+def api_exec_tick(dry_run: bool = Query(False)):
+ """择时出手一跳。dry_run=true 只试算不下发, 用来在盘中先看「现在会怎么动」。"""
+ from app.services import executor
+ return ok(executor.run_tick, dry_run=dry_run)
+
+
+@app.post("/api/ops/sweep-windows")
+def api_sweep_windows():
+ from app.services import executor
+ return ok(executor.sweep_windows)
+
+
+@app.post("/api/instructions/{instruction_id}/cancel")
+def api_cancel_instruction(instruction_id: str, payload: dict = Body(default={})):
+ from app.services import executor
+ return ok(executor.cancel_instruction, instruction_id,
+ payload.get("reason") or "页面人工撤销")
+
+
+@app.get("/api/dispatch-mode")
+def api_dispatch_mode():
+ from app.services import dispatcher
+ return ok(dispatcher.describe)
+
+
@app.get("/api/ops/downstream-schema")
def api_downstream_schema():
"""导出下游三表的实际列定义 —— 用于回填 QMT_INTERFACE_REQUIREMENTS D1。"""
diff --git a/app/web/static/index.html b/app/web/static/index.html
index 5c60bfe..62c9740 100644
--- a/app/web/static/index.html
+++ b/app/web/static/index.html
@@ -50,6 +50,8 @@
执行 {{ ov.exec_halt ? '已暂停':'正常' }}
自主档位 {{ ov.autonomy || '-' }}
+
+ 下发 {{ dm.mode || '-' }}
日报
运维
刷新
@@ -65,6 +67,8 @@
:title="'以下持仓取不到实时价, 已用摊薄成本兜底: ' + ov.price_missing.join(', ')"/>
+
@@ -290,20 +294,58 @@
在途指令
-
+
+
+
+
+
+ 最近判定 [{{ s.row.progress.last_decision.at }}]
+ {{ s.row.progress.last_decision.action }} —
+ {{ s.row.progress.last_decision.reason }}
+
+
+ 执行窗口截止: {{ s.row.progress.deadline }}
+
+ · {{ s.row.progress.window_verdict.note }}
+
+
+
+
+
+
+
+
+ {{ c.row.forced ? '是' : '—' }}
+
+
+
+
尚无出手记录
+
- 指令下发通道待 QMT 侧协商落地 (B1); 通道未通前减持类只记账不下发 (影子运行)。
+ {{ dm.hint }} 展开可看每日子单与择时依据; 成交量以下游回放为准, 不拿下发量当成交量。
@@ -367,14 +409,25 @@
调度器 (--profile sched) 会自动按调度总表执行; 此处按钮用于未接调度或临时验证时手动触发。
+ 推进待规划命令
+ 方案转指令
+
+ 出手试算 (不下发)
+ 出手一跳
+ 窗口收口
+
+
成交回放
账本对账
盘前准备
日终结算
生成日报
- 推进待规划命令
导出下游表结构
+
+ 当前下发通道: {{ dm.mode }} —— {{ dm.hint }}
+ (通道在「参数设置 → PMS_DISPATCH_MODE」切换)
+
{{ opsResult }}
@@ -404,7 +457,7 @@ createApp({
const catalog = ref([]), commands = ref([]), plans = ref([]), plansOf = ref('');
const positions = ref([]), lots = ref([]), lotsOf = ref('');
const instructions = ref([]), ledger = ref([]), proposals = ref([]);
- const report = ref({}), reportDrawer = ref(false);
+ const report = ref({}), reportDrawer = ref(false), dm = ref({});
const opsDrawer = ref(false), opsResult = ref(''), opsLoading = ref(false);
const issuing = ref(false);
const form = reactive({ cmd_type: '', params: {}, note: '' });
@@ -433,6 +486,10 @@ createApp({
const t = Number(p.target_amount || 0), d = Number(p.done_amount || 0);
return t > 0 ? Math.min(100, Math.round(d / t * 100)) : 0;
};
+ const insPct = r => {
+ const q = Number(r.qty || 0), e = Number(r.exec_qty || 0);
+ return q > 0 ? Math.min(100, Math.round(e / q * 100)) : 0;
+ };
async function call(method, url, body) {
try {
@@ -491,12 +548,25 @@ createApp({
lotsOf.value = code;
const d = await call('get', '/api/positions/' + code + '/lots?status='); lots.value = d.data || [];
}
+ async function loadDispatchMode() {
+ const d = await call('get', '/api/dispatch-mode'); dm.value = d.data || d || {};
+ }
async function loadAll() {
loading.value = true; err.value = '';
await Promise.all([loadOverview(), loadParams(), loadCatalog(), loadCommands(),
- loadPositions(), loadInstructions(), loadLedger(), loadProposals()]);
+ loadPositions(), loadInstructions(), loadLedger(), loadProposals(),
+ loadDispatchMode()]);
loading.value = false;
}
+ async function cancelIns(iid) {
+ try {
+ await ElementPlus.ElMessageBox.confirm('撤销指令 ' + iid + '?', '确认',
+ { type: 'warning' });
+ } catch (e) { return; }
+ const d = await call('post', '/api/instructions/' + iid + '/cancel', {});
+ ElementPlus.ElMessage[(d.ok ? 'success' : 'error')](d.message || d.error);
+ await Promise.all([loadInstructions(), loadLedger()]);
+ }
function onCmdChange() {
form.params = {};
@@ -585,10 +655,10 @@ createApp({
onMounted(loadAll);
return { tab, loading, err, health, ov, params, catalog, commands, plans, plansOf,
positions, lots, lotsOf, instructions, ledger, proposals, report, reportDrawer,
- opsDrawer, opsResult, opsLoading, issuing, form, curSpec, dirtyCount,
- money, pct, groupLabel, fieldLabel, stTag, cuTag, canCancel, progPct,
+ opsDrawer, opsResult, opsLoading, issuing, form, curSpec, dirtyCount, dm,
+ money, pct, groupLabel, fieldLabel, stTag, cuTag, canCancel, progPct, insPct,
loadAll, loadParams, saveParams, loadPlans, loadLots, onCmdChange, issue,
- cancelCmd, replan, decide, ops, loadSchema, openReport };
+ cancelCmd, replan, decide, ops, loadSchema, openReport, cancelIns };
}
}).use(ElementPlus).mount('#app');
diff --git a/config/settings.py b/config/settings.py
index 15a7beb..ecaee2f 100644
--- a/config/settings.py
+++ b/config/settings.py
@@ -84,6 +84,8 @@ class Settings(BaseSettings):
PMS_EOD_FORCE_DISCOUNT: float = 0.998 # 兜底限价 = 现价 × 此系数 (卖出)
PMS_MIN_LOT_MERGE: bool = True # 一手检查: 批次自动合并
PMS_DISPATCH_EXPIRE_MIN: int = 30 # 指令下发后未被接受的过期时间
+ PMS_DISPATCH_MODE: str = "shadow" # 下发通道: shadow(影子,默认) / plan_x / channel_y
+ PMS_EXEC_SLICES: int = 1 # 当日配额分几笔出手 (设计「分笔卖出配额」)
# --- 风险披露与刹车 ---
PMS_RISK_WARN_ENTRY: float = 0.01 # 单笔敞口告警线 (占规模)
diff --git a/ddl_pms_v1.sql b/ddl_pms_v1.sql
index a3d33b0..7673629 100644
--- a/ddl_pms_v1.sql
+++ b/ddl_pms_v1.sql
@@ -38,7 +38,7 @@ CREATE TABLE IF NOT EXISTS pms_plan (
priority INT NOT NULL DEFAULT 100 COMMENT '越小越先执行',
deadline DATE NULL COMMENT '执行窗口截止日',
status VARCHAR(16) NOT NULL DEFAULT 'PENDING'
- COMMENT 'PENDING/EXECUTING/DONE/PARTIAL/CANCELLED',
+ COMMENT 'PENDING/GATED(建仓补足加仓批,待动作引擎解锁)/EXECUTING/DONE/PARTIAL/CANCELLED',
filled_qty INT NOT NULL DEFAULT 0,
reason VARCHAR(300) NULL COMMENT '进方案的理由 (弱票清仓/收利润/等比减 等)',
created_at DATETIME NOT NULL,
diff --git a/scripts/run_tests.py b/scripts/run_tests.py
index 193eeed..700c1e2 100644
--- a/scripts/run_tests.py
+++ b/scripts/run_tests.py
@@ -7,7 +7,8 @@
包含:
test_core_units.py 仓位规划器 / 安全垫与成本账 (14 例)
test_batch2_units.py 命令状态机 / 方案生成器 / 回放对账纯逻辑 (35 例)
- test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (18 例)
+ test_batch3_units.py 规则闸 / 择时执行器实现B 纯逻辑 (18 例)
+ test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (25 例)
任一子集失败即整体失败 (退出码 1)。
"""
import os
@@ -16,7 +17,8 @@ import sys
HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
-SUITES = ["test_core_units.py", "test_batch2_units.py", "test_wiring.py"]
+SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py",
+ "test_wiring.py"]
def main():
diff --git a/scripts/test_batch3_units.py b/scripts/test_batch3_units.py
new file mode 100644
index 0000000..ed42f41
--- /dev/null
+++ b/scripts/test_batch3_units.py
@@ -0,0 +1,293 @@
+# -*- coding: utf-8 -*-
+"""
+第三批模块单测 (实机运行, 零外部依赖)
+======================================
+运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch3_units.py
+覆盖: exec_timing 分日配额/分笔/买卖出手判定/兜底/顺延/窗口收口;
+ rule_gate 减持放行口径、增持全约束、命令与自主的刹车差别。
+约定同前: 全过输出 "ALL PASS (n cases)" 退出码 0。
+"""
+import os
+import sys
+import traceback
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from app.core import exec_timing as et # noqa: E402
+from app.core import rule_gate as rg # noqa: E402
+
+RESULTS = []
+
+
+def case(name):
+ def deco(fn):
+ RESULTS.append((name, fn))
+ return fn
+ return deco
+
+
+PRM = {"sell_avoid_open_min": 30, "buy_halt_dayup": 0.05,
+ "eod_force_time": "14:45", "eod_force_discount": 0.998}
+
+
+def day(**kw):
+ d = {"price": 10.0, "vwap": 10.0, "high": 10.5, "low": 9.5, "open": 10.0,
+ "day_chg_from_open": 0.0, "bars": 60}
+ d.update(kw)
+ return d
+
+
+# ================================================================ exec_timing
+@case("分日配额·整除/上取整到一手/最后一日全出/零股尾巴并入")
+def _():
+ assert et.daily_quota(6000, 3) == 2000
+ assert et.daily_quota(5000, 3) == 1700 # 1666.7 → 上取整到一手
+ assert et.daily_quota(1000, 1) == 1000 # 最后一日全出
+ assert et.daily_quota(150, 3) == 150 # 尾巴不足一手 → 一次出完
+ assert et.daily_quota(0, 3) == 0
+ assert et.daily_quota(100, 5) == 100
+ # 整票清仓允许零股
+ assert et.daily_quota(14050, 3, allow_odd_tail=True) == 4700
+ assert et.daily_quota(14050, 1, allow_odd_tail=True) == 14050
+
+
+@case("分笔·配额切成 N 笔, 余数并入最后一笔")
+def _():
+ assert et.slice_qty(3000, 3) == [1000, 1000, 1000]
+ assert et.slice_qty(1000, 3) == [300, 300, 400]
+ assert et.slice_qty(100, 3) == [100]
+ assert et.slice_qty(0, 3) == []
+ assert et.slice_qty(2500, 1) == [2500]
+
+
+@case("卖出择时·避开开盘30分钟 → 站上均价才出手 → 限价打折")
+def _():
+ r = et.decide(side="sell", now="09:45", day=day(), params=PRM, is_last_day=False, quota=2000)
+ assert r["action"] == et.ACT_WAIT and "避开开盘" in r["reason"], r
+ r = et.decide(side="sell", now="10:05", day=day(price=10.2, vwap=10.0), params=PRM,
+ is_last_day=False, quota=2000)
+ assert r["action"] == et.ACT_FIRE and r["limit_price"] == 10.18, r # 10.2×0.998
+ r = et.decide(side="sell", now="10:05", day=day(price=9.8, vwap=10.0), params=PRM,
+ is_last_day=False, quota=2000)
+ assert r["action"] == et.ACT_WAIT and "等更好的价" in r["reason"], r
+
+
+@case("卖出择时·14:45 兜底强制出手 (不管均价)")
+def _():
+ r = et.decide(side="sell", now="14:45", day=day(price=9.5, vwap=10.0), params=PRM,
+ is_last_day=False, quota=2000)
+ assert r["action"] == et.ACT_FIRE and r["forced"] is True, r
+ assert r["limit_price"] == 9.48, r # 9.5×0.998
+ assert et.decide(side="sell", now="14:44", day=day(price=9.5, vwap=10.0), params=PRM,
+ is_last_day=False, quota=2000)["action"] == et.ACT_WAIT
+
+
+@case("买入择时·跌破均价才买 / 回踩带买 / 不追高停手")
+def _():
+ r = et.decide(side="buy", now="10:05", day=day(price=9.8, vwap=10.0), params=PRM,
+ is_last_day=False, quota=3000)
+ assert r["action"] == et.ACT_FIRE and r["limit_price"] == 9.82, r # 9.8×1.002
+ r = et.decide(side="buy", now="10:05", day=day(price=10.3, vwap=10.0), params=PRM,
+ is_last_day=False, quota=3000)
+ assert r["action"] == et.ACT_WAIT and "等回调" in r["reason"], r
+ r = et.decide(side="buy", now="10:05", day=day(price=10.3, vwap=10.0, support=10.2),
+ params=PRM, is_last_day=False, quota=3000)
+ assert r["action"] == et.ACT_FIRE and "回踩带" in r["reason"], r
+ r = et.decide(side="buy", now="10:05", day=day(price=11.0, vwap=10.0,
+ day_chg_from_open=0.06),
+ params=PRM, is_last_day=False, quota=3000)
+ assert r["action"] == et.ACT_STOP and "不追高" in r["reason"], r
+
+
+@case("买入择时·窗口末日 14:45 强制完成, 非末日则顺延")
+def _():
+ r = et.decide(side="buy", now="14:50", day=day(price=10.5, vwap=10.0), params=PRM,
+ is_last_day=True, quota=3000)
+ assert r["action"] == et.ACT_FIRE and r["forced"] and r["limit_price"] == 10.52, r
+ r = et.decide(side="buy", now="14:50", day=day(price=10.5, vwap=10.0), params=PRM,
+ is_last_day=False, quota=3000)
+ assert r["action"] == et.ACT_WAIT and "顺延次日" in r["reason"], r
+
+
+@case("择时·停牌/一字板/非交易时段/配额出完 一律不动")
+def _():
+ assert et.decide(side="sell", now="10:05", day=day(price=0), params=PRM,
+ is_last_day=False, quota=100)["action"] == et.ACT_SKIP
+ assert et.decide(side="sell", now="10:05", day=day(halted=True), params=PRM,
+ is_last_day=False, quota=100)["action"] == et.ACT_SKIP
+ assert et.decide(side="buy", now="10:05", day=day(limit_up=True), params=PRM,
+ is_last_day=False, quota=100)["action"] == et.ACT_SKIP
+ r = et.decide(side="sell", now="10:05", day=day(limit_down=True), params=PRM,
+ is_last_day=False, quota=100)
+ assert r["action"] == et.ACT_SKIP and "顺延" in r["reason"], r
+ assert et.decide(side="sell", now="12:00", day=day(), params=PRM,
+ is_last_day=False, quota=100)["action"] == et.ACT_WAIT
+ r = et.decide(side="sell", now="10:05", day=day(price=10.2), params=PRM,
+ is_last_day=False, quota=1000, fired_today=1000)
+ assert r["action"] == et.ACT_WAIT and "配额已出完" in r["reason"], r
+
+
+@case("择时·时点解析与交易时段判定")
+def _():
+ assert et.hm_to_min("14:45") == 885 and et.hm_to_min((9, 30)) == 570
+ assert et.in_session(570) and et.in_session(690) and et.in_session(780)
+ assert not et.in_session(569) and not et.in_session(700) and not et.in_session(901)
+
+
+@case("窗口收口·命令类置部分完成, 自主类作废, 未到期继续")
+def _():
+ assert et.window_verdict(remaining_qty=0, tdays_left=0, is_command=True)["verdict"] == "DONE"
+ assert et.window_verdict(remaining_qty=500, tdays_left=2,
+ is_command=True)["verdict"] == "RUNNING"
+ v = et.window_verdict(remaining_qty=500, tdays_left=0, is_command=True)
+ assert v["verdict"] == "PARTIAL" and "告警" in v["note"], v
+ assert et.window_verdict(remaining_qty=500, tdays_left=0,
+ is_command=False)["verdict"] == "EXPIRED"
+
+
+# ================================================================ rule_gate
+def ctx(**kw):
+ d = {"ts_code": "600000.SH",
+ "position": {"total_qty": 6000, "avail_qty": 6000, "frozen_reason": "NONE"},
+ "day": {"price": 10.0, "vwap": 10.0, "ma5": 10.0, "day_chg_from_open": 0.01},
+ "params": {"no_chase_ma5": 0.06, "buy_halt_dayup": 0.05},
+ "flags": {"buy_halt": False, "exec_halt": False, "brake_active": False,
+ "blacklisted": False, "is_command": False},
+ "caps": dict(scale=2_000_000, portfolio_cap=0.60, stock_cap=0.08, max_names=15,
+ portfolio_mv=800_000, names_count=5, stock_mv=60_000, is_new_name=False,
+ sector=None, sector_names=0, sector_mv=0.0, sector_max_names=4,
+ sector_max_ratio=0.40, cash_reserve=0.0, sector_source_ready=True)}
+ for k, v in kw.items():
+ if isinstance(v, dict) and isinstance(d.get(k), dict):
+ d[k] = {**d[k], **v}
+ else:
+ d[k] = v
+ return d
+
+
+@case("规则闸·卖出正常放行 + 硬数字留痕")
+def _():
+ r = rg.check(side="sell", action="TRIM", qty=2000, price=10.0, ctx=ctx())
+ assert r["passed"] and r["failed"] == [], r
+ assert r["hard_numbers"]["avail_qty"] == 6000 and r["hard_numbers"]["qty"] == 2000
+
+
+@case("规则闸·卖出超持仓/超可卖/零股非清仓 三种拦截")
+def _():
+ r = rg.check(side="sell", action="EXIT", qty=9000, price=10.0, ctx=ctx())
+ assert any(x.startswith("OVER_SELL") for x in r["failed"]), r
+ r = rg.check(side="sell", action="TRIM", qty=5000, price=10.0,
+ ctx=ctx(position={"avail_qty": 3000}))
+ assert any(x.startswith("T1_UNAVAILABLE") for x in r["failed"]), r
+ r = rg.check(side="sell", action="TRIM", qty=1050, price=10.0, ctx=ctx())
+ assert any(x.startswith("LOT_INVALID") for x in r["failed"]), r
+ # 清仓允许零股
+ r = rg.check(side="sell", action="EXIT", qty=6050, price=10.0,
+ ctx=ctx(position={"total_qty": 6050, "avail_qty": 6050}))
+ assert r["passed"], r
+
+
+@case("规则闸·减持不受冻结/刹车/上限影响 (只挡增持)")
+def _():
+ c = ctx(position={"frozen_reason": "COMMAND_HALT"},
+ flags={"brake_active": True, "buy_halt": True})
+ r = rg.check(side="sell", action="EXIT", qty=6000, price=10.0, ctx=c)
+ assert r["passed"], r
+ r2 = rg.check(side="buy", action="ADD", qty=1000, price=10.0, ctx=c)
+ assert not r2["passed"] and any(x.startswith("FROZEN") for x in r2["failed"]), r2
+
+
+@case("规则闸·买入 全局暂停/黑名单/非整百 拦截")
+def _():
+ r = rg.check(side="buy", action="ADD", qty=1000, price=10.0,
+ ctx=ctx(flags={"buy_halt": True}))
+ assert any(x.startswith("BUY_HALT") for x in r["failed"]), r
+ r = rg.check(side="buy", action="OPEN", qty=1000, price=10.0,
+ ctx=ctx(flags={"blacklisted": True}))
+ assert any(x.startswith("BLACKLIST") for x in r["failed"]), r
+ r = rg.check(side="buy", action="OPEN", qty=150, price=10.0, ctx=ctx())
+ assert any(x.startswith("LOT_INVALID") for x in r["failed"]), r
+
+
+@case("规则闸·刹车对自主是拦截, 对命令只是提示 (命令至上)")
+def _():
+ auto = rg.check(side="buy", action="ADD", qty=1000, price=10.0,
+ ctx=ctx(flags={"brake_active": True}))
+ assert not auto["passed"] and any(x.startswith("BRAKE_ACTIVE") for x in auto["failed"])
+ cmd = rg.check(side="buy", action="OPEN", qty=1000, price=10.0,
+ ctx=ctx(flags={"brake_active": True, "is_command": True}))
+ assert cmd["passed"], cmd
+ assert any(w.startswith("BRAKE_ACTIVE") for w in cmd["warnings"]), cmd
+
+
+@case("规则闸·不追高两道 (当日涨幅 / 距 MA5) 与 MA5 缺失告警")
+def _():
+ r = rg.check(side="buy", action="OPEN", qty=1000, price=10.0,
+ ctx=ctx(day={"day_chg_from_open": 0.07}))
+ assert any(x.startswith("NO_CHASE_DAYUP") for x in r["failed"]), r
+ r = rg.check(side="buy", action="OPEN", qty=1000, price=11.0,
+ ctx=ctx(day={"ma5": 10.0}))
+ assert any(x.startswith("NO_CHASE_MA5") for x in r["failed"]), r # 距 MA5 10% > 6%
+ r = rg.check(side="buy", action="OPEN", qty=1000, price=10.3,
+ ctx=ctx(day={"ma5": 10.0}))
+ assert r["passed"], r # 3% 在容忍内
+ r = rg.check(side="buy", action="OPEN", qty=1000, price=10.0, ctx=ctx(day={"ma5": None}))
+ assert r["passed"] and any(w.startswith("MA5_MISSING") for w in r["warnings"]), r
+
+
+@case("规则闸·买入过组合上限 (复用 planner 同一份口径)")
+def _():
+ r = rg.check(side="buy", action="ADD", qty=12000, price=10.0, ctx=ctx())
+ assert any(x.startswith("STOCK_CAP") for x in r["failed"]), r # 6万+12万 > 16万
+ r = rg.check(side="buy", action="ADD", qty=1000, price=10.0, ctx=ctx(caps=None))
+ assert any(x.startswith("CAPS_MISSING") for x in r["failed"]), r
+
+
+@case("规则闸·必要输入缺失与全局暂停执行一律拒绝 (宁可不动)")
+def _():
+ r = rg.check(side="sell", action="EXIT", qty=1000, price=0, ctx=ctx())
+ assert any(x.startswith("PRICE_MISSING") for x in r["failed"]), r
+ r = rg.check(side="sell", action="EXIT", qty=0, price=10.0, ctx=ctx())
+ assert any(x.startswith("QTY_INVALID") for x in r["failed"]), r
+ r = rg.check(side="sell", action="EXIT", qty=1000, price=10.0,
+ ctx=ctx(flags={"exec_halt": True}))
+ assert any(x.startswith("EXEC_HALT") for x in r["failed"]), r
+ r = rg.check(side="sell", action="EXIT", qty=1000, price=10.0,
+ ctx=ctx(day={"halted": True}))
+ assert any(x.startswith("HALTED") for x in r["failed"]), r
+ r = rg.check(side="hold", action="X", qty=100, price=10.0, ctx=ctx())
+ assert any(x.startswith("SIDE_INVALID") for x in r["failed"]), r
+
+
+@case("规则闸·批量统计 (日报关注区用)")
+def _():
+ rs = [rg.check(side="sell", action="EXIT", qty=1000, price=10.0, ctx=ctx()),
+ rg.check(side="buy", action="OPEN", qty=150, price=10.0, ctx=ctx()),
+ rg.check(side="buy", action="OPEN", qty=100, price=10.0,
+ ctx=ctx(flags={"buy_halt": True}))]
+ s = rg.summarize(rs)
+ assert s["total"] == 3 and s["passed"] == 1 and s["rejected"] == 2, s
+ assert s["by_reason"].get("LOT_INVALID") == 1 and s["by_reason"].get("BUY_HALT") == 1, s
+
+
+# ---------------------------------------------------------------- runner
+def main():
+ passed, failed = 0, 0
+ for name, fn in RESULTS:
+ try:
+ fn()
+ print(f" PASS {name}")
+ passed += 1
+ except Exception:
+ print(f" FAIL {name}")
+ traceback.print_exc()
+ failed += 1
+ print("-" * 60)
+ if failed:
+ print(f"FAILED: {failed} / {passed + failed}")
+ sys.exit(1)
+ print(f"ALL PASS ({passed} cases)")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py
index 51615a4..8322f0a 100644
--- a/scripts/test_wiring.py
+++ b/scripts/test_wiring.py
@@ -322,6 +322,10 @@ def install_fakes(prices=None, positions=None, params=None):
market.get_prices = lambda cs_: {c: (prices or {}).get(c) for c in (cs_ or [])}
market.get_refs = lambda c, **kw: {"support": None, "pressure": None, "stop": None,
"source": "none"}
+ market.get_ma5 = lambda c: (prices or {}).get(c)
+ market.day_snapshot = lambda c: ({} if not (prices or {}).get(c) else {
+ "price": prices[c], "vwap": prices[c], "open": prices[c], "high": prices[c] * 1.02,
+ "low": prices[c] * 0.98, "day_chg_from_open": 0.0, "bars": 60})
industry.get_many = lambda cs_: {c: None for c in (cs_ or [])}
industry.get = lambda c: None
industry.ready = lambda: False
@@ -708,6 +712,178 @@ def _():
assert fake.positions["600000.SH"]["avail_qty"] == 6000
+@case("执行器·方案转指令 (GATED 批不转, HALT 直接完结)")
+def _():
+ from app.services import command_service as csvc, executor
+ fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
+ positions=[{"ts_code": "600000.SH", "total_qty": 14000,
+ "base_qty": 7000, "fill_qty": 3500, "add_qty": 3500,
+ "avail_qty": 14000, "avg_cost": 9.0}])
+ r = csvc.issue("REDUCE_EXPOSURE", {"pct": "3%"})
+ assert r["ok"], r
+ # 掺一条 GATED 方案 (建仓补足批) 与一条 HALT 方案
+ fake.insert_plans([
+ {"plan_id": "P_G", "command_id": r["command_id"], "ts_code": "600000.SH",
+ "action": "FILL", "qty": 3000, "amount": 30000, "priority": 20,
+ "deadline": "2026-07-30", "status": "GATED", "reason": "回踩补足批"},
+ {"plan_id": "P_H", "command_id": r["command_id"], "ts_code": "600000.SH",
+ "action": "HALT", "qty": 0, "amount": 0, "priority": 10,
+ "deadline": "2026-07-30", "status": "PENDING", "reason": "撤在途买入"}])
+ m = executor.materialize_plans()
+ assert m["created"], m
+ codes = {fake.instructions[i]["ts_code"] for i in m["created"]}
+ assert codes == {"600000.SH"}, codes
+ assert all(fake.instructions[i]["status"] == "PROPOSED" for i in m["created"])
+ gated = [p for p in fake.plans if p["plan_id"] == "P_G"][0]
+ assert gated["status"] == "GATED", gated # 未解锁的批次不转指令
+ halt = [p for p in fake.plans if p["plan_id"] == "P_H"][0]
+ assert halt["status"] == "DONE", halt
+ assert any("HALT" in s["why"] for s in m["skipped"]), m["skipped"]
+ # 幂等: 再跑一次不重复建指令
+ n1 = len(fake.instructions)
+ executor.materialize_plans()
+ assert len(fake.instructions) == n1
+
+
+@case("执行器·出手一跳 (影子模式): 择时→规则闸→下发→子单→评审留痕")
+def _():
+ from datetime import datetime as _dt
+ from app.services import executor
+ fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
+ positions=[{"ts_code": "600000.SH", "total_qty": 6000,
+ "avail_qty": 6000, "avg_cost": 9.0}])
+ fake.insert_instruction(instruction_id="INS_S", origin_type="plan", origin_id="P1",
+ ts_code="600000.SH", action="TRIM", side="sell", qty=3000,
+ window_tdays=3, status="PROPOSED",
+ progress={"deadline": "2026-07-29", "is_command": True,
+ "children": []})
+ # 周一 10:05, 距截止 3 个交易日 → 当日配额 1000
+ r = executor.run_tick(now=_dt(2026, 7, 27, 10, 5))
+ assert r["ok"] and len(r["fired"]) == 1, r
+ fired = r["fired"][0]
+ assert fired["qty"] == 1000 and fired["mode"] == "shadow", fired
+ ins = fake.instructions["INS_S"]
+ assert ins["status"] == "DISPATCHED"
+ assert len(ins["progress"]["children"]) == 1
+ assert ins["progress"]["children"][0]["qty"] == 1000
+ assert ins["progress"]["dispatched_at"]
+ assert any(x["verdict"] == "PASS" and x["arbiter"] == "rule" for x in fake.ledger)
+ # 同一天再跳一次: 配额已出完, 不重复下发
+ r2 = executor.run_tick(now=_dt(2026, 7, 27, 10, 6))
+ assert not r2["fired"] and r2["waited"], r2
+ assert len(fake.instructions["INS_S"]["progress"]["children"]) == 1
+
+
+@case("执行器·规则闸拦截时不下发且落拒绝留痕")
+def _():
+ from datetime import datetime as _dt
+ from app.services import executor
+ fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
+ positions=[{"ts_code": "600000.SH", "total_qty": 6000,
+ "avail_qty": 6000, "avg_cost": 9.0}])
+ fake.insert_instruction(instruction_id="INS_B", origin_type="plan", origin_id="P2",
+ ts_code="600000.SH", action="OPEN", side="buy", qty=1000,
+ window_tdays=1, status="PROPOSED",
+ progress={"deadline": "2026-07-27", "is_command": True,
+ "children": []})
+ from app.services import param_store
+ param_store.set_param("PMS_GLOBAL_BUY_HALT", True, "test")
+ r = executor.run_tick(now=_dt(2026, 7, 27, 10, 5))
+ assert not r["fired"] and r["rejected"], r
+ assert any("BUY_HALT" in f for f in r["rejected"][0]["failed"]), r["rejected"]
+ assert fake.instructions["INS_B"]["status"] == "PROPOSED" # 未下发
+ assert any(x["verdict"] == "REJECT" for x in fake.ledger)
+
+
+@case("执行器·试算模式只算不发不落库")
+def _():
+ from datetime import datetime as _dt
+ from app.services import executor
+ fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
+ positions=[{"ts_code": "600000.SH", "total_qty": 6000,
+ "avail_qty": 6000, "avg_cost": 9.0}])
+ fake.insert_instruction(instruction_id="INS_D", origin_type="plan", origin_id="P3",
+ ts_code="600000.SH", action="EXIT", side="sell", qty=2000,
+ window_tdays=1, status="PROPOSED",
+ progress={"deadline": "2026-07-27", "children": []})
+ r = executor.run_tick(now=_dt(2026, 7, 27, 10, 5), dry_run=True)
+ assert r["fired"] and r["fired"][0]["dry_run"] is True, r
+ assert r["fired"][0]["qty"] == 2000 # 末日全出
+ assert fake.instructions["INS_D"]["status"] == "PROPOSED"
+ assert not fake.instructions["INS_D"]["progress"]["children"]
+ assert not fake.ledger
+
+
+@case("执行器·窗口收口: 成交回写方案 + 足额置确认 + 命令类部分完成")
+def _():
+ from datetime import datetime as _dt
+ from app.services import executor
+ fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"},
+ positions=[{"ts_code": "600000.SH", "total_qty": 6000,
+ "avail_qty": 6000, "avg_cost": 9.0}])
+ fake.insert_plans([{"plan_id": "P_F", "command_id": "CMD_X", "ts_code": "600000.SH",
+ "action": "TRIM", "qty": 2000, "amount": 20000, "priority": 30,
+ "deadline": "2026-07-27", "status": "EXECUTING", "reason": "收利润"}])
+ fake.insert_instruction(instruction_id="INS_F", origin_type="plan", origin_id="P_F",
+ ts_code="600000.SH", action="TRIM", side="sell", qty=2000,
+ status="DISPATCHED", exec_qty=2000,
+ progress={"deadline": "2026-07-27", "is_command": True,
+ "children": []})
+ fake.insert_instruction(instruction_id="INS_P", origin_type="plan", origin_id="P_F",
+ ts_code="600000.SH", action="TRIM", side="sell", qty=2000,
+ status="DISPATCHED", exec_qty=500,
+ progress={"deadline": "2026-07-20", "is_command": True,
+ "children": []})
+ fake.insert_instruction(instruction_id="INS_A2", origin_type="proposal", origin_id="PR1",
+ ts_code="600000.SH", action="ADD", side="buy", qty=1000,
+ status="DISPATCHED", exec_qty=0,
+ progress={"deadline": "2026-07-20", "is_command": False,
+ "children": []})
+ r = executor.sweep_windows(now=_dt(2026, 7, 27, 15, 10))
+ assert fake.instructions["INS_F"]["status"] == "CONFIRMED", fake.instructions["INS_F"]
+ assert [p for p in fake.plans if p["plan_id"] == "P_F"][0]["filled_qty"] == 500
+ assert any(x["instruction_id"] == "INS_P" for x in r["partial"]), r
+ assert fake.instructions["INS_P"]["progress"]["window_verdict"]["verdict"] == "PARTIAL"
+ assert "INS_A2" in r["expired"], r # 自主类窗口耗尽即作废
+ assert fake.instructions["INS_A2"]["status"] == "EXPIRED"
+
+
+@case("下发通道·三模式描述与影子回执; 撤销走本地置状态")
+def _():
+ from app.services import dispatcher, executor, param_store
+ fake = install_fakes()
+ assert dispatcher.mode() == "shadow"
+ d = dispatcher.dispatch(instruction_id="INS_1", ts_code="600000.SH", side="sell",
+ qty=1000, limit_price=9.98)
+ assert d["ok"] and d["ref"] == "manual:INS_1" and "人工" in d["note"], d
+ assert param_store.set_param("PMS_DISPATCH_MODE", "bad_mode")["ok"] is False
+ assert param_store.set_param("PMS_DISPATCH_MODE", "plan_x")["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"] and d2["mode"] == "shadow" and "无卖出通道" in d2["note"], d2
+ param_store.set_param("PMS_DISPATCH_MODE", "shadow")
+
+ fake.insert_instruction(instruction_id="INS_C", origin_type="plan", origin_id="P1",
+ ts_code="600000.SH", action="TRIM", side="sell", qty=1000,
+ status="DISPATCHED", progress={"children": []})
+ r = executor.cancel_instruction("INS_C")
+ assert r["ok"] and fake.instructions["INS_C"]["status"] == "CANCELLED", r
+ assert executor.cancel_instruction("INS_C")["ok"] is False
+
+
+@case("装配·执行相关路由与调度接线到位")
+def _():
+ from app.web.main import app
+ from app import scheduler as sch
+ paths = {r.path for r in app.routes}
+ for p in ("/api/ops/materialize", "/api/ops/exec-tick", "/api/ops/sweep-windows",
+ "/api/instructions/{instruction_id}/cancel", "/api/dispatch-mode"):
+ assert p in paths, p
+ import inspect
+ src = inspect.getsource(sch.intraday_exec)
+ assert "executor" in src and "run_tick" in src, "调度器未接执行器"
+
+
# ---------------------------------------------------------------- runner
def main():
passed, failed = 0, 0