tradingSystem/app/core/tech_rules.py

# -*- 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",
                  "开口向上未确认（现价未站上均价或量能不足/取不到），进观察等确认")