213 lines
10 KiB
Python
213 lines
10 KiB
Python
# -*- 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)}
|