diff --git a/README.md b/README.md index 8c9fb0f..2ed2143 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,7 @@ app/ recon.py 成交认领与入账映射 / 对账差异与修正 / 除权检测 / T+1 可用量 rule_gate.py 规则闸终检: 上限/一手/可卖/冻结/刹车/行业/不追高 (减持只放行不阻拦) exec_timing.py 择时实现B: 分日配额 / 分笔 / 买卖出手判定 / 14:45 兜底 / 窗口收口 + action_engine.py 动作引擎: FILL 回踩补足 / ADD 盈利加仓 / DCA 补仓 / TRIM 保垫减仓 tradedays.py 交易日历: 调度守卫与执行窗口计算 db/session.py 三库连接 + **严格单表访问守卫** (JOIN/逗号连表/跨表子查询一律拒绝) repo/ 单表数据访问: pms_repo (自有 10 表) / downstream_repo (下游只读三表) @@ -36,6 +37,8 @@ app/ command_service.py 命令下达→校验→冲突→生效/规划→进度推进→撤销 executor.py 方案→指令→分日出手→窗口收口 (规则闸与择时的编排落点) dispatcher.py 下发通道三适配器: shadow(默认) / plan_x / channel_y + proposal_service.py 自主提议: 扫描→规则闸→研判闸→按自主档位分流 (执行/入队) + judge.py 研判闸客户端 (决策系统未接通时自动降级为人工确认) ledger_service.py 成交回放 / 对账 / 除权 / 盘前 / 日终结算 / 运营日报 market.py 行情 (Redis db13) 与参考位 (决策系统主口径 + 兜底自算) industry.py 行业划分可插拔适配器 (custom_table / gp_stock_category / 停用) @@ -46,7 +49,8 @@ scripts/ test_core_units.py 仓位与安全垫核心逻辑 14 例 test_batch2_units.py 命令 / 方案 / 回放对账 纯逻辑 35 例 test_batch3_units.py 规则闸 / 择时执行器实现B 纯逻辑 18 例 - test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 25 例 + test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 11 例 + test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 30 例 init_db.py 建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练) check_db.py 实机连通性与表结构自检 (需真实 .env) ``` @@ -96,7 +100,7 @@ git pull && docker compose build && docker compose up -d | 盘前准备 | 交易日 08:50 | T+1 可卖重置 / 参考位取数 / 刹车结算 | ✅ | | 命令轮询 | 每 1 分钟（全天） | 新命令解析 → 方案生成 → 状态机推进 | ✅ | | 成交回放 | 交易时段每 5 分钟 | `trading_order` 增量回放 + 盘中轻对账 | ✅ | -| 盘中执行 | 交易时段每 1 分钟 | 方案转指令 → 择时出手（规则闸终检 → 下发 → 记子单） | ✅（自主提议扫描待下一批） | +| 盘中执行 | 交易时段每 1 分钟 | 方案转指令 → 自主提议扫描 → 择时出手（规则闸终检 → 下发 → 记子单） | ✅ | | 信号消化 | 交易时段每 1 分钟 | 订阅决策系统盘中信号 | 🔜 下一批 | | T 仓平回 | 14:50 | 做T强制平回 | 🔜 二期（现只自证 T 仓为 0） | | 日终结算 | 15:10 | 除权检测 / 全量对账 / 安全垫 / 命令进度日结 | ✅ | @@ -114,11 +118,21 @@ git pull && docker compose build && docker compose up -d 影子模式下的完整闭环：页面下命令 → 方案落表 → 方案转指令 → 择时按日配额给出「今天该出多少、什么价」→ 你照着在 QMT 下单 → 5 分钟一次的回放把成交认领回批次账本 → 命令进度自动推进。整条链路除了「人手下单」这一步，其余与实盘接管后完全一致。 +## 自主提议的分流（设计 §6 / §7） + +动作引擎每分钟扫一遍持仓，产出 FILL / ADD / DCA / TRIM 四类候选，然后依次过三道： + +1. **规则闸**（一级，纯代码）——不过就拒，未通过项落评审账本。自主动作受组合刹车约束（命令驱动不受）。 +2. **研判闸**（二级，仅补足/加仓/补仓，委托决策系统）——`PMS_JUDGE_API_BASE` 为空即视为未接通，按设计**自动降级为人工确认**并记 ERROR，绝不把「研判拿不到」当成「研判通过」。 +3. **按自主档位分流**——`full` 执行、`propose_only` 入队（一期默认）、`off` 不扫描。 + +两条无条件覆盖档位的规矩：**减持方向不设确认门槛**（TRIM 保垫减仓任何档位都直接落指令）；**−15% 及更深的补仓永远需用户确认**（即便档位是 full 也强制入队）。同一只票的同一动作若已有在途提议或在途指令，不重复提。 + ## 已实现 / 待开发 -**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；**规则闸终检**；**择时执行器实现 B**（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口）；**三模式下发通道**；管理页面四块 + 运维/日报抽屉；调度器八个调度位；单测 92 例。 +**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口）；三模式下发通道；**动作引擎四类自主动作 + 研判闸客户端 + 提议分流**；管理页面四块 + 运维/日报抽屉；调度器八个调度位；单测 108 例。 -**待开发（下一批）**：动作引擎（FILL/ADD/DCA/TRIM 自主提议扫描 + 提议入队）、决策系统信号订阅（风控 SELL / 止盈 / 反转 → 卖出方案）、研判闸对接与择时实现 A、T0 做T（二期）。 +**待开发（下一批）**：决策系统盘中信号订阅（风控 SELL / 止盈 / 反转 → 卖出方案或提议）、择时实现 A（委托决策系统盘中择时）、T0 做T（二期）。研判闸客户端已就位，等 bionic 侧 `process_intraday_audit` 新增 PMS 请求 direction 后，在页面填 `PMS_JUDGE_API_BASE` 即接通。 **待外部协商**：`QMT_INTERFACE_REQUIREMENTS.md` 的 A/B/C/D 各项——尤其 A1（`trading_position` 完整 DDL 与可用数量列）、A2（`trading_order` 状态枚举与**来源标识**）、B1（统一指令通道）。在来源标识到位前，回放按「同股同向 + 下发早于成交 + FIFO」贪心认领，认领不上即判外部成交并告警；持仓数量列用候选名探测，探测结果可经页面「运维 → 导出下游表结构」查看，也是回填 D1 的现成材料。 diff --git a/app/core/action_engine.py b/app/core/action_engine.py new file mode 100644 index 0000000..0afb176 --- /dev/null +++ b/app/core/action_engine.py @@ -0,0 +1,219 @@ +# -*- coding: utf-8 -*- +""" +动作引擎 · 自主提议扫描 (纯逻辑, 无外部依赖, 可单测) +===================================================== +设计 POSITION_MGMT_DESIGN.md §6 的四类自主动作。每条规则的触发口径与约束都照抄设计表: + +| 引擎 | 触发 | 关键约束 | +|---|---|---| +| FILL 回踩补足 | 建仓期(≤10 交易日)内回踩支撑带不破, 浮亏 < 3% | 每票 1 次; 补后 ≤ 目标仓位 | +| ADD 盈利加仓 | 安全垫 ≥ +3% 且创 5 日新高或站上压力位 | 距上次 ≥2 交易日; ≤ 单股上限; 距 MA5 <+6% | +| DCA 补仓 | 浮亏触及 −8%/−15% 评估档 (各评估一次, 执行终身一次) | ≤ 底仓 50%; −15% 及更深永远需用户确认 | +| TRIM 保垫减仓 | 安全垫峰值 ≥6% 且回吐过半 → 减 1/3 锁盈 | 纯规则自动执行 (减持方向不设确认门槛) | + +本模块只回答「该不该动、动多少、为什么」, 不查库不下发: + * 交易日相关的输入 (建仓天数、距上次加仓天数) 由 services 层用交易日历算好传进来, + 避免把日历依赖塞进纯逻辑。 + * 上限/一手/冻结等硬约束**不在这里重复判**, 统一由规则闸终检 (职责单一, 口径唯一)。 + 这里只做「引擎自身的触发条件」与「批次额度」计算。 + +输出候选统一结构, 供 proposal_service 走 规则闸 → 研判闸 → 按自主档位分流。 +""" +from __future__ import annotations + +from app.core.cushion import dca_stage, trim_trigger +from app.core.sizer import LOT, lot_qty + +A_FILL, A_ADD, A_DCA, A_TRIM = "FILL", "ADD", "DCA", "TRIM" +BUY, SELL = "buy", "sell" + +# 需要送研判闸的动作 (设计 §7: 仅自主提议的补足/加仓/补仓/调仓) +JUDGE_ACTIONS = {A_FILL, A_ADD, A_DCA} + + +def _f(v, d=0.0): + try: + return float(v) + except (TypeError, ValueError): + return d + + +def _cand(p, action, side, qty, reason, hard, *, confirm=False): + return {"ts_code": p["ts_code"], "action": action, "side": side, "qty": int(qty), + "reason": reason, "hard_numbers": hard, "needs_user_confirm": bool(confirm), + "judge_required": action in JUDGE_ACTIONS} + + +def batch_amount(p: dict, params: dict, idx: int) -> float: + """第 idx 批 (0=底仓 1=回踩补足 2=盈利加仓) 的金额额度 = 目标仓位 × 该批比例。""" + scale = _f(params.get("scale")) + target_pct = _f(p.get("target_pct")) or _f(params.get("stock_target_default"), 0.06) + split = params.get("batch_split") or (0.5, 0.25, 0.25) + ratio = split[idx] if idx < len(split) else 0.0 + return scale * target_pct * ratio + + +def room_to_target(p: dict, params: dict) -> float: + """距目标仓位还差多少钱 (已达标返回 0)。""" + scale = _f(params.get("scale")) + target_pct = _f(p.get("target_pct")) or _f(params.get("stock_target_default"), 0.06) + return max(0.0, target_pct * scale - _f(p.get("market_value"))) + + +# ================================================================ 四类动作 +def eval_fill(p: dict, params: dict, mkt: dict): + """回踩补足: 建仓期内回踩到支撑带但没破, 浮亏还浅 —— 把底仓补到位。""" + if int(p.get("fill_count") or 0) > 0: + return None # 每票 1 次 + tdays = mkt.get("tdays_since_open") + if tdays is None or tdays > int(params.get("build_window_tdays", 10)): + return None # 只在建仓期内 + cushion = p.get("cushion_pct") + if cushion is None: + return None + floor = _f(params.get("fill_max_loss"), -0.03) + if not (floor < _f(cushion) <= 0): + return None # 浮亏 <3% 且尚未转盈 + support = _f(p.get("support_ref")) + price = _f(p.get("price")) + if support <= 0 or price <= 0: + return None # 没有支撑参考位就不做这件事 + if price < support: + return None # 支撑已破, 不补 + room = room_to_target(p, params) + amt = min(batch_amount(p, params, 1), room) + qty = lot_qty(amt, price) + if qty < LOT: + return None + return _cand(p, A_FILL, BUY, qty, + f"回踩补足: 建仓第 {tdays} 个交易日, 浮亏 {_f(cushion):.2%} 未破支撑 " + f"{support}, 补 {qty} 股至目标仓位", + {"cushion_pct": cushion, "support": support, "price": price, + "tdays_since_open": tdays, "room_to_target": round(room, 2)}) + + +def eval_add(p: dict, params: dict, mkt: dict): + """盈利加仓: 垫子够厚 + 走出新高或站上压力位 —— 顺势加。""" + cushion = p.get("cushion_pct") + solid = _f(params.get("cushion_solid"), 0.03) + if cushion is None or _f(cushion) < solid: + return None # 垫子不厚不加 + price = _f(p.get("price")) + if price <= 0: + return None + high5, pressure = _f(mkt.get("high5")), _f(p.get("pressure_ref")) + breakout = (high5 > 0 and price >= high5) or (pressure > 0 and price >= pressure) + if not breakout: + return None + since_add = mkt.get("tdays_since_last_add") + if since_add is not None and since_add < 2: + return None # 距上次加仓 ≥2 交易日 + ma5 = _f(mkt.get("ma5")) + no_chase = _f(params.get("no_chase_ma5"), 0.06) + if ma5 > 0 and price / ma5 - 1 > no_chase: + return None # 不追高 (规则闸还会再拦一次) + room = room_to_target(p, params) + amt = min(batch_amount(p, params, 2), room) if room > 0 else batch_amount(p, params, 2) + qty = lot_qty(amt, price) + if qty < LOT: + return None + why = "创 5 日新高" if (high5 > 0 and price >= high5) else "站上压力位" + return _cand(p, A_ADD, BUY, qty, + f"盈利加仓: 安全垫 {_f(cushion):.2%} ≥ {solid:.0%} 且{why} " + f"({price} vs 高点 {high5 or '—'}/压力 {pressure or '—'}), 加 {qty} 股", + {"cushion_pct": cushion, "price": price, "high5": high5, + "pressure": pressure, "ma5": ma5, "tdays_since_last_add": since_add}) + + +def eval_dca(p: dict, params: dict, mkt: dict): + """补仓: 浮亏触及评估档才评估, 各档评估一次、执行终身一次, 深档永远要用户点头。""" + cushion = p.get("cushion_pct") + if cushion is None or _f(cushion) >= 0: + return None + if int(p.get("dca_qty") or 0) > 0: + return None # 终身一次, 已执行过 + triggers = params.get("dca_triggers") or (-0.08, -0.15) + stage = dca_stage(_f(cushion), triggers) + if stage <= 0: + return None + evaluated = int(p.get("dca_count") or 0) + if stage <= evaluated: + return None # 该档已评估过, 不重复提 + price = _f(p.get("price")) + base_qty = int(p.get("base_qty") or 0) + if price <= 0 or base_qty <= 0: + return None + max_ratio = _f(params.get("dca_max_ratio"), 0.5) + qty = int(base_qty * max_ratio // LOT) * LOT + if qty < LOT: + return None + deep = _f(params.get("dca_deep_confirm"), -0.15) + is_deep = _f(cushion) <= deep + 1e-12 + depth = sorted(triggers, reverse=True)[stage - 1] + return _cand(p, A_DCA, BUY, qty, + f"补仓评估: 浮亏 {_f(cushion):.2%} 触及 {depth:.0%} 档 (第 {stage} 档), " + f"拟补 {qty} 股 (≤底仓 {max_ratio:.0%})" + + ("; **深档: 必须用户确认**, 研判须回答杀逻辑还是杀情绪" if is_deep else ""), + {"cushion_pct": cushion, "stage": stage, "trigger": depth, + "base_qty": base_qty, "price": price, "stop_ref": p.get("stop_ref")}, + confirm=is_deep) + + +def eval_trim(p: dict, params: dict, mkt: dict = None): + """保垫减仓: 垫子冲高后回吐过半 —— 减 1/3 把利润锁住。减持不设确认门槛, 规则自动执行。""" + peak = _f(p.get("cushion_peak")) + now = p.get("cushion_pct") + if now is None: + return None + if not trim_trigger(peak, _f(now), _f(params.get("trim_peak"), 0.06), + _f(params.get("trim_giveback"), 0.5)): + return None + total = int(p.get("total_qty") or 0) + qty = int(total / 3 // LOT) * LOT + if qty < LOT: + return None + return _cand(p, A_TRIM, SELL, qty, + f"保垫减仓: 安全垫峰值 {peak:.2%} 回吐至 {_f(now):.2%} (过半), " + f"减 {qty} 股锁盈", + {"cushion_peak": peak, "cushion_pct": now, "total_qty": total, + "price": p.get("price")}) + + +EVALUATORS = ((A_TRIM, eval_trim), (A_ADD, eval_add), (A_FILL, eval_fill), (A_DCA, eval_dca)) + + +# ================================================================ 扫描入口 +def scan(*, positions: list, params: dict, market: dict, skip: set = None) -> dict: + """扫描全部持仓, 产出候选动作。 + + positions: portfolio.positions_view()["held"] 的口径 + market: {ts_code: {ma5, high5, tdays_since_open, tdays_since_last_add}} + skip: 已有在途提议/指令的 (ts_code, action) 集合 —— 不重复提 + 返回 {"candidates": [...], "skipped": [...]} + """ + skip = skip or set() + out, skipped = [], [] + for p in positions or []: + code = p.get("ts_code") + if not code or int(p.get("total_qty") or 0) <= 0: + continue + mkt = (market or {}).get(code) or {} + frozen = (p.get("frozen_reason") or "NONE") != "NONE" + for action, fn in EVALUATORS: + if (code, action) in skip: + skipped.append({"ts_code": code, "action": action, "why": "已有在途提议/指令"}) + continue + # 冻结只禁增持, 减仓照评 (与规则闸同一口径, 这里先剪枝少算一遍) + if frozen and action != A_TRIM: + skipped.append({"ts_code": code, "action": action, + "why": f"该股 {p['frozen_reason']}, 禁增持"}) + continue + try: + c = fn(p, params, mkt) + except Exception as e: # 单票异常不能拖垮整轮扫描 + skipped.append({"ts_code": code, "action": action, + "why": f"评估异常 {type(e).__name__}: {e}"}) + continue + if c: + out.append(c) + return {"candidates": out, "skipped": skipped} diff --git a/app/scheduler.py b/app/scheduler.py index e5035c3..d64e3e8 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -134,10 +134,11 @@ def replay_fills(): def intraday_exec(): """盘中执行: 方案转指令 → 择时出手 (规则闸终检 → 下发 → 记子单)。 - 自主提议扫描 (FILL/ADD/DCA/TRIM 动作引擎) 仍为下一批交付。 + 含自主提议扫描 (FILL/ADD/DCA/TRIM 动作引擎 → 规则闸 → 研判闸 → 按档位分流)。 """ - from app.services import executor + from app.services import executor, proposal_service r = {"materialized": executor.materialize_plans()} + r["proposals"] = proposal_service.scan_and_route() r.update(executor.run_tick()) return r diff --git a/app/services/judge.py b/app/services/judge.py new file mode 100644 index 0000000..26d46ca --- /dev/null +++ b/app/services/judge.py @@ -0,0 +1,95 @@ +# -*- coding: utf-8 -*- +""" +研判闸客户端 · 二级关口 (设计 §7) +================================== +「PMS 备齐硬数字与账本流水作为 context → 请求决策系统研判 (其 process_intraday_audit + 新增 PMS 请求类 direction, 复用既有仲裁哲学: 代码算数、AI 裁定性、越界收口) + → 结构化答复 通过/驳回+理由。**研判服务不可用 → 自动降级 propose_only + (人工确认替位) + ERROR 告警**。命令驱动动作不过研判闸 (不用 AI 审用户)。」 + +当前状态: 决策系统侧的 PMS 请求 direction 尚未开发 (bionic 仓库的配套改造)。 +因此本模块默认 `PMS_JUDGE_API_BASE` 为空 = 未接通, 一律返回 UNAVAILABLE, 调用方按设计 +降级为「人工确认替位」。接通后只需在页面把 base 填上, 无需改代码。 + +返回结构固定: {"verdict": PASS|REJECT|UNAVAILABLE, "reason", "degraded", "raw"} +""" +from __future__ import annotations + +import logging + +from app.services import param_store + +logger = logging.getLogger("pms.judge") + +PASS, REJECT, UNAVAILABLE = "PASS", "REJECT", "UNAVAILABLE" + + +def enabled() -> bool: + return param_store.get_bool("PMS_JUDGE_ENABLED", True) + + +def base_url() -> str: + return (param_store.get("PMS_JUDGE_API_BASE", "") or "").strip().rstrip("/") + + +def available() -> bool: + return bool(enabled() and base_url()) + + +def judged_actions() -> set: + return set(param_store.get_list("PMS_JUDGE_ACTIONS", ["FILL", "ADD", "DCA", "SWITCH"])) + + +def status() -> dict: + if not enabled(): + return {"available": False, "reason": "研判闸已关闭 (PMS_JUDGE_ENABLED=False), " + "自主提议一律走人工确认"} + if not base_url(): + return {"available": False, "reason": "决策系统 PMS 研判接口未配置 " + "(PMS_JUDGE_API_BASE 为空) —— 待 bionic 侧配套改造" + "完成后填入, 当前自动降级为人工确认"} + return {"available": True, "base": base_url(), "actions": sorted(judged_actions())} + + +def request(candidate: dict, context: dict = None, *, timeout: int = None) -> dict: + """请求一次研判。任何异常/超时/未接通都返回 UNAVAILABLE (绝不把提议当成通过)。""" + action = str(candidate.get("action") or "").upper() + if action not in judged_actions(): + return {"verdict": PASS, "reason": f"{action} 不在研判范围, 规则闸通过即可", + "degraded": False, "raw": None} + if not available(): + st = status() + logger.error("[研判闸] 不可用, 降级人工确认: %s", st["reason"]) + return {"verdict": UNAVAILABLE, "reason": st["reason"], "degraded": True, "raw": None} + + payload = { + "direction": "PMS_JUDGE", "action": action, "ts_code": candidate.get("ts_code"), + "qty": candidate.get("qty"), "reason": candidate.get("reason"), + "hard_numbers": candidate.get("hard_numbers") or {}, + "context": context or {}, + # 设计要求: 补仓类研判必须回答「下跌是杀逻辑还是杀情绪」 + "must_answer": (["下跌是杀逻辑还是杀情绪"] if action == "DCA" else []), + } + to = int(timeout or param_store.get_int("PMS_JUDGE_TIMEOUT", 90)) + url = base_url() + (param_store.get("PMS_JUDGE_PATH", "/api/intraday/pms_judge") or "") + try: + import requests + r = requests.post(url, json=payload, timeout=to) + r.raise_for_status() + data = r.json() or {} + except Exception as e: + logger.error("[研判闸] 请求失败, 降级人工确认: %s: %s", type(e).__name__, e) + return {"verdict": UNAVAILABLE, "reason": f"研判请求失败: {type(e).__name__}: {e}", + "degraded": True, "raw": None} + + verdict = str(data.get("verdict") or data.get("decision") or "").upper() + if verdict in ("PASS", "APPROVE", "APPROVED", "ALLOW", "通过"): + v = PASS + elif verdict in ("REJECT", "DENY", "DENIED", "BLOCK", "驳回"): + v = REJECT + else: + logger.error("[研判闸] 答复无法识别 (%s), 按不可用降级", verdict or data) + return {"verdict": UNAVAILABLE, "reason": f"研判答复无法识别: {verdict or data}", + "degraded": True, "raw": data} + return {"verdict": v, "reason": data.get("reason") or data.get("rationale") or "", + "degraded": False, "raw": data} diff --git a/app/services/market.py b/app/services/market.py index 6c1f0ed..32ac2e0 100644 --- a/app/services/market.py +++ b/app/services/market.py @@ -189,6 +189,27 @@ def get_ma5(ts_code: str): return val +def get_high5(ts_code: str): + """近 5 个交易日最高价 (动作引擎「创 5 日新高」用)。同样按日缓存。""" + today = datetime.now().strftime("%Y%m%d") + key = f"H5:{ts_code}" + if _ma_cache["day"] != today: + _ma_cache.update({"day": today, "data": {}}) + if key in _ma_cache["data"]: + return _ma_cache["data"][key] + val = None + try: + rows = _factor_rows(ts_code, days=10) + highs = [float(r.get("high_qfq") or r.get("close_qfq") or 0) for r in rows] + highs = [h for h in highs if h > 0] + if len(highs) >= 5: + val = round(max(highs[-5:]), 3) + except Exception as e: + logger.warning("5日高点取数失败 [%s]: %s", ts_code, e) + _ma_cache["data"][key] = val + return val + + def get_refs(ts_code: str, *, base_cost=None) -> dict: """参考位: 决策系统主口径 → 日龄超期/缺失时兜底自算 → 都拿不到返回 source=none。""" stale_days = param_store.get_int("PMS_REF_STALE_TDAYS", 3) diff --git a/app/services/param_store.py b/app/services/param_store.py index d3efbfb..b596a54 100644 --- a/app/services/param_store.py +++ b/app/services/param_store.py @@ -73,6 +73,8 @@ DESC = { "PMS_REF_STALE_TDAYS": "决策系统结论日龄超此转自算兜底", "PMS_JUDGE_ENABLED": "研判闸开关", "PMS_JUDGE_ACTIONS": "需过研判闸的动作", "PMS_JUDGE_TIMEOUT": "研判超时 (秒) → 降级 propose_only", + "PMS_JUDGE_API_BASE": "决策系统 PMS 研判接口根地址; 留空=未接通, 自动降级人工确认", + "PMS_JUDGE_PATH": "研判接口路径 (bionic 侧配套改造后确定)", "PMS_T0_RATIO_MAX": "T 仓硬上限 (占持仓)", "PMS_T0_PULLBACK_PCT": "正T: 距当日高点回落触发", "PMS_T0_RALLY_PCT": "反T: 日内涨幅触发", "PMS_T0_ROUND_TARGET": "单次T目标价差", "PMS_T0_CLOSE_TIME": "T仓强制平回时点", "PMS_T0_STOCK_DAY_LOSS": "单票当日T亏熔断", diff --git a/app/services/proposal_service.py b/app/services/proposal_service.py new file mode 100644 index 0000000..62fd10e --- /dev/null +++ b/app/services/proposal_service.py @@ -0,0 +1,264 @@ +# -*- coding: utf-8 -*- +""" +自主提议: 扫描 → 规则闸 → 研判闸 → 按自主档位分流 (设计 §6 / §7) +================================================================== +分流规则 (设计原文): + full 闸门与研判通过即执行 → 直接落指令 + propose_only 增持类全部待用户确认 (一期默认) → 落 pms_proposal 队列 + off 不扫描 + +两条无条件覆盖档位的规矩: + * **减持方向不设确认门槛** —— TRIM 保垫减仓属纯规则自动执行, 任何档位都直接落指令。 + * **−15% 及更深的补仓永远需用户确认** —— 即便档位是 full, 也强制入队。 + +研判闸不可用时 (决策系统未接通/超时), 按设计自动降级为 propose_only + ERROR 告警, +**绝不把「研判拿不到」当成「研判通过」**。 +""" +from __future__ import annotations + +import logging +from datetime import datetime, timedelta + +from app.core import action_engine as ae +from app.core import command_spec as cs +from app.core import rule_gate +from app.core import tradedays as td +from app.repo import pms_repo +from app.services import (command_service, executor, judge, market, param_store, portfolio) + +logger = logging.getLogger("pms.proposal") + +AUTONOMY_FULL, AUTONOMY_PROPOSE, AUTONOMY_OFF = "full", "propose_only", "off" + + +def scan_and_route(*, now=None, dry_run: bool = False) -> dict: + """自主提议扫描一轮。dry_run=True 只出候选与判定, 不落任何表。""" + now = now or datetime.now() + out = {"ok": True, "autonomy": None, "candidates": 0, "executed": [], "queued": [], + "rejected": [], "skipped": [], "errors": [], "degraded": False, "dry_run": dry_run} + + autonomy = param_store.get("PMS_AUTONOMY", AUTONOMY_PROPOSE) + out["autonomy"] = autonomy + if autonomy == AUTONOMY_OFF: + out["skipped"].append({"why": "自主档位 off, 不扫描"}) + return out + if param_store.get_bool("PMS_GLOBAL_EXEC_HALT", False): + out["skipped"].append({"why": "全局暂停执行 (休假模式)"}) + return out + + try: + view = portfolio.positions_view() + params = _scan_params(view) + mkt = _market_ctx(view["held"], now) + params["_mkt"] = mkt # 规则闸要用同一份 MA5, 不再重取 + skip = _inflight_keys() + scanned = ae.scan(positions=view["held"], params=params, market=mkt, skip=skip) + except Exception as e: + logger.exception("提议扫描失败") + return {**out, "ok": False, "errors": [f"扫描失败: {type(e).__name__}: {e}"]} + + out["candidates"] = len(scanned["candidates"]) + out["skipped"].extend(scanned["skipped"]) + stock_params = command_service.effective_stock_params() + brake_active = td.ymd() < param_store.get_int("PMS_BRAKE_UNTIL", 0) + + for c in scanned["candidates"]: + try: + _route_one(c, view, params, stock_params, brake_active, now, dry_run, out) + except Exception as e: + logger.exception("提议分流失败 %s", c.get("ts_code")) + out["errors"].append(f"{c.get('ts_code')} {c.get('action')}: " + f"{type(e).__name__}: {e}") + out["ok"] = not out["errors"] + return out + + +def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out): + code, action, side = c["ts_code"], c["action"], c["side"] + pos = _pos_of(view, code) + price = float(pos.get("price") or 0) + + # ---- 一级: 规则闸 (自主动作受刹车约束, is_command=False) ---- + gate = rule_gate.check( + side=side, action=action, qty=c["qty"], price=price, + ctx={"ts_code": code, "position": pos, + "day": {"price": price, "vwap": price, + "ma5": (params.get("_mkt") or {}).get(code, {}).get("ma5"), + "day_chg_from_open": None}, + "params": {"no_chase_ma5": params.get("no_chase_ma5"), + "buy_halt_dayup": params.get("buy_halt_dayup"), + "sector_source_ready": view["sector_ready"]}, + "caps": portfolio.caps_ctx(view, ts_code=code) if side == "buy" else None, + "flags": {"buy_halt": params.get("buy_halt"), "exec_halt": params.get("exec_halt"), + "brake_active": brake_active, + "blacklisted": bool(stock_params.get(code, {}).get("black")), + "is_command": False}}) + if not gate["passed"]: + out["rejected"].append({"ts_code": code, "action": action, "by": "rule", + "failed": gate["failed"]}) + if not dry_run: + pms_repo.insert_ledger(ts_code=code, action=action, arbiter="rule", verdict="REJECT", + price_at=price, hard_numbers=c["hard_numbers"], + failed_checks=gate["failed"], reason="自主提议未过规则闸") + return + + # ---- 二级: 研判闸 (仅补足/加仓/补仓; 减持不送研判) ---- + verdict = {"verdict": judge.PASS, "reason": "", "degraded": False} + if c.get("judge_required"): + verdict = judge.request(c, context={"position": _judge_ctx(pos), + "recent_ledger": _recent_ledger(code)}) + if verdict["verdict"] == judge.REJECT: + out["rejected"].append({"ts_code": code, "action": action, "by": "judge", + "failed": [verdict.get("reason") or "研判驳回"]}) + if not dry_run: + pms_repo.insert_ledger(ts_code=code, action=action, arbiter="judge", + verdict="REJECT", price_at=price, + hard_numbers=c["hard_numbers"], + reason=verdict.get("reason") or "研判驳回") + return + if verdict.get("degraded"): + out["degraded"] = True + + # ---- 三级: 按档位分流 ---- + autonomy = out["autonomy"] + force_queue = bool(c.get("needs_user_confirm")) or verdict.get("degraded") + auto_exec = (side == "sell") or (autonomy == AUTONOMY_FULL and not force_queue) + + if dry_run: + (out["executed"] if auto_exec else out["queued"]).append( + {**_brief(c), "route": "auto" if auto_exec else "queue", + "judge": verdict["verdict"], "dry_run": True}) + return + + if auto_exec: + iid = _make_instruction(c, price, now) + pms_repo.insert_ledger(ts_code=code, action=action, + arbiter="judge" if c.get("judge_required") else "rule", + verdict="PASS", price_at=price, hard_numbers=c["hard_numbers"], + ref_id=iid, + reason=(verdict.get("reason") or c["reason"])[:500]) + out["executed"].append({**_brief(c), "instruction_id": iid, + "why": "减持方向自动执行" if side == "sell" else "档位 full"}) + else: + pid = _make_proposal(c, price, verdict) + why = ("深档补仓强制确认" if c.get("needs_user_confirm") + else ("研判不可用, 降级人工确认" if verdict.get("degraded") + else "档位 propose_only")) + out["queued"].append({**_brief(c), "proposal_id": pid, "why": why}) + + +# ================================================================ 落表 +def _make_instruction(c, price, now) -> str: + ymd = td.ymd(now) + seq = int(now.strftime("%H%M%S")) + iid = cs.make_instruction_id(ymd, c["ts_code"], c["action"], seq % 1000) + window = param_store.get_int("PMS_EXEC_WINDOW_TDAYS", 3) + pms_repo.insert_instruction( + instruction_id=iid, origin_type="proposal", origin_id=None, ts_code=c["ts_code"], + action=c["action"], side=c["side"], qty=c["qty"], limit_price=None, + window_tdays=window, status=executor.ST_PROPOSED, + progress={"deadline": str(td.window_deadline(now.date(), window)), + "is_command": False, "children": [], "auto": True, + "reason": c["reason"]}) + return iid + + +def _make_proposal(c, price, verdict) -> str: + ttl = param_store.get_int("PMS_PROPOSAL_TTL_HOURS", 24) + pid = f"PRP_{td.ymd()}_{c['ts_code'].replace('.', '')}_{c['action']}" + hn = {**(c.get("hard_numbers") or {}), "price": price, "reason": c["reason"], + "needs_user_confirm": c.get("needs_user_confirm", False)} + pms_repo.insert_proposal( + proposal_id=pid, ts_code=c["ts_code"], action=c["action"], qty=c["qty"], + hard_numbers=hn, expire_at=datetime.now() + timedelta(hours=ttl), + judge_verdict=verdict.get("verdict"), + judge_reason=(verdict.get("reason") or "")[:500]) + return pid + + +# ================================================================ 上下文 +def _scan_params(view: dict) -> dict: + p = dict(view["params"]) + p.update({ + "cushion_solid": param_store.get_float("PMS_CUSHION_SOLID", 0.03), + "trim_peak": param_store.get_float("PMS_TRIM_PEAK", 0.06), + "trim_giveback": param_store.get_float("PMS_TRIM_GIVEBACK", 0.5), + "dca_triggers": param_store.get_tuple_floats("PMS_DCA_TRIGGERS", (-0.08, -0.15)), + "dca_deep_confirm": param_store.get_float("PMS_DCA_DEEP_CONFIRM", -0.15), + "dca_max_ratio": param_store.get_float("PMS_DCA_MAX_RATIO", 0.5), + "no_chase_ma5": param_store.get_float("PMS_NO_CHASE_MA5", 0.06), + "buy_halt_dayup": param_store.get_float("PMS_BUY_HALT_DAYUP", 0.05), + "build_window_tdays": param_store.get_int("PMS_BUILD_WINDOW_TDAYS", 10), + "fill_max_loss": param_store.get_float("PMS_FILL_MAX_LOSS", -0.03), + }) + return p + + +def _market_ctx(held: list, now) -> dict: + """每票的 MA5 / 5日高点 / 建仓天数 / 距上次加仓天数 (交易日口径)。""" + out = {} + today = now.date() if hasattr(now, "date") else now + for p in held or []: + code = p["ts_code"] + d = {"ma5": market.get_ma5(code), "high5": market.get_high5(code)} + opened, last_add = p.get("opened_date"), p.get("last_add_date") + d["tdays_since_open"] = _tdays_between(opened, today) + d["tdays_since_last_add"] = _tdays_between(last_add, today) + out[code] = d + return out + + +def _tdays_between(start, today): + """start(含) 到 today 的交易日数; start 为空返回 None (调用方按「无约束」处理)。""" + if not start: + return None + try: + return max(0, td.trade_days_left(today, start) - 1) + except Exception: + return None + + +def _inflight_keys() -> set: + """已有在途提议或在途指令的 (代码, 动作) —— 同一件事不重复提。""" + keys = set() + try: + for p in pms_repo.list_proposals(statuses=("WAIT_USER",), limit=200): + keys.add((p["ts_code"], p["action"])) + except Exception as e: + logger.warning("读提议队列失败: %s", e) + try: + for i in pms_repo.list_instructions(statuses=list(executor.LIVE), limit=300): + keys.add((i["ts_code"], i.get("action"))) + except Exception as e: + logger.warning("读在途指令失败: %s", e) + return keys + + +def _pos_of(view: dict, ts_code: str) -> dict: + for x in view["positions"]: + if x["ts_code"] == ts_code: + return x + return {"ts_code": ts_code, "total_qty": 0, "avail_qty": 0, "frozen_reason": "NONE"} + + +def _judge_ctx(pos: dict) -> dict: + """送研判的账本快照 (设计 §7: PMS 备齐硬数字与账本流水作为 context)。""" + keep = ("ts_code", "price", "avg_cost", "total_qty", "base_qty", "add_qty", "dca_qty", + "cushion_pct", "cushion_peak", "pct_of_scale", "target_pct", "support_ref", + "pressure_ref", "stop_ref", "ref_source", "sector", "neg_cushion_days") + return {k: pos.get(k) for k in keep} + + +def _recent_ledger(ts_code: str, limit: int = 10) -> list: + try: + rows = pms_repo.list_ledger(ts_code=ts_code, limit=limit) + except Exception: + return [] + return [{"at": str(r.get("decided_at")), "action": r.get("action"), + "arbiter": r.get("arbiter"), "verdict": r.get("verdict"), + "price": r.get("price_at"), "reason": r.get("reason")} for r in rows] + + +def _brief(c: dict) -> dict: + return {"ts_code": c["ts_code"], "action": c["action"], "side": c["side"], + "qty": c["qty"], "reason": c["reason"]} diff --git a/app/web/main.py b/app/web/main.py index ed614a5..dd1a94c 100644 --- a/app/web/main.py +++ b/app/web/main.py @@ -309,10 +309,17 @@ def api_cancel_instruction(instruction_id: str, payload: dict = Body(default={}) payload.get("reason") or "页面人工撤销") +@app.post("/api/ops/scan-proposals") +def api_scan_proposals(dry_run: bool = Query(False)): + """自主提议扫描一轮 (动作引擎 → 规则闸 → 研判闸 → 按档位分流)。""" + from app.services import proposal_service + return ok(proposal_service.scan_and_route, dry_run=dry_run) + + @app.get("/api/dispatch-mode") def api_dispatch_mode(): - from app.services import dispatcher - return ok(dispatcher.describe) + from app.services import dispatcher, judge + return ok(lambda: {"ok": True, **dispatcher.describe(), "judge": judge.status()}) @app.get("/api/ops/downstream-schema") diff --git a/app/web/static/index.html b/app/web/static/index.html index 62c9740..204e1fb 100644 --- a/app/web/static/index.html +++ b/app/web/static/index.html @@ -382,6 +382,9 @@ propose_only 档位下, 增持类自主动作全部在此等待裁决; 减持方向不设确认门槛。 采纳后先落指令表 (先记账后动作), 由择时执行器分日出手。 + @@ -416,6 +419,11 @@ 出手一跳 窗口收口 +
+ + 提议扫描试算 (不落表) + 提议扫描 +
成交回放 账本对账 diff --git a/config/settings.py b/config/settings.py index ecaee2f..7e773c7 100644 --- a/config/settings.py +++ b/config/settings.py @@ -99,6 +99,8 @@ class Settings(BaseSettings): PMS_JUDGE_ENABLED: bool = True PMS_JUDGE_ACTIONS: str = "FILL,ADD,DCA,SWITCH" PMS_JUDGE_TIMEOUT: int = 90 # 超时 → 降级 propose_only + PMS_JUDGE_API_BASE: str = "" # 决策系统 PMS 研判接口根地址; 空=未接通(自动降级人工确认) + PMS_JUDGE_PATH: str = "/api/intraday/pms_judge" # 研判接口路径 (bionic 侧配套改造后确定) # --- T0 做T (命令授权制) --- PMS_T0_RATIO_MAX: float = 0.333 # T仓硬上限 (占持仓) diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 700c1e2..d56c911 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -8,7 +8,8 @@ test_core_units.py 仓位规划器 / 安全垫与成本账 (14 例) test_batch2_units.py 命令状态机 / 方案生成器 / 回放对账纯逻辑 (35 例) test_batch3_units.py 规则闸 / 择时执行器实现B 纯逻辑 (18 例) - test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (25 例) + test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 (11 例) + test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (30 例) 任一子集失败即整体失败 (退出码 1)。 """ import os @@ -18,7 +19,7 @@ import sys HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py", - "test_wiring.py"] + "test_batch4_units.py", "test_wiring.py"] def main(): diff --git a/scripts/test_batch4_units.py b/scripts/test_batch4_units.py new file mode 100644 index 0000000..a92a775 --- /dev/null +++ b/scripts/test_batch4_units.py @@ -0,0 +1,201 @@ +# -*- coding: utf-8 -*- +""" +第四批模块单测 (实机运行, 零外部依赖) +====================================== +运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch4_units.py +覆盖: action_engine 四类自主动作的触发边界与数量口径 (FILL/ADD/DCA/TRIM) 及扫描剪枝。 +""" +import os +import sys +import traceback + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.core import action_engine as ae # noqa: E402 + +RESULTS = [] + + +def case(name): + def deco(fn): + RESULTS.append((name, fn)) + return fn + return deco + + +PARAMS = {"scale": 2_000_000, "stock_target_default": 0.06, + "batch_split": (0.5, 0.25, 0.25), "cushion_solid": 0.03, + "trim_peak": 0.06, "trim_giveback": 0.5, + "dca_triggers": (-0.08, -0.15), "dca_deep_confirm": -0.15, "dca_max_ratio": 0.5, + "no_chase_ma5": 0.06, "build_window_tdays": 10, "fill_max_loss": -0.03} + + +def pos(**kw): + p = {"ts_code": "600000.SH", "price": 10.0, "avg_cost": 10.0, "total_qty": 6000, + "base_qty": 6000, "add_qty": 0, "dca_qty": 0, "market_value": 60_000, + "cushion_pct": 0.0, "cushion_peak": 0.0, "target_pct": 0.06, + "support_ref": None, "pressure_ref": None, "stop_ref": None, + "fill_count": 0, "dca_count": 0, "frozen_reason": "NONE"} + p.update(kw) + return p + + +def mkt(**kw): + m = {"ma5": 10.0, "high5": 10.0, "tdays_since_open": 3, "tdays_since_last_add": 5} + m.update(kw) + return m + + +# ================================================================ FILL +@case("回踩补足·建仓期内浅亏未破支撑 → 按补足批额度补到目标") +def _(): + c = ae.eval_fill(pos(price=9.8, cushion_pct=-0.02, support_ref=9.7, market_value=58_800), + PARAMS, mkt()) + assert c and c["action"] == "FILL" and c["side"] == "buy", c + assert c["qty"] == 3000, c # 补足批 3 万 ÷ 9.8 → 3000 股 + assert c["judge_required"] and not c["needs_user_confirm"] + assert "未破支撑" in c["reason"] + + +@case("回踩补足·每票一次 / 出建仓期 / 亏太深 / 破支撑 / 无支撑 一律不提") +def _(): + base = dict(price=9.8, cushion_pct=-0.02, support_ref=9.7, market_value=58_800) + assert ae.eval_fill(pos(**base, fill_count=1), PARAMS, mkt()) is None + assert ae.eval_fill(pos(**base), PARAMS, mkt(tdays_since_open=11)) is None + assert ae.eval_fill(pos(price=9.5, cushion_pct=-0.05, support_ref=9.0, + market_value=57_000), PARAMS, mkt()) is None + assert ae.eval_fill(pos(price=9.5, cushion_pct=-0.02, support_ref=9.7, + market_value=57_000), PARAMS, mkt()) is None # 破支撑 + assert ae.eval_fill(pos(**{**base, "support_ref": None}), PARAMS, mkt()) is None + # 已转盈就不是「回踩补足」的场景了 + assert ae.eval_fill(pos(price=10.5, cushion_pct=0.05, support_ref=9.7, + market_value=63_000), PARAMS, mkt()) is None + + +# ================================================================ ADD +@case("盈利加仓·厚垫 + 创5日新高 → 按加仓批额度加") +def _(): + c = ae.eval_add(pos(price=11.0, cushion_pct=0.10, market_value=66_000), + PARAMS, mkt(ma5=10.6, high5=11.0)) + assert c and c["action"] == "ADD" and c["qty"] == 2700, c # 3万 ÷ 11 → 2700 股 + assert "创 5 日新高" in c["reason"], c + # 站上压力位也算 + c2 = ae.eval_add(pos(price=11.0, cushion_pct=0.10, pressure_ref=10.9, + market_value=66_000), PARAMS, mkt(ma5=10.6, high5=12.0)) + assert c2 and "压力位" in c2["reason"], c2 + + +@case("盈利加仓·薄垫/未突破/距上次<2日/追高 一律不提") +def _(): + assert ae.eval_add(pos(price=11.0, cushion_pct=0.02, market_value=66_000), + PARAMS, mkt(high5=11.0)) is None # 垫子不厚 + assert ae.eval_add(pos(price=10.5, cushion_pct=0.10, market_value=63_000), + PARAMS, mkt(high5=12.0)) is None # 没新高没压力位 + assert ae.eval_add(pos(price=11.0, cushion_pct=0.10, market_value=66_000), + PARAMS, mkt(high5=11.0, tdays_since_last_add=1)) is None + assert ae.eval_add(pos(price=11.0, cushion_pct=0.10, market_value=66_000), + PARAMS, mkt(ma5=10.0, high5=11.0)) is None # 距 MA5 10% > 6% + + +# ================================================================ DCA +@case("补仓·触及 -8% 首档 → 按底仓一半提, 不强制确认") +def _(): + c = ae.eval_dca(pos(price=9.1, cushion_pct=-0.09, market_value=54_600), PARAMS, mkt()) + assert c and c["action"] == "DCA" and c["qty"] == 3000, c # 底仓 6000 × 50% + assert c["needs_user_confirm"] is False and c["judge_required"], c + assert c["hard_numbers"]["stage"] == 1 + + +@case("补仓·-15% 及更深 永远需用户确认 + 研判须答杀逻辑还是杀情绪") +def _(): + c = ae.eval_dca(pos(price=8.4, cushion_pct=-0.16, market_value=50_400), PARAMS, mkt()) + assert c and c["needs_user_confirm"] is True, c + assert c["hard_numbers"]["stage"] == 2 and "必须用户确认" in c["reason"], c + + +@case("补仓·各档只评估一次 / 终身只执行一次 / 浮盈不评估") +def _(): + assert ae.eval_dca(pos(cushion_pct=-0.09, dca_count=1), PARAMS, mkt()) is None + assert ae.eval_dca(pos(cushion_pct=-0.16, dca_count=1), PARAMS, mkt()) is not None + assert ae.eval_dca(pos(cushion_pct=-0.16, dca_count=2), PARAMS, mkt()) is None + assert ae.eval_dca(pos(cushion_pct=-0.20, dca_qty=1000), PARAMS, mkt()) is None + assert ae.eval_dca(pos(cushion_pct=-0.05), PARAMS, mkt()) is None + assert ae.eval_dca(pos(cushion_pct=0.05), PARAMS, mkt()) is None + + +# ================================================================ TRIM +@case("保垫减仓·峰值≥6%且回吐过半 → 减 1/3, 不需研判不需确认") +def _(): + c = ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=0.04), PARAMS) + assert c and c["action"] == "TRIM" and c["side"] == "sell", c + assert c["qty"] == 2000, c # 6000 的 1/3 + assert not c["judge_required"] and not c["needs_user_confirm"], c + + +@case("保垫减仓·峰值不足或回吐不到一半 不提") +def _(): + assert ae.eval_trim(pos(cushion_peak=0.05, cushion_pct=-0.01), PARAMS) is None + assert ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=0.0401), PARAMS) is None + assert ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=None), PARAMS) is None + assert ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=0.04, total_qty=200), + PARAMS) is None # 1/3 不足一手 + + +# ================================================================ 扫描 +@case("扫描·冻结票只评减仓 / 已在途不重复提 / 单票异常不拖垮整轮") +def _(): + ps = [pos(ts_code="600000.SH", cushion_peak=0.08, cushion_pct=0.04, + frozen_reason="COMMAND_HALT"), + pos(ts_code="000001.SZ", price=11.0, cushion_pct=0.10, market_value=66_000), + pos(ts_code="600519.SH", price=None, cushion_pct=0.10, market_value=66_000)] + r = ae.scan(positions=ps, params=PARAMS, market={ + "600000.SH": mkt(), "000001.SZ": mkt(ma5=10.6, high5=11.0), "600519.SH": mkt()}) + acts = {(c["ts_code"], c["action"]) for c in r["candidates"]} + assert ("600000.SH", "TRIM") in acts, acts # 冻结不挡减仓 + assert not any(c[0] == "600000.SH" and c[1] != "TRIM" for c in acts), acts + assert ("000001.SZ", "ADD") in acts, acts + assert any(s["why"].endswith("禁增持") for s in r["skipped"]), r["skipped"] + + r2 = ae.scan(positions=ps, params=PARAMS, + market={"600000.SH": mkt(), "000001.SZ": mkt(ma5=10.6, high5=11.0), + "600519.SH": mkt()}, + skip={("000001.SZ", "ADD")}) + assert ("000001.SZ", "ADD") not in {(c["ts_code"], c["action"]) for c in r2["candidates"]} + assert any(s.get("why") == "已有在途提议/指令" for s in r2["skipped"]), r2["skipped"] + + # 空仓票直接跳过, 不进候选 + r3 = ae.scan(positions=[pos(ts_code="300750.SZ", total_qty=0)], params=PARAMS, market={}) + assert r3["candidates"] == [] + + +@case("扫描·批次额度与距目标空间口径") +def _(): + p = pos(market_value=100_000) + assert ae.batch_amount(p, PARAMS, 0) == 60_000.0 # 底仓 50% + assert ae.batch_amount(p, PARAMS, 1) == 30_000.0 # 补足 25% + assert ae.batch_amount(p, PARAMS, 2) == 30_000.0 # 加仓 25% + assert ae.room_to_target(p, PARAMS) == 20_000.0 + assert ae.room_to_target(pos(market_value=130_000), PARAMS) == 0.0 + + +# ---------------------------------------------------------------- runner +def main(): + passed, failed = 0, 0 + for name, fn in RESULTS: + try: + fn() + print(f" PASS {name}") + passed += 1 + except Exception: + print(f" FAIL {name}") + traceback.print_exc() + failed += 1 + print("-" * 60) + if failed: + print(f"FAILED: {failed} / {passed + failed}") + sys.exit(1) + print(f"ALL PASS ({passed} cases)") + + +if __name__ == "__main__": + main() diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index 8322f0a..d2b516a 100644 --- a/scripts/test_wiring.py +++ b/scripts/test_wiring.py @@ -298,7 +298,7 @@ class FakeRepo: return len(rows) -def install_fakes(prices=None, positions=None, params=None): +def install_fakes(prices=None, positions=None, params=None, high5=None): """把内存桩装到各模块上, 返回 FakeRepo 实例。""" from app.repo import downstream_repo, pms_repo from app.services import industry, market, param_store, portfolio @@ -323,6 +323,7 @@ def install_fakes(prices=None, positions=None, params=None): market.get_refs = lambda c, **kw: {"support": None, "pressure": None, "stop": None, "source": "none"} market.get_ma5 = lambda c: (prices or {}).get(c) + market.get_high5 = lambda c: (high5 or prices or {}).get(c) market.day_snapshot = lambda c: ({} if not (prices or {}).get(c) else { "price": prices[c], "vwap": prices[c], "open": prices[c], "high": prices[c] * 1.02, "low": prices[c] * 0.98, "day_chg_from_open": 0.0, "bars": 60}) @@ -871,21 +872,160 @@ def _(): assert executor.cancel_instruction("INS_C")["ok"] is False +def _prop_fakes(**kw): + """自主提议用的组合: A 票厚垫创新高(可加仓), B 票垫子回吐过半(可保垫减仓)。""" + return install_fakes( + prices={"600000.SH": 11.0, "000001.SZ": 10.4}, + high5={"600000.SH": 11.0, "000001.SZ": 12.0}, # B 没创新高, 只该出 TRIM + params={"PMS_TOTAL_SCALE": "2000000", **(kw.get("params") or {})}, + positions=[{"ts_code": "600000.SH", "total_qty": 6000, "avail_qty": 6000, + "base_qty": 6000, "avg_cost": 10.0, "cushion_peak": 0.0, + "target_pct": 0.06}, + {"ts_code": "000001.SZ", "total_qty": 6000, "avail_qty": 6000, + "base_qty": 6000, "avg_cost": 10.0, "cushion_peak": 0.08, + "target_pct": 0.06}]) + + +@case("自主提议·propose_only: 减持自动执行 / 增持入队 / 研判未接通即降级留痕") +def _(): + from app.services import proposal_service as ps + fake = _prop_fakes(params={"PMS_AUTONOMY": "propose_only"}) + r = ps.scan_and_route() + assert r["ok"], r + ex = {(x["ts_code"], x["action"]) for x in r["executed"]} + qd = {(x["ts_code"], x["action"]) for x in r["queued"]} + assert ("000001.SZ", "TRIM") in ex, r # 减持方向不设确认门槛 + assert ("600000.SH", "ADD") in qd, r # 增持入队 + assert r["degraded"] is True, r # 研判未接通 → 降级 + assert any("研判不可用" in x["why"] for x in r["queued"]), r["queued"] + # 减持落了指令, 增持落了提议 + trim_ins = [i for i in fake.instructions.values() if i["action"] == "TRIM"] + assert trim_ins and trim_ins[0]["side"] == "sell" and trim_ins[0]["qty"] == 2000, trim_ins + add_prop = [p for p in fake.proposals.values() if p["action"] == "ADD"] + assert add_prop and add_prop[0]["qty"] == 2700, add_prop + assert add_prop[0]["hard_numbers"]["price"] == 11.0 + # 再扫一轮不重复提 (在途去重) + r2 = ps.scan_and_route() + assert not r2["executed"] and not r2["queued"], r2 + assert any(s.get("why") == "已有在途提议/指令" for s in r2["skipped"]), r2["skipped"] + + +@case("自主提议·full + 研判通过: 增持直接落指令; 深档补仓仍强制确认") +def _(): + from app.services import judge, proposal_service as ps + orig = judge.request + try: + judge.request = lambda c, context=None, **kw: {"verdict": judge.PASS, + "reason": "研判通过(桩)", + "degraded": False, "raw": None} + fake = _prop_fakes(params={"PMS_AUTONOMY": "full"}) + r = ps.scan_and_route() + assert r["degraded"] is False, r + ex = {(x["ts_code"], x["action"]) for x in r["executed"]} + assert ("600000.SH", "ADD") in ex and ("000001.SZ", "TRIM") in ex, r + assert not r["queued"], r + add_ins = [i for i in fake.instructions.values() if i["action"] == "ADD"] + assert add_ins and add_ins[0]["side"] == "buy" and add_ins[0]["qty"] == 2700, add_ins + assert add_ins[0]["progress"]["is_command"] is False + + # 深档补仓: 即使档位 full、研判通过, 也必须入队等用户点头 + fake2 = install_fakes(prices={"600519.SH": 8.4}, high5={"600519.SH": 9.9}, + params={"PMS_TOTAL_SCALE": "2000000", "PMS_AUTONOMY": "full"}, + positions=[{"ts_code": "600519.SH", "total_qty": 6000, + "avail_qty": 6000, "base_qty": 6000, + "avg_cost": 10.0, "cushion_peak": 0.0, + "target_pct": 0.06}]) + r2 = ps.scan_and_route() + qd = {(x["ts_code"], x["action"]) for x in r2["queued"]} + assert ("600519.SH", "DCA") in qd, r2 + assert any("深档" in x["why"] for x in r2["queued"]), r2["queued"] + prop = [p for p in fake2.proposals.values() if p["action"] == "DCA"][0] + assert prop["qty"] == 3000 and prop["hard_numbers"]["stage"] == 2, prop + finally: + judge.request = orig + + +@case("自主提议·研判驳回与规则闸拦截各自留痕") +def _(): + from app.services import judge, proposal_service as ps + orig = judge.request + try: + judge.request = lambda c, context=None, **kw: {"verdict": judge.REJECT, + "reason": "形态走坏, 不宜加仓", + "degraded": False, "raw": None} + fake = _prop_fakes(params={"PMS_AUTONOMY": "full"}) + r = ps.scan_and_route() + rej = {(x["ts_code"], x["action"], x["by"]) for x in r["rejected"]} + assert ("600000.SH", "ADD", "judge") in rej, r + assert any(x["arbiter"] == "judge" and x["verdict"] == "REJECT" for x in fake.ledger) + assert not any(i["action"] == "ADD" for i in fake.instructions.values()) + finally: + judge.request = orig + + # 规则闸拦截: 全局暂停买入 + fake2 = _prop_fakes(params={"PMS_AUTONOMY": "full", "PMS_GLOBAL_BUY_HALT": "true"}) + r2 = ps.scan_and_route() + rej2 = {(x["ts_code"], x["action"], x["by"]) for x in r2["rejected"]} + assert ("600000.SH", "ADD", "rule") in rej2, r2 + assert any("BUY_HALT" in f for x in r2["rejected"] for f in x["failed"]), r2 + assert any(x["arbiter"] == "rule" and x["verdict"] == "REJECT" for x in fake2.ledger) + # 减持不受暂停买入影响, 照样执行 + assert ("000001.SZ", "TRIM") in {(x["ts_code"], x["action"]) for x in r2["executed"]}, r2 + + +@case("自主提议·档位 off 与休假模式不扫描; 试算不落表") +def _(): + from app.services import proposal_service as ps + _prop_fakes(params={"PMS_AUTONOMY": "off"}) + r = ps.scan_and_route() + assert r["candidates"] == 0 and any("off" in s["why"] for s in r["skipped"]), r + + _prop_fakes(params={"PMS_AUTONOMY": "full", "PMS_GLOBAL_EXEC_HALT": "true"}) + r2 = ps.scan_and_route() + assert any("休假" in s["why"] for s in r2["skipped"]), r2 + + fake = _prop_fakes(params={"PMS_AUTONOMY": "propose_only"}) + r3 = ps.scan_and_route(dry_run=True) + assert (r3["executed"] or r3["queued"]) and all( + x.get("dry_run") for x in r3["executed"] + r3["queued"]), r3 + assert not fake.instructions and not fake.proposals and not fake.ledger + + +@case("研判闸·未配置即不可用, 动作不在研判范围则直接放行") +def _(): + from app.services import judge + install_fakes() + assert judge.available() is False + st = judge.status() + assert st["available"] is False and "未配置" in st["reason"], st + r = judge.request({"action": "ADD", "ts_code": "600000.SH", "qty": 100}) + assert r["verdict"] == judge.UNAVAILABLE and r["degraded"] is True, r + r2 = judge.request({"action": "TRIM", "ts_code": "600000.SH", "qty": 100}) + assert r2["verdict"] == judge.PASS and r2["degraded"] is False, r2 # TRIM 不在研判范围 + + @case("装配·执行相关路由与调度接线到位") def _(): from app.web.main import app from app import scheduler as sch paths = {r.path for r in app.routes} for p in ("/api/ops/materialize", "/api/ops/exec-tick", "/api/ops/sweep-windows", - "/api/instructions/{instruction_id}/cancel", "/api/dispatch-mode"): + "/api/instructions/{instruction_id}/cancel", "/api/dispatch-mode", + "/api/ops/scan-proposals"): assert p in paths, p import inspect src = inspect.getsource(sch.intraday_exec) assert "executor" in src and "run_tick" in src, "调度器未接执行器" + assert "proposal_service" in src, "调度器未接自主提议扫描" # ---------------------------------------------------------------- runner def main(): + # 静音日志: 本套里有好几条用例**故意**触发异常与告警来验证「守成」行为 + # (调度守卫吞异常、外部成交告警、连续对账升级 ERROR、窗口耗尽告警), + # 这些 ERROR 栈打在测试输出里会被误读成失败。判定标准只看断言, 不看日志。 + import logging + logging.disable(logging.CRITICAL) passed, failed = 0, 0 for name, fn in RESULTS: try: