调度器 (--profile sched) 会自动按调度总表执行; 此处按钮用于未接调度或临时验证时手动触发。
@@ -1887,6 +2037,9 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
计划不带价格与金额 —— 买多少/什么价一律 PMS 自己算; 行情取不到价的票会被剔除。
+
+ 正在读上游选股计划…… 上游装配这份计划要读论断与催化事件，慢的时候要几十秒。
+ 这段时间抽屉是空的，属正常；真读不到会在这里显示原因。
@@ -3027,6 +3180,95 @@ createApp({
out.sort((a, b) => (b.held - a.held) || String(b.time).localeCompare(String(a.time)));
return out;
});
+ // ══════════════ 单票信号抽屉 (2026-09-09, 台账 055) ══════════════
+ // 信号栏一行 = 上游流里的一条原始消息, 一只活跃票一天十来条, 散在好几个分组里。
+ // 点一下票名, 把它今天的全部信号按时间倒序摊开。
+ //
+ // 两段取数, 顺序不能反:
+ // 第一段 —— 用手上已有的快照就地渲染, 所以**第一帧就有内容**, 永远不会出现
+ // "弹了个遮罩但什么都没有"。
+ // 第二段 —— 同时发按代码回扫的请求, 回来后整份替换成完整版。
+ // 快照口径注定不全 (告警每类只留最新 20 条), 到下午一只活跃票早盘的信号早被挤掉了,
+ // 所以第一段只是垫场, 表头会写明它是"可能不全"的。
+ // 补全失败就保留快照并挂一条黄条, 不清空 —— 有一半总比没有强。
+ const csOpen = ref(false), csCode = ref(''), csLoading = ref(false),
+ csErr = ref(''), csData = ref(null), csFromSnap = ref(false);
+ function _csSnapRows(code) {
+ const k = code6(code), out = [];
+ const push = (src, label, r, extra) => out.push(Object.assign(
+ { src, src_label: label, time: r.time, ts: 0, dup: 1, ts_code: r.ts_code }, extra || {}));
+ for (const r of (SIG.value.decision || []))
+ if (code6(r.ts_code) === k) push('decision', '决策系统广播', r,
+ { cat_label: r.action === 'SELL' ? '建议卖出' : '建议买入', direction: r.action === 'SELL' ? '看跌' : '看涨' });
+ for (const r of (SIG.value.timing || []))
+ if (code6(r.ts_code) === k) push('timing', '盘中择时层入场', r, { cat_label: '择时层入场', price: r.price, confidence: r.confidence });
+ for (const r of (SIG.value.sell || []))
+ if (code6(r.ts_code) === k && r.today !== false) push('sell', '风控卖出动作', r,
+ { cat_label: '风控卖出', direction: '看跌', reason: r.reason, confidence: r.confidence });
+ for (const r of (alertItems.value || []))
+ if (code6(r.ts_code) === k) push('alert', '盘中告警', r,
+ { cat: r.cat, cat_label: r.cat, direction: r.direction, level: r.level, value: r.value, stale: r.stale, age_min: r.age_min });
+ out.sort((a, b) => String(b.time || '').localeCompare(String(a.time || '')));
+ return out;
+ }
+ function openCodeSignals(code) {
+ if (!code) return;
+ csCode.value = String(code);
+ csErr.value = '';
+ csFromSnap.value = true;
+ csData.value = { ts_code: String(code), timeline: _csSnapRows(code), sources: {},
+ snapshot: { metrics: (SIG.value.metrics || []).filter(r => code6(r.ts_code) === code6(code)),
+ mr: { watch: ((SIG.value.mr || {}).watch || []).find(r => code6(r.ts_code) === code6(code)) || null,
+ avoid: ((SIG.value.mr || {}).avoid || []).find(r => code6(r.ts_code) === code6(code)) || null } } };
+ // 浮层写死 2010、遮罩 2005, 抽屉 3000 盖得住; 但窄屏上浮层铺满整块, 底下再压一层
+ // 抽屉读起来很乱, 所以先把浮层收掉。双保险的另一半是那条写死的层级。
+ if (railOpen.value) closeRail();
+ csOpen.value = true;
+ csLoadFull();
+ csLoadLedger();
+ }
+ function closeCodeSignals() { csOpen.value = false; }
+ async function csLoadFull() {
+ const code = csCode.value;
+ csLoading.value = true; csErr.value = '';
+ try {
+ const d = await call('get', '/api/upstream/signals/' + encodeURIComponent(code));
+ const body = (d && d.data) ? d.data : d;
+ if (csCode.value !== code) return; // 期间换了票, 丢弃这份回包
+ if (body && body.ok && Array.isArray(body.timeline)) {
+ csData.value = Object.assign({}, csData.value, body);
+ csFromSnap.value = false;
+ } else {
+ csErr.value = (body && body.error) || (d && d.error) || '补全失败';
+ }
+ } catch (e) { csErr.value = e.message || String(e); }
+ finally { if (csCode.value === code) csLoading.value = false; }
+ }
+ // 账本按代码查的接口早就支持, 页面从没用过 —— 后端零改动。
+ const csLedger = ref([]), csLedgerErr = ref('');
+ async function csLoadLedger() {
+ const code = csCode.value;
+ csLedger.value = []; csLedgerErr.value = '';
+ const d = await call('get', '/api/ledger?ts_code=' + encodeURIComponent(code) + '&limit=40');
+ if (csCode.value !== code) return;
+ if (d && d.ok === false) { csLedgerErr.value = d.error || '读不到账本'; return; }
+ const ymd = new Date().toISOString().slice(0, 10);
+ csLedger.value = (d.data || []).filter(r => String(r.created_at || r.decided_at || '').slice(0, 10) === ymd);
+ }
+ const csTimeline = computed(() => (csData.value && csData.value.timeline) || []);
+ const csSrcFails = computed(() => {
+ const src = (csData.value && csData.value.sources) || {};
+ return Object.keys(src).filter(k => src[k] && src[k].ok === false);
+ });
+ const csTruncated = computed(() => {
+ const t = (csData.value && csData.value.truncated) || {};
+ return Object.keys(t).filter(k => t[k]);
+ });
+ const csSrcLabel = { intraday: '决策与择时层', alerts: '盘中告警', sell_actions: '风控卖出',
+ metrics: '资金异动', mr: '实况评分' };
+ function csHeld(code) { return positions.value.some(p => code6(p.ts_code) === code6(code)); }
+ function csInPlan(code) { return planCodeSet.value.has(code6(code)); }
+
// 「今天的信号」列与持仓角标都只数今天的 (2026-08-28): 昨天的风控卖出不再冒充今天
const sigBriefHeldCount = computed(() => sigBrief.value.filter(x => x.held && x.today !== false).length);
const heldSigMap = computed(() => {
@@ -3112,10 +3354,20 @@ createApp({
});
function macroChipClick() { if (mzAdvicePending.value && mzSig.value) macroAdopt(mzSig.value.key); }
+ // 上游信号六路: 五个 Redis 源 + 实况分。**整体失败时不许抹平成空数据** ——
+ // 原来失败时 sources 被换成 {}, 模板里的"读取失败"角标一个都渲染不出来,
+ // 页面于是显示"今天没有信号"。风控面板上"没有信号"和"读不到信号"是两回事。
+ const UP_SRC_KEYS = ['intraday', 'alerts', 'metrics', 'sell_actions', 'mr'];
async function loadUpstream() {
const d = await call('get', '/api/upstream-signals');
- upstream.value = (d && d.sources) ? d : { sources: {} };
+ if (d && d.sources) { upstream.value = d; return; }
+ const why = (d && d.error) || '接口没有返回数据';
+ const src = {};
+ for (const k of UP_SRC_KEYS) src[k] = { ok: false, error: why };
+ upstream.value = { sources: src, at: '', fail: why };
}
+ // 某一路读不到时返回它的错误对象, 否则 null。模板用它把"今天没有"换成"读取失败"。
+ function srcFail(k) { const a = usrc(k); return (a && a.ok === false) ? a : null; }
async function loadPlans(cid) {
plansOf.value = cid;
const d = await call('get', '/api/plans?command_id=' + cid); plans.value = d.data || [];
@@ -3665,7 +3917,10 @@ createApp({
commandsShown, strategiesShown, instrLive, showAllCommands, showAllStrategies, showAllInstr,
cmdArchivable, stratArchivable, archiveRow,
showOpLog, showSettings, watchCandidates, marketOpen, marketState, lastRefresh, autoRefresh,
- upstream, showUpstream, usrc, loadUpstream,
+ upstream, showUpstream, usrc, loadUpstream, srcFail,
+ csOpen, csCode, csLoading, csErr, csData, csFromSnap, csTimeline, csSrcFails,
+ csTruncated, csSrcLabel, csLedger, csLedgerErr, csHeld, csInPlan,
+ openCodeSignals, closeCodeSignals, csLoadFull,
macro, macroGateOn, macroUnavailable, mzLabel, mzColor, mzBarStyle,
loadMacro, macroScan, macroAdopt,
mzSig, mzAdvicePending, mzChipCls, mzChipText, macroChipClick,
diff --git a/docs/盘中改判与信号聚合方案_2026-09-09.md b/docs/盘中改判与信号聚合方案_2026-09-09.md
new file mode 100644
index 0000000..4860f8a
--- /dev/null
+++ b/docs/盘中改判与信号聚合方案_2026-09-09.md
@@ -0,0 +1,159 @@
+# 盘中改判与信号聚合方案（2026-09-09，已批准）
+
+> 两件事一份方案。批准后落盘为 PMS 与择时决策系统两个仓库的 docs/盘中改判与信号聚合方案_2026-09-09.md，台账记 055。
+> 09-09 三条拍板已吸收：盘中改判做到**解锁重问加主动扫描**两层，不改追高原则；重问后仍驳回**静默记账**，页面不单列；择时侧盘中资金流分发**一并打开**。
+
+---
+
+## 一、上下文：为什么做
+
+你提了两件事。第一件是信号栏里同一只股票反复出现，想点开看这只票今天的全部信号。第二件是四方股份的"决策系统判暂不建仓"看起来是一次性的，而这只票今天资金流多次转强、涨幅超过六个点，系统该不该在盘中改判。
+
+我把 09-09 这只票的全过程查了一遍：
+
+| 时间 | 发生了什么 | 谁做的 |
+|---|---|---|
+| 10:04 | 资金流强度告警触发盘中审计，判**反转买入**（主力净占比由 -12.5% 翻正到 +13.41%），广播买入信号置信 95% | 择时决策系统 |
+| 10:05 | 消化这条信号，写留痕 | PMS |
+| 10:12 | 送建仓研判，现价 43.15 → **驳回**（把握度 72，理由是量能未确认、提纯量不足） | 择时决策系统 |
+| 10:12 起 | 当日死锁，扫描每分钟仍在跑但直接跳过这只票 | PMS |
+| 13:02 | 资金流告警再次触发，现价 42.47 → **维持**（价格虽破 42.0，但量能闸 1.5 没过） | 择时决策系统 |
+
+**结论分两半。** 择时侧盘中复查了两次，昨夜布下的关注条件也在工作，所以不是完全没复查；但 PMS 的建仓侧确实是一次性的，驳回后当天不再问，13:02 那次复查的结论进不了建仓这条路。
+
+**而且这次系统没判错。** 从驳回到下午，这只票从 43.15 跌到 42.44，跌了 1.65%；今天被驳回的 11 只票里 5 只在跌、4 只基本没动、只有 1 只涨超四个点。前一天（09-08）被驳回的那批更是 11 只里 10 只下跌。研判连续三天百分之百驳回，在这个弱市里是合理的，不是过严。
+
+所以要修的是机制，不是结论：
+
+1. **研判驳回后的当日死锁没有解除条件。** 这道闸的注释写明了它的假设——"这只票今天不该从零建仓这个结论当天基本不会翻转"，以及防账本刷屏（2026-07-29 那次每分钟写一行、一天几千行的教训）。假设合理但不是绝对，缺一个"确实有新信息"的例外。
+2. **盘中转多信号只改候选排序、不改建仓资格。** 今天三只票被判转多，一只被死锁挡住、一只候选卡判决是仅展示、一只不在候选池，一个出口都没有。
+3. **择时侧没有任何任务主动扫"今天谁在涨、谁在进钱"。** 四个盘中定时任务分别只看持仓、已有意向单的票、昨夜布过防的票、门控白名单。强势票要被看见，必须上游先推告警。
+
+---
+
+## 二、一个必须先讲清的事实：改完之后不会立刻追高买入
+
+这决定了整个方案的取向。我把两只票的择时买入区间算了出来：
+
+| 股票 | 昨夜支撑 / 压力 | 买入区间 | 现价 | 择时会怎么说 |
+|---|---|---|---|---|
+| 四方股份 | 40.00 / 42.00 | 39.60 到 40.60 | 42.60 | 高于上沿，等待不追 |
+| 长源电力 | 24.50 / 27.42 | 24.25 到 25.38 | 26.18 | 高于上沿，等待不追 |
+
+也就是说，即使解锁重问、研判放行、你在页面上点了头，落成指令之后择时仍会说"现价高于买入区间上沿，不追"，指令会在窗口期里挂着等价格回落到区间内才真的下单。系统本身也是只提议不自动执行的档位。
+
+**所以解锁重问买到的不是"立刻追高"，而是"让这只票进入待办、等回调时自动买入"。** 这与系统写死的那条原则（可以接受买不上）不冲突，也正是你要的"重新评估"。追高那条线本方案不动。
+
+顺带澄清一个容易误会的地方：不追高那道闸算的是**较今日开盘**的涨幅，不是较昨收。四方股份今天较昨收涨 6.2%，但较开盘是 -1.5%，所以它其实没有被这道闸拦过，是研判驳的。
+
+---
+
+## 三、第一件：点股票看它今天的全部信号
+
+### 3.1 现状与量级
+
+信号栏是左侧浮层里的扁平行列表，一行等于上游流里的一条原始消息，不按股票聚合。数据来自 Redis 五个源：盘中告警、决策系统买卖广播、择时层入场、风控卖出动作、资金异动，外加一份实况评分榜。今天告警流 157 条，四方股份一只就占 11 条，正是你说的反复出现。
+
+页面已经有一个按代码分组的实现（持仓表那列"今天的信号"），但被限死在持仓票、只看今天、只能悬停看前三条，而且明确不可点。
+
+### 3.2 做法：抽屉，两段取数
+
+**后端加一个按代码回扫的接口。** 快照口径注定不全（告警每类只留最新 20 条），到下午一只活跃票早盘的信号早被挤掉了。新增 `upstream_signals.by_code(ts_code)` 与 `GET /api/upstream/signals/{ts_code}`：分页回扫（每页 500 条，上限可配，默认 5000），两条按日期分键的流整键就是今天，两条不分键的流（风控卖出、资金异动）扫到今天零点为止；同源同分钟同值的合并成一条并显示重复次数，跨源永不合并；按票缓存八秒防连点；单源失败不拖垮整体，返回里逐源标明哪路读不到。只用倒序读取，不建消费组、不写任何东西，且**只能点击触发，绝不能进轮询**。
+
+**前端抽屉两段取数。** 点开的瞬间用手上的快照渲染，所以第一帧就有内容，永远不会变成"只有遮罩没有信息"；同时发按代码回扫的请求，回来后替换成完整版，表头的口径说明从"最近快照，可能不全"翻成"完整，已回扫到今天开盘"。补全失败就保留快照并在抽屉里挂一条黄条，带重试按钮。
+
+抽屉里分七段：补全失败提示、回扫未见底提示、分源失败角标、时间线、当前快照（资金异动与实况评分这两路上游只给当前值不带时间，单独成段，不编造时间戳）、这只票今天的账本记录、持仓与候选池处置。账本那条接口早就支持按代码查，只是页面从没用过，后端零改动。
+
+**三处入口可点**：信号栏每行的股票名、持仓表那列（现在不可点，连"今天没有"也要可点，因为速览口径下的"没有"可能是假的）、消息栏的股票名。风控卖出那一行的股票名要用阻止冒泡，否则点名字会顺手把理由折叠条也翻一下。
+
+### 3.3 顺带修遮罩空白
+
+你看到的"弹出遮罩层但没有任何信息"，成因是四条，一并修：
+
+1. **浮层打开时不等数据、不看空、失败不兜底。** 改成：面板先开（关闭随时可用），再发请求，加载态显示在面板里，失败显示原因加重试按钮，面板不自动关（自动关看起来像"点了没反应"，比空面板更糟）。
+2. **接口整体失败时把错误抹平成空数据**，导致模板里那些"读取失败"角标一个都显示不出来，页面看起来就是"今天什么信号都没有"——在风控面板上这是最危险的一种谎。改成保留错误并给每一路造一条失败标记，模板一个字不用改，角标立刻就亮。另外有三处配套：告警派生的四个分组本来就没有失败角标要补上；各组的空文案要让位给失败态；那个失败角标的样式选择器写漏了信号栏，即使显示出来也看不出是错误。
+3. **错误横幅被遮罩压住。** 顶部那条横幅在文档流里，遮罩、浮层、弹层的层级全都压得住它，而且往下滚一屏就看不见。改成在遮罩之上再显示一份浮动的。
+4. **新抽屉必须显式抬层级。** 弹层库的层级从 2000 起自增，而信号栏浮层写死 2010、遮罩 2005，新抽屉不抬层就会落在它们底下，屏幕上正好只剩一层灰——这是同一个症状的另一张脸。双保险：打开抽屉前先收浮层，抽屉本身写死更高的层级。选股计划抽屉也是同一种病（先开抽屉再等数据，上游不通时空白挂满一分钟），一并加加载提示。
+
+---
+
+## 四、第二件：盘中改判
+
+### 4.1 解锁重问（PMS 侧）
+
+**插入点**是那个返回"今天已被研判驳回"键集合的函数。它现在无条件返回被驳回的票，改成在返回前逐只判一次解锁条件，满足的从集合里剔掉——没有解锁条件时行为逐字相同。
+
+**解锁条件一次定死**，四条满足任一即解锁（写台账，不按读数回调）：
+
+| 条件 | 判据 | 为什么 |
+|---|---|---|
+| 择时判转多 | 当日账本里这只票有转多留痕，且时间**晚于**驳回时刻 | 驳回之前就有的信号研判已经看过了，不算新信息 |
+| 价格显著上行 | 现价较驳回时涨幅不低于 2%，且当前有看涨方向的资金流告警 | 单看价格容易被噪音触发，要资金面同时确认 |
+| 关注条件升级 | 择时侧审计表当日出现反转买入或升级判定，且晚于驳回时刻 | 昨夜布下的预案在盘中被兑现，是最硬的新信息 |
+| 主动扫描命中 | 第二件里新增的强势扫描判定这只票转强，且晚于驳回时刻 | 与上一条同源，走新入口 |
+
+**限次：每票每天最多解一次。** 判据是当日账本里这只票的新建仓驳回行数——第一次驳回一行，重问后再驳回就是两行，两行及以上不再解锁。不用新表、不用新参数。这直接回应了原设计防账本刷屏的顾虑：最坏情况每票每天多写一行。
+
+**试算过今天的实际量级**：11 只被驳回的票里只有 1 只会被重问（长源电力，较驳回涨 4.05%，三条资金流看涨告警）。四方股份虽然有五条看涨告警，但价格较驳回时跌了 1.5%，不会被重问——这是对的，它的情况没有变好。
+
+**重问要绕过择时侧的三十分钟缓存。** 那层缓存的键里没有任何输入指纹，价格资金全变了三十分钟内仍回同一份裁决。做法是研判请求里带一个强制标记，择时侧读缓存前先看这个标记。
+
+**重问仍驳回就静默记账**（你的拍板）：账本记一行并标明是重问后的结论，页面不单列。
+
+**开关** `PMS_JUDGE_RECHECK_ENABLED`，默认开；解锁条件的阈值各有参数，都要同时登记进运行参数白名单，否则页面上改不了。
+
+### 4.2 主动扫描（择时侧）
+
+新增一个每五分钟的盘中定时任务，扫两件事：当日涨幅与资金流转强。命中的走已有的盘中审计入口，终点还是那条已经通了的管道（判反转买入 → 广播信号 → 触发重算 → 翻案上游拒单）。
+
+**扫哪些票。** 择时侧读不到 PMS 的候选池（不同库），取它自己的：昨夜有结论且定性不是卖出、回避、剔除三类的票。昨天这批是 506 只，去掉卖出类还剩四百多只，每轮读四百多个实时快照，Redis 上是秒级，可接受。这个池天然覆盖了 PMS 的候选（候选来自同一批昨夜结论）。
+
+**当日涨幅要自己算**（现在系统里没有）：现价从已有的实时快照函数取，昨收从日线取最近一根。
+
+**阈值一次定死**：当日涨幅不低于 3%、当日量比不低于 1.5、资金流方向为看涨，三条同时满足才派单。节流用现有的锁族，同一只票两小时内不重复上大脑；每轮派单数量另设上限（初值 10），防止普涨日一次性打出几十次模型调用。
+
+**扫描结果分两种落法**：达到阈值的派单上研判；只达到涨幅一条、资金没跟上的只写一行日志，不占模型额度。
+
+### 4.3 资金流分发打开（择时侧）
+
+把盘中资金异动的分发模式从"只写时间线"改成全量，让资金流转强能直接触发碰撞与研判。这是原设计里标注的灰度状态，注释写着"观察一天量级正常后改 full"。
+
+**风险要正视**：资金异动流是上游每十五分钟推一整批全市场，今天累计已有十万条。打开后要盯三个读数：每天触发研判的次数、模型调用量、有没有把别的链路挤慢。回退就是把开关改回去、重启一个容器。建议与前两件分开一天上线，出问题分得清是谁。
+
+### 4.4 顺带修三处小问题
+
+- **驳回行没记两个期限的头。** 提议行记了，驳回行只记了把握度。补上，顺便把基线快照（驳回时的当日涨幅、量比、资金流方向）一起记进去，解锁判断要用。
+- **低把握驳回转交人的阈值参数没登记进运行参数白名单**（09-08 就发现了），页面上改不了、写不进去，代码走默认值 60。补登记。
+- **不追高那道闸的口径名不副实**：变量名与错误信息都说"当日涨幅"，实际算的是较今日开盘的涨幅。高开八个点然后横盘的票会被放行，低开三个点涨到四个点的票反而被拦。这次**只记台账不改**——改它会影响所有买入，要单独评估。
+
+---
+
+## 五、判收
+
+**第一件**：抽屉从三处入口打开都拿到同一份内容；补全失败时抽屉里有内容加黄条，不是空白；拦掉上游接口后每一路都显示"读取失败"、没有任何一处说"今天没有"；窄屏从浮层里点股票名，抽屉盖在最上面而不是只剩一层灰。
+
+**第二件**：连续三个交易日观察——每天解锁重问几次（预期个位数）、重问后放行几次、放行的票最终有没有成交；主动扫描每天命中几只、其中几只本来就在候选池里；资金流打开后每天触发研判的次数与模型用量。两周后拿读数决定要不要调阈值。
+
+---
+
+## 六、风险与回退
+
+- **解锁重问导致账本刷屏**：每票每天限一次，最坏情况多写一行。这是原设计最担心的事，限次是硬约束。
+- **重问放行后买不进**：这是预期行为（现价高于买入区间上沿），指令挂在窗口期里等回调。要在页面上说清楚，不然人会以为系统没执行。
+- **主动扫描增加模型调用**：只扫候选池、三条阈值同时满足、两小时节流，预计每天个位数。
+- **资金流分发打开量级失控**：单独一天上线，盯三个读数，回退是改开关加重启。
+- **前端没有测试框架**：新增静态断言把三处入口与层级设置钉死，另有六条回归手工步骤。
+
+每件事各有开关，全部可以单独关掉回到今天的行为。
+
+---
+
+## 附录 · 实现要点（可不读）
+
+**第一件（PMS）**：`app/services/upstream_signals.py` 加 `by_code` 与分页回扫辅助函数；`app/web/main.py` 加一个路由；`app/services/param_store.py` 登记回扫上限；`app/web/static/index.html` 加抽屉、三处入口、遮罩四处修复与一条链接样式；新增 `scripts/test_batch24_units.py`（代码归一、分页边界、按今天截断、去重口径、单源失败、缓存、只读守卫、页面静态断言），在 `scripts/run_tests.py` 的清单与例数表登记，路由清单也要加。
+
+**第二件（PMS）**：`app/services/proposal_service.py` 的驳回键函数加解锁判断、驳回落账补基线快照与两个头；`app/services/judge.py` 的请求体加强制标记；`app/services/param_store.py` 登记开关与阈值；新增单测批次。
+
+**第二件（择时决策系统）**：`workers/` 下新增强势扫描任务并在调度表登记；当日涨幅自算复用现有的实时快照函数与日线取数；`workers/tasks_intraday.py` 的缓存读取处认强制标记；`config/settings.py` 改分发模式与新增阈值；新增 `scripts/selftest_*.py`。
+
+**验证命令**：PMS 在 155 跑 `make test` 与页面手工步骤；择时在 188 跑自检脚本与看日志读数。容器重启与镜像重建各自需要你点头，PMS 侧仍先跑测试、避开开盘。
diff --git a/scripts/run_tests.py b/scripts/run_tests.py
index 2071dec..a3636d2 100644
--- a/scripts/run_tests.py
+++ b/scripts/run_tests.py
@@ -56,6 +56,9 @@
强制入队对卖出同样有效/研判不可用的减持入队/研究证据走弱
来源必定交人裁决/风控高置信卖出与命令清仓不经提议分流/
到价提议挂着时止损照落并作废该提议 (17 例)
+ test_batch24_units.py 点股票看它今天全部信号 (2026-09-09 方案第三节): 代码归一与跨市场不误配/
+ 分页回扫边界与按今天零点截断/五源解析与超龄标陈旧/去重同源合并跨源不合并/
+ 单源失败逐源标明与按票缓存/只读守卫/页面三处入口与层级 (9 例)
test_batch23_units.py 公司质地与建议方案接入 (2026-09-09 接入方案): 解析层收公司深度两键/整句送研判可关/建议
方案矩阵与研判补档/持仓页公司块。
test_batch22_units.py 逻辑状态四态接入 (2026-09-07 第三件): 解析层收逻辑状态/硬数字带它而研判
@@ -113,6 +116,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py",
"test_batch16_units.py", "test_batch17_units.py",
"test_batch18_units.py", "test_batch19_units.py", "test_batch20_units.py",
"test_batch21_units.py", "test_batch22_units.py", "test_batch23_units.py",
+ "test_batch24_units.py",
"test_page_enum_guard.py",
"test_wiring.py"]
diff --git a/scripts/test_batch24_units.py b/scripts/test_batch24_units.py
new file mode 100644
index 0000000..f70f0f5
--- /dev/null
+++ b/scripts/test_batch24_units.py
@@ -0,0 +1,281 @@
+# -*- coding: utf-8 -*-
+"""点股票看它今天全部信号 (2026-09-09 方案第三节, 台账 055)。全部离线, 不连 Redis 不连库。
+
+ 1. 代码归一与匹配: 四种写法归一、跨市场同六位不误配、认不出的回错误不抛。
+ 2. 分页回扫边界: 手工减一的流 id 边界、按今天零点截断、上限截断标记。
+ 3. 五源解析: 决策与择时层分流、告警超龄标陈旧、风控卖出载荷在 data 里、资金异动时间取流 id。
+ 4. 去重口径: 同源同分钟同值合并并计数、跨源永不合并。
+ 5. 韧性: 单源失败不拖垮整体且逐源标明、按票缓存、实况分不编时间戳。
+ 6. 只读守卫与页面静态断言。
+"""
+import io
+import os
+import re
+import sys
+import time
+import traceback
+
+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 upstream_signals as us # noqa: E402
+
+RESULTS = []
+
+
+def case(name):
+ def deco(fn):
+ RESULTS.append((name, fn))
+ return fn
+ return deco
+
+
+class _Patch:
+ def __init__(self): self._saved = []
+ def __enter__(self): return self
+ def __call__(self, obj, name, value):
+ self._saved.append((obj, name, getattr(obj, name))); setattr(obj, name, value)
+ def __exit__(self, *a):
+ for obj, name, v in reversed(self._saved):
+ setattr(obj, name, v)
+
+
+def _today_ms(h, m):
+ t = time.localtime()
+ return int(time.mktime((t.tm_year, t.tm_mon, t.tm_mday, h, m, 0, 0, 0, -1)) * 1000)
+
+
+class _FakeRedis:
+ """只实现 xrevrange 的假客户端。streams: {key: [(id, fields), ...]} 按 id 升序存, 读时倒序。"""
+
+ def __init__(self, streams=None, zsets=None, boom=None):
+ self.streams = streams or {}
+ self.zsets = zsets or {}
+ self.boom = boom or set()
+ self.calls = []
+
+ def xrevrange(self, key, max="+", min="-", count=None):
+ if key in self.boom:
+ raise RuntimeError("stream down: %s" % key)
+ self.calls.append((key, max, count))
+ rows = sorted(self.streams.get(key, []), key=lambda r: _idk(r[0]), reverse=True)
+ if max != "+":
+ rows = [r for r in rows if _idk(r[0]) <= _idk(max)]
+ return rows[:count] if count else rows
+
+ def zrevrange(self, key, s, e, withscores=False):
+ return self.zsets.get(key, [])
+
+
+def _idk(sid):
+ a, b = str(sid).split("-")
+ return (int(a), int(b))
+
+
+@case("代码归一: 四种写法同归、跨市场同六位不误配、认不出回错误")
+def test_code_norm():
+ assert us._norm_code("SH601126") == "601126.SH"
+ assert us._norm_code("601126.SH") == "601126.SH"
+ assert us._norm_code("sh601126") == "601126.SH"
+ assert us._norm_code("601126") == "601126", us._norm_code("601126")
+ assert us._norm_code("洋垃圾") == "" and us._norm_code(None) == ""
+ assert us._same_code("SH601126", "601126.SH") and us._same_code("601126", "601126.SZ")
+ assert not us._same_code("601126.SH", "601126.SZ"), "带后缀且不同市场必须判不同"
+ assert not us._same_code("601126.SH", "000001.SZ") and not us._same_code("", "601126.SH")
+ out = us.by_code("洋垃圾")
+ assert out["ok"] is False and out["timeline"] == [], out
+
+
+@case("分页回扫边界: 手工减一、翻页不重不漏、上限截断标记")
+def test_scan_paging():
+ rows = [("%d-0" % (1000 + i), {"i": i}) for i in range(12)]
+ c = _FakeRedis({"k": rows})
+ got, trunc = us._scan_stream(c, "k", cap=100, page=5)
+ assert len(got) == 12 and not trunc, (len(got), trunc)
+ assert [f["i"] for _s, f in got] == list(range(11, -1, -1)), "必须倒序且不重不漏"
+ assert len(c.calls) == 3 and c.calls[1][1] == "1006-18446744073709551615", c.calls
+ got2, trunc2 = us._scan_stream(c, "k", cap=7, page=5)
+ assert len(got2) == 7 and trunc2 is True, (len(got2), trunc2)
+ assert us._prev_id("100-3") == "100-2" and us._prev_id("100-0") == "99-18446744073709551615"
+ assert us._prev_id("0-0") == "-" and us._prev_id("garbage") == "-"
+ assert us._id_ms("1757000000000-5") == 1757000000000 and us._id_ms("x") == 0
+ got3, trunc3 = us._scan_stream(c, "缺这条流", cap=100)
+ assert got3 == [] and not trunc3
+
+
+@case("不分键的流扫到今天零点为止, 且不算截断")
+def test_stop_before_today():
+ y = us._today_start_ms() - 3600_000
+ rows = [("%d-0" % y, {"a": "昨天"}), ("%d-0" % _today_ms(9, 40), {"a": "今天"})]
+ c = _FakeRedis({"k": rows})
+ got, trunc = us._scan_stream(c, "k", cap=100, page=50, stop_before_ms=us._today_start_ms())
+ assert len(got) == 1 and got[0][1]["a"] == "今天", got
+ assert trunc is False, "今天的已经扫全, 不能报截断"
+
+
+@case("五源解析: 决策与择时层分流、告警超龄标陈旧、卖出载荷在 data 里、资金异动取流 id")
+def test_parse_sources():
+ ymd = "2026-09-09"
+ itd = [("1-0", {"ts_code": "601126.SH", "producer_id": "bionic_decider", "action": "BUY",
+ "trigger_time": _today_ms(10, 4), "suggested_price": "43.15", "confidence": "0.95",
+ "signal_type": "REVERSAL_BUY"}),
+ ("2-0", {"ts_code": "601126.SH", "producer_id": "intraday_timing_v2", "action": "BUY",
+ "trigger_time": _today_ms(10, 30), "suggested_price": "43.0", "entry_score": "78"}),
+ ("3-0", {"ts_code": "000001.SZ", "producer_id": "bionic_decider", "action": "BUY",
+ "trigger_time": _today_ms(10, 31)})]
+ rows, n, _t = us._bycode_intraday(_FakeRedis({"intraday_signals:%s" % ymd: itd}), ymd, "601126.SH", 5000)
+ assert n == 3 and len(rows) == 2, (n, len(rows))
+ assert rows[0]["src"] == "decision" and rows[0]["time"] == "10:04" and rows[0]["confidence"] == 0.95
+ assert rows[1]["src"] == "timing" and rows[1]["entry_score"] == 78.0, rows[1]
+
+ old = _today_ms(9, 35)
+ with _Patch() as p:
+ p(us, "_alert_max_age_sec", lambda: 60)
+ al = [("1-0", {"ts_code": "601126.SH", "source": "fund_flow_strength", "level": "HIGH",
+ "value": "13.41", "trigger_time": old}),
+ ("2-0", {"ts_code": "601126.SH", "source": "未知源", "level": "MID", "value": "1",
+ "trigger_time": int(time.time() * 1000), "metadata": '{"direction": "down"}'})]
+ arows, an, _ = us._bycode_alerts(_FakeRedis({"intraday_alerts:%s" % ymd: al}), ymd, "SH601126", 5000)
+ assert an == 2 and len(arows) == 2
+ stale = [r for r in arows if r["time"] == "09:35"][0]
+ assert stale["stale"] is True and stale["cat"] and stale["age_min"] >= 0, stale
+ unknown = [r for r in arows if not r["stale"]][0]
+ assert unknown["cat"] == "other" and unknown["direction"] == "看跌", unknown
+
+ sell = [("1-0", {"data": '{"ts_code": "601126.SH", "action": "SELL", "confidence": 0.8,'
+ ' "dominant_signal": "BREAK_DOWN", "llm_reason": "跌破支撑",'
+ ' "timestamp": %d}' % _today_ms(14, 10)})]
+ srows, _n, _t = us._bycode_sell(_FakeRedis({"bionic:signals:llm_sell_actions": sell}), "601126.SH", 5000)
+ assert len(srows) == 1 and srows[0]["time"] == "14:10" and srows[0]["reason"] == "跌破支撑", srows
+
+ mid = _today_ms(13, 41)
+ met = [("%d-0" % mid, {"data": '{"ts_code": "601126.SH", "direction": "up", "z_dd": 2.4, "window_net": 1200}'})]
+ mrows, _n, _t = us._bycode_metrics(_FakeRedis({"mtf:intraday:stream:metrics": met}), "601126.SH", 5000)
+ assert len(mrows) == 1 and mrows[0]["time"] == "13:41" and mrows[0]["z_dd"] == 2.4
+ assert mrows[0]["ts_from"] == "stream_id", "时间取自流 id 是写入时刻, 必须标出来"
+
+
+@case("去重: 同源同分钟同值合并并计数; 跨源永不合并")
+def test_dedup():
+ a = us._row("alert", "盘中告警", _today_ms(10, 4), cat="fund_flow_strength", level="HIGH", value=13.41)
+ b = us._row("alert", "盘中告警", _today_ms(10, 4), cat="fund_flow_strength", level="HIGH", value=13.41)
+ c = us._row("alert", "盘中告警", _today_ms(10, 4), cat="fund_flow_strength", level="HIGH", value=20.0)
+ d = us._row("decision", "决策系统广播", _today_ms(10, 4), action="BUY", price=43.15)
+ out = us._dedup([a, b, c, d])
+ assert len(out) == 3, [(x["src"], x["value"], x["dup"]) for x in out]
+ assert out[0]["dup"] == 2 and out[1]["dup"] == 1 and out[2]["src"] == "decision"
+
+
+@case("单源失败不拖垮整体且逐源标明; 缓存按票生效; 实况分不编时间戳")
+def test_resilience_and_cache():
+ ymd = time.strftime("%Y-%m-%d")
+ good = _FakeRedis({"intraday_signals:%s" % ymd: [
+ ("1-0", {"ts_code": "601126.SH", "producer_id": "bionic_decider", "action": "BUY",
+ "trigger_time": _today_ms(10, 4), "suggested_price": "43.15"})]},
+ boom={"intraday_alerts:%s" % ymd, "mtf:intraday:stream:metrics"})
+ hits = {"n": 0}
+
+ def fake_c208(db):
+ hits["n"] += 1
+ return good
+
+ def fake_mr(_c):
+ return {"watch": [{"ts_code": "601126.SH", "score": 88}], "avoid": []}
+
+ us._BYCODE_CACHE.clear()
+ with _Patch() as p:
+ p(us, "_c208", fake_c208)
+ p(us, "_c214", lambda: object())
+ p(us, "_read_mr", fake_mr)
+ out = us.by_code("SH601126")
+ n1 = hits["n"]
+ out2 = us.by_code("601126.SH") # 归一后同一只票, 走缓存
+ n2 = hits["n"]
+ out3 = us.by_code("601126.SH", force=True)
+ assert out["ok"] and len(out["timeline"]) == 1, out
+ assert out["sources"]["intraday"]["ok"] and out["sources"]["sell_actions"]["ok"]
+ assert out["sources"]["alerts"]["ok"] is False and "error" in out["sources"]["alerts"], out["sources"]
+ assert out["sources"]["metrics"]["ok"] is False and out["sources"]["mr"]["ok"] is True
+ assert n2 == n1 and out2 is out, "同一只票八秒内必须命中缓存"
+ assert out3 is not out, "force 必须绕开缓存"
+ assert out["snapshot"]["mr"]["watch"]["score"] == 88 and out["snapshot"]["mr"]["avoid"] is None
+ assert all(r.get("ts") for r in out["timeline"]), "时间线每行都要有真实时间戳"
+ assert "mr" not in {r["src"] for r in out["timeline"]}, "实况分没有时间, 绝不许排进时间线"
+
+
+@case("回扫上限参数已登记 (页面能改) 且有范围")
+def test_param_registered():
+ assert "PMS_UPSTREAM_BYCODE_SCAN" in param_store.RUNTIME_EXTRA
+ assert "PMS_UPSTREAM_BYCODE_SCAN" in param_store._RANGES
+ lo, hi = param_store._RANGES["PMS_UPSTREAM_BYCODE_SCAN"]
+ assert lo >= 100 and hi <= 50000
+ with _Patch() as p:
+ p(param_store, "get_int", lambda k, d=0: 1)
+ assert us._bycode_scan() >= 100, "下限兜底防把回扫调成零"
+
+
+@case("只读守卫: 按代码回扫这段不许出现任何写入或消费组调用")
+def test_readonly_guard():
+ src = io.open(us.__file__.replace(".pyc", ".py"), encoding="utf-8").read()
+ seg = src[src.index("# ================================================================ 按代码回扫"):]
+ for bad in ("xadd", "xack", "xgroup", "xreadgroup", "xdel", "xtrim", "delete(", ".set(", "zadd"):
+ assert bad not in seg.lower().replace("_bycode_cache.pop", ""), bad
+ assert "xrevrange" in seg and "zrevrange" not in seg
+
+
+@case("页面静态断言: 三处入口、抽屉层级压得住浮层与遮罩、不进轮询、失败态不说今天没有")
+def test_page_static():
+ from app.web import main as web
+ html = io.open(os.path.join(web.STATIC_DIR, "index.html"), encoding="utf-8").read()
+ assert "/api/upstream/signals/" in html, "页面没接单票信号接口"
+ assert html.count("openCodeSignals(") >= 6, "信号栏、持仓表、消息栏三处入口都要接上"
+ # 层级: 弹层库从 2000 起自增, 浮层写死 2010、遮罩 2005。抽屉不显式抬层就落在它们底下,
+ # 屏幕上正好只剩一层灰 —— 这就是遮罩空白那个症状的另一张脸。
+ rail = int(re.search(r"\.rail\.open\{[^}]*z-index:\s*(\d+)", html).group(1))
+ scrim = int(re.search(r"\.rail-scrim\{[^}]*z-index:\s*(\d+)", html).group(1))
+ dz = int(re.search(r"\.cs-drawer\{[^}]*z-index:\s*(\d+)", html).group(1))
+ sz = int(re.search(r"\.cs-scrim\{[^}]*z-index:\s*(\d+)", html).group(1))
+ assert dz > rail and dz > scrim and sz > rail, (dz, sz, rail, scrim)
+ assert "if (railOpen.value) closeRail();" in html, "打开抽屉前要先收浮层 (双保险的另一半)"
+ # 只能点击触发: 回扫接口全页只出现一次, 且不在整页刷新与任何定时器里
+ assert html.count("'/api/upstream/signals/'") == 1
+ load_all = html[html.index("async function loadAll()"):][:1600]
+ assert "csLoadFull" not in load_all and "openCodeSignals" not in load_all, "回扫绝不许进整页刷新"
+ for fn in ("startAutoRefresh", "wsPollSync"):
+ if fn in html:
+ seg = html[html.index("function " + fn):][:1200]
+ assert "csLoadFull" not in seg, fn + " 里不许调回扫"
+ # 读不到时不许说"今天没有"
+ assert html.count("srcFail(") >= 8, "各路空文案都要给失败态让位"
+ assert "别当成平安无事" in html
+
+
+@case("路由清单收了新接口")
+def test_route_registered():
+ from app.web.main import app
+ paths = {r.path for r in app.routes}
+ assert "/api/upstream/signals/{ts_code}" in paths
+ src = io.open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "test_wiring.py"), encoding="utf-8").read()
+ assert "/api/upstream/signals/{ts_code}" in src, "test_wiring 的路由清单也要加, 否则漏了没人发现"
+
+
+def main():
+ ok = 0
+ for name, fn in RESULTS:
+ try:
+ fn()
+ ok += 1
+ print(" ok " + name)
+ except Exception:
+ print(" FAIL " + name)
+ traceback.print_exc()
+ print("-" * 60)
+ if ok == len(RESULTS):
+ print("ALL PASS (%d cases)" % ok)
+ return 0
+ print("FAILED %d/%d" % (len(RESULTS) - ok, len(RESULTS)))
+ return 1
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py
index 36a2275..becc39b 100644
--- a/scripts/test_wiring.py
+++ b/scripts/test_wiring.py
@@ -647,6 +647,7 @@ def _():
"/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/signals/{ts_code}", # 2026-09-09 点股票看它今天全部信号
"/api/ops/strategy-attach-scan"] # 2026-08-25 策略自动挂载试算位
missing = [p for p in need if p not in paths]
assert not missing, missing