解锁重问的判定逻辑与取数：纯逻辑模块 + 当日成交量与五日均量

量比的折算分母是**分钟线实际覆盖的时段**，不是从开盘算的已过时段。实测 09-09 15:00
四方股份的分钟线首根是 10:11 —— db13 那张 key 是滚动窗口，不保证从开盘存起；按当前
时刻算分母会把没存进来的那段也算上，量比被系统性压低，量能确认这条永远过不了。

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

app/core/reask_rules.py

@@ -0,0 +1,212 @@
+# -*- coding: utf-8 -*-
+"""研判驳回后的「解锁重问」判定 (2026-09-09 方案第四之一节, 台账 055)。
+
+纯逻辑, 零外部依赖, 可离线单测 —— 与 signal_rules、exec_timing 同一手法。
+
+要解决什么: 新建仓被研判闸驳回后, 那只票当天就进了跳过集合, 扫描每分钟仍在跑但直接
+略过它。这道闸挡的是真问题 (2026-07-29 那次每分钟写一行驳回、一天几千行, 把判分锚
+淹掉), 假设也合理 ——「这只票今天不该从零建仓这个结论当天基本不会翻转」—— 但它不是
+绝对的, 缺一个「确实有新信息」的例外。这里就是那个例外。
+
+判定式:  解锁 = G0 且 G1 且 G2 且 G3 且 (U1 或 U2 或 U3)
+
+四条护栏 (全要):
+  G0 总开关开着
+  G1 距驳回已过 20 分钟        —— 不是等择时侧那层 30 分钟缓存过期 (重问带强制标记绕开它),
+                                  是等量能确认成形: 20 分钟等于四根五分钟线
+  G2 这只票今天还没重问过      —— 每票每天一次。两次以上就等于取消当日闸
+  G3 当前在 09:45 到 14:30 之间 —— 早于 09:45 时段折算量比不可靠; 晚于 14:30 即便放行,
+                                  提议到人拍板到落指令到择时区间, 来不及
+
+三条解锁条件 (任一成立):
+  U1 择时决策系统当日判过这只票转多, 且**时刻晚于驳回**
+     —— 另一套系统自己改了口, 是最硬的一种新信息。"晚于驳回"这个限定是关键: 驳回之前
+        就有的信号研判当时已经看过了 (它作为盘中转多留痕进了提示词), 拿它当理由是循环论证
+  U2 昨夜结论被盘中重算改写, 且方向向好 (定性翻成看多, 或支撑压力漂了 3%)
+     —— 研判的底牌就是那一行。它被改写意味着研判当时依据的东西已经不在了。
+        3% 直接取 PMS_OPEN_REF_DRIFT_MAX 同值: 那道闸说「底牌漂了就别买」, 这条说
+        「底牌漂了就重问一次」, 同一件事的两个方向用同一个数, 复盘能直接对照
+  U3 价升且量能确认: 现价较驳回涨 2%, 且时段折算量比 ≥ 1.5, 且量比涨到驳回时的 1.3 倍
+     —— 直冲「今日突破量能未确认」那类驳回理由。三条**并列**才算: 最常见的假信号正是
+        「价格再涨一点但量没跟上」, 只看价格等于在追高; 只看量比会被早盘脉冲骗
+
+**取数失败一律判该条不成立** (fail-closed): 方向是漏解锁, 不是误解锁。与规则闸
+「缺输入必须留痕」、研判闸「拿不到不等于通过」是同一条纪律。
+
+明确不做的两条 (记下来免得日后当成遗漏):
+  · 「现价回落进择时买入区间」—— 区间公式在择时侧, PMS 复制一份就是第二个事实源;
+    而且 09-09 拍板是「不改追高原则」, 这条属于另一件事
+  · 纯时间兜底 (比如过一小时无条件重问) —— 那是取消当日闸, 不是解除
+"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Optional
+
+ELAPSED_FULL_MIN = 240          # 全日交易分钟: 09:30-11:30 与 13:00-15:00 各 120
+VOL_SANITY_MAX = 30.0           # 折算前的原始比超过它判数据异常 (口径与择时侧同源)
+# 覆盖时段占全日不足这个比例就不判量比。0.08 约合 20 分钟 —— 实测 db13 的分钟线是滚动
+# 窗口, 不保证从开盘存起 (09-09 四方股份首根是 10:11), 覆盖太短时量比抖得没有意义。
+FRAC_MIN = 0.08
+
+
+def _f(v, d=None):
+    try:
+        x = float(v)
+    except (TypeError, ValueError):
+        return d
+    return x
+
+
+def _minutes(t: datetime) -> int:
+    return t.hour * 60 + t.minute
+
+
+def elapsed_trading_frac(now: datetime) -> float:
+    """当日已进行的交易时间占比。盘前 0, 收盘后 1。
+
+    口径与择时侧的两处逐字对齐 (台账 055 记了「三处同源」这条纪律)。
+    """
+    m = _minutes(now)
+    if m < 9 * 60 + 30:
+        return 0.0
+    if m <= 11 * 60 + 30:
+        return (m - (9 * 60 + 30)) / float(ELAPSED_FULL_MIN)
+    if m < 13 * 60:
+        return 120 / float(ELAPSED_FULL_MIN)
+    if m <= 15 * 60:
+        return (120 + (m - 13 * 60)) / float(ELAPSED_FULL_MIN)
+    return 1.0
+
+
+def vol_ratio(cum_vol_shares, vol_base_lots, frac) -> Optional[float]:
+    """时段折算量比 = (当日累计量 ÷ 100) ÷ 近五日日均量 ÷ 已过时段占比。
+
+    两个单位不一样: 分钟线的量是**股**, 因子表的均量是**手**, 所以要除以 100。
+    这个除法是最容易在日后被人"顺手删掉"的一行, 两侧单测各钉了同一组边界数。
+
+    三条哨兵, 任一触发回 None (由调用方当作「读不到」, 于是该条解锁不成立):
+      · 时段占比太小 —— 开盘头几分钟的量比没有意义
+      · 折算前的原始比超过 30 —— 多半是上游改了单位, 宁可不判也不误判
+      · 任一入参取不到或不是正数
+    """
+    v, base, fr = _f(cum_vol_shares), _f(vol_base_lots), _f(frac)
+    if v is None or base is None or fr is None:
+        return None
+    if base <= 0 or v < 0 or fr < FRAC_MIN:
+        return None
+    raw = (v / 100.0) / base
+    if raw > VOL_SANITY_MAX:
+        return None
+    return round(raw / fr, 3)
+
+
+def in_window(now: datetime, window: str) -> bool:
+    """窗口写成 "0945-1430" 这种。写坏了返回 True (不因为一个格式错误把整条路堵死)。"""
+    try:
+        a, b = str(window or "").split("-")
+        lo = int(a[:2]) * 60 + int(a[2:])
+        hi = int(b[:2]) * 60 + int(b[2:])
+    except Exception:
+        return True
+    return lo <= _minutes(now) <= hi
+
+
+def _u1(base, now_read) -> tuple:
+    """择时决策系统当日判过转多, 且晚于驳回时刻。"""
+    at = str((now_read or {}).get("bionic_buy_at") or "")
+    base_at = str((base or {}).get("at") or "")
+    if not at:
+        return False, "U1 当日没有转多留痕"
+    if not base_at:
+        return False, "U1 驳回时刻读不到, 判不了先后"
+    if at <= base_at:
+        return False, "U1 转多留痕在驳回之前, 研判当时已经看过"
+    return True, "决策系统 %s 判这只票转多 (晚于 %s 的驳回)" % (at[11:16] or at, base_at[11:16] or base_at)
+
+
+def _u2(base, now_read, drift_max) -> tuple:
+    """昨夜结论被盘中重算改写且方向向好。"""
+    y0 = str((base or {}).get("y_signal") or "").upper()
+    y1 = str((now_read or {}).get("y_signal") or "").upper()
+    good = ("BUY", "MAINTAIN")
+    if y1 and y1 in good and y0 and y0 not in good:
+        return True, "昨夜定性由 %s 改写成 %s" % (y0, y1)
+    for k, label in (("support", "支撑"), ("pressure", "压力")):
+        a, b = _f((base or {}).get(k)), _f((now_read or {}).get(k))
+        if a and b and a > 0:
+            d = abs(b / a - 1.0)
+            if d > drift_max:
+                return True, "%s位由 %.2f 漂到 %.2f (%.1f%%, 超过 %.0f%%)" % (
+                    label, a, b, d * 100, drift_max * 100)
+    return False, "U2 底牌没被改写"
+
+
+def _u3(base, now_read, up_min, vol_min, vol_mult) -> tuple:
+    """价升且量能确认, 三条并列。"""
+    p0, p1 = _f((base or {}).get("price")), _f((now_read or {}).get("price"))
+    if not p0 or not p1 or p0 <= 0:
+        return False, "U3 价格读不到"
+    up = p1 / p0 - 1.0
+    if up < up_min:
+        return False, "U3 较驳回只走了 %.2f%% (要 %.0f%%)" % (up * 100, up_min * 100)
+    v0, v1 = _f((base or {}).get("vol_ratio")), _f((now_read or {}).get("vol_ratio"))
+    if v1 is None:
+        return False, "U3 量比取不到"
+    if v1 < vol_min:
+        return False, "U3 量比 %.2f 不到 %.1f" % (v1, vol_min)
+    if v0 is None or v0 <= 0:
+        return False, "U3 驳回时的量比没记下, 判不了是不是相对变强"
+    if v1 < v0 * vol_mult:
+        return False, "U3 量比 %.2f 没到驳回时 %.2f 的 %.1f 倍" % (v1, v0, vol_mult)
+    return True, "现价 %.2f→%.2f (+%.2f%%), 时段折算量比 %.2f→%.2f" % (p0, p1, up * 100, v0, v1)
+
+
+def evaluate(*, base: dict, now_read: dict, params: dict, used_today: int, now: datetime) -> dict:
+    """唯一的判定函数。零副作用, 不读库不写库 —— 取数由调用方备齐。
+
+    返回 {"unlock": bool, "hits": [...], "why": 一句中文, "blocked_by": "G1"|None, "notes": [...]}
+    """
+    notes = []
+    if not params.get("enabled", True):
+        return {"unlock": False, "hits": [], "why": "解锁重问总开关关着", "blocked_by": "G0", "notes": notes}
+
+    gap_min = int(params.get("min_gap_min", 20))
+    base_at = str((base or {}).get("at") or "")
+    if not base_at:
+        # 旧的驳回行没有基线快照 (这个块是今天才开始写的)。U1、U2 还能判, U3 判不了。
+        notes.append("这条驳回行没有基线快照，U3 用不了")
+    else:
+        try:
+            t0 = datetime.fromisoformat(base_at)
+            if (now - t0).total_seconds() < gap_min * 60:
+                return {"unlock": False, "hits": [], "blocked_by": "G1", "notes": notes,
+                        "why": "距驳回还不到 %d 分钟" % gap_min}
+        except ValueError:
+            notes.append("驳回时刻格式认不出：" + base_at)
+
+    if used_today >= int(params.get("max_per_day", 1)):
+        return {"unlock": False, "hits": [], "blocked_by": "G2", "notes": notes,
+                "why": "今天已经重问过 %d 次，一天就这一次" % used_today}
+    if not in_window(now, params.get("window", "0945-1430")):
+        return {"unlock": False, "hits": [], "blocked_by": "G3", "notes": notes,
+                "why": "不在重问时段 %s 内" % params.get("window", "0945-1430")}
+
+    hits, whys, misses = [], [], []
+    for tag, fn in (("U1", lambda: _u1(base, now_read)),
+                    ("U2", lambda: _u2(base, now_read, float(params.get("ref_drift", 0.03)))),
+                    ("U3", lambda: _u3(base, now_read, float(params.get("up_min", 0.02)),
+                                       float(params.get("vol_min", 1.5)),
+                                       float(params.get("vol_mult", 1.3))))):
+        try:
+            ok, why = fn()
+        except Exception as e:                      # 判定本身出错也算不成立
+            ok, why = False, "%s 判定出错: %s" % (tag, e)
+        (whys if ok else misses).append(why)
+        if ok:
+            hits.append(tag)
+    if not hits:
+        return {"unlock": False, "hits": [], "blocked_by": None, "notes": notes,
+                "why": "输入没有向好的变化：" + "；".join(misses)}
+    return {"unlock": True, "hits": hits, "blocked_by": None, "notes": notes,
+            "why": "；".join(whys)}

app/services/market.py

@@ -203,6 +203,102 @@ def get_last_close(ts_code: str):
     return val
 
 
+def day_volume(ts_code: str) -> dict:
+    """当日累计成交量与它实际覆盖的交易时段 (2026-09-09, 解锁重问的量能确认要用)。
+
+    返回 {"vol": 股, "frac": 覆盖占全日的比例, "from": "10:11", "to": "15:00", "minutes": 199}
+    取不到返回 {}。
+
+    **为什么要带 frac 出来, 而不是让调用方按当前时刻算**: 实测 09-09 15:00,
+    601126.SH 的分钟线首根是 10:11 不是 09:30 —— db13 里这张 key 是滚动窗口, 不保证
+    从开盘存起。按当前时刻算已过时段, 分母会把没存进来的那段也算上, 量比被系统性压低,
+    「量能确认」这道条件于是永远过不了。所以折算的分母只能是**这份数据真正覆盖的时段**。
+
+    单位: 分钟线的 vol 是**股**, 因子表的日线 vol 是**手**。两边差 100 倍, 换算在
+    reask_rules.vol_ratio 里做, 那里有单测钉着。
+    """
+    bars = _bars(ts_code)
+    if not bars:
+        return {}
+    try:
+        vol = sum(float(b.get("vol") or 0) for b in bars)
+        t0, t1 = str(bars[0].get("time") or ""), str(bars[-1].get("time") or "")
+        m0, m1 = _hhmm_min(t0), _hhmm_min(t1)
+        if m0 is None or m1 is None or m1 < m0:
+            return {}
+        mins = _trading_minutes_between(m0, m1)
+        if mins <= 0:
+            return {}
+        return {"vol": vol, "minutes": mins, "frac": round(mins / 240.0, 4),
+                "from": t0[11:16], "to": t1[11:16], "bars": len(bars)}
+    except (TypeError, ValueError) as e:
+        logger.warning("当日成交量聚合失败 [%s]: %s", ts_code, e)
+        return {}
+
+
+def _hhmm_min(t: str):
+    """"2026-09-09 10:11:00" → 611 (当天第几分钟)。认不出回 None。"""
+    try:
+        hh, mm = t[11:13], t[14:16]
+        return int(hh) * 60 + int(mm)
+    except (ValueError, IndexError):
+        return None
+
+
+def _trading_minutes_between(m0: int, m1: int) -> int:
+    """两个时刻之间的**交易**分钟数 (跨午休不算休市那 90 分钟)。"""
+    AM0, AM1, PM0, PM1 = 9 * 60 + 30, 11 * 60 + 30, 13 * 60, 15 * 60
+    lo, hi = max(m0, AM0), min(m1, PM1)
+    if hi <= lo:
+        return 0
+    n = 0
+    for a, b in ((AM0, AM1), (PM0, PM1)):
+        n += max(0, min(hi, b) - max(lo, a))
+    return n
+
+
+def get_vol_base5(ts_code: str):
+    """近五个交易日的日均成交量 (单位: **手**, 因子表口径)。按日缓存, 取不到回 None。
+
+    单独一条 SQL 而不是扩 _factor_rows 的 SELECT: 那个函数在候选池装配的热路径上被几十只
+    票各调一次, 给它加一列会让每一次都多搬一份用不上的数据。这条只在解锁判定里按票调一次。
+    """
+    today = datetime.now().strftime("%Y%m%d")
+    key = f"VB5:{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, failed = None, False
+    since = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")
+    try:
+        rows = []
+        for tbl in _shard_tables():
+            # 因子表的 symbol 实测是前缀式 (SH601126), 但两种写法都试 —— 与 _factor_rows 同手法,
+            # 上游哪天改了写法这里不至于静默取空。
+            for sym in (downstream_repo.to_prefix(ts_code), ts_code):
+                try:
+                    part = fetch_all(
+                        f"SELECT trade_date, vol FROM {tbl} "
+                        f"WHERE symbol = :sym AND trade_date >= :since", {"sym": sym, "since": since})
+                except Exception:
+                    part = []
+                if part:
+                    rows.extend(part)
+                    break
+        rows = [r for r in rows if r.get("vol")]
+        rows.sort(key=lambda r: str(r.get("trade_date")))
+        vols = [float(r["vol"]) for r in rows[-5:]]
+        if len(vols) >= 3:              # 三根就够算日均; 少于三根宁可不给
+            val = round(sum(vols) / len(vols), 2)
+    except Exception as e:
+        failed = True
+        logger.warning("五日均量取数失败 [%s] (本次不缓存): %s", ts_code, e)
+    if not failed:
+        _ma_cache["data"][key] = val
+    return val
+
+
 def plan_price(ts_code: str) -> dict:
     """规划用价: 实时价优先, 盘前/停更回落昨收。返回 {price, source}。