添加交易逻辑

DEVLOG.md

@@ -30,6 +30,47 @@
 
 ---
 
+## 2026-08-25（续）· 策略自动挂载：个股打法状态机（吸筹挂网格 / 高热挂止盈 / 派发停买腿 / 接力切换）
+
+**背景**
+用户要把个股策略从「手工挂」推进到「默认自动用」：有吸筹标志就开始网格建仓，热度很高就挂跟踪止盈。方案文件 STRATEGY_AUTO_ATTACH_PLAN.md（V2 定稿）已入库，五处拍板均已当面定下：只认「明确吸筹」档触发；建仓资格不放宽（吸筹只影响候选排序，不给入场资格）；不设观察期直接全自动；实现骨架取「状态机含接力切换」（网格票站上区间上界且热度达标时自动换挂止盈）；判分闭环进 V1。上线前在桥机跑过一轮只读探测（scripts/probe_strategy_signals.py）：最新结论日吸筹定性 420 只（明确 164），热度全市场 5195 只、不低于 0.80 的共 194 只（约前 4%，阈值 0.80 因此保留），候选池 31 只中 17 只带明确吸筹（拍板②成立），两信号源日龄 1 天。探测同时暴露一个问题：62 只「词表外」定性，根因是首版归类用前缀匹配，而 state 常带前后缀修饰——数据底座 feed.py 用的是子串包含，本次全部改为与它同法（子串 + 派发优先的保守次序）。
+
+**做了什么**
+一，新模块 app/services/strategy_advisor.py：每交易日 09:40 对每只持仓票判阶段、走四条边——明确吸筹且新鲜挂网格（区间锚支撑压力、锚不住退百分比带，锚不出合法区间就放弃不硬凑）；热度超阈值且安全垫为正挂跟踪止盈（双命中也取止盈，负垫永不挂止盈，把位置留给补仓评估）；定性转派发或标志失效只暂停网格买入腿（来源标 accum，回到明确吸筹自动解除，不动卖出侧）；自动网格站上区间上界且热度达标撤网格换挂止盈（接力），高水位从当日现价起算。挂载与切换只做配置动作，真正买卖仍由 strategy_runner 逐笔过规则闸与全部熔断。自动策略的身份全靠 note 约定承载（「自动挂载: 」「自动挂载(接力): 」「[接力撤下]」），每日新挂上限、人工撤下冷却（同票同规则 10 个交易日）、接力冷却（同票网格边 10 个交易日）全部从策略表推导，不加新表不加状态参数。排除项：人工策略不碰、冻结与黑名单不挂网格、有在途不挂、热度表停更时热度边整轮不动、词表外定性按无标志处理并整轮报样本。
+二，strategy_service 补 clear_buypause（按来源解除买入暂停，advisor 只清自己 accum 停的，不放开风控停的）；scheduler 新任务 pms.strategy_attach（09:40，在宏观扫描 09:35 之后，宏观要降仓时在途票自动让路）；web 补试算端点 POST /api/ops/strategy-attach-scan?dry_run=true（只判不写，供判收与日常复核）。
+三，13 个 PMS_AUTO_* 参数进 settings 与参数页（总开关、边清单、日上限 2、吸筹日龄 3、网格带宽/档距/投入比例、热度阈值 0.80、止盈回撤/卖出比例、接力开关与两种冷却），全带中文说明与范围校验，边清单在写入口拦未知边名；总开关进 param_store.FAIL_CLOSED——参数表读不到时按关处理，故障期间宁可不挂。
+四，探测脚本升 v1.1（归类改子串包含并打印词表外原文样本；对读注释改口径说明：探测是单日分布、前端清单是 30 天窗口，比结构不比绝对数）。
+
+**动了哪些文件**
+app/services/strategy_advisor.py（新增）；app/services/strategy_service.py（clear_buypause）；app/services/param_store.py（DESC 13 条 / _RANGES 10 条 / FAIL_CLOSED 加总开关 / 边清单校验）；config/settings.py（PMS_AUTO_* 参数块）；app/scheduler.py（strategy_attach 任务 + beat 09:40 + 调度总表注释）；app/web/main.py（strategy-attach-scan 端点）；scripts/probe_strategy_signals.py（v1.1）；scripts/test_batch17_units.py（新增 34 例：定性归类子串与保守优先级、网格参数生成与三条放弃路径、连边矩阵含双命中与负垫、note 字面量钉死、日上限接力不占、两种冷却推导、接力判定、clear_buypause 来源匹配、编排冒烟 dry_run 滴水不写/名额/边三/边四全链/三道总闸）；scripts/test_wiring.py（beat 哨兵集合加 strategy_attach，路由清单加试算端点）；scripts/run_tests.py（注册 batch17，总数 519 → 553）。
+
+**部署方式**
+桥机 factorevaluation 上收盘后 make deploy（源码打进镜像），make test 见 ALL SUITES PASS。不动 .env，不需要迁移。上线即全自动（拍板③），随时可在参数页把 PMS_AUTO_STRATEGY_ENABLED 或单独把接力 PMS_AUTO_HANDOFF_ENABLED 关掉，即时生效。
+
+**真机判收**
+未判收。开发容器全量单测 ALL SUITES PASS（553 例）。判收建议：部署当天先 curl -X POST 'http://127.0.0.1:38100/api/ops/strategy-attach-scan?dry_run=true' 看它想挂什么、排除原因对不对（尤其 unknown_states 是否为空）；次日 09:40 后看 beat 日志 strategy_attach 一跳的 attached/handoffs/paused，页面策略列表出现「自动挂载: 」开头的条目、台账出现 ATTACH 留痕；挂出的网格由 strategy_runner 正常逐档发单、每笔过闸。
+
+**还欠着什么**
+一，判分闭环的读数脚本（report_strategy_score.py：自动挂载 vs 人工挂载 vs 不挂的对照统计）本次未写——留痕字段（ATTACH/HANDOFF/NOTE 台账 + note 约定）已齐，脚本只读统计，等跑出两周数据再交，先欠着。二，探测顺带发现候选池计划日停在 2026-08-21（探测日 08-25，日龄已 2 个交易日以上），疑似上游桥的日更链又停了，且 PMS_PLAN_STALE_TDAYS 可能被调宽过——与本特性无关，单独排查。三，热度阈值 0.80 与日上限 2 都是首版拍的，判分数据出来后用读数回调。
+
+---
+
+## 2026-08-25 · 软归档：四类记录加 archived_at，已完成默认从在办视图移除（补记）
+
+**做了什么**
+命令 / 策略 / 指令 / 提议四张表加 archived_at 列做软归档：只归**终态**记录（各表终态集在 pms_repo._ARCHIVE_TERMINAL 一处钉死，前端「可移除」判据用同一份），归档只写时间戳不删行，列表默认带 archived_at IS NULL、页面勾「显示已完成」传 include_archived=true 拉回来。repo 层 archive_*/unarchive_* 均为单表 UPDATE、过单表守卫；web 层四类各一对归档/取消归档端点。第十六批单测 5 例钉死：只归终态、默认排除、单表守卫、终态集与设计一致。
+
+**动了哪些文件**
+app/repo/pms_repo.py（archive_*/unarchive_*/list_* 的 include_archived）；app/web/main.py（归档端点与列表参数）；app/web/static/index.html（显示已完成开关与可移除判据）；scripts/ddl_pms_v1.sql（建表带列）；scripts/migrate_archived_at.py（新增：给**已有库**补列，幂等，先演练后 --yes）；scripts/test_batch16_units.py（5 例）；scripts/run_tests.py（batch16 注册，总数 519）。
+
+**部署方式**
+桥机 make deploy 之外**必须先跑一次迁移**：`docker compose run --rm pms-web python scripts/migrate_archived_at.py`（演练）确认四条 ALTER，再加 `--yes` 执行。init_db 只认 CREATE TABLE IF NOT EXISTS，不会给旧表加列——不跑迁移，新代码的列表 SQL 一执行就报未知列。新库不需要（建表语句已带列）。
+
+**真机判收**
+未判收。判收建议：迁移脚本两跑（演练/实做）各表 added 或 skip 清楚；页面把一条 DONE 命令归档后从在办列表消失，勾「显示已完成」回来且标已归档；ALL SUITES PASS 含 batch16。
+
+---
+
 ## 2026-08-20（续）· 清仓与减仓对策略的优先级理顺：减仓不掐网格、清仓补撤在途买单、在途文案分开
 
 **背景**

app/scheduler.py

@@ -10,6 +10,7 @@
 | 盘中执行  | 交易时段每 1 分钟    | 择时出手 + 自主提议扫描 (执行器下一批交付)   |
 | 信号消化  | 交易时段每 1 分钟    | 订阅决策系统盘中信号 (下一批交付)            |
 | 宏观择时  | 交易日 09:35         | 股汇对冲指数 → 升降仓命令或建议 + 宏观闸     |
+| 策略挂载  | 交易日 09:40         | 个股打法状态机: 吸筹挂网格/高热挂止盈/接力   |
 | 成交回放  | 交易时段每 1 分钟    | ws 逐笔入账 + trading_order 增量 + 轻对账     |
 | T 仓平回  | 14:50 (二期)         | 做T强制平回                                  |
 | 日终结算  | 15:10                | 全量对账 / 除权 / 安全垫 / 命令进度日结      |
@@ -195,6 +196,20 @@ def macro_scan():
     return macro_service.scan()
 
 
+@celery_app.task(name="pms.strategy_attach")
+@guard(trade_day=True, respect_exec_halt=True)
+def strategy_attach():
+    """策略自动挂载 (09:40): 个股打法状态机 —— 吸筹挂网格 / 高热挂止盈 / 派发停买腿 / 接力。
+
+    定在 09:40: 在宏观扫描 (09:35) 之后 —— 若宏观当天要降仓, 先让降仓命令占住在途,
+    有在途的票本轮自动挂载会主动让路 (缓到下一个扫描日); 也避开开盘前 15 分钟的
+    竞价噪声, 此刻 positions_view 已有实时价, 网格区间锚得住。只做配置动作不下单,
+    真正买卖由 strategy_runner 逐笔过闸。详见 STRATEGY_AUTO_ATTACH_PLAN.md。
+    """
+    from app.services import strategy_advisor
+    return strategy_advisor.scan()
+
+
 @celery_app.task(name="pms.t0_close")
 @guard(trade_day=True)
 def t0_close():
@@ -250,6 +265,7 @@ celery_app.conf.beat_schedule = {
     "intraday_exec": {"task": "pms.intraday_exec", "schedule": crontab(minute="*")},
     "signal_digest": {"task": "pms.signal_digest", "schedule": crontab(minute="*")},
     "macro_scan": {"task": "pms.macro_scan", "schedule": crontab(hour=9, minute=35)},
+    "strategy_attach": {"task": "pms.strategy_attach", "schedule": crontab(hour=9, minute=40)},
     "t0_close": {"task": "pms.t0_close", "schedule": crontab(hour=14, minute=50)},
     "daily_settle": {"task": "pms.daily_settle", "schedule": crontab(hour=15, minute=10)},
     "daily_report": {"task": "pms.daily_report", "schedule": crontab(hour=15, minute=30)},

app/services/param_store.py

@@ -73,6 +73,9 @@ FAIL_CLOSED = {
     # 基础设施故障期间照常想下命令。安全方向是关: 宁可少动一轮。取数环节同库先挂是第一道
     # 保险, 这条名单是第二道。
     "PMS_MACRO_ENABLED": False,
+    # 2026-08-25 策略自动挂载: 同宏观层的理由 —— 文件初值 True (模拟仓拍板全自动),
+    # 表读不到时退文件初值等于故障期间照常自动挂策略。安全方向是关, 本轮不挂不撤。
+    "PMS_AUTO_STRATEGY_ENABLED": False,
 }
 
 # 页面展示用的中文说明 (settings.py 用行尾注释, pydantic 取不到, 故在此集中维护)
@@ -151,6 +154,19 @@ DESC = {
     "PMS_T0_CLOSE_TIME": "T仓强制平回时点", "PMS_T0_STOCK_DAY_LOSS": "单票当日T亏熔断",
     "PMS_T0_GLOBAL_DAY_LOSS": "全局当日T亏熔断",
     "PMS_STRATEGY_ENABLED": "个股交易方案(策略)层总开关 (关=strategy_runner 空转)",
+    "PMS_AUTO_STRATEGY_ENABLED": "策略自动挂载总开关 (个股打法状态机; 关=09:40 调度位空转不挂不撤)",
+    "PMS_AUTO_STRATEGY_RULES": "启用的边 (逗号分隔): accum_grid 吸筹挂网格 / heat_trail 热度挂止盈 / accum_exit 派发暂停买腿 / handoff 网格换止盈接力",
+    "PMS_AUTO_STRATEGY_DAILY_MAX": "每日自动新挂上限 (只数; 接力换挂是换不是增, 不计入)",
+    "PMS_AUTO_ACCUM_STALE_TDAYS": "吸筹结论日龄超此 (交易日) 视为过期不触发",
+    "PMS_AUTO_GRID_BAND": "网格区间退化带宽: 支撑/压力锚不住时用 现价×(1±带宽)",
+    "PMS_AUTO_GRID_STEP_PCT": "自动网格步长 (占中枢价)",
+    "PMS_AUTO_GRID_CAP_RATIO": "自动网格最大投入占单股上限余量的比例",
+    "PMS_AUTO_HEAT_TH": "热度阈值 0~1 (2026-08-25 实测: ≥0.80 约取全市场前 4%, 194 只)",
+    "PMS_AUTO_TRAIL_GIVEBACK": "自动跟踪止盈: 距高水位回撤触发比例",
+    "PMS_AUTO_TRAIL_SELL_RATIO": "自动跟踪止盈: 触发时卖出比例",
+    "PMS_AUTO_HANDOFF_ENABLED": "边四接力独立开关 (自动网格站上区间上界且热度达标→撤网格换挂止盈)",
+    "PMS_AUTO_HANDOFF_COOLDOWN_TDAYS": "接力后同票网格边冷却 (交易日; 防区间上沿震荡把两种策略翻来覆去换)",
+    "PMS_AUTO_OPTOUT_COOLDOWN_TDAYS": "人工撤下自动策略后同票同规则冷却 (交易日; 尊重人的否决, 防人机拉锯)",
     "PMS_REPLAY_INTERVAL_MIN": "成交回放间隔 (分钟)", "PMS_RECON_ALARM_DAYS": "连续不一致升级天数",
     "PMS_SIGNAL_ENABLED": "是否消化决策系统盘中信号",
     "PMS_SIGNAL_GROUP": "信号消费组名 (独立于 trading_service, 互不抢消息)",
@@ -364,6 +380,12 @@ _RANGES = {
     "PMS_MACRO_MAX_PCT": (0, 1), "PMS_MACRO_WINDOW_TDAYS": (1, 20),
     "PMS_MACRO_STALE_TDAYS": (1, 20), "PMS_MACRO_RET_WIN": (5, 60),
     "PMS_MACRO_Z_WIN": (10, 250), "PMS_MACRO_SHIBOR_BETA": (0, 1),
+    # 策略自动挂载
+    "PMS_AUTO_STRATEGY_DAILY_MAX": (0, 20), "PMS_AUTO_ACCUM_STALE_TDAYS": (1, 20),
+    "PMS_AUTO_GRID_BAND": (0.01, 0.5), "PMS_AUTO_GRID_STEP_PCT": (0.005, 0.2),
+    "PMS_AUTO_GRID_CAP_RATIO": (0, 1), "PMS_AUTO_HEAT_TH": (0, 1),
+    "PMS_AUTO_TRAIL_GIVEBACK": (0.01, 0.5), "PMS_AUTO_TRAIL_SELL_RATIO": (0.1, 1),
+    "PMS_AUTO_HANDOFF_COOLDOWN_TDAYS": (0, 60), "PMS_AUTO_OPTOUT_COOLDOWN_TDAYS": (0, 60),
 }
 
 
@@ -388,6 +410,14 @@ def _range_check(key, v):
         return "PMS_DISPATCH_MODE 只能是 shadow / ws"
     if key == "PMS_CANDIDATE_SOURCE" and v not in ("plan_api", "buy_plan", "both"):
         return "PMS_CANDIDATE_SOURCE 只能是 plan_api / buy_plan / both"
+    if key == "PMS_AUTO_STRATEGY_RULES":
+        # 页面上打错一个边名不会报错、只会静默少一条边 —— 在写入口就拦下来
+        bad = [t.strip() for t in str(v).split(",")
+               if t.strip() and t.strip() not in ("accum_grid", "heat_trail",
+                                                  "accum_exit", "handoff")]
+        if bad:
+            return (f"PMS_AUTO_STRATEGY_RULES 含未知边 {bad}; "
+                    f"可选 accum_grid / heat_trail / accum_exit / handoff")
     lo_hi = _RANGES.get(key)
     if lo_hi and isinstance(v, (int, float)) and not isinstance(v, bool):
         lo, hi = lo_hi

app/services/strategy_advisor.py

@@ -0,0 +1,649 @@
+# -*- coding: utf-8 -*-
+"""
+策略自动挂载 · 个股打法状态机 (STRATEGY_AUTO_ATTACH_PLAN.md V2)
+================================================================
+每交易日 09:40 跑一次 (scheduler.strategy_attach)。对每只持仓票判「阶段」、走「边」:
+
+  阶段: 吸筹震荡 / 启动拉升 / 高位派发 / 深亏修复 / 中性
+  边一  中性→吸筹震荡     明确吸筹且结论新鲜 → 挂网格
+  边二  中性→启动拉升     热度超阈值且安全垫为正 → 挂跟踪止盈 (双命中也取止盈)
+  边三  吸筹震荡→高位派发  定性转派发或标志失效 → 暂停网格买入腿 (来源 accum, 回明确自动解除)
+  边四  吸筹震荡→启动拉升  自动网格票站上区间上界且热度超阈值 → 撤网格换挂止盈 (接力)
+
+**挂载和切换本身不下单。** 真正的买卖仍由 strategy_runner 每分钟评估、每一笔过规则闸与
+全部熔断。本模块只做「配置动作」: 读信号、判阶段、调 strategy_service.attach / set_status,
+全程留痕 (pms_action_ledger: ATTACH / HANDOFF / NOTE)。
+
+两个信号源 (153 代理, 严格单表; 口径同源声明见方案第二节):
+  吸筹定性  strategy_daily_results.raw_logic_json 的 fund_flow.state —— 决策系统每晚产出。
+            判定用**子串包含**而不是前缀 (与数据底座 feed.py 的 _ACCUM_KEEP 同手法):
+            实测 state 会带前后缀修饰, 前缀匹配漏了 2026-08-25 探测里 62 只「词表外」。
+            含「派发」优先于含「明确吸筹」—— 两个词同现时按保守方向算派发。
+            词表外的 state 一律当无标志 (宁可不挂), 并在返回里报出来供核对契约。
+  热度分    stock_fund_heat_scores 最新交易日最大批次, 0~1。全市场约 5200 只,
+            阈值 0.80 约取前 4% (2026-08-25 探测: ≥0.8 共 194 只)。
+
+自动策略的「身份」全靠 note 约定承载, **不加任何新表新参数状态** (可从策略表完整推导,
+审计与测试都只看得见的东西):
+  "自动挂载: ..."        本模块常规挂出的 (计入每日新挂上限)
+  "自动挂载(接力): ..."  边四换挂出来的止盈 (不占每日上限 —— 它是换不是增)
+  note 里含 "[接力撤下]"  被边四撤掉的网格 (据此算接力冷却, 不算人工撤下)
+由此派生的两种冷却 (方案第七节小口径):
+  人工撤下: 自动策略被撤、票还持有、note 无接力标记 → 同票同规则 N 个交易日不再自动挂
+  接力之后: 同票网格边 N 个交易日不再挂 (防区间上沿来回震把两种策略翻来覆去换)
+"""
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timedelta
+
+from app.core import tradedays as td
+from app.db.session import fetch_all, fetch_one
+from app.repo import pms_repo
+from app.repo.downstream_repo import to_dot, to_prefix
+from app.services import param_store
+
+logger = logging.getLogger("pms.strategy_advisor")
+
+# 决策系统定性词表 (契约)。判定看子串, 次序即优先级 —— 派发最先 (保守方向)。
+CLS_DISTRIB = "高位派发"
+CLS_CLEAR, CLS_MAYBE = "明确吸筹", "潜在吸筹"
+CLS_NONE_SIGN, CLS_UNCLEAR = "无吸筹迹象", "信号不明"
+CLS_UNKNOWN, CLS_NOFIELD = "词表外", "无字段"
+_CLS_ORDER = (("派发", CLS_DISTRIB), ("明确吸筹", CLS_CLEAR), ("潜在吸筹", CLS_MAYBE),
+              ("无吸筹迹象", CLS_NONE_SIGN), ("不明", CLS_UNCLEAR))
+
+# 边 (规则注册表键名, PMS_AUTO_STRATEGY_RULES 里逗号列出即启用)
+R_GRID, R_TRAIL, R_EXIT, R_HANDOFF = "accum_grid", "heat_trail", "accum_exit", "handoff"
+
+# note 约定 (见模块头; 改这里必须同步改 test_batch17 钉住的字面量)
+NOTE_AUTO = "自动挂载: "
+NOTE_HANDOFF = "自动挂载(接力): "
+MARK_HANDOFF_OUT = "[接力撤下]"
+
+ACCUM_WINDOW_DAYS = 45          # 每票取近 45 自然日内最新一条结论
+HEAT_MAX_AGE_DAYS = 4           # 热度表末日落后超此自然日 → 热度信号本轮不可用
+_RULE_OF_TYPE = {"GRID": R_GRID, "TRAIL": R_TRAIL}
+
+
+def _f(v, d=None):
+    try:
+        return float(v)
+    except (TypeError, ValueError):
+        return d
+
+
+# ================================================================ 纯逻辑
+def classify_accum(state) -> str:
+    """定性字符串 → 档位。子串包含 + 固定优先级; 判不出的当词表外 (调用方按无标志处理)。"""
+    s = str(state or "").strip()
+    if not s:
+        return CLS_NOFIELD
+    for key, cls in _CLS_ORDER:
+        if key in s:
+            return cls
+    return CLS_UNKNOWN
+
+
+def grid_params(*, price, support, pressure, band, step_pct, cap_room, cap_ratio):
+    """网格参数自动生成 (方案附录二)。返回 (params, why); params=None 时 why 说明放弃原因。
+
+    区间优先锚支撑压力, 锚不住退百分比带; 任何一步不满足 0<下界<中枢<上界 就放弃不硬凑。
+    """
+    p = _f(price, 0.0)
+    if not p or p <= 0:
+        return None, "取不到实时价, 网格区间无从定"
+    cap = _f(cap_room, 0.0) or 0.0
+    max_capital = round(cap * _f(cap_ratio, 0.5), 2)
+    if max_capital < p * 100:
+        return None, (f"单股上限余量 {cap:,.0f} 元按投入比例折出 {max_capital:,.0f} 元, "
+                      f"买不起一手, 不挂")
+    r, s = _f(pressure, 0.0) or 0.0, _f(support, 0.0) or 0.0
+    upper = round(r * 1.01, 3) if r > p else round(p * (1 + band), 3)
+    lower = round(s * 0.99, 3) if 0 < s < p else round(p * (1 - band), 3)
+    if not (0 < lower < p < upper):
+        return None, (f"区间不成立 (下界 {lower} / 现价 {p} / 上界 {upper}), "
+                      f"支撑压力形态不适合网格, 本轮放弃")
+    step = max(0.005, _f(step_pct, 0.02))
+    n_below = max(1, int((p - lower) / (p * step)))
+    per_lot = int(max_capital / n_below / p / 100) * 100
+    if per_lot < 100:
+        per_lot = 100
+    return ({"center": p, "lower": lower, "upper": upper, "step_pct": step,
+             "per_lot": per_lot, "max_capital": max_capital}, "")
+
+
+def plan_edge(*, cls, fresh, heat, cushion, prm):
+    """无策略持仓票该走哪条边。返回 (边|None, 原因)。
+
+    双命中取止盈 (保住利润优先于做波段); 负垫不挂止盈 (挂了永远不武装, 还把该票从
+    动作引擎排除, 白挡掉深亏补仓的评估) —— 方案第七节小口径。
+    """
+    rules = prm["rules"]
+    hot = heat is not None and heat >= prm["heat_th"]
+    pos = cushion is not None and _f(cushion, 0.0) > 0
+    clear = (cls == CLS_CLEAR and fresh)
+    if R_TRAIL in rules and hot and pos:
+        return R_TRAIL, ("明确吸筹与高热度双命中, 按口径取跟踪止盈" if clear
+                         else f"热度 {heat:.3f} 超阈值 {prm['heat_th']:.2f} 且有浮盈")
+    if R_GRID in rules and clear:
+        return R_GRID, "明确吸筹且结论新鲜, 适合网格吃震荡"
+    if hot and not pos:
+        return None, "热度够但安全垫不正, 不挂止盈 (把位置留给补仓评估)"
+    if cls == CLS_CLEAR and not fresh:
+        return None, "吸筹结论超日龄, 视为无标志"
+    return None, ""
+
+
+def count_auto_today(rows, today_ymd: int) -> int:
+    """今天已常规自动挂载几条 (接力挂出的不算 —— 它是换不是增)。rows=策略行 (含已归档)。"""
+    n = 0
+    for r in rows or []:
+        note = str(r.get("note") or "")
+        if not note.startswith(NOTE_AUTO):
+            continue
+        if _ymd_of(r.get("created_at")) == today_ymd:
+            n += 1
+    return n
+
+
+def cooldowns_from_cancelled(rows, held_codes, today_ymd: int, *,
+                             optout_tdays: int, handoff_tdays: int):
+    """从已撤销的自动策略行推导两种冷却 (无状态, 全部可从表推导)。
+
+    返回 (optout: {(code, rule)}, handoff_cool: {code})。
+    人工撤下 = 自动策略被撤、票还持有、note 无接力标记 —— 撤它的可能是你、也可能是
+    清仓命令或清场; 后两种情形持仓多半已归零, 「票还持有」这一条把它们天然排除,
+    剩下的按"有人特意撤过"处理, 冷却期内不再自动挂, 杜绝人机拉锯。
+    交易日换算按自然日乘二宽松 (与全库惯例一致), 冷却只会偏长不会偏短。
+    """
+    optout, hand = set(), set()
+    held = set(held_codes or ())
+    for r in rows or []:
+        note = str(r.get("note") or "")
+        if not note.startswith("自动挂载"):
+            continue
+        code = r.get("ts_code")
+        age = _age_days_of(r.get("updated_at"), today_ymd)
+        if age is None:
+            continue
+        if MARK_HANDOFF_OUT in note:
+            if age <= handoff_tdays * 2:
+                hand.add(code)
+            continue
+        rule = _RULE_OF_TYPE.get(str(r.get("type") or "").upper())
+        if rule and code in held and age <= optout_tdays * 2:
+            optout.add((code, rule))
+    return optout, hand
+
+
+def handoff_ready(*, price, price_ok, upper, heat, cushion, prm):
+    """边四触发判定 (不含「无在途委托」那条 —— 那要查库, 由编排层补)。返回 (bool, why)。"""
+    if not prm.get("handoff_enabled") or R_HANDOFF not in prm["rules"]:
+        return False, ""
+    if not price_ok or _f(price, 0.0) <= 0:
+        return False, ""
+    if _f(upper, 0.0) <= 0 or _f(price) < _f(upper):
+        return False, ""
+    if heat is None or heat < prm["heat_th"]:
+        return False, f"已站上网格上界但热度 {heat if heat is not None else '缺失'} 未达阈值, 继续网格"
+    if cushion is None or _f(cushion, 0.0) <= 0:
+        return False, "已站上网格上界但安全垫不正, 不接力"
+    return True, (f"现价 {price} 站上网格上界 {upper} 且热度 {heat:.3f} 超阈值 —— "
+                  f"吸筹震荡期转启动拉升期, 网格换跟踪止盈")
+
+
+def _ymd_of(ts) -> int:
+    try:
+        if hasattr(ts, "strftime"):
+            return int(ts.strftime("%Y%m%d"))
+        s = str(ts or "").strip().replace("-", "")[:8]
+        return int(s) if s.isdigit() and len(s) == 8 else 0
+    except (TypeError, ValueError):
+        return 0
+
+
+def _age_days_of(ts, today_ymd: int):
+    y = _ymd_of(ts)
+    if not y:
+        return None
+    try:
+        a = datetime.strptime(str(y), "%Y%m%d").date()
+        b = datetime.strptime(str(today_ymd), "%Y%m%d").date()
+        return (b - a).days
+    except ValueError:
+        return None
+
+
+# ================================================================ 取数 (153 代理, 单表)
+def _variants(dot_codes):
+    rev = {}
+    for c in dot_codes or []:
+        d = to_dot(c)
+        if not d:
+            continue
+        rev[d] = d
+        rev[to_prefix(d)] = d
+        rev[d.split(".")[0]] = d
+    return rev
+
+
+def _in(values, prefix, params):
+    keys = []
+    for i, v in enumerate(values):
+        keys.append(f":{prefix}{i}")
+        params[f"{prefix}{i}"] = v
+    return ", ".join(keys)
+
+
+def accum_of(dot_codes) -> dict:
+    """{点式: {cls, state, score, ymd, age_days}} —— 窗口内每票最新一条定性。"""
+    if not dot_codes:
+        return {}
+    rev = _variants(dot_codes)
+    since = int((datetime.now().date() - timedelta(days=ACCUM_WINDOW_DAYS)).strftime("%Y%m%d"))
+    p = {"since": since}
+    rows = fetch_all(
+        "SELECT stock_code, trade_date, raw_logic_json FROM strategy_daily_results "
+        f"WHERE trade_date >= :since AND stock_code IN ({_in(list(rev), 'c', p)})", p)
+    best = {}
+    for r in rows:
+        dot = rev.get(str(r.get("stock_code") or "").strip())
+        ymd = _ymd_of(r.get("trade_date"))
+        if not dot or not ymd or (dot in best and best[dot]["ymd"] >= ymd):
+            continue
+        try:
+            raw = r.get("raw_logic_json")
+            d = raw if isinstance(raw, dict) else json.loads(raw or "{}")
+            ff = d.get("fund_flow") or {}
+        except (ValueError, TypeError):
+            ff = {}
+        state = ff.get("state") if isinstance(ff, dict) else None
+        best[dot] = {"ymd": ymd, "age_days": _age_days_of(ymd, td.ymd()),
+                     "cls": classify_accum(state), "state": state,
+                     "score": _f((ff or {}).get("score"))}
+    return best
+
+
+def heat_of(dot_codes):
+    """({点式: score}, meta)。meta.stale=True 表示热度表停更, 调用方按热度不可用处理。"""
+    meta = {"td": None, "batch": None, "stale": True}
+    if not dot_codes:
+        return {}, meta
+    r = fetch_one("SELECT MAX(trade_date) AS td FROM stock_fund_heat_scores")
+    tdd = (r or {}).get("td")
+    if tdd is None:
+        return {}, meta
+    r2 = fetch_one("SELECT MAX(batch_no) AS b FROM stock_fund_heat_scores "
+                   "WHERE trade_date = :td", {"td": tdd})
+    rev = _variants(dot_codes)
+    p = {"td": tdd, "b": (r2 or {}).get("b")}
+    rows = fetch_all(
+        "SELECT stock_code, score FROM stock_fund_heat_scores "
+        f"WHERE trade_date = :td AND batch_no = :b AND stock_code IN ({_in(list(rev), 'c', p)})",
+        p)
+    out = {}
+    for x in rows:
+        dot = rev.get(str(x.get("stock_code") or "").strip())
+        v = _f(x.get("score"))
+        if dot and v is not None:
+            out[dot] = v
+    age = _age_days_of(_ymd_of(tdd), td.ymd())
+    meta.update({"td": str(tdd), "batch": (r2 or {}).get("b"),
+                 "stale": age is None or age > HEAT_MAX_AGE_DAYS, "age_days": age})
+    return out, meta
+
+
+# ================================================================ 参数
+def _params() -> dict:
+    g = param_store
+    return {
+        "enabled": g.get_bool("PMS_AUTO_STRATEGY_ENABLED", False),
+        "rules": set(g.get_list("PMS_AUTO_STRATEGY_RULES",
+                                [R_GRID, R_TRAIL, R_EXIT, R_HANDOFF])),
+        "daily_max": g.get_int("PMS_AUTO_STRATEGY_DAILY_MAX", 2),
+        "accum_stale_tdays": g.get_int("PMS_AUTO_ACCUM_STALE_TDAYS", 3),
+        "grid_band": g.get_float("PMS_AUTO_GRID_BAND", 0.08),
+        "grid_step_pct": g.get_float("PMS_AUTO_GRID_STEP_PCT", 0.02),
+        "grid_cap_ratio": g.get_float("PMS_AUTO_GRID_CAP_RATIO", 0.5),
+        "heat_th": g.get_float("PMS_AUTO_HEAT_TH", 0.80),
+        "trail_giveback": g.get_float("PMS_AUTO_TRAIL_GIVEBACK", 0.05),
+        "trail_sell_ratio": g.get_float("PMS_AUTO_TRAIL_SELL_RATIO", 0.5),
+        "handoff_enabled": g.get_bool("PMS_AUTO_HANDOFF_ENABLED", True),
+        "handoff_cooldown_tdays": g.get_int("PMS_AUTO_HANDOFF_COOLDOWN_TDAYS", 10),
+        "optout_cooldown_tdays": g.get_int("PMS_AUTO_OPTOUT_COOLDOWN_TDAYS", 10),
+    }
+
+
+# ================================================================ 编排
+def scan(*, dry_run: bool = False, now=None) -> dict:
+    """每交易日一跳 (09:40)。dry_run=True 只判不写 —— 不挂、不留痕、不动暂停表。"""
+    now = now or datetime.now()
+    today = td.ymd(now)
+    prm = _params()
+    out = {"ok": True, "enabled": prm["enabled"], "dry_run": dry_run, "checked": 0,
+           "attached": [], "handoffs": [], "paused": [], "resumed": [],
+           "blocked": [], "skipped": [], "errors": [],
+           "unknown_states": []}
+    if not prm["enabled"]:
+        out["skipped"].append({"why": "策略自动挂载总开关关闭 (PMS_AUTO_STRATEGY_ENABLED)"})
+        return out
+    if not param_store.get_bool("PMS_STRATEGY_ENABLED", False):
+        out["skipped"].append({"why": "策略层总开关关闭 (PMS_STRATEGY_ENABLED) —— "
+                                      "挂了也不会跑, 本轮不挂"})
+        return out
+    if param_store.get_bool("PMS_GLOBAL_EXEC_HALT", False):
+        out["skipped"].append({"why": "全局暂停执行 (休假模式)"})
+        return out
+
+    from app.services import command_service, portfolio, strategy_service
+    try:
+        view = portfolio.positions_view()
+    except Exception as e:  # noqa: BLE001 —— 守成: 读不到持仓不动任何东西
+        return {**out, "ok": False, "errors": [f"读持仓失败: {type(e).__name__}: {e}"]}
+    held = [p for p in view["held"] if int(p.get("total_qty") or 0) > 0]
+    if not held:
+        out["skipped"].append({"why": "当前无持仓"})
+        return out
+    codes = [p["ts_code"] for p in held]
+
+    # ---- 信号与既有状态一次取齐 (任一失败只废对应的边, 不废整轮) ----
+    try:
+        accum = accum_of(codes)
+    except Exception as e:  # noqa: BLE001
+        accum = {}
+        out["errors"].append(f"吸筹取数失败 (边一/三/四本轮不动): {type(e).__name__}: {e}")
+    try:
+        heat, heat_meta = heat_of(codes)
+        if heat_meta.get("stale"):
+            heat = {}
+            out["skipped"].append({"why": f"热度表停更 (最新 {heat_meta.get('td')}), "
+                                          f"热度相关的边本轮不动"})
+    except Exception as e:  # noqa: BLE001
+        heat = {}
+        out["errors"].append(f"热度取数失败 (边二/四本轮不动): {type(e).__name__}: {e}")
+    for c in codes:
+        a = accum.get(c)
+        if a and a["cls"] == CLS_UNKNOWN:
+            out["unknown_states"].append({"ts_code": c, "state": a.get("state")})
+    if out["unknown_states"]:
+        logger.warning("[自动挂载] %s 只票的吸筹定性在词表外 (按无标志处理), 样本: %s —— "
+                       "五档契约可能变了, 与决策系统核对", len(out["unknown_states"]),
+                       out["unknown_states"][:3])
+
+    try:
+        strat_all = pms_repo.list_strategies(statuses=["ACTIVE", "PAUSED"], limit=500,
+                                             include_archived=True)
+    except Exception as e:  # noqa: BLE001
+        return {**out, "ok": False, "errors": out["errors"] +
+                [f"读策略表失败, 整轮守成不动: {type(e).__name__}: {e}"]}
+    strat_by_code = {}
+    for s in strat_all:
+        strat_by_code.setdefault(s.get("ts_code"), s)
+    try:
+        cancelled = pms_repo.list_strategies(statuses=["CANCELLED"], limit=300,
+                                             include_archived=True)
+    except Exception as e:  # noqa: BLE001
+        cancelled = []
+        out["errors"].append(f"读已撤策略失败 (冷却按无算): {e}")
+    optout, handoff_cool = cooldowns_from_cancelled(
+        cancelled, codes, today, optout_tdays=prm["optout_cooldown_tdays"],
+        handoff_tdays=prm["handoff_cooldown_tdays"])
+    attached_today = count_auto_today(
+        (list(strat_all) + list(cancelled)), today)
+
+    try:
+        black = command_service.blacklist()
+    except Exception:  # noqa: BLE001
+        black = set()
+    try:
+        from app.services import executor
+        live_codes = {i.get("ts_code") for i in
+                      pms_repo.list_instructions(statuses=list(executor.LIVE), limit=300)}
+        live_codes |= {pl.get("ts_code") for pl in
+                       pms_repo.list_plans(statuses=["PENDING", "GATED", "EXECUTING"],
+                                           limit=300)}
+    except Exception as e:  # noqa: BLE001
+        live_codes = set()
+        out["errors"].append(f"读在途失败 (在途排除按无算): {e}")
+    buypause = {}
+    try:
+        buypause = strategy_service.buypause_map()
+    except Exception:  # noqa: BLE001
+        pass
+
+    # ---- 逐票走状态机 ----
+    for p in held:
+        out["checked"] += 1
+        code = p["ts_code"]
+        try:
+            a = accum.get(code) or {}
+            fresh = (a.get("age_days") is not None
+                     and a["age_days"] <= prm["accum_stale_tdays"] * 2)
+            hv = heat.get(code)
+            st = strat_by_code.get(code)
+            if st:
+                _tend_existing(st, p, a, fresh, hv, prm, buypause, handoff_cool,
+                               today, dry_run, out, strategy_service)
+                continue
+            edge, why = plan_edge(cls=a.get("cls"), fresh=fresh, heat=hv,
+                                  cushion=p.get("cushion_pct"), prm=prm)
+            if not edge:
+                if why:
+                    out["skipped"].append({"ts_code": code, "why": why})
+                continue
+            _attach_one(edge, why, p, a, hv, prm, view, black, live_codes, optout,
+                        attached_today, today, dry_run, out, strategy_service)
+            attached_today = out["_attached_today"]
+        except Exception as e:  # noqa: BLE001 —— 单票异常不拖垮整轮
+            logger.exception("[自动挂载] %s 处理失败", code)
+            out["errors"].append(f"{code}: {type(e).__name__}: {e}")
+    out.pop("_attached_today", None)
+    out["ok"] = not out["errors"]
+    return out
+
+
+def _tend_existing(st, p, a, fresh, hv, prm, buypause, handoff_cool, today,
+                   dry_run, out, strategy_service):
+    """已挂策略的票: 边三 (派发停买/回明确恢复) 与 边四 (接力切换)。"""
+    code = p["ts_code"]
+    note = str(st.get("note") or "")
+    is_auto = note.startswith("自动挂载")
+    stype = str(st.get("type") or "").upper()
+    if not is_auto:
+        out["skipped"].append({"ts_code": code, "why": f"挂着人工策略 ({stype}), 自动挂载不碰它"})
+        return
+    if stype == "GRID":
+        # ---- 边四: 接力切换 (先于边三判 —— 都成立时说明已在拉升, 换止盈比停买更对) ----
+        ok_h, why_h = handoff_ready(price=p.get("price"), price_ok=p.get("price_ok"),
+                                    upper=(st.get("params") or {}).get("upper"),
+                                    heat=hv, cushion=p.get("cushion_pct"), prm=prm)
+        if ok_h and code in handoff_cool:
+            ok_h, why_h = False, "接力条件到了但在接力冷却期内, 不动"
+        if ok_h and _has_pending(st):
+            ok_h, why_h = False, "接力条件到了但策略有在途委托, 等它走完 (下一跳再看)"
+        if ok_h:
+            if dry_run:
+                out["handoffs"].append({"ts_code": code, "from": st["strategy_id"],
+                                        "dry_run": True, "why": why_h})
+                return
+            _do_handoff(st, p, hv, prm, why_h, out, strategy_service)
+            return
+        if why_h:
+            out["skipped"].append({"ts_code": code, "why": why_h})
+        # ---- 边三: 派发/失效 → 停买腿; 回明确 → 解除 accum 来源的暂停 ----
+        if R_EXIT not in prm["rules"]:
+            return
+        cls = a.get("cls")
+        pause_ent = (buypause or {}).get(code) or {}
+        if cls == CLS_DISTRIB or (cls == CLS_CLEAR and not fresh) or cls in (
+                CLS_NONE_SIGN, CLS_UNKNOWN, CLS_NOFIELD):
+            why = ("吸筹定性转高位派发" if cls == CLS_DISTRIB
+                   else "吸筹标志消失或超日龄")
+            if pause_ent:
+                return                          # 已经停着 (accum 或风控来源), 不重复
+            if dry_run:
+                out["paused"].append({"ts_code": code, "dry_run": True, "why": why})
+                return
+            ids = strategy_service.pause_buy(code, reason=f"{why}, 网格买入腿暂停 (卖出照常)",
+                                             source="accum")
+            out["paused"].append({"ts_code": code, "strategies": ids, "why": why})
+            _ledger(code, "NOTE", f"边三: {why} —— 网格买入腿已暂停, 卖出与已买档位照常",
+                    {"accum_state": a.get("state"), "accum_ymd": a.get("ymd")},
+                    st["strategy_id"], out, action="NOTE")
+        elif cls == CLS_CLEAR and fresh and pause_ent.get("source") == "accum":
+            if dry_run:
+                out["resumed"].append({"ts_code": code, "dry_run": True})
+                return
+            r = strategy_service.clear_buypause(code, only_source="accum") or {}
+            if r.get("cleared"):
+                out["resumed"].append({"ts_code": code})
+                _ledger(code, "NOTE", "边三解除: 定性回到明确吸筹, 网格买入腿恢复",
+                        {"accum_state": a.get("state")}, st["strategy_id"], out,
+                        action="NOTE")
+            elif r.get("error"):
+                out["errors"].append(f"{code} 解除买入暂停失败: {r['error']}")
+    else:
+        out["skipped"].append({"ts_code": code, "why": f"已挂自动 {stype}, 本轮无事"})
+
+
+def _attach_one(edge, why, p, a, hv, prm, view, black, live_codes, optout,
+                attached_today, today, dry_run, out, strategy_service):
+    """无策略票走边一/边二: 排除项 → 参数生成 → 校验挂载 → 留痕。"""
+    code = p["ts_code"]
+    out["_attached_today"] = attached_today
+    if (code, edge) in optout:
+        out["skipped"].append({"ts_code": code, "why":
+                               f"人工撤下过同类自动策略, 冷却期内不再自动挂 "
+                               f"({prm['optout_cooldown_tdays']} 个交易日)"})
+        return
+    if code in live_codes:
+        out["skipped"].append({"ts_code": code, "why": "有在途指令或未完成方案, 缓到下一个扫描日"})
+        return
+    if edge == R_GRID:
+        if (p.get("frozen_reason") or "NONE") != "NONE":
+            out["skipped"].append({"ts_code": code, "why": "该股被冻结 (禁增持), 网格只剩半条腿, 不挂"})
+            return
+        if code in black:
+            out["skipped"].append({"ts_code": code, "why": "该股在黑名单, 网格买腿必被闸拦, 不挂"})
+            return
+    if attached_today >= prm["daily_max"]:
+        out["blocked"].append({"ts_code": code, "edge": edge,
+                               "why": f"今日新挂名额已满 ({prm['daily_max']} 条), 留到明天"})
+        if not dry_run:
+            _ledger(code, "NOTE", f"想挂 {edge} 但今日新挂名额已满, 留到明天",
+                    _snap(a, hv, p), None, out)
+        return
+
+    prm_view = view["params"]
+    if edge == R_GRID:
+        cap_room = max(0.0, _f(prm_view.get("stock_cap"), 0.08)
+                       * _f(prm_view.get("scale"), 0.0)
+                       - _f(p.get("market_value"), 0.0))
+        gp, gwhy = grid_params(price=(p.get("price") if p.get("price_ok") else None),
+                               support=p.get("support_ref"), pressure=p.get("pressure_ref"),
+                               band=prm["grid_band"], step_pct=prm["grid_step_pct"],
+                               cap_room=cap_room, cap_ratio=prm["grid_cap_ratio"])
+        if not gp:
+            out["blocked"].append({"ts_code": code, "edge": edge, "why": gwhy})
+            if not dry_run:
+                _ledger(code, "NOTE", f"想挂网格但放弃: {gwhy}", _snap(a, hv, p), None, out)
+            return
+        stype, params = "GRID", gp
+        note = (f"{NOTE_AUTO}明确吸筹(评分 {a.get('score')}, 结论日 {a.get('ymd')}) → 网格 "
+                f"[{gp['lower']}~{gp['upper']}] 档距 {gp['step_pct']:.1%} "
+                f"上限 {gp['max_capital']:,.0f} 元")
+    else:
+        stype = "TRAIL"
+        params = {"giveback": prm["trail_giveback"], "sell_ratio": prm["trail_sell_ratio"]}
+        note = (f"{NOTE_AUTO}热度 {hv:.3f} 超阈值 {prm['heat_th']:.2f}"
+                + (", 与明确吸筹双命中取止盈" if a.get("cls") == CLS_CLEAR else "")
+                + f" → 跟踪止盈 回撤 {prm['trail_giveback']:.1%} 卖 {prm['trail_sell_ratio']:.0%}")
+
+    if dry_run:
+        out["attached"].append({"ts_code": code, "type": stype, "dry_run": True,
+                                "why": why, "params": params})
+        out["_attached_today"] = attached_today + 1
+        return
+    r = strategy_service.attach({"ts_code": code, "type": stype, "autonomy": "auto",
+                                 "params": params, "note": note[:280]}, by="auto") or {}
+    if not r.get("ok"):
+        errs = "; ".join(str(x) for x in (r.get("errors") or ["挂载校验未过"]))
+        out["blocked"].append({"ts_code": code, "edge": edge, "why": errs})
+        _ledger(code, "NOTE", f"想挂 {stype} 被挂载校验挡下: {errs}", _snap(a, hv, p), None, out)
+        return
+    out["attached"].append({"ts_code": code, "type": stype,
+                            "strategy_id": r.get("strategy_id"), "why": why})
+    out["_attached_today"] = attached_today + 1
+    _ledger(code, "PASS", f"{note[:200]} —— {why}",
+            {**_snap(a, hv, p), "params": params}, r.get("strategy_id"), out,
+            action="ATTACH")
+    logger.warning("[自动挂载] %s 挂 %s: %s", code, stype, why)
+
+
+def _do_handoff(st, p, hv, prm, why, out, strategy_service):
+    """边四落地: 撤网格 (打接力标记) → 挂止盈 → 高水位从当日高点起算 → 留痕。"""
+    code = p["ts_code"]
+    old_note = str(st.get("note") or "")
+    r1 = strategy_service.set_status(st["strategy_id"], "CANCELLED", by="auto") or {}
+    if not r1.get("ok"):
+        out["errors"].append(f"{code} 接力第一步撤网格失败: {r1.get('error')}")
+        return
+    n = pms_repo.update_strategy(st["strategy_id"],
+                                 note=(old_note + " " + MARK_HANDOFF_OUT)[:280])
+    if not n:
+        # 标记没打上: 冷却推导会把这次接力误判成人工撤下 —— 只是偏保守 (多冷却), 但要留痕
+        out["errors"].append(f"{code} 接力标记没写上 (影响 0 行), 冷却推导会偏保守")
+    params = {"giveback": prm["trail_giveback"], "sell_ratio": prm["trail_sell_ratio"]}
+    note = (f"{NOTE_HANDOFF}{why}"[:200] + f" 回撤 {prm['trail_giveback']:.1%}")
+    r2 = strategy_service.attach({"ts_code": code, "type": "TRAIL", "autonomy": "auto",
+                                  "params": params, "note": note[:280]}, by="auto") or {}
+    if not r2.get("ok"):
+        errs = "; ".join(str(x) for x in (r2.get("errors") or ["挂载校验未过"]))
+        out["errors"].append(f"{code} 接力第二步挂止盈失败 (网格已撤!): {errs} —— "
+                             f"该票此刻无策略保护, 请人工处理")
+        _ledger(code, "REJECT", f"接力半途而废: 网格已撤但止盈没挂上 ({errs})",
+                {"heat": hv}, st["strategy_id"], out, action="HANDOFF")
+        return
+    hw0 = _f((p.get("price") if p.get("price_ok") else None), 0.0)
+    if hw0:
+        nn = pms_repo.update_strategy(r2["strategy_id"], state={"high_water": round(hw0, 3)})
+        if not nn:
+            out["errors"].append(f"{code} 止盈高水位初始化没写上 (影响 0 行), "
+                                 f"将从下一跳现价起算, 只偏保守")
+    out["handoffs"].append({"ts_code": code, "from": st["strategy_id"],
+                            "to": r2.get("strategy_id"), "why": why})
+    _ledger(code, "PASS", f"接力切换: {why}",
+            {"heat": hv, "from": st["strategy_id"], "to": r2.get("strategy_id")},
+            r2.get("strategy_id"), out, action="HANDOFF")
+    logger.warning("[自动挂载] %s 接力: 网格 %s → 止盈 %s", code, st["strategy_id"],
+                   r2.get("strategy_id"))
+
+
+def _has_pending(st) -> bool:
+    """策略上一笔委托还在途? 查不到按无在途 (接力多等一天没有代价, 误停一天有)。"""
+    pend = (st.get("state") or {}).get("pending") or {}
+    iid = pend.get("iid")
+    if not iid:
+        return False
+    try:
+        ins = pms_repo.get_instruction(iid)
+    except Exception:  # noqa: BLE001
+        return False
+    return bool(ins and ins.get("status") in ("PROPOSED", "RULE_PASSED", "DISPATCHED"))
+
+
+def _snap(a, hv, p) -> dict:
+    return {"accum_state": (a or {}).get("state"), "accum_score": (a or {}).get("score"),
+            "accum_ymd": (a or {}).get("ymd"), "heat": hv,
+            "price": (p or {}).get("price"), "cushion_pct": (p or {}).get("cushion_pct")}
+
+
+def _ledger(code, verdict, reason, hard, ref_id, out, *, action="ATTACH"):
+    try:
+        pms_repo.insert_ledger(ts_code=code, action=action, arbiter="rule", verdict=verdict,
+                               price_at=_f((hard or {}).get("price"), 0.0) or 0,
+                               hard_numbers=hard, ref_id=ref_id, reason=str(reason)[:500])
+    except Exception as e:  # noqa: BLE001 —— 留痕失败不拦动作本体, 但要说
+        out["errors"].append(f"{code} 留痕失败: {type(e).__name__}: {e}")

app/services/strategy_service.py

@@ -194,3 +194,25 @@ def resume_buy(strategy_id: str, by: str = "user") -> dict:
         if not r.get("ok"):
             return {"ok": False, "error": r.get("error")}
     return {"ok": True, "strategy_id": strategy_id, "ts_code": code}
+
+
+def clear_buypause(ts_code: str, only_source: str = None) -> dict:
+    """按票解除买入暂停 (strategy_advisor 边三恢复用, 2026-08-25)。
+
+    only_source 给了就只清对应来源的暂停项 —— advisor 传 "accum", 这样它只解除
+    自己停的, 不会把决策系统风控 (source="signal") 停的顺手放开; 页面上人停的同理。
+    cleared=False 且 ok=True 表示没有可清的项 (没停过, 或来源不匹配), 不算错。"""
+    if not ts_code:
+        return {"ok": False, "cleared": False, "error": "ts_code 为空"}
+    m = buypause_map()
+    ent = m.get(ts_code)
+    if not ent:
+        return {"ok": True, "cleared": False}
+    if only_source and ent.get("source") != only_source:
+        return {"ok": True, "cleared": False,
+                "why": f"暂停来源是 {ent.get('source')}, 不动 (只清 {only_source})"}
+    m.pop(ts_code, None)
+    r = _save_buypause(m)
+    if not r.get("ok"):
+        return {"ok": False, "cleared": False, "error": r.get("error")}
+    return {"ok": True, "cleared": True, "ts_code": ts_code}

app/web/main.py

@@ -536,6 +536,15 @@ def api_macro_adopt(payload: dict = Body(default={})):
     return ok_logged("macro_adopt", macro_service.adopt, key, params={"signal_key": key})
 
 
+@app.post("/api/ops/strategy-attach-scan")
+def api_strategy_attach_scan(dry_run: bool = Query(False)):
+    """手动策略自动挂载 (= 09:40 调度位)。dry_run=true 只判不写 —— 不挂、不留痕、
+    不动买入暂停表, 返回本轮会做什么, 供上线判收与日常复核。"""
+    from app.services import strategy_advisor
+    return ok_logged("strategy_attach_scan", strategy_advisor.scan, dry_run=dry_run,
+                     params={"dry_run": dry_run})
+
+
 @app.get("/api/upstream/plan")
 def api_upstream_plan(limit: int = Query(30), date: str = Query(None),
                       bucket: str = Query("main")):

config/settings.py

@@ -238,6 +238,25 @@ class Settings(BaseSettings):
     PMS_T0_GLOBAL_DAY_LOSS: float = 0.01   # 全局当日T亏熔断
     PMS_STRATEGY_ENABLED: bool = False     # 个股交易方案(策略)层总开关; False=strategy_runner 空转不评估
 
+    # --- 策略自动挂载 (个股打法状态机, STRATEGY_AUTO_ATTACH_PLAN.md) ---
+    #     每交易日 09:40 由 scheduler.strategy_attach 跑一次, 只做「配置动作」(挂/换/暂停买腿),
+    #     真正的买卖仍由 strategy_runner 逐分钟评估、逐笔过规则闸与全部熔断。
+    #     模拟仓拍板 (2026-08-25): 全自动上线不设观察期; ENABLED 已进 param_store.FAIL_CLOSED
+    #     (参数表读不到时按关处理, 基础设施故障期间宁可不挂)。
+    PMS_AUTO_STRATEGY_ENABLED: bool = True     # 总开关; 关 = 09:40 调度位空转不挂不撤
+    PMS_AUTO_STRATEGY_RULES: str = "accum_grid,heat_trail,accum_exit,handoff"  # 启用的边
+    PMS_AUTO_STRATEGY_DAILY_MAX: int = 2       # 每日自动新挂上限 (接力换挂不计入)
+    PMS_AUTO_ACCUM_STALE_TDAYS: int = 3        # 吸筹结论日龄超此 (交易日) 视为过期不触发
+    PMS_AUTO_GRID_BAND: float = 0.08           # 支撑压力锚不住时网格区间退化为 现价×(1±带宽)
+    PMS_AUTO_GRID_STEP_PCT: float = 0.02       # 自动网格步长 (占中枢价)
+    PMS_AUTO_GRID_CAP_RATIO: float = 0.5       # 网格最大投入 = 单股上限余量 × 此比例
+    PMS_AUTO_HEAT_TH: float = 0.80             # 热度阈值 0~1 (2026-08-25 实测 ≥0.8 约前 4%)
+    PMS_AUTO_TRAIL_GIVEBACK: float = 0.05      # 自动跟踪止盈: 距高水位回撤触发
+    PMS_AUTO_TRAIL_SELL_RATIO: float = 0.5     # 自动跟踪止盈: 触发时卖出比例
+    PMS_AUTO_HANDOFF_ENABLED: bool = True      # 边四接力 (网格→止盈) 独立开关, 页面可单独关
+    PMS_AUTO_HANDOFF_COOLDOWN_TDAYS: int = 10  # 接力后同票网格边冷却 (交易日)
+    PMS_AUTO_OPTOUT_COOLDOWN_TDAYS: int = 10   # 人工撤下后同票同规则冷却 (交易日)
+
     # --- 决策系统信号消化 (设计 §10) ---
     PMS_SIGNAL_ENABLED: bool = True
     PMS_SIGNAL_GROUP: str = "pms_signal_consumer"   # 独立消费组, 不与 trading_service 抢消息

scripts/probe_strategy_signals.py

@@ -30,7 +30,13 @@ from app.repo.downstream_repo import to_dot, to_prefix    # noqa: E402
 from app.services import param_store                      # noqa: E402
 
 # 决策系统定性词表 (契约; 见方案「口径同源声明」)。不在表里的词一律当无标志。
+# v1.1 (2026-08-25): 归类从前缀匹配改成**子串包含** —— 首跑 62 只「词表外」的根因是
+# state 带前后缀修饰 (如 "→ 高位派发风险"), 数据底座 feed.py 的 _ACCUM_KEEP 用的
+# 正是 any(k in state)。次序即优先级, 派发最先 (两词同现按保守方向), 与 strategy_advisor
+# 的 classify_accum 完全同法 —— 探测读数和线上判定必须是同一双眼睛。
 FIVE_STATES = ("明确吸筹", "潜在吸筹", "无吸筹迹象", "高位派发", "信号不明")
+_CLS_ORDER = (("派发", "高位派发"), ("明确吸筹", "明确吸筹"), ("潜在吸筹", "潜在吸筹"),
+              ("无吸筹迹象", "无吸筹迹象"), ("不明", "信号不明"))
 ACCUM_WINDOW_DAYS = 45          # 每票取近 45 自然日内最新一条 (方案口径是 30 天, 取宽探测)
 
 
@@ -66,10 +72,12 @@ def _fund_flow(raw):
 
 def _classify(state):
     s = str(state or "").strip()
-    for k in FIVE_STATES:
+    if not s:
-        if s.startswith(k):
+        return "无字段"
-            return k
+    for key, cls in _CLS_ORDER:
-    return "词表外" if s else "无字段"
+        if key in s:
+            return cls
+    return "词表外"
 
 
 def _variants(dot_codes):
@@ -100,15 +108,20 @@ def latest_accum_day():
 
 
 def accum_distribution(td):
-    """最新一日的定性分布 (全池)。"""
+    """最新一日的定性分布 (全池)。返回 (总数, 分布, 词表外原文样本)。"""
     rows = fetch_all("SELECT stock_code, raw_logic_json FROM strategy_daily_results "
                      "WHERE trade_date = :td", {"td": td})
-    dist, total = {}, 0
+    dist, total, oov = {}, 0, {}
     for r in rows:
         total += 1
         ff = _fund_flow(r.get("raw_logic_json"))
-        dist[_classify((ff or {}).get("state"))] = dist.get(_classify((ff or {}).get("state")), 0) + 1
+        state = (ff or {}).get("state")
-    return total, dist
+        cls = _classify(state)
+        dist[cls] = dist.get(cls, 0) + 1
+        if cls == "词表外":
+            oov[str(state)] = oov.get(str(state), 0) + 1
+    samples = sorted(oov.items(), key=lambda kv: -kv[1])[:10]
+    return total, dist, samples
 
 
 def accum_of(dot_codes):
@@ -181,16 +194,18 @@ def main():
         if accum_td is None:
             print("  结论表里一行都没有 —— 先确认决策系统夜间链在跑")
         else:
-            total, dist = accum_distribution(accum_td)
+            total, dist, oov_samples = accum_distribution(accum_td)
             age = _age_days(_ymd_int(accum_td))
             print(f"  最新结论日 {accum_td} (距今 {age} 个自然日), 当日共 {total} 只")
             for k in list(FIVE_STATES) + ["词表外", "无字段"]:
                 if dist.get(k):
                     print(f"    {k:<6} {dist[k]} 只")
             if dist.get("词表外"):
-                print("  ⚠ 出现词表外定性 —— 五档契约可能漂了, 需与决策系统核对")
+                print("  ⚠ 仍有子串也认不出的定性 —— 五档契约可能真漂了, 拿下面原文与决策系统核对:")
-            print("  对读: 8 月 24 日前端清单读数为 明确192 / 潜在278 / 不明82 / 派发8, "
+                for s, n in oov_samples:
-                  "同量级即两个读者一致")
+                    print(f"      「{s}」 × {n}")
+            print("  对读口径: 本段是**单日**分布; 数据底座前端清单是 30 天窗口每票最新一条,"
+                  " 总数天然更大, 两边比的是档位结构而不是绝对数")
     except Exception as e:
         print(f"  ✗ 本段失败: {type(e).__name__}: {e}")
 

scripts/run_tests.py

@@ -26,8 +26,11 @@
                         误判配额已出完/清仓命令即时撤策略/买入提议/在途买单;
                         在途与配额文案拆分; 减仓不掐策略腿                (12 例)
   test_batch16_units.py 软归档 archived_at: 单表守卫/只归终态/列表默认排除    (5 例)
+  test_batch17_units.py 策略自动挂载: 定性归类(子串+保守优先)/网格参数生成/
+                        连边矩阵/note 约定与冷却推导/接力判定/clear_buypause/
+                        编排冒烟(dry_run 滴水不写/名额/边三/边四全链)      (34 例)
   test_wiring.py        装配自检: 服务层→核心→落表 全链路 (内存桩)      (58 例)
-                                                                  共 519 例
+                                                                  共 553 例
 任一子集失败即整体失败 (退出码 1)。
 """
 import os
@@ -41,7 +44,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py",
           "test_batch7_units.py", "test_batch8_units.py", "test_batch9_units.py",
           "test_batch10_units.py", "test_batch11_units.py", "test_batch12_units.py",
           "test_batch13_units.py", "test_batch14_units.py", "test_batch15_units.py",
-          "test_batch16_units.py", "test_wiring.py"]
+          "test_batch16_units.py", "test_batch17_units.py", "test_wiring.py"]
 
 
 def main():

scripts/test_batch17_units.py

@@ -0,0 +1,684 @@
+# -*- coding: utf-8 -*-
+"""
+第十七批: 策略自动挂载 (个股打法状态机) —— STRATEGY_AUTO_ATTACH_PLAN.md 步骤二
+==============================================================================
+零外部依赖, 不连库。四块内容:
+  1. 纯逻辑: 定性归类 (子串包含+保守优先级) / 网格参数生成与放弃路径 /
+     连边矩阵 (双命中取止盈、负垫不挂) / 日上限计数 (接力不占) / 冷却推导 / 接力判定;
+  2. note 约定字面量钉死 —— 冷却与计数全靠它, 改一个字就把无状态推导改坏;
+  3. strategy_service.clear_buypause 的来源匹配 (只清自己停的, 不放开风控停的);
+  4. 编排冒烟 (全打桩): dry_run 滴水不写 / 真挂走 attach+留痕 / 名额满挡下 /
+     边三停买腿与恢复 / 边四接力全链 (撤网格→打标记→挂止盈→初始化高水位)。
+运行: python scripts/test_batch17_units.py
+"""
+import os
+import sys
+import traceback
+from types import SimpleNamespace
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from app.services import param_store                                            # noqa: E402
+from app.services import strategy_advisor as adv                                # noqa: E402
+from app.services import strategy_service as svc                                # noqa: E402
+from app.repo import pms_repo                                                   # noqa: E402
+
+RESULTS = []
+TODAY = 20260825
+
+
+def case(name):
+    def deco(fn):
+        RESULTS.append((name, fn))
+        return fn
+    return deco
+
+
+def _prm(**kw):
+    d = {"enabled": True,
+         "rules": {adv.R_GRID, adv.R_TRAIL, adv.R_EXIT, adv.R_HANDOFF},
+         "daily_max": 2, "accum_stale_tdays": 3, "grid_band": 0.08,
+         "grid_step_pct": 0.02, "grid_cap_ratio": 0.5, "heat_th": 0.80,
+         "trail_giveback": 0.05, "trail_sell_ratio": 0.5, "handoff_enabled": True,
+         "handoff_cooldown_tdays": 10, "optout_cooldown_tdays": 10}
+    d.update(kw)
+    return d
+
+
+def _out():
+    return {"ok": True, "checked": 0, "attached": [], "handoffs": [], "paused": [],
+            "resumed": [], "blocked": [], "skipped": [], "errors": [],
+            "unknown_states": []}
+
+
+def _pos(**kw):
+    d = {"ts_code": "600000.SH", "total_qty": 1000, "avail_qty": 1000,
+         "price": 10.0, "price_ok": True, "market_value": 10000.0,
+         "cushion_pct": 0.05, "support_ref": 9.0, "pressure_ref": 11.0,
+         "frozen_reason": "NONE"}
+    d.update(kw)
+    return d
+
+
+# ================================================================
+# [一] 定性归类: 子串包含 + 保守优先级
+# ================================================================
+@case("[归类] 五档标准词各归各位 (契约钉死)")
+def _():
+    assert adv.classify_accum("明确吸筹") == adv.CLS_CLEAR
+    assert adv.classify_accum("潜在吸筹") == adv.CLS_MAYBE
+    assert adv.classify_accum("无吸筹迹象") == adv.CLS_NONE_SIGN
+    assert adv.classify_accum("高位派发") == adv.CLS_DISTRIB
+    assert adv.classify_accum("信号不明") == adv.CLS_UNCLEAR
+
+
+@case("[归类] 子串包含: 带前后缀修饰照样认 (探测实测 62 只词表外的根因)")
+def _():
+    # 数据底座 feed.py 的 _ACCUM_KEEP 用 any(k in state), 前缀匹配会把这些全漏掉
+    assert adv.classify_accum("明确吸筹(强)") == adv.CLS_CLEAR
+    assert adv.classify_accum("近期呈潜在吸筹迹象") == adv.CLS_MAYBE
+    assert adv.classify_accum("→ 高位派发风险") == adv.CLS_DISTRIB
+    assert adv.classify_accum("走势信号不明朗") == adv.CLS_UNCLEAR
+
+
+@case("[归类] 两词同现按保守方向: 派发优先于明确吸筹")
+def _():
+    assert adv.classify_accum("前期明确吸筹, 现转高位派发") == adv.CLS_DISTRIB
+    assert adv.classify_accum("高位派发后再现潜在吸筹") == adv.CLS_DISTRIB
+
+
+@case("[归类] 空值→无字段, 认不出→词表外 (调用方一律按无标志处理)")
+def _():
+    assert adv.classify_accum(None) == adv.CLS_NOFIELD
+    assert adv.classify_accum("") == adv.CLS_NOFIELD
+    assert adv.classify_accum("   ") == adv.CLS_NOFIELD
+    assert adv.classify_accum("横盘整理") == adv.CLS_UNKNOWN
+    assert adv.classify_accum("chip_concentration_up") == adv.CLS_UNKNOWN
+
+
+# ================================================================
+# [二] 网格参数生成 (方案附录二)
+# ================================================================
+@case("[网格] 支撑压力锚得住: 上界=压力×1.01 下界=支撑×0.99, 手数为整百")
+def _():
+    gp, why = adv.grid_params(price=10.0, support=9.0, pressure=11.0, band=0.08,
+                              step_pct=0.02, cap_room=70000.0, cap_ratio=0.5)
+    assert gp, why
+    assert gp["lower"] == 8.91 and gp["upper"] == 11.11 and gp["center"] == 10.0, gp
+    assert gp["max_capital"] == 35000.0, gp
+    assert gp["per_lot"] % 100 == 0 and gp["per_lot"] >= 100, gp
+    # 下方档数 5 档 (1.09 / 0.2), 每档预算 7000 → 700 股
+    assert gp["per_lot"] == 700, gp
+
+
+@case("[网格] 锚不住 (支撑压力缺失/不合形) 退百分比带")
+def _():
+    gp, _ = adv.grid_params(price=10.0, support=0, pressure=0, band=0.08,
+                            step_pct=0.02, cap_room=70000.0, cap_ratio=0.5)
+    assert gp and gp["lower"] == 9.2 and gp["upper"] == 10.8, gp
+    # 支撑在现价上方 (不合形) 同样退带宽, 不硬凑
+    gp2, _ = adv.grid_params(price=10.0, support=12.0, pressure=11.0, band=0.08,
+                             step_pct=0.02, cap_room=70000.0, cap_ratio=0.5)
+    assert gp2 and gp2["lower"] == 9.2 and gp2["upper"] == 11.11, gp2
+
+
+@case("[网格] 三条放弃路径: 无实时价 / 买不起一手 / 区间不成立")
+def _():
+    gp, why = adv.grid_params(price=None, support=9, pressure=11, band=0.08,
+                              step_pct=0.02, cap_room=70000, cap_ratio=0.5)
+    assert gp is None and "实时价" in why, why
+    gp, why = adv.grid_params(price=10.0, support=9, pressure=11, band=0.08,
+                              step_pct=0.02, cap_room=1500.0, cap_ratio=0.5)
+    assert gp is None and "买不起一手" in why, why       # 750 元 < 一手 1000 元
+    gp, why = adv.grid_params(price=10.0, support=0, pressure=0, band=1.5,
+                              step_pct=0.02, cap_room=70000, cap_ratio=0.5)
+    assert gp is None and "区间不成立" in why, why       # 带宽>1 → 下界为负 (防御路径)
+
+
+@case("[网格] 档距下限 0.5%: 传入更小的步长被抬起来")
+def _():
+    gp, _ = adv.grid_params(price=10.0, support=9.0, pressure=11.0, band=0.08,
+                            step_pct=0.001, cap_room=70000.0, cap_ratio=0.5)
+    assert gp and gp["step_pct"] == 0.005, gp
+
+
+# ================================================================
+# [三] 连边矩阵 (plan_edge)
+# ================================================================
+@case("[连边] 双命中 (明确吸筹+高热+正垫) 取止盈 —— 保利润优先于做波段")
+def _():
+    e, why = adv.plan_edge(cls=adv.CLS_CLEAR, fresh=True, heat=0.9, cushion=0.05,
+                           prm=_prm())
+    assert e == adv.R_TRAIL and "双命中" in why, (e, why)
+
+
+@case("[连边] 单命中各走各边: 高热正垫→止盈; 明确吸筹新鲜→网格")
+def _():
+    e, why = adv.plan_edge(cls=adv.CLS_NOFIELD, fresh=False, heat=0.85,
+                           cushion=0.02, prm=_prm())
+    assert e == adv.R_TRAIL and "超阈值" in why, (e, why)
+    e, why = adv.plan_edge(cls=adv.CLS_CLEAR, fresh=True, heat=0.3, cushion=-0.02,
+                           prm=_prm())
+    assert e == adv.R_GRID, (e, why)
+
+
+@case("[连边] 负垫永不挂止盈 (把位置留给深亏补仓评估) —— 小口径钉死")
+def _():
+    e, why = adv.plan_edge(cls=adv.CLS_NOFIELD, fresh=False, heat=0.95,
+                           cushion=-0.08, prm=_prm())
+    assert e is None and "补仓评估" in why, (e, why)
+    # 垫子缺失 (盘前无价) 同样不挂
+    e, _ = adv.plan_edge(cls=adv.CLS_NOFIELD, fresh=False, heat=0.95,
+                         cushion=None, prm=_prm())
+    assert e is None
+
+
+@case("[连边] 吸筹结论超日龄视为无标志; 潜在吸筹不触发 (拍板①只认明确)")
+def _():
+    e, why = adv.plan_edge(cls=adv.CLS_CLEAR, fresh=False, heat=None, cushion=0.05,
+                           prm=_prm())
+    assert e is None and "超日龄" in why, (e, why)
+    e, _ = adv.plan_edge(cls=adv.CLS_MAYBE, fresh=True, heat=None, cushion=0.05,
+                         prm=_prm())
+    assert e is None
+
+
+@case("[连边] 边清单裁剪生效: 去掉 heat_trail 后双命中落回网格")
+def _():
+    p = _prm(rules={adv.R_GRID, adv.R_EXIT})
+    e, _ = adv.plan_edge(cls=adv.CLS_CLEAR, fresh=True, heat=0.9, cushion=0.05, prm=p)
+    assert e == adv.R_GRID, e
+    p = _prm(rules={adv.R_TRAIL})
+    e, _ = adv.plan_edge(cls=adv.CLS_CLEAR, fresh=True, heat=0.3, cushion=0.05, prm=p)
+    assert e is None, e
+
+
+# ================================================================
+# [四] note 约定与无状态推导 (日上限 / 两种冷却)
+# ================================================================
+@case("[约定] note 字面量钉死 —— 冷却与计数全靠它, 改一个字就坏")
+def _():
+    assert adv.NOTE_AUTO == "自动挂载: "
+    assert adv.NOTE_HANDOFF == "自动挂载(接力): "
+    assert adv.MARK_HANDOFF_OUT == "[接力撤下]"
+    assert adv.NOTE_HANDOFF.startswith("自动挂载")   # 冷却推导认「自动挂载」开头
+
+
+@case("[计数] 今日新挂只数常规挂计入、接力与人工与昨日不计入")
+def _():
+    rows = [
+        {"note": "自动挂载: 明确吸筹→网格", "created_at": "2026-08-25 09:40:01"},
+        {"note": "自动挂载: 高热→止盈", "created_at": "2026-08-25 09:40:02"},
+        {"note": "自动挂载(接力): 站上上界", "created_at": "2026-08-25 09:41:00"},  # 接力不算
+        {"note": "手工挂的网格", "created_at": "2026-08-25 10:00:00"},              # 人工不算
+        {"note": "自动挂载: 昨天挂的", "created_at": "2026-08-24 09:40:00"},        # 昨日不算
+        {"note": None, "created_at": "2026-08-25 09:40:00"},
+    ]
+    assert adv.count_auto_today(rows, TODAY) == 2
+
+
+@case("[冷却] 人工撤下→同票同规则进冷却; 接力标记→只进接力冷却不算人工")
+def _():
+    rows = [
+        # 人工撤下的自动网格, 票还持有, 5 天前 → optout
+        {"ts_code": "600000.SH", "type": "GRID", "note": "自动挂载: 网格",
+         "updated_at": "2026-08-20 10:00:00"},
+        # 接力撤下的网格 → 只进 handoff_cool
+        {"ts_code": "600519.SH", "type": "GRID", "note": "自动挂载: 网格 [接力撤下]",
+         "updated_at": "2026-08-20 11:00:00"},
+    ]
+    opt, hand = adv.cooldowns_from_cancelled(rows, ["600000.SH", "600519.SH"], TODAY,
+                                             optout_tdays=10, handoff_tdays=10)
+    assert ("600000.SH", adv.R_GRID) in opt, opt
+    assert "600519.SH" not in {c for c, _ in opt}, opt
+    assert hand == {"600519.SH"}, hand
+
+
+@case("[冷却] 票已不持有不算人工撤下 (清仓清场撤的不该罚) / 超窗口出冷却 / 人工策略不掺和")
+def _():
+    rows = [
+        {"ts_code": "600000.SH", "type": "GRID", "note": "自动挂载: 网格",
+         "updated_at": "2026-08-20 10:00:00"},              # 不在持仓 → 不算
+        {"ts_code": "600519.SH", "type": "TRAIL", "note": "自动挂载: 止盈",
+         "updated_at": "2026-07-10 10:00:00"},              # 46 天 > 10td×2 → 出冷却
+        {"ts_code": "000001.SZ", "type": "GRID", "note": "手工网格",
+         "updated_at": "2026-08-24 10:00:00"},              # 人工挂的, 撤了也不进冷却
+    ]
+    opt, hand = adv.cooldowns_from_cancelled(rows, ["600519.SH", "000001.SZ"], TODAY,
+                                             optout_tdays=10, handoff_tdays=10)
+    assert opt == set() and hand == set(), (opt, hand)
+
+
+# ================================================================
+# [五] 接力判定 (handoff_ready)
+# ================================================================
+@case("[接力] 齐活才走: 站上上界 + 热度达标 + 正垫 → True")
+def _():
+    ok, why = adv.handoff_ready(price=10.6, price_ok=True, upper=10.5, heat=0.85,
+                                cushion=0.06, prm=_prm())
+    assert ok and "站上网格上界" in why, (ok, why)
+
+
+@case("[接力] 没站上上界静默不动; 站上了但热度不够/垫不正要说清")
+def _():
+    ok, why = adv.handoff_ready(price=10.2, price_ok=True, upper=10.5, heat=0.9,
+                                cushion=0.05, prm=_prm())
+    assert not ok and why == "", (ok, why)
+    ok, why = adv.handoff_ready(price=10.6, price_ok=True, upper=10.5, heat=0.5,
+                                cushion=0.05, prm=_prm())
+    assert not ok and "未达阈值" in why, why
+    ok, why = adv.handoff_ready(price=10.6, price_ok=True, upper=10.5, heat=0.9,
+                                cushion=-0.01, prm=_prm())
+    assert not ok and "不接力" in why, why
+
+
+@case("[接力] 独立开关与边清单双闸; 价格拿成本顶的 (price_ok=False) 不判")
+def _():
+    ok, _ = adv.handoff_ready(price=10.6, price_ok=True, upper=10.5, heat=0.9,
+                              cushion=0.05, prm=_prm(handoff_enabled=False))
+    assert not ok
+    ok, _ = adv.handoff_ready(price=10.6, price_ok=True, upper=10.5, heat=0.9,
+                              cushion=0.05,
+                              prm=_prm(rules={adv.R_GRID, adv.R_TRAIL, adv.R_EXIT}))
+    assert not ok
+    ok, _ = adv.handoff_ready(price=10.6, price_ok=False, upper=10.5, heat=0.9,
+                              cushion=0.05, prm=_prm())
+    assert not ok
+
+
+# ================================================================
+# [六] clear_buypause: 只清自己停的
+# ================================================================
+def _with_buypause(entry, fn):
+    import json as _json
+    saved = {}
+    orig_get, orig_set = pms_repo.get_param, pms_repo.set_param
+    pms_repo.get_param = lambda k: _json.dumps(entry) if entry is not None else None
+    pms_repo.set_param = lambda k, v, by="system": saved.update({"key": k, "val": v}) or {"ok": True}
+    try:
+        r = fn()
+    finally:
+        pms_repo.get_param, pms_repo.set_param = orig_get, orig_set
+    return r, saved
+
+
+@case("[停买] clear_buypause 来源匹配才清: accum 停的清得掉, 风控停的不动")
+def _():
+    r, saved = _with_buypause({"600000.SH": {"source": "accum", "reason": "x"}},
+                              lambda: svc.clear_buypause("600000.SH", only_source="accum"))
+    assert r == {"ok": True, "cleared": True, "ts_code": "600000.SH"}, r
+    assert saved and "600000.SH" not in saved["val"], saved
+    r, saved = _with_buypause({"600000.SH": {"source": "signal", "reason": "风控预警"}},
+                              lambda: svc.clear_buypause("600000.SH", only_source="accum"))
+    assert r["ok"] and not r["cleared"] and "signal" in r.get("why", ""), r
+    assert not saved, saved                                  # 没清 → 一个字都没写
+
+
+@case("[停买] 没停过→ok 且 cleared=False; 空码→ok=False")
+def _():
+    r, saved = _with_buypause({}, lambda: svc.clear_buypause("600000.SH",
+                                                             only_source="accum"))
+    assert r["ok"] and not r["cleared"] and not saved, (r, saved)
+    r, _ = _with_buypause({}, lambda: svc.clear_buypause(""))
+    assert not r["ok"], r
+
+
+# ================================================================
+# [七] 参数登记 (param_store)
+# ================================================================
+@case("[参数] 总开关进 FAIL_CLOSED=False; 13 个键全有中文说明")
+def _():
+    assert param_store.FAIL_CLOSED.get("PMS_AUTO_STRATEGY_ENABLED") is False
+    for k in ("PMS_AUTO_STRATEGY_ENABLED", "PMS_AUTO_STRATEGY_RULES",
+              "PMS_AUTO_STRATEGY_DAILY_MAX", "PMS_AUTO_ACCUM_STALE_TDAYS",
+              "PMS_AUTO_GRID_BAND", "PMS_AUTO_GRID_STEP_PCT", "PMS_AUTO_GRID_CAP_RATIO",
+              "PMS_AUTO_HEAT_TH", "PMS_AUTO_TRAIL_GIVEBACK", "PMS_AUTO_TRAIL_SELL_RATIO",
+              "PMS_AUTO_HANDOFF_ENABLED", "PMS_AUTO_HANDOFF_COOLDOWN_TDAYS",
+              "PMS_AUTO_OPTOUT_COOLDOWN_TDAYS"):
+        assert k in param_store.DESC, k
+        from config.settings import settings as _s
+        assert hasattr(_s, k), k
+
+
+@case("[参数] 边清单写入口校验: 打错边名被拦, 合法值与越界数值各归各")
+def _():
+    err = param_store._range_check("PMS_AUTO_STRATEGY_RULES", "accum_grid,typo_edge")
+    assert err and "typo_edge" in err, err
+    assert param_store._range_check("PMS_AUTO_STRATEGY_RULES",
+                                    "accum_grid,heat_trail,accum_exit,handoff") is None
+    assert param_store._range_check("PMS_AUTO_HEAT_TH", 1.5), "热度阈值 1.5 应越界"
+    assert param_store._range_check("PMS_AUTO_HEAT_TH", 0.8) is None
+
+
+# ================================================================
+# [八] 编排冒烟 (全打桩, 不连库)
+# ================================================================
+class _Rec:
+    """记录调用的假 strategy_service / pms_repo 面板。"""
+
+    def __init__(self):
+        self.attach_calls, self.status_calls, self.pause_calls = [], [], []
+        self.clear_calls, self.ledger, self.upd_calls = [], [], []
+
+
+def _scan_stubbed(rec, *, held, accum, heat, strategies=None, cancelled=None,
+                  dry_run, prm=None, attach_ret=None):
+    """把 scan() 的所有外部依赖打桩后跑一轮, 返回 out。"""
+    from app.services import command_service, portfolio, strategy_service
+    orig = {}
+
+    def keep(mod, name, fake):
+        orig[(mod, name)] = getattr(mod, name)
+        setattr(mod, name, fake)
+
+    view = {"held": held, "positions": held,
+            "params": {"scale": 1000000.0, "stock_cap": 0.08}}
+    n_attach = {"n": 0}
+
+    def fake_attach(cfg, by="user"):
+        rec.attach_calls.append((cfg, by))
+        n_attach["n"] += 1
+        return dict(attach_ret or {"ok": True, "strategy_id": f"S_NEW{n_attach['n']}"})
+
+    def fake_list_strategies(**kw):
+        sts = kw.get("statuses") or []
+        if "CANCELLED" in sts:
+            return list(cancelled or [])
+        return list(strategies or [])
+
+    keep(adv, "_params", lambda: prm or _prm())
+    keep(param_store, "get_bool",
+         lambda k, d=False: {"PMS_STRATEGY_ENABLED": True,
+                             "PMS_GLOBAL_EXEC_HALT": False}.get(k, d))
+    keep(portfolio, "positions_view", lambda: view)
+    keep(adv, "accum_of", lambda codes: dict(accum))
+    keep(adv, "heat_of", lambda codes: (dict(heat), {"stale": False, "td": TODAY}))
+    keep(command_service, "blacklist", lambda: set())
+    keep(pms_repo, "list_strategies", fake_list_strategies)
+    keep(pms_repo, "list_instructions", lambda **kw: [])
+    keep(pms_repo, "list_plans", lambda **kw: [])
+    keep(pms_repo, "insert_ledger",
+         lambda **kw: rec.ledger.append(kw) or 1)
+    keep(pms_repo, "update_strategy",
+         lambda sid, **kw: rec.upd_calls.append((sid, kw)) or 1)
+    keep(strategy_service, "buypause_map", lambda: {})
+    keep(strategy_service, "attach", fake_attach)
+    keep(strategy_service, "set_status",
+         lambda sid, st, by="user": rec.status_calls.append((sid, st, by)) or
+         {"ok": True, "strategy_id": sid})
+    keep(strategy_service, "pause_buy",
+         lambda code, reason="", source="signal":
+         rec.pause_calls.append((code, source)) or ["SID"])
+    keep(strategy_service, "clear_buypause",
+         lambda code, only_source=None:
+         rec.clear_calls.append((code, only_source)) or {"ok": True, "cleared": True})
+    try:
+        return adv.scan(dry_run=dry_run)
+    finally:
+        for (mod, name), fn in orig.items():
+            setattr(mod, name, fn)
+
+
+@case("[冒烟] dry_run 滴水不写: 报出会挂网格, 但 attach/留痕/停买全没动")
+def _():
+    rec = _Rec()
+    out = _scan_stubbed(rec, held=[_pos()], heat={},
+                        accum={"600000.SH": {"cls": adv.CLS_CLEAR, "state": "明确吸筹",
+                                             "score": 80, "ymd": TODAY, "age_days": 1}},
+                        dry_run=True)
+    assert out["ok"], out["errors"]
+    assert out["attached"] and out["attached"][0]["dry_run"], out["attached"]
+    assert out["attached"][0]["ts_code"] == "600000.SH"
+    assert not rec.attach_calls and not rec.ledger and not rec.pause_calls, "dry_run 写了东西"
+
+
+@case("[冒烟] 真挂网格: attach 收到区间与上限, note 以「自动挂载: 」开头, 留痕 ATTACH/PASS")
+def _():
+    rec = _Rec()
+    out = _scan_stubbed(rec, held=[_pos()], heat={},
+                        accum={"600000.SH": {"cls": adv.CLS_CLEAR, "state": "明确吸筹",
+                                             "score": 80, "ymd": TODAY, "age_days": 1}},
+                        dry_run=False)
+    assert out["ok"] and out["attached"], (out["errors"], out["blocked"], out["skipped"])
+    cfg, by = rec.attach_calls[0]
+    assert by == "auto" and cfg["type"] == "GRID" and cfg["autonomy"] == "auto"
+    assert cfg["note"].startswith(adv.NOTE_AUTO), cfg["note"]
+    assert 0 < cfg["params"]["lower"] < 10.0 < cfg["params"]["upper"], cfg["params"]
+    assert cfg["params"]["max_capital"] == 35000.0, cfg["params"]
+    led = [x for x in rec.ledger if x.get("action") == "ATTACH"]
+    assert led and led[0]["verdict"] == "PASS" and led[0]["ref_id"] == "S_NEW1", led
+
+
+@case("[冒烟] 每日名额: 第三只被挡并留痕说明, dry_run 同样受限")
+def _():
+    held = [_pos(ts_code=c) for c in ("600000.SH", "600519.SH", "000001.SZ")]
+    accum = {c: {"cls": adv.CLS_CLEAR, "state": "明确吸筹", "score": 80,
+                 "ymd": TODAY, "age_days": 1} for c in ("600000.SH", "600519.SH",
+                                                        "000001.SZ")}
+    rec = _Rec()
+    out = _scan_stubbed(rec, held=held, accum=accum, heat={}, dry_run=True)
+    assert len(out["attached"]) == 2, out
+    assert len(out["blocked"]) == 1 and "名额已满" in out["blocked"][0]["why"], out["blocked"]
+    rec2 = _Rec()
+    out2 = _scan_stubbed(rec2, held=held, accum=accum, heat={}, dry_run=False)
+    assert len(rec2.attach_calls) == 2, rec2.attach_calls
+    assert any("名额已满" in x.get("reason", "") for x in rec2.ledger), rec2.ledger
+
+
+@case("[冒烟] 排除项: 冻结票与热度停更都不挂; 词表外定性浮到 unknown_states")
+def _():
+    held = [_pos(frozen_reason="RISK_FREEZE")]
+    accum = {"600000.SH": {"cls": adv.CLS_CLEAR, "state": "明确吸筹", "score": 80,
+                           "ymd": TODAY, "age_days": 1}}
+    rec = _Rec()
+    out = _scan_stubbed(rec, held=held, accum=accum, heat={}, dry_run=False)
+    assert not out["attached"] and not rec.attach_calls
+    assert any("冻结" in s.get("why", "") for s in out["skipped"]), out["skipped"]
+    # 词表外
+    rec2 = _Rec()
+    out2 = _scan_stubbed(rec2, held=[_pos()], heat={},
+                         accum={"600000.SH": {"cls": adv.CLS_UNKNOWN, "state": "横盘整理",
+                                              "score": None, "ymd": TODAY, "age_days": 1}},
+                         dry_run=False)
+    assert out2["unknown_states"] and out2["unknown_states"][0]["state"] == "横盘整理"
+    assert not rec2.attach_calls
+
+
+@case("[冒烟] 边三: 已挂自动网格遇派发 → pause_buy(source=accum) 一次, 已停不重复")
+def _():
+    st = {"strategy_id": "S_G", "ts_code": "600000.SH", "type": "GRID",
+          "note": "自动挂载: 明确吸筹→网格", "params": {"upper": 10.5}, "state": {}}
+    rec = _Rec()
+    fake_svc = SimpleNamespace(
+        pause_buy=lambda code, reason="", source="signal":
+        rec.pause_calls.append((code, source)) or ["S_G"],
+        clear_buypause=lambda code, only_source=None:
+        rec.clear_calls.append((code, only_source)) or {"ok": True, "cleared": True})
+    orig_led = pms_repo.insert_ledger
+    pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) or 1
+    try:
+        out = _out()
+        adv._tend_existing(st, _pos(price=10.0), {"cls": adv.CLS_DISTRIB, "state": "高位派发"},
+                           True, 0.3, _prm(), {}, set(), TODAY, False, out, fake_svc)
+        assert rec.pause_calls == [("600000.SH", "accum")], rec.pause_calls
+        assert out["paused"] and any(x.get("action") == "NOTE" for x in rec.ledger)
+        # 已停着 (无论谁停的) 不重复停
+        out2 = _out()
+        adv._tend_existing(st, _pos(price=10.0), {"cls": adv.CLS_DISTRIB}, True, 0.3,
+                           _prm(), {"600000.SH": {"source": "signal"}}, set(), TODAY,
+                           False, out2, fake_svc)
+        assert len(rec.pause_calls) == 1 and not out2["paused"]
+    finally:
+        pms_repo.insert_ledger = orig_led
+
+
+@case("[冒烟] 边三解除: 定性回明确且新鲜, 只解除 accum 来源的停买")
+def _():
+    st = {"strategy_id": "S_G", "ts_code": "600000.SH", "type": "GRID",
+          "note": "自动挂载: 网格", "params": {"upper": 10.5}, "state": {}}
+    rec = _Rec()
+    fake_svc = SimpleNamespace(
+        pause_buy=lambda *a, **k: [],
+        clear_buypause=lambda code, only_source=None:
+        rec.clear_calls.append((code, only_source)) or {"ok": True, "cleared": True})
+    orig_led = pms_repo.insert_ledger
+    pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) or 1
+    try:
+        out = _out()
+        adv._tend_existing(st, _pos(price=10.0),
+                           {"cls": adv.CLS_CLEAR, "state": "明确吸筹"}, True, 0.3,
+                           _prm(), {"600000.SH": {"source": "accum"}}, set(), TODAY,
+                           False, out, fake_svc)
+        assert rec.clear_calls == [("600000.SH", "accum")], rec.clear_calls
+        assert out["resumed"] == [{"ts_code": "600000.SH"}], out["resumed"]
+        # 风控停的不归边三管
+        out2 = _out()
+        adv._tend_existing(st, _pos(price=10.0), {"cls": adv.CLS_CLEAR}, True, 0.3,
+                           _prm(), {"600000.SH": {"source": "signal"}}, set(), TODAY,
+                           False, out2, fake_svc)
+        assert len(rec.clear_calls) == 1 and not out2["resumed"]
+    finally:
+        pms_repo.insert_ledger = orig_led
+
+
+@case("[冒烟] 边四全链: 撤网格→老 note 打[接力撤下]→挂止盈(接力 note)→高水位从现价起算")
+def _():
+    st = {"strategy_id": "S_G", "ts_code": "600000.SH", "type": "GRID",
+          "note": "自动挂载: 网格", "params": {"upper": 10.5}, "state": {}}
+    rec = _Rec()
+    fake_svc = SimpleNamespace(
+        set_status=lambda sid, s, by="user": rec.status_calls.append((sid, s, by)) or
+        {"ok": True},
+        attach=lambda cfg, by="user": rec.attach_calls.append((cfg, by)) or
+        {"ok": True, "strategy_id": "S_T"},
+        pause_buy=lambda *a, **k: [], clear_buypause=lambda *a, **k: {"ok": True})
+    orig_led, orig_upd = pms_repo.insert_ledger, pms_repo.update_strategy
+    pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) or 1
+    pms_repo.update_strategy = lambda sid, **kw: rec.upd_calls.append((sid, kw)) or 1
+    try:
+        out = _out()
+        adv._tend_existing(st, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
+                           _prm(), {}, set(), TODAY, False, out, fake_svc)
+        assert rec.status_calls == [("S_G", "CANCELLED", "auto")], rec.status_calls
+        marks = [kw for sid, kw in rec.upd_calls if sid == "S_G"]
+        assert marks and adv.MARK_HANDOFF_OUT in marks[0]["note"], rec.upd_calls
+        cfg, by = rec.attach_calls[0]
+        assert cfg["type"] == "TRAIL" and cfg["note"].startswith(adv.NOTE_HANDOFF)
+        hw = [kw for sid, kw in rec.upd_calls if sid == "S_T"]
+        assert hw and hw[0]["state"] == {"high_water": 10.6}, rec.upd_calls
+        assert out["handoffs"] and out["handoffs"][0]["to"] == "S_T"
+        assert any(x.get("action") == "HANDOFF" and x.get("verdict") == "PASS"
+                   for x in rec.ledger)
+    finally:
+        pms_repo.insert_ledger, pms_repo.update_strategy = orig_led, orig_upd
+
+
+@case("[冒烟] 边四让路: 接力冷却期 / 有在途委托 / dry_run 都不真动")
+def _():
+    st = {"strategy_id": "S_G", "ts_code": "600000.SH", "type": "GRID",
+          "note": "自动挂载: 网格", "params": {"upper": 10.5}, "state": {}}
+    rec = _Rec()
+    fake_svc = SimpleNamespace(
+        set_status=lambda *a, **k: rec.status_calls.append(a) or {"ok": True},
+        attach=lambda *a, **k: rec.attach_calls.append(a) or {"ok": True,
+                                                              "strategy_id": "X"},
+        pause_buy=lambda *a, **k: [], clear_buypause=lambda *a, **k: {"ok": True})
+    # 冷却期
+    out = _out()
+    adv._tend_existing(st, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
+                       _prm(), {}, {"600000.SH"}, TODAY, False, out, fake_svc)
+    assert not rec.status_calls and any("冷却" in s["why"] for s in out["skipped"])
+    # 在途委托
+    st2 = dict(st, state={"pending": {"iid": "INS_1"}})
+    orig_get = pms_repo.get_instruction
+    pms_repo.get_instruction = lambda iid: {"instruction_id": iid, "status": "DISPATCHED"}
+    try:
+        out2 = _out()
+        adv._tend_existing(st2, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
+                           _prm(), {}, set(), TODAY, False, out2, fake_svc)
+        assert not rec.status_calls and any("在途" in s["why"] for s in out2["skipped"])
+    finally:
+        pms_repo.get_instruction = orig_get
+    # dry_run 只报不动
+    out3 = _out()
+    adv._tend_existing(st, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
+                       _prm(), {}, set(), TODAY, True, out3, fake_svc)
+    assert out3["handoffs"] and out3["handoffs"][0]["dry_run"] and not rec.status_calls
+
+
+@case("[冒烟] 人工策略不碰; 自动止盈无事; 接力半途而废要报错并留痕 REJECT")
+def _():
+    rec = _Rec()
+    fake_svc = SimpleNamespace(
+        set_status=lambda sid, s, by="user": {"ok": True},
+        attach=lambda cfg, by="user": {"ok": False, "errors": ["跟踪止盈触发时要卖出, 但当前 T+1 可卖为 0"]},
+        pause_buy=lambda *a, **k: [], clear_buypause=lambda *a, **k: {"ok": True})
+    manual = {"strategy_id": "S_M", "ts_code": "600000.SH", "type": "GRID",
+              "note": "手工网格", "params": {"upper": 10.5}, "state": {}}
+    out = _out()
+    adv._tend_existing(manual, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
+                       _prm(), {}, set(), TODAY, False, out, fake_svc)
+    assert any("人工策略" in s["why"] for s in out["skipped"]), out["skipped"]
+    auto_trail = dict(manual, note="自动挂载: 止盈", type="TRAIL")
+    out2 = _out()
+    adv._tend_existing(auto_trail, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
+                       _prm(), {}, set(), TODAY, False, out2, fake_svc)
+    assert any("本轮无事" in s["why"] for s in out2["skipped"]), out2["skipped"]
+    # 半途而废: 撤成挂败 → errors 里有「人工处理」, 留痕 HANDOFF/REJECT
+    orig_led, orig_upd = pms_repo.insert_ledger, pms_repo.update_strategy
+    pms_repo.insert_ledger = lambda **kw: rec.ledger.append(kw) or 1
+    pms_repo.update_strategy = lambda sid, **kw: 1
+    try:
+        grid = dict(manual, note="自动挂载: 网格")
+        out3 = _out()
+        adv._tend_existing(grid, _pos(price=10.6), {"cls": adv.CLS_CLEAR}, True, 0.9,
+                           _prm(), {}, set(), TODAY, False, out3, fake_svc)
+        assert any("人工处理" in e for e in out3["errors"]), out3["errors"]
+        assert any(x.get("action") == "HANDOFF" and x.get("verdict") == "REJECT"
+                   for x in rec.ledger), rec.ledger
+    finally:
+        pms_repo.insert_ledger, pms_repo.update_strategy = orig_led, orig_upd
+
+
+@case("[冒烟] 三道总闸: 自动开关关 / 策略层关 / 休假模式, 各自明说并整轮不动")
+def _():
+    rec = _Rec()
+    out = _scan_stubbed(rec, held=[_pos()], accum={}, heat={}, dry_run=False,
+                        prm=_prm(enabled=False))
+    assert not rec.attach_calls and any("总开关关闭" in s["why"] for s in out["skipped"])
+    orig = param_store.get_bool
+    param_store.get_bool = lambda k, d=False: {"PMS_STRATEGY_ENABLED": False,
+                                               "PMS_GLOBAL_EXEC_HALT": False}.get(k, d)
+    orig_p = adv._params
+    adv._params = lambda: _prm()
+    try:
+        out2 = adv.scan(dry_run=False)
+        assert any("策略层总开关" in s["why"] for s in out2["skipped"]), out2
+        param_store.get_bool = lambda k, d=False: {"PMS_STRATEGY_ENABLED": True,
+                                                   "PMS_GLOBAL_EXEC_HALT": True}.get(k, d)
+        out3 = adv.scan(dry_run=False)
+        assert any("休假" in s["why"] for s in out3["skipped"]), out3
+    finally:
+        param_store.get_bool, adv._params = orig, orig_p
+
+
+def main():
+    passed, failed = 0, []
+    for name, fn in RESULTS:
+        try:
+            fn()
+            passed += 1
+            print(f"  ✓ {name}")
+        except Exception as e:
+            failed.append((name, e))
+            print(f"  ✗ {name}: {type(e).__name__}: {e}")
+            traceback.print_exc()
+    print()
+    if failed:
+        print(f"FAILED {len(failed)}/{len(RESULTS)}")
+        sys.exit(1)
+    print(f"ALL PASS ({passed} cases)")
+
+
+if __name__ == "__main__":
+    main()

scripts/test_wiring.py

@@ -542,7 +542,8 @@ def _():
             "/api/proposals", "/api/proposals/{proposal_id}/decide", "/api/ops/replay",
             "/api/ops/reconcile", "/api/ops/daily-settle", "/api/ops/report",
             "/api/ops/downstream-schema", "/api/industry",
-            "/api/upstream/plan", "/api/ops/plan-refresh"]
+            "/api/upstream/plan", "/api/ops/plan-refresh",
+            "/api/ops/strategy-attach-scan"]   # 2026-08-25 策略自动挂载试算位
     missing = [p for p in need if p not in paths]
     assert not missing, missing
     import os as _os
@@ -557,11 +558,12 @@ def _():
     assert names == {"plan_pull", "premarket", "command_poll", "replay_fills",
                      "intraday_exec", "signal_digest", "t0_close", "daily_settle",
                      "daily_report",
-                     "macro_scan"}, names   # 2026-08-18 +宏观择时 09:35
+                     "macro_scan",          # 2026-08-18 +宏观择时 09:35
+                     "strategy_attach"}, names   # 2026-08-25 +策略自动挂载 09:40
     tasks = set(sch.celery_app.tasks)
     for t in ("pms.premarket", "pms.command_poll", "pms.replay_fills", "pms.daily_settle",
               "pms.daily_report", "pms.t0_close", "pms.intraday_exec", "pms.signal_digest",
-              "pms.plan_pull", "pms.macro_scan"):
+              "pms.plan_pull", "pms.macro_scan", "pms.strategy_attach"):
         assert t in tasks, t
     assert sch.in_session(datetime(2026, 7, 27, 10, 0)) is True
     assert sch.in_session(datetime(2026, 7, 27, 12, 0)) is False