216 lines
12 KiB
Python
216 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""技术面相位合成 (2026-09-11 技术面接入, 纯逻辑, 零外部依赖)。
|
||
|
||
输入是一只票按 data_date 升序的近若干个交易日技术面读数 (tech_repo 落表的行, 字段见
|
||
ddl_pms_v1.sql 的 pms_tech_daily), 最后一行是最新那天。数据新鲜度 —— 有没有当天行、读数
|
||
陈不陈旧 —— 由 tech_service 在调用前保证 (它握着交易日历); 本模块只认传进来的 rows 内容。
|
||
|
||
合成三步, 与《技术面接入与三源合议方案》第三节一一对应:
|
||
一, 判有没有读数: 没有最新行 / 未除权重锚 / 数据质量非 OK / 关键读数缺失, 一律无读数。
|
||
无读数是弃权, 绝不折成看空 (设计原则二「无读数等于弃权」)。
|
||
二, 判是不是震荡市: 最近 20 个交易日 SAR 翻向达到 choppy_flips 次即震荡。震荡市里 SAR
|
||
翻向不当信号, 这是决策系统文档明说的指标特性。
|
||
三, 判相位: 按固定次序逐条比对, 第一个命中的就是结论 (方案第三节相位表)。
|
||
|
||
立场取值只有四个中文词: 看多 / 看空 / 中性 / 无读数。强弱只给看多看空用, 中性与无读数没有。
|
||
阈值默认值见 DEFAULTS, tech_service 会从参数中心读实际值覆盖; 本模块不读参数中心, 以便脱库单测。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
# 盘中确认包 (2026-09-14): 时段判断复用解锁重问那一份 in_window, 不另写 —— 台账 003
|
||
# 「三处同源」的哨兵 (解锁重问、择时、盘中确认共用同一套时段与量比算法)。量比由调用方算,
|
||
# 也走 reask_rules.vol_ratio。这里只导入 in_window, 是同一个函数对象。
|
||
from app.core.reask_rules import in_window
|
||
|
||
# 数翻向的窗口「最近 20 个交易日」(方案第三节)。窗口本身不设页面参数, 翻向次数阈值才设。
|
||
CHOPPY_WINDOW = 20
|
||
|
||
DEFAULTS = {
|
||
"squeeze_lookback": 5, # 近几个交易日内收过口算「刚收口过」(PMS_TECH_SQUEEZE_LOOKBACK)
|
||
"open_bw_growth": 0.20, # 今日带宽比收口日至少扩这么多算开口 (PMS_TECH_OPEN_BW_GROWTH)
|
||
"choppy_flips": 4, # 20 日 SAR 翻向达到这么多次算震荡 (PMS_TECH_CHOPPY_FLIPS)
|
||
"flip_fresh_days": 2, # SAR 翻向后这么多个交易日内算「刚转向」(PMS_TECH_FLIP_FRESH_DAYS)
|
||
}
|
||
|
||
# 价格「贴下轨」的派生判据 (带内位置 ≤ 此值)。转空确认的次要条件, 主条件是多空布林线空头区。
|
||
# 这不是方案里一次定死的五个阈值之一, 是实现派生量; 要调改这里即可。
|
||
_NEAR_LOWER = 0.20
|
||
|
||
_LONG = "多"
|
||
_SHORT = "空"
|
||
|
||
|
||
def _num(v):
|
||
try:
|
||
return None if v is None else float(v)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _truthy(v) -> bool:
|
||
"""TINYINT / 布尔 / 字符串都收: 1 / True / "是" 为真。"""
|
||
if isinstance(v, str):
|
||
return v.strip() in ("1", "true", "True", "是")
|
||
return bool(v)
|
||
|
||
|
||
def _no_read(why: str, latest=None) -> dict:
|
||
return {"stance": "无读数", "strength": None, "phase": None, "confirm": None,
|
||
"sar_side": None, "sar_value": None, "sar_flip_days": None, "choppy": None,
|
||
"reason": why, "no_read_why": why,
|
||
"data_date": (latest or {}).get("data_date")}
|
||
|
||
|
||
def count_sar_flips(rows) -> int:
|
||
"""rows 升序, 数相邻两行 SAR 方向变化的次数 (只看最近 CHOPPY_WINDOW 行)。"""
|
||
seq = [str(r.get("sar_side") or "") for r in (rows or [])[-CHOPPY_WINDOW:]]
|
||
seq = [s for s in seq if s in (_LONG, _SHORT)]
|
||
return sum(1 for a, b in zip(seq, seq[1:]) if a != b)
|
||
|
||
|
||
def _recent_squeeze_bw(rows, lookback):
|
||
"""近 lookback 个交易日内 (不含今日) 最近一次收口那天的带宽; 没收口过返回 None。"""
|
||
window = (rows or [])[-(lookback + 1):-1] # 去掉今日那一行
|
||
for r in reversed(window): # 由近及远
|
||
if _truthy(r.get("boll_squeeze")):
|
||
return _num(r.get("boll_bw_pct"))
|
||
return None
|
||
|
||
|
||
def synthesize(rows, *, params=None) -> dict:
|
||
"""把一只票的近 N 日读数 (升序) 合成技术面立场。返回含 stance/strength/phase/confirm/
|
||
sar_value/choppy/reason/data_date/no_read_why 的字典。"""
|
||
p = dict(DEFAULTS)
|
||
if params:
|
||
p.update({k: params[k] for k in DEFAULTS if params.get(k) is not None})
|
||
rows = list(rows or [])
|
||
latest = rows[-1] if rows else None
|
||
|
||
# ---- 第一步: 有没有读数 (无读数是弃权) ----
|
||
if not latest:
|
||
return _no_read("没有当天的技术面读数")
|
||
if not _truthy(latest.get("reanchored")):
|
||
return _no_read("未做除权重锚，读数不可比", latest)
|
||
if str(latest.get("quality") or "").upper() != "OK":
|
||
return _no_read(f"数据质量非正常（{latest.get('quality') or '缺'}）", latest)
|
||
side = str(latest.get("sar_side") or "")
|
||
pos = _num(latest.get("boll_pos"))
|
||
if side not in (_LONG, _SHORT) or pos is None:
|
||
return _no_read("关键读数缺失（SAR 方向或带内位置）", latest)
|
||
|
||
squeeze = _truthy(latest.get("boll_squeeze"))
|
||
bbi_state = str(latest.get("bbi_state") or "")
|
||
bw = _num(latest.get("boll_bw_pct"))
|
||
flip_days = latest.get("sar_flip_days")
|
||
flip_days = int(flip_days) if flip_days is not None else None
|
||
sar_value = _num(latest.get("sar_value"))
|
||
|
||
base = {"sar_side": side, "sar_value": sar_value, "sar_flip_days": flip_days,
|
||
"boll_upper": _num(latest.get("boll_upper")), # 盘中确认包: 收口突破要拿昨日上轨
|
||
"data_date": latest.get("data_date"), "no_read_why": None}
|
||
|
||
# ---- 第二步: 震荡市 ----
|
||
flips = count_sar_flips(rows)
|
||
choppy = flips >= int(p["choppy_flips"])
|
||
base["choppy"] = choppy
|
||
|
||
def out(stance, strength, phase, reason, confirm=None):
|
||
r = dict(base)
|
||
r.update({"stance": stance, "strength": strength, "phase": phase,
|
||
"confirm": confirm, "reason": reason})
|
||
return r
|
||
|
||
long_, short_ = side == _LONG, side == _SHORT
|
||
fresh_flip = flip_days is not None and flip_days <= int(p["flip_fresh_days"])
|
||
bbi_bull = bbi_state == "多头区"
|
||
bbi_bear = bbi_state == "空头区"
|
||
upper_half = pos >= 0.5
|
||
|
||
# ---- 第三步: 相位 (固定次序, 第一个命中为结论) ----
|
||
# 1 收口等待
|
||
if squeeze:
|
||
return out("中性", None, "收口等待", "布林带收口，等开口再定方向")
|
||
# 2 / 3 开口向上 / 开口向下 (今天已不收口; 近 lookback 日内收过口且带宽扩够)
|
||
sq_bw = _recent_squeeze_bw(rows, int(p["squeeze_lookback"]))
|
||
opened = (sq_bw is not None and sq_bw > 0 and bw is not None
|
||
and (bw / sq_bw - 1) >= p["open_bw_growth"])
|
||
if opened and upper_half and long_:
|
||
return out("看多", "强", "开口向上", "刚收口后放开、价在带内上半部、SAR 多：向上变盘")
|
||
if opened and (not upper_half) and short_:
|
||
return out("看空", "强", "开口向下", "刚收口后放开、价在带内下半部、SAR 空：向下变盘")
|
||
# 4 转空 (非震荡, SAR 刚翻空)
|
||
if short_ and fresh_flip and not choppy:
|
||
confirm = bbi_bear or pos <= _NEAR_LOWER
|
||
why = "SAR 刚翻空" + ("、多空布林线空头区或贴下轨（确认）" if confirm else "（待确认）")
|
||
return out("看空", "强" if confirm else "弱", "转空", why, confirm=confirm)
|
||
# 5 转多 (非震荡, SAR 刚翻多, 多空布林线不在空头区)
|
||
if long_ and fresh_flip and not bbi_bear and not choppy:
|
||
return out("看多", "弱", "转多", "SAR 刚翻多、多空布林线未在空头区")
|
||
# 6 趋势多 (SAR 多且多空布林线在多头区为强, 中性区为弱)
|
||
if long_ and bbi_bull:
|
||
return out("看多", "强", "趋势多", "SAR 多且多空布林线多头区：上升趋势延续")
|
||
if long_ and not bbi_bear:
|
||
return out("看多", "弱", "趋势多", "SAR 多、多空布林线中性区：偏多但不强")
|
||
# 7 趋势空 (SAR 空且多空布林线不在多头区)
|
||
if short_ and not bbi_bull:
|
||
return out("看空", "强" if bbi_bear else "弱", "趋势空", "SAR 空且多空布林线未在多头区：下行")
|
||
# 8 分歧 (SAR 与多空布林线方向相反)
|
||
if (long_ and bbi_bear) or (short_ and bbi_bull):
|
||
return out("中性", None, "分歧", "SAR 与多空布林线方向相反，不表态")
|
||
# 9 震荡 (震荡市且以上都不命中)
|
||
if choppy:
|
||
return out("中性", None, "震荡", f"近 20 日 SAR 翻向 {flips} 次，震荡市不表态")
|
||
# 兜底: side 已限定在 {多, 空}, 上面各分支理应已覆盖全部组合; 到这里保守中性。
|
||
return out("中性", None, "未归类", "指标组合未归入任何相位，保守中性")
|
||
|
||
|
||
def intraday_confirm(state, *, price, vwap, upper_prev, vol_ratio, now,
|
||
window="0945-1430", vol_min=1.5, breakout_on=True) -> dict:
|
||
"""给技术面这一票加盘中确认, **只用于入场** (方案第三节、附录乙)。纯逻辑, 零外部依赖。
|
||
|
||
state 是 synthesize 出的技术面块 (至少含 phase)。返回:
|
||
{"state": 新块或原块, "changed": bool,
|
||
"kind": None | "breakout" | "opened_ok" | "opened_wait" | "outside_window",
|
||
"why": str, "reads": {price, vwap, upper_prev, vol_ratio, at}}
|
||
|
||
判定次序 (附录乙):
|
||
1 相位不是收口等待也不是开口向上 → 原样返回, kind 空。
|
||
2 不在时段内: 开口向上 → outside_window (进观察等确认); 收口等待 → 原样返回。
|
||
3 收口等待: breakout_on 且 现价/昨日上轨/量比都有值 且 现价>上轨 且 量比≥vol_min
|
||
→ breakout, 立场看多、强、相位「盘中开口向上」; 任一不满足原样返回。
|
||
4 开口向上: 现价/均价/量比都有值 且 现价≥均价 且 量比≥vol_min → opened_ok (块不变);
|
||
任一取不到或不满足 → opened_wait。**取不到按等待而不是放行** —— 入场的门, 方向保守。"""
|
||
st = state if isinstance(state, dict) else {}
|
||
phase = st.get("phase")
|
||
at = now.strftime("%H:%M") if now is not None else None
|
||
reads = {"price": price, "vwap": vwap, "upper_prev": upper_prev,
|
||
"vol_ratio": vol_ratio, "at": at}
|
||
|
||
def result(new_state, changed, kind, why):
|
||
return {"state": new_state, "changed": changed, "kind": kind, "why": why, "reads": reads}
|
||
|
||
if phase not in ("收口等待", "开口向上"):
|
||
return result(st, False, None, "")
|
||
if now is None or not in_window(now, window):
|
||
if phase == "开口向上":
|
||
return result(st, False, "outside_window",
|
||
"开口向上候选不在确认时段 %s 内，进观察等确认" % window)
|
||
return result(st, False, None, "") # 收口等待在时段外照旧观察 (原样)
|
||
if phase == "收口等待":
|
||
if (breakout_on and price is not None and upper_prev is not None
|
||
and vol_ratio is not None and price > upper_prev and vol_ratio >= vol_min):
|
||
new = dict(st)
|
||
new.update({"stance": "看多", "strength": "强", "phase": "盘中开口向上",
|
||
"reason": "盘中站上昨日布林上轨且放量：收口突破视为开口向上"})
|
||
return result(new, True, "breakout",
|
||
"现价 %s 站上昨日上轨 %s、量比 %s≥%s：收口突破视为开口向上"
|
||
% (price, upper_prev, vol_ratio, vol_min))
|
||
return result(st, False, None, "收口等待，盘中未突破")
|
||
# phase == 开口向上, 在时段内
|
||
if (price is not None and vwap is not None and vol_ratio is not None
|
||
and price >= vwap and vol_ratio >= vol_min):
|
||
return result(st, False, "opened_ok",
|
||
"现价 %s≥当日均价 %s、量比 %s≥%s：开口确认" % (price, vwap, vol_ratio, vol_min))
|
||
return result(st, False, "opened_wait",
|
||
"开口向上未确认（现价未站上均价或量能不足/取不到），进观察等确认")
|