From f4123d36d30cdf35dd56ccc094db93caa782788f Mon Sep 17 00:00:00 2001 From: zlt Date: Tue, 25 Aug 2026 13:45:51 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E4=BA=A4=E6=98=93=E9=80=BB?= =?UTF-8?q?=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DEVLOG.md | 41 ++ app/scheduler.py | 16 + app/services/param_store.py | 30 ++ app/services/strategy_advisor.py | 649 ++++++++++++++++++++++++++++ app/services/strategy_service.py | 22 + app/web/main.py | 9 + config/settings.py | 19 + scripts/probe_strategy_signals.py | 39 +- scripts/run_tests.py | 7 +- scripts/test_batch17_units.py | 684 ++++++++++++++++++++++++++++++ scripts/test_wiring.py | 8 +- 11 files changed, 1507 insertions(+), 17 deletions(-) create mode 100644 app/services/strategy_advisor.py create mode 100644 scripts/test_batch17_units.py diff --git a/DEVLOG.md b/DEVLOG.md index 09896c0..4caf340 100644 --- a/DEVLOG.md +++ b/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（续）· 清仓与减仓对策略的优先级理顺：减仓不掐网格、清仓补撤在途买单、在途文案分开 **背景** diff --git a/app/scheduler.py b/app/scheduler.py index cf00535..f8e6ea3 100644 --- a/app/scheduler.py +++ b/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)}, diff --git a/app/services/param_store.py b/app/services/param_store.py index c7e0e3c..7f95bc3 100644 --- a/app/services/param_store.py +++ b/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 diff --git a/app/services/strategy_advisor.py b/app/services/strategy_advisor.py new file mode 100644 index 0000000..ee3fe61 --- /dev/null +++ b/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}") diff --git a/app/services/strategy_service.py b/app/services/strategy_service.py index 22dcda0..9f13f90 100644 --- a/app/services/strategy_service.py +++ b/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} diff --git a/app/web/main.py b/app/web/main.py index ece54f7..ac70d16 100644 --- a/app/web/main.py +++ b/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")): diff --git a/config/settings.py b/config/settings.py index c26b992..b23d130 100644 --- a/config/settings.py +++ b/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 抢消息 diff --git a/scripts/probe_strategy_signals.py b/scripts/probe_strategy_signals.py index 05f8b05..f101b74 100644 --- a/scripts/probe_strategy_signals.py +++ b/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 s.startswith(k): - return k - return "词表外" if s else "无字段" + if not s: + return "无字段" + for key, cls in _CLS_ORDER: + 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 - return total, dist + state = (ff or {}).get("state") + 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(" 对读: 8 月 24 日前端清单读数为 明确192 / 潜在278 / 不明82 / 派发8, " - "同量级即两个读者一致") + print(" ⚠ 仍有子串也认不出的定性 —— 五档契约可能真漂了, 拿下面原文与决策系统核对:") + for s, n in oov_samples: + print(f" 「{s}」 × {n}") + print(" 对读口径: 本段是**单日**分布; 数据底座前端清单是 30 天窗口每票最新一条," + " 总数天然更大, 两边比的是档位结构而不是绝对数") except Exception as e: print(f" ✗ 本段失败: {type(e).__name__}: {e}") diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 588e125..f9fe600 100644 --- a/scripts/run_tests.py +++ b/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(): diff --git a/scripts/test_batch17_units.py b/scripts/test_batch17_units.py new file mode 100644 index 0000000..3df1516 --- /dev/null +++ b/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() diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index 4e337e2..1e16135 100644 --- a/scripts/test_wiring.py +++ b/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