diff --git a/app/core/reask_rules.py b/app/core/reask_rules.py new file mode 100644 index 0000000..353e68f --- /dev/null +++ b/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)} diff --git a/app/services/market.py b/app/services/market.py index 15e0ae6..0b6a98d 100644 --- a/app/services/market.py +++ b/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}。