159 lines
8.1 KiB
Python
159 lines
8.1 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
|
||
|
||
# 数翻向的窗口「最近 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,
|
||
"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, "未归类", "指标组合未归入任何相位，保守中性")
|