PMS 文案第二批：规则闸、仓位约束、资金、计划取数

抽了一个公共的显示层函数 app/core/copy.py，把句首那个大写判据码剥掉再给人看。

这一步是有讲究的。那些判据码是承重的：账本靠它取去重键（同一条被拒指令当日只记一条），
命令进度靠它做原因聚类。2026-07-29 账本被同一条指令按分钟灌满，病根就是去重键失效。
所以码必须留在句首，只能在显示层剥。此前后端两处各写了一份剥前缀的逻辑，现在共用一份。

改写的文案：

规则闸六句。冻结原因原先写「该股处于 COMMAND_HALT, 禁止增持」，既印了枚举值，也没分开
三种来路——你下的命令、你手设的、系统刹车期间，对交易员是三件不同的事。昨夜信号原先写
「决策系统昨夜定性 SELL (形态恶化)」，SELL 是枚举值，而「形态恶化」是给三种信号硬安的
定性，回避未必是形态问题。

仓位约束六句，原先全是「加后总仓 82.3% > 上限 80%」这种电报体。大于号要人翻译，而且
没说清百分比是占什么。行业占比那条还漏了设计要点——两条都超才拦，不写会被当成误拦
来投诉。

资金四句改在源头。这个字段有三个下游都直接显示给人看，改一处三处一起干净。异常原文
只进日志。

计划取数两句会渲染成页面顶部整条红色横幅。原先网络抖一下就顶一条异常原文给基金经理
看。改成说清三件事：现在拿不到什么、多久之后重试有意义、还不行找谁。其中「过 1 分钟
再试」不能省——失败结果会被缓存一分钟，早于这个点重读拿到的还是同一条失败。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

app/core/action_engine.py

@@ -32,6 +32,7 @@
 from __future__ import annotations
 
 from app.core.cushion import dca_stage, trim_trigger
+from app.core import copy
 from app.core.sizer import LOT, lot_of, lot_qty, split_batches
 # 新建仓要在这里滚动校验上限, 用的必须是规则闸那一份 check_all_caps, 不能另写一套。
 # _new_name_ctx / _ctx_after 是命令驱动建仓 (planner.plan_increase_exposure) 滚动更新组合
@@ -96,6 +97,21 @@ WHY_WATCH_CONFIRM = "选股系统判为关注（无法判断），交人裁决"
 WHY_DISPLAY_ONLY = "选股系统判为仅展示"
 
 
+# 冻结的三种来路，对交易员是三件不同的事：自己下的命令、自己手设的、系统刹车期间。
+# 原先一句「该股 COMMAND_HALT, 禁增持」三种都不分，还把枚举值印在了页面上。
+_FROZEN_WHY = {
+    "COMMAND_HALT": "你下过命令把这只票设成只卖不买",
+    "MANUAL": "你手动把这只票设成只卖不买",
+    "BRAKE": "组合刹车期间这只票只卖不买",
+}
+
+
+def frozen_why(reason) -> str:
+    """这只票为什么不能加仓。认不出的来路说一句实话，不硬猜。"""
+    r = str(reason or "").strip()
+    return _FROZEN_WHY.get(r, "这只票现在只卖不买") + "，所以不评估加仓"
+
+
 def display_only_why(row) -> str:
     """这一栏回答的是「为什么今天没买这只票」。所以它只写原因, 不写系统内部的判决名。
 
@@ -425,7 +441,7 @@ def scan(*, positions: list, params: dict, market: dict, skip=None,
             # 冻结只禁增持, 减仓照评 (与规则闸同一口径, 这里先剪枝少算一遍)
             if frozen and action not in SELL_SIDE_ACTIONS:
                 skipped.append({"ts_code": code, "action": action,
-                                "why": f"该股 {p['frozen_reason']}, 禁增持"})
+                                "why": frozen_why(p.get("frozen_reason"))})
                 continue
             try:
                 c = fn(p, params, mkt)
@@ -514,7 +530,10 @@ def eval_open(c: dict, params: dict, caps: dict, room_amt: float):
     full_amt = sum(b["qty"] * price for b in sp["batches"])
     bad = check_all_caps(ts_code=code, add_amount=full_amt, ctx=_new_name_ctx(caps, c))
     if bad:
-        return None, "; ".join(bad)
+        # 这一串是纯显示用途 —— 它只被塞进 by_code[].why 给页面看, 没有任何地方拿它
+        # 归类或做键 (做键的是 executor 那条路, 用的是另一份带前缀的原串)。所以在这里
+        # 剥掉大写判据码, 页面上就不会出现「PORTFOLIO_CAP: …」这种东西。
+        return None, copy.join_reasons(bad)
 
     base = sp["batches"][0]
     qty = int(base["qty"])

app/core/copy.py

@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+"""给人看的文字的统一处理。后端与前端共用同一套口径。
+
+## 为什么要有这个文件
+
+系统里有一批字符串**同时背两份职责**：既给机器用，又给人看。典型是规则闸的未通过项，
+形如 `PORTFOLIO_CAP: 买了这一笔，总持仓会占到…`。句首那个大写代码是承重的：
+
+  app/services/executor.py 靠它取账本去重键（同一条被拒指令当日只记一条）
+  app/core/planner.py      靠它做原因聚类（命令进度那行注记）
+
+所以**代码不能从句首挪走，也不能删**——2026-07-29 那次账本被同一条指令按分钟灌满的
+事故，病根就是去重键失效。正确的做法是：码留在句首，到了显示层再剥掉。
+
+这个文件就是那个显示层。后端两处与前端各一份，全部走同一口径，不再各写各的。
+
+## 长远的根治
+
+更彻底的做法是让产出方直接返回 {"code": ..., "text": ...} 两个字段：机器只读 code，
+页面只读 text，代码名泄漏这件事在结构上就消失了。那是更大的改动，本文件是它之前的
+过渡口径，也是那之后前端仍然需要的兜底。
+"""
+from __future__ import annotations
+
+
+def strip_code(text) -> str:
+    """剥掉句首那个给运维看的大写代码前缀，只留中文说明。
+
+    判定条件写得严：冒号之前必须全部是大写字母、数字或下划线，且非空。这样
+    「PORTFOLIO_CAP: 总持仓…」会被剥掉前缀，而「预期空间: 算不出」这种正常带冒号的
+    中文句子不会被误伤。认不出的原样返回——宁可多显示几个字，也不要把人家的话截半句。
+    """
+    s = str(text if text is not None else "").strip()
+    head, sep, tail = s.partition(":")
+    if sep and head.strip() and all(
+            ch.isupper() or ch.isdigit() or ch == "_" for ch in head.strip()):
+        return tail.strip() or s
+    return s
+
+
+def join_reasons(items, sep: str = "；") -> str:
+    """把多条原因拼成一句给人读的话：逐条剥掉代码前缀，用中文分号连起来。
+
+    分隔符用中文分号而不是 " | "，因为这一栏是句子不是表格。空条目丢掉，
+    全空时返回空串，让调用方自己决定显示什么。
+    """
+    out = [strip_code(x) for x in (items or [])]
+    return sep.join(x for x in out if x)

app/core/planner.py

@@ -370,7 +370,7 @@ def plan_open_target(*, ts_code: str, target_pct: float, price: float, ctx: dict
 
     if scale <= 0:
         return {"ok": False, "items": [], "notes": [], "rejects": [
-            {"ts_code": ts_code, "reasons": ["SCALE_INVALID: 总规模未设置"]}]}
+            {"ts_code": ts_code, "reasons": ["SCALE_INVALID: 系统里还没填总规模，仓位上限算不出来，先去参数设置里填总规模"]}]}
     if add_amount <= 0:
         return {"ok": True, "items": [], "target_amount": 0.0, "planned_amount": 0.0,
                 "gap": 0.0, "notes": [f"{ts_code} 现有仓位已达/超过目标 {target_pct:.1%}, 无需建仓"],
@@ -613,8 +613,10 @@ def check_all_caps(*, ts_code: str, add_amount: float, ctx: dict) -> list:
         usable = scale * (1 - reserve)
         after = float(ctx.get("portfolio_mv") or 0) + float(add_amount)
         if after > usable + 1e-9:
-            v.append(f"CASH_RESERVE: 加后市值 {after:,.0f} > 可用额度 {usable:,.0f} "
+            # 因果别说反了：拦的不是「预留现金不够」，是「买完会动到留着不用的那笔钱」。
-                     f"(预留现金 {reserve:.0%})")
+            v.append(f"CASH_RESERVE: 买了这一笔，持仓市值到 {after:,.0f} 元，"
+                     f"超过可动用的 {usable:,.0f} 元"
+                     f"（总规模里留了 {reserve:.0%} 现金不动用）")
     return v
 
 

app/core/rule_gate.py

@@ -36,6 +36,28 @@ BUY, SELL = "buy", "sell"
 # 那是"拿不到意见"的正常降级, 不是漏拦。
 BAD_Y_SIGNALS = ("SELL", "AVOID", "DROPPED")
 
+# 冻结的三种来路对交易员是三件不同的事，原先一句话三种都不分, 还把枚举值印出来了。
+_FROZEN_CN = {"COMMAND_HALT": "你下过命令把这只票设成只卖不买",
+              "MANUAL": "你手动把这只票设成只卖不买",
+              "BRAKE": "组合刹车期间这只票只卖不买"}
+
+# 昨夜信号的枚举值。它们是决策系统与本系统之间传递用的, 不该出现在给人看的句子里。
+# 「已剔除」单独成句 —— 「给出已剔除信号」不成话。
+# 这份表与选股系统 akg-factor-bridge/card.py 的 SIGNAL_CN 同源: 两个系统各自独立部署,
+# 进程之间 import 不到, 只能各存一份同名同值的, 改一处必须两处同改。
+_SIGNAL_CN = {"SELL": "择时决策系统昨天收盘后给这只票打了卖出信号",
+              "AVOID": "择时决策系统昨天收盘后给这只票打了回避信号",
+              "DROPPED": "择时决策系统昨天收盘后把这只票从名单里去掉了"}
+
+
+def _frozen_cn(reason) -> str:
+    return _FROZEN_CN.get(str(reason or "").strip(), "这只票现在只卖不买")
+
+
+def _signal_cn(sig) -> str:
+    s = str(sig or "").strip().upper()
+    return _SIGNAL_CN.get(s, f"择时决策系统昨天收盘后给这只票打了不看好的信号（{s}）")
+
 
 def _num(v, default=0.0):
     try:
@@ -149,24 +171,24 @@ def _check_buy(failed, warns, qty, price, ctx, pos, day, prm, flg, is_cmd, hard,
                min_lot: int = LOT):
     """增持方向: 全部硬约束都要过。"""
     if qty % LOT != 0:
-        failed.append(f"LOT_INVALID: 买入 {qty} 股非整百")
+        failed.append(f"LOT_INVALID: 买 {qty} 股不是一百的整数倍，交易所不收")
     elif min_lot > LOT and 0 < qty < min_lot:
         failed.append(f"LOT_INVALID: 科创板买入申报最少 {min_lot} 股 (买 {qty} 股会被交易所拒单)")
     if (pos.get("frozen_reason") or "NONE") != "NONE":
-        failed.append(f"FROZEN: 该股处于 {pos['frozen_reason']}, 禁止增持")
+        failed.append(f"FROZEN: {_frozen_cn(pos.get('frozen_reason'))}，这笔买 {qty} 股不下单")
     if flg.get("buy_halt"):
-        failed.append("BUY_HALT: 全局暂停买入生效中")
+        failed.append("BUY_HALT: 现在全局暂停买入，所有票都不新增")
     if flg.get("blacklisted"):
-        failed.append("BLACKLIST: 该股在黑名单 (永不买入)")
+        failed.append("BLACKLIST: 这只票在黑名单里，任何情况都不买")
     # 决策系统昨夜定性为形态恶化 —— 无论有没有拿到定性, 都写进 hard_numbers, 这样
     # 评审账本上"放行的那些当时定性是什么"同样查得到, 而不是只有被拒的才留下痕迹。
     y_sig = str(flg.get("y_signal") or "").strip().upper()
     hard["y_signal"] = y_sig or None
     if y_sig in BAD_Y_SIGNALS:
-        failed.append(f"BAD_SIGNAL: 决策系统昨夜定性 {y_sig} (形态恶化), 不新增买入")
+        failed.append(f"BAD_SIGNAL: {_signal_cn(y_sig)}，今天不买")
     if flg.get("brake_active"):
         if is_cmd:
-            warns.append("BRAKE_ACTIVE: 组合刹车中 —— 命令驱动不受限, 仅提示")
+            warns.append("BRAKE_ACTIVE: 组合正在刹车期，但这是你下的命令，不受限制，这里只作提示")
         else:
             failed.append("BRAKE_ACTIVE: 组合刹车中, 自主增持暂停")
 

app/core/sizer.py

@@ -106,22 +106,27 @@ def check_caps(*, ts_code: str, add_amount: float, ctx: dict) -> list:
     v = []
     scale = float(ctx["scale"])
     if scale <= 0:
-        return ["SCALE_INVALID: 总规模未设置"]
+        return ["SCALE_INVALID: 系统里还没填总规模，仓位上限算不出来，先去参数设置里填总规模"]
 
     if (ctx["portfolio_mv"] + add_amount) / scale > ctx["portfolio_cap"] + 1e-9:
-        v.append(f"PORTFOLIO_CAP: 加后总仓 {(ctx['portfolio_mv'] + add_amount) / scale:.1%} "
+        v.append(f"PORTFOLIO_CAP: 买了这一笔，总持仓会占到总规模的 "
-                 f"> 上限 {ctx['portfolio_cap']:.0%}")
+                 f"{(ctx['portfolio_mv'] + add_amount) / scale:.1%}，"
+                 f"超过 {ctx['portfolio_cap']:.0%} 的总仓上限")
     if (ctx["stock_mv"] + add_amount) / scale > ctx["stock_cap"] + 1e-9:
-        v.append(f"STOCK_CAP: {ts_code} 加后 {(ctx['stock_mv'] + add_amount) / scale:.1%} "
+        # 句首不写股票代码：页面左边一列已经是股票名和代码，重复只是噪音。
-                 f"> 单股上限 {ctx['stock_cap']:.0%}")
+        v.append(f"STOCK_CAP: 买了这一笔，这只票会占到总规模的 "
+                 f"{(ctx['stock_mv'] + add_amount) / scale:.1%}，"
+                 f"超过单只 {ctx['stock_cap']:.0%} 的上限")
     if ctx.get("is_new_name") and ctx["names_count"] + 1 > ctx["max_names"]:
-        v.append(f"MAX_NAMES: 持仓数将达 {ctx['names_count'] + 1} > 上限 {ctx['max_names']}")
+        v.append(f"MAX_NAMES: 再开一只就是 {ctx['names_count'] + 1} 只持仓，"
+                 f"超过最多 {ctx['max_names']} 只的上限")
 
     sector = ctx.get("sector")
     if sector:  # None/"" = 行业数据源未配置, 约束停用 (调用方负责页面提示)
         if ctx.get("is_new_name") and ctx.get("sector_names", 0) + 1 > ctx["sector_max_names"]:
-            v.append(f"SECTOR_NAMES: 行业[{sector}]将达 {ctx['sector_names'] + 1} 只 "
+            v.append(f"SECTOR_NAMES: {sector}已经持有 {ctx.get('sector_names', 0)} 只，"
-                     f"> 上限 {ctx['sector_max_names']}")
+                     f"再开一只就是 {ctx['sector_names'] + 1} 只，"
+                     f"超过同一行业最多 {ctx['sector_max_names']} 只的上限")
         # 行业占比: **占组合超上限 且 绝对敞口够大**, 两条同时成立才拦 (2026-07-31 定)
         # ---------------------------------------------------------------------
         # 只看"占组合"的话, 这道闸在建仓期是**结构性不可满足**的: 空账本买第一只票,
@@ -144,8 +149,11 @@ def check_caps(*, ts_code: str, add_amount: float, ctx: dict) -> list:
         of_scale = sector_after / scale
         floor = ratio_max * float(ctx.get("portfolio_cap") or 1.0)
         if of_port > ratio_max + 1e-9 and of_scale > floor + 1e-9:
-            v.append(f"SECTOR_RATIO: 行业[{sector}]占组合将达 {of_port:.1%} > 上限 "
+            # 末尾那句"两条都超才拦"必须写出来：这是这道闸的设计要点，不说的话
-                     f"{ratio_max:.0%}, 且占规模 {of_scale:.1%} > 绝对线 {floor:.1%}")
+            # 会被当成误拦来投诉。
+            v.append(f"SECTOR_RATIO: 买了这一笔，{sector}会占到持仓市值的 {of_port:.1%}，"
+                     f"超过 {ratio_max:.0%} 的上限；同时占到总规模的 {of_scale:.1%}，"
+                     f"超过 {floor:.1%}。这两条都超了才拦")
     return v
 
 

app/services/plan_feed.py

@@ -501,6 +501,30 @@ def parse_query_extra(text) -> dict:
         return {}
 
 
+# 接口地址没填时的那句话。环境变量名退到括号里做精确指认, 不做句子主干 ——
+# 原先那句以变量名开头, 读的人不知道该找谁、去哪儿填。
+_NO_BASE_WHY = ("选股系统的接口地址没填，PMS 拿不到今天的股票名单。"
+                "请运维在「运维 → 参数设置」里补上地址（PMS_PLAN_API_BASE），填完点右上角刷新。")
+
+
+def _fetch_fail_why(e, timeout_sec) -> str:
+    """拉不到计划时给人看的那句话。按失败的类型分三种, 因为处置不同。
+
+    「过 1 分钟再试」这句不能省: 失败结果会被缓存住 FAIL_CACHE_SEC 秒 (默认 60),
+    早于这个点重读拿到的还是缓存里同一条失败, 人会以为重试没用。
+    """
+    name = type(e).__name__
+    wait = int(FAIL_CACHE_SEC // 60) or 1
+    if "Timeout" in name or "Connection" in name:
+        return (f"选股系统没有回应，等了 {timeout_sec} 秒。现在拿不到今天的候选票。"
+                f"过 {wait} 分钟，点右上角「选股计划」里的「只重读」再试一次；"
+                f"还是不行就找运维。")
+    code = getattr(getattr(e, "response", None), "status_code", None)
+    if code:
+        return f"选股系统返回了错误，状态码 {code}。详细原因已记进运行日志。"
+    return "选股系统返回的内容不是正常的计划数据。详细原因已记进运行日志。"
+
+
 def fetch(*, date=None, base=None, path=None, timeout=None, extra_params=None) -> dict:
     """拉一次并解析 (不走缓存、不校新鲜度)。任何失败抛 PlanFeedError。
 
@@ -514,7 +538,7 @@ def fetch(*, date=None, base=None, path=None, timeout=None, extra_params=None) -
         timeout = p["timeout"] if timeout is None else timeout
     base = (base or "").strip().rstrip("/")
     if not base:
-        raise PlanFeedError("上游计划接口未配置 (PMS_PLAN_API_BASE 为空)")
+        raise PlanFeedError(_NO_BASE_WHY)
     path = (path or "/plan").strip() or "/plan"
     if not path.startswith("/"):
         path = "/" + path
@@ -531,7 +555,11 @@ def fetch(*, date=None, base=None, path=None, timeout=None, extra_params=None) -
     except PlanFeedError:
         raise
     except Exception as e:
-        raise PlanFeedError(f"拉取上游计划失败 {url}: {type(e).__name__}: {e}") from e
+        # 这条会渲染成页面顶部整条红色横幅, 交易员一进页面就看到。所以句子里不许出现
+        # 接口地址、异常类名、异常原文 —— 那些只进日志。取而代之要说清三件事:
+        # 现在拿不到什么、多久之后重试有意义、还不行找谁。
+        logger.warning("拉取上游计划失败 %s: %s: %s", url, type(e).__name__, e)
+        raise PlanFeedError(_fetch_fail_why(e, to)) from e
     plan = parse_plan(payload, requested=q)
     plan["url"] = url
     plan["fetched_at"] = time.time()

app/services/portfolio.py

@@ -48,19 +48,23 @@ def cash_view() -> dict:
     try:
         snap = qmt_repo.latest_snapshot("funds")
     except Exception as e:
-        out["why"] = f"读 ws 资金快照失败: {type(e).__name__}: {e}"
+        # 异常细节只进日志。这个字段有三个下游都直接显示给人看
+        # （建仓跳过原因、日报、待办清单），不能把异常原文送到人眼前。
+        logger.warning("读交易通道的资金快照失败: %s: %s", type(e).__name__, e)
+        out["why"] = "查询账户资金时程序出错，详细原因已记进运行日志"
         return out
     if not snap:
-        out["why"] = "ws 从未回过 funds 快照 (对端未实现 query_funds, 或 pms-ws 没在跑)"
+        out["why"] = "交易通道从没送回过资金数据，通道可能没启动"
         return out
     age = snap.get("age_sec")
     if age is not None and age > max_age:
-        out.update({"why": f"ws 资金快照已过期 ({age:.0f}s > {max_age}s)", "age_sec": age})
+        out.update({"why": f"资金数据停在 {age / 60:.0f} 分钟前，"
+                           f"超过了 {max_age / 60:.0f} 分钟的有效期", "age_sec": age})
         return out
     f = wsc.parse_funds_snapshot(snap["payload"])
     avail = f.get("available_cash")
     if avail is None:
-        out["why"] = "ws 资金快照里认不出可用资金字段 (补进 ws_codec.parse_funds_snapshot)"
+        out["why"] = "拿到了资金数据，但里面找不到可用余额这一项"
         return out
     out.update({"source": CASH_WS, "age_sec": age, "total_asset": f.get("total_asset"),
                 "cash_avail": round(float(avail) + float(f.get("sell_return_today") or 0), 2)})

app/services/proposal_service.py

@@ -33,6 +33,7 @@ import time
 from datetime import datetime, timedelta
 
 from app.core import action_engine as ae
+from app.core import copy
 from app.core import command_spec as cs
 from app.core import rule_gate
 from app.core import signal_rules as sr
@@ -248,8 +249,9 @@ def _scan_open(view, params, stock_params, skip, mkt, out) -> list:
         room = min(room, float(t["cash_avail"]))
     elif param_store.get_bool("PMS_OPEN_REQUIRE_WS_CASH", True):
         out["skipped"].append({"action": ae.A_OPEN,
-                               "why": f"拿不到 ws 资金快照 ({t.get('cash_why') or '原因未知'}), "
+                               "why": f"读不到券商账户的可用资金，这一轮不开新仓；"
-                                      f"本轮不自动新建仓 (PMS_OPEN_REQUIRE_WS_CASH=True)"})
+                                      f"已持有的票照常处理。"
+                                      f"原因：{t.get('cash_why') or '原因未知'}"})
         return []
 
     res = ae.scan_open(candidates=cands, params=params, caps=portfolio.caps_ctx(view),
@@ -346,12 +348,12 @@ def _inflight_open_commitments(view) -> dict:
 
 
 def _plain_check(c) -> str:
-    """规则闸的 failed 串形如 "T1_UNAVAILABLE: 卖出…" —— 去掉给运维看的大写代码前缀, 只留中文说明。"""
+    """规则闸的 failed 串形如 "T1_UNAVAILABLE: 卖出…" —— 去掉给运维看的大写代码前缀, 只留中文说明。
-    s = str(c)
+
-    head, sep, tail = s.partition(":")
+    实现搬到 app/core/copy.py 了: 同一套口径后端两处 + 前端一份共用, 不再各写各的。
-    if sep and head.strip() and all(ch.isupper() or ch.isdigit() or ch == "_" for ch in head.strip()):
+    这里留一层薄壳, 是因为本文件里已经有多处按名字调它。
-        return tail.strip()
+    """
-    return s.strip()
+    return copy.strip_code(c)
 
 
 def _today_open_reject_reasons(now):