chan_engine.py

"""
缠论引擎 v2.1 (原始版, 用于P0-P3改造的基线)
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional
import numpy as np
import pandas as pd

PIVOT_MAX_EXTEND_SEGS = 6


@dataclass
class Fractal:
    idx: int
    date: pd.Timestamp
    kind: str
    price: float
    k_high: float
    k_low: float


@dataclass
class Bi:
    start: Fractal
    end: Fractal
    direction: str
    bars: int
    high: float
    low: float
    @property
    def amplitude(self) -> float:
        return self.high - self.low


@dataclass
class Seg:
    start: Fractal
    end: Fractal
    direction: str
    bis: list
    high: float
    low: float
    confirmed: bool = True


@dataclass
class Pivot:
    start_date: pd.Timestamp
    end_date: pd.Timestamp
    zg: float
    zd: float
    gg: float
    dd: float
    bis: list
    direction: str
    zg_date: Optional[pd.Timestamp] = None
    zd_date: Optional[pd.Timestamp] = None
    gg_date: Optional[pd.Timestamp] = None
    dd_date: Optional[pd.Timestamp] = None
    g: Optional[float] = None
    d: Optional[float] = None
    state: str = 'new'
    death_combo: str = ''
    capped: bool = False
    upgraded_level: str = ''


@dataclass
class DivergenceGrade:
    grade: str
    area_ok: bool
    dif_ok: bool
    area_ratio: float
    a_area: float
    c_area: float
    a_dif: float
    c_dif: float
    direction: str
    reason: str
    is_trend_divergence: bool = False
    n_trend_pivots: int = 0


@dataclass
class Signal:
    kind: str
    date: pd.Timestamp
    price: float
    reason: str
    pivot_zg: Optional[float] = None
    pivot_zd: Optional[float] = None
    macd_ratio: Optional[float] = None
    dif_value: Optional[float] = None
    n_pivots: int = 0
    trend: str = ''
    extras: dict = field(default_factory=dict)
    diverge_grade: Optional[DivergenceGrade] = None


def merge_klines(df: pd.DataFrame) -> pd.DataFrame:
    if len(df) == 0:
        return df.copy()
    h = df['high'].values; l = df['low'].values; d = df['date'].values
    out_h, out_l, out_d, out_idx = [h[0]], [l[0]], [d[0]], [0]
    for i in range(1, len(df)):
        ph, pl = out_h[-1], out_l[-1]; ch, cl = h[i], l[i]
        direction = 1 if (len(out_h) >= 2 and out_h[-1] >= out_h[-2]) else (1 if len(out_h) < 2 else -1)
        contained_a = ph >= ch and pl <= cl
        contained_b = ch >= ph and cl <= pl
        if contained_a or contained_b:
            if direction >= 0:
                out_h[-1] = max(ph, ch); out_l[-1] = max(pl, cl)
            else:
                out_h[-1] = min(ph, ch); out_l[-1] = min(pl, cl)
            out_idx[-1] = i
        else:
            out_h.append(ch); out_l.append(cl); out_d.append(d[i]); out_idx.append(i)
    return pd.DataFrame({'date': out_d, 'high': out_h, 'low': out_l, 'orig_idx': out_idx})


def find_fractals(merged: pd.DataFrame) -> list:
    res = []
    n = len(merged)
    if n < 3:
        return res
    h = merged['high'].values; l = merged['low'].values; d = merged['date'].values
    hi = h[1:-1]; hp = h[:-2]; hn = h[2:]
    li = l[1:-1]; lp = l[:-2]; ln = l[2:]
    top = (hi > hp) & (hi > hn) & (li >= lp) & (li >= ln)
    bot = (li < lp) & (li < ln) & (hi <= hp) & (hi <= hn)
    idxs = np.nonzero(top | bot)[0]
    if len(idxs) == 0:
        return res
    # 仅对被选中的(稀疏)分型点构造 Timestamp, 避免对全序列逐根转换
    is_top = top  # 局部别名
    for j in idxs:
        i = j + 1
        if is_top[j]:
            res.append(Fractal(i, pd.Timestamp(d[i]), 'top', float(h[i]), float(h[i]), float(l[i])))
        else:
            res.append(Fractal(i, pd.Timestamp(d[i]), 'bottom', float(l[i]), float(h[i]), float(l[i])))
    return res


def find_bis(fractals: list, min_k: int = 4) -> list:
    if len(fractals) < 2:
        return []
    cleaned = [fractals[0]]
    for fx in fractals[1:]:
        last = cleaned[-1]
        if fx.kind == last.kind:
            if fx.kind == 'top' and fx.price > last.price:
                cleaned[-1] = fx
            elif fx.kind == 'bottom' and fx.price < last.price:
                cleaned[-1] = fx
        else:
            cleaned.append(fx)
    alt = [cleaned[0]]
    for fx in cleaned[1:]:
        if fx.kind != alt[-1].kind and fx.idx - alt[-1].idx >= min_k - 1:
            alt.append(fx)
        elif fx.kind != alt[-1].kind:
            continue
    bis = []
    for i in range(len(alt) - 1):
        a, b = alt[i], alt[i+1]
        if a.kind == b.kind:
            continue
        direction = 'up' if b.kind == 'top' else 'down'
        bis.append(Bi(start=a, end=b, direction=direction, bars=b.idx - a.idx,
                      high=max(a.price, b.price), low=min(a.price, b.price)))
    return bis


def _find_feature_fractal(std: list, seg_dir: str):
    """[保留] 供调试/对照用的非增量实现; 主路径已改用 _first_feature_fractal_incremental。"""
    up = (seg_dir == 'up')
    for i in range(1, len(std) - 1):
        a = std[i-1]; b = std[i]; c = std[i+1]
        if up:
            if b['high'] > a['high'] and b['high'] > c['high'] \
               and b['low'] > a['low'] and b['low'] > c['low']:
                return (i, b.get('has_gap_before', False))
        else:
            if b['low'] < a['low'] and b['low'] < c['low'] \
               and b['high'] < a['high'] and b['high'] < c['high']:
                return (i, b.get('has_gap_before', False))
    return None


def _first_feature_fractal_incremental(bis, start_i, end_i, seg_dir):
    """增量构建特征序列, 在第一个特征分型被"锁定"时立即返回。

    锁定条件: 出现特征分型(a,b,c)后, 再追加一个新的标准元素(即 c 之后
    已有一个不被包含的元素 d)。此时 c 不会再被向后合并改变, 分型 b 的
    左右高低关系已固定, 与"先把整段展开再找首个分型"语义等价。
    返回 (b_first_bi_idx, b_has_gap_before) 或 None。
    """
    feat_dir = 'down' if seg_dir == 'up' else 'up'
    up = (seg_dir == 'up')
    std = []  # 每元素: [high, low, bi_idx, has_gap_before, first_bi_idx]
    for k in range(start_i, end_i + 1):
        b = bis[k]
        if b.direction != feat_dir:
            continue
        ch = b.high; cl = b.low
        if not std:
            std.append([ch, cl, k, False, k]); continue
        prev = std[-1]
        ph = prev[0]; pl = prev[1]
        contained = (ph >= ch and pl <= cl) or (ch >= ph and cl <= pl)
        if contained:
            if up:
                prev[0] = ph if ph > ch else ch
                prev[1] = pl if pl > cl else cl
            else:
                prev[0] = ph if ph < ch else ch
                prev[1] = pl if pl < cl else cl
            prev[2] = k
            continue
        std.append([ch, cl, k, gap_flag(cl, ch, ph, pl)] + [k])
        # 锁定检查: 需要至少4个已定型元素, 才能保证倒数第3个(候选分型b)
        # 的右邻c已被其后元素d终结、不会再被向后合并。
        if len(std) >= 4:
            a = std[-4]; bm = std[-3]; c = std[-2]
            if up:
                ok = (bm[0] > a[0] and bm[0] > c[0] and bm[1] > a[1] and bm[1] > c[1])
            else:
                ok = (bm[1] < a[1] and bm[1] < c[1] and bm[0] < a[0] and bm[0] < c[0])
            if ok:
                return (bm[4], bm[3])
    # 收尾: 末端无后继元素, 用最终 std 找首个内部分型(与原实现等价)
    for i in range(1, len(std) - 1):
        a = std[i-1]; bm = std[i]; c = std[i+1]
        if up:
            ok = (bm[0] > a[0] and bm[0] > c[0] and bm[1] > a[1] and bm[1] > c[1])
        else:
            ok = (bm[1] < a[1] and bm[1] < c[1] and bm[0] < a[0] and bm[0] < c[0])
        if ok:
            return (bm[4], bm[3])
    return None


def gap_flag(cl, ch, ph, pl):
    return (cl > ph) or (ch < pl)


def _seq_fractal_confirms_reversal(bis, start_i, end_i, cur_dir):
    fr = _first_feature_fractal_incremental(bis, start_i, end_i, cur_dir)
    if fr is None:
        return (False, None)
    feat_first_bi, has_gap = fr
    seg_end_bi = feat_first_bi - 1
    if seg_end_bi <= start_i:
        return (False, None)
    if has_gap:
        opp = 'down' if cur_dir == 'up' else 'up'
        fr2 = _first_feature_fractal_incremental(bis, feat_first_bi, end_i, opp)
        if fr2 is None:
            return (False, None)
    return (True, seg_end_bi)


def find_segs(bis: list) -> list:
    n = len(bis)
    if n < 3:
        return []
    base = bis[0].start.price
    look = min(3, n)
    net = bis[look - 1].end.price - base
    cur_dir = 'up' if net > 0 else 'down'
    segs = []
    i = 0
    while i < n - 2:
        confirmed, seg_end_bi = _seq_fractal_confirms_reversal(bis, i, n - 1, cur_dir)
        if confirmed and seg_end_bi > i:
            seg_bis = bis[i:seg_end_bi + 1]
            net_up = seg_bis[-1].end.price > seg_bis[0].start.price
            if (cur_dir == 'up') == net_up:
                segs.append(Seg(start=bis[i].start, end=seg_bis[-1].end, direction=cur_dir,
                                bis=seg_bis, high=max(b.high for b in seg_bis),
                                low=min(b.low for b in seg_bis), confirmed=True))
                i = seg_end_bi + 1
                cur_dir = 'down' if cur_dir == 'up' else 'up'
                continue
        alt = 'down' if cur_dir == 'up' else 'up'
        confirmed2, seg_end_bi2 = _seq_fractal_confirms_reversal(bis, i, n - 1, alt)
        if confirmed2 and seg_end_bi2 > i:
            seg_bis = bis[i:seg_end_bi2 + 1]
            net_up = seg_bis[-1].end.price > seg_bis[0].start.price
            if (alt == 'up') == net_up:
                segs.append(Seg(start=seg_bis[0].start, end=seg_bis[-1].end, direction=alt,
                                bis=seg_bis, high=max(b.high for b in seg_bis),
                                low=min(b.low for b in seg_bis), confirmed=True))
                i = seg_end_bi2 + 1
                cur_dir = 'down' if alt == 'up' else 'up'
                continue
        break
    if i < n - 1 and (n - i) >= 1:
        seg_bis = bis[i:]
        if len(seg_bis) >= 1:
            net_up = seg_bis[-1].end.price > seg_bis[0].start.price
            d = 'up' if net_up else 'down'
            segs.append(Seg(start=seg_bis[0].start, end=seg_bis[-1].end, direction=d,
                            bis=seg_bis, high=max(b.high for b in seg_bis),
                            low=min(b.low for b in seg_bis), confirmed=False))
    return segs


PIVOT_UPGRADE_SPAN_DAYS = 540


def find_pivots(bis: list, segs: Optional[list] = None) -> list:
    confirmed_segs = [s for s in (segs or []) if getattr(s, 'confirmed', True)]
    units = confirmed_segs if len(confirmed_segs) >= 3 else bis
    using_segs = units is confirmed_segs
    pivots = []; n = len(units)
    if n < 3:
        return pivots

    bi_pos = {id(b): k for k, b in enumerate(bis)}

    def _unit_bi_range(u):
        if hasattr(u, 'bis'):
            idxs = [bi_pos[id(b)] for b in u.bis if id(b) in bi_pos]
            return (min(idxs), max(idxs)) if idxs else (0, 0)
        k = bi_pos.get(id(u), 0)
        return (k, k)

    def _unit_bi_indices(unit_list):
        out = []
        for u in unit_list:
            a, b = _unit_bi_range(u)
            out.extend(range(a, b + 1))
        return sorted(set(out))

    def _unit_high_date(u):
        return u.start.date if u.start.price >= u.end.price else u.end.date

    def _unit_low_date(u):
        return u.start.date if u.start.price <= u.end.price else u.end.date

    max_ext = PIVOT_MAX_EXTEND_SEGS if PIVOT_MAX_EXTEND_SEGS else 10 ** 9

    i = 0
    while i <= n - 3:
        b1, b2, b3 = units[i], units[i+1], units[i+2]
        r1 = (b1.low, b1.high)
        r2 = (b2.low, b2.high)
        r3 = (b3.low, b3.high)
        zg = min(r1[1], r2[1], r3[1]); zd = max(r1[0], r2[0], r3[0])
        if zg > zd:
            direction = b1.direction; zg_orig, zd_orig = zg, zd; gg, dd = zg, zd
            highs = [r1[1], r2[1], r3[1]]; lows = [r1[0], r2[0], r3[0]]
            zg_bi = (b1, b2, b3)[highs.index(zg_orig)]
            zd_bi = (b1, b2, b3)[lows.index(zd_orig)]
            zg_d = _unit_high_date(zg_bi)
            zd_d = _unit_low_date(zd_bi)
            gg_d, dd_d = zg_d, zd_d
            zn_dir = direction
            gn_list = []; dn_list = []
            for bb in (b1, b2, b3):
                if bb.direction == zn_dir:
                    gn_list.append(max(bb.start.price, bb.end.price))
                    dn_list.append(min(bb.start.price, bb.end.price))
            piv_units = [i, i+1, i+2]; j = i + 3
            capped = False
            while j < n:
                if (len(piv_units) - 3) >= max_ext:
                    capped = True
                    break
                bj = units[j]; lo_j = bj.low; hi_j = bj.high
                if hi_j >= zd_orig and lo_j <= zg_orig:
                    if hi_j > gg:
                        gg = hi_j; gg_d = _unit_high_date(bj)
                    if lo_j < dd:
                        dd = lo_j; dd_d = _unit_low_date(bj)
                    if bj.direction == zn_dir:
                        gn_list.append(hi_j); dn_list.append(lo_j)
                    piv_units.append(j); j += 1
                else:
                    break
            g_val = min(gn_list) if gn_list else zg_orig
            d_val = max(dn_list) if dn_list else zd_orig
            piv_bis = _unit_bi_indices([units[k] for k in piv_units]) if using_segs else piv_units
            p_start = b1.start.date
            p_end = units[piv_units[-1]].end.date
            piv = Pivot(start_date=p_start, end_date=p_end,
                        zg=zg_orig, zd=zd_orig, gg=gg, dd=dd, bis=piv_bis, direction=direction,
                        zg_date=zg_d, zd_date=zd_d, gg_date=gg_d, dd_date=dd_d,
                        g=g_val, d=d_val, capped=capped)
            try:
                span_days = (pd.Timestamp(p_end) - pd.Timestamp(p_start)).days
            except Exception:
                span_days = 0
            if capped or span_days > PIVOT_UPGRADE_SPAN_DAYS:
                piv.upgraded_level = 'weekly'
            pivots.append(piv)
            i = piv_units[-1] + 1
        else:
            i += 1

    for k in range(1, len(pivots)):
        prev, cur = pivots[k-1], pivots[k]
        no_overlap = (cur.dd > prev.gg) or (cur.gg < prev.dd)
        if no_overlap:
            cur.state = 'new'
        else:
            cur.state = 'expand'

    for k in range(len(pivots) - 1):
        cur = pivots[k]
        nxt = pivots[k + 1]
        gap_start = cur.bis[-1] + 1
        gap_end = nxt.bis[0]
        gap_bis = bis[gap_start:gap_end] if gap_end > gap_start else []
        if len(gap_bis) >= 2:
            leave = gap_bis[:max(1, len(gap_bis) // 2)]
            pull = gap_bis[max(1, len(gap_bis) // 2):]
            leave_trend = len(leave) >= 3
            pull_trend = len(pull) >= 3
            if leave_trend and not pull_trend:
                cur.death_combo = 'trend+consol'
            elif leave_trend and pull_trend:
                cur.death_combo = 'trend+counter'
            else:
                cur.death_combo = 'consol+counter'
    return pivots


def classify_trend(pivots: list) -> str:
    if len(pivots) < 2:
        return 'consolidation'
    p1, p2 = pivots[-2], pivots[-1]
    if p2.dd > p1.gg:
        return 'up_trend'
    if p2.gg < p1.dd:
        return 'down_trend'
    if (p2.zg < p1.zd and p2.gg >= p1.dd) or (p2.zd > p1.zg and p2.dd <= p1.gg):
        return 'expanding'
    return 'consolidation'


def count_trend_pivots(pivots: list) -> int:
    if not pivots:
        return 0
    if len(pivots) == 1:
        return 1
    cnt = 1
    for k in range(len(pivots) - 1, 0, -1):
        p_prev, p_cur = pivots[k - 1], pivots[k]
        if p_cur.dd > p_prev.gg:
            cnt += 1
        elif p_cur.gg < p_prev.dd:
            cnt += 1
        else:
            break
    return cnt


def calc_macd(close: pd.Series, fast=12, slow=26, signal=9):
    ema_fast = close.ewm(span=fast, adjust=False).mean()
    ema_slow = close.ewm(span=slow, adjust=False).mean()
    dif = ema_fast - ema_slow
    dea = dif.ewm(span=signal, adjust=False).mean()
    macd_bar = 2 * (dif - dea)
    return dif, dea, macd_bar


def macd_area_between(start_date, end_date, bar_series, date_series, direction):
    mask = (date_series >= start_date) & (date_series <= end_date)
    vals = bar_series[mask]
    if len(vals) == 0:
        return 0.0
    if direction == 'up':
        return float(vals.clip(lower=0).sum())
    return float(vals.clip(upper=0).abs().sum())


def dif_extreme_in(start_date, end_date, dif_series, date_series, kind='peak'):
    mask = (date_series >= start_date) & (date_series <= end_date)
    vals = dif_series[mask]
    if len(vals) == 0:
        return 0.0
    return float(vals.max()) if kind == 'peak' else float(vals.min())


class ChanAnalyzer:
    DIVERGE_RATIO = 0.80
    PIVOT_TOLERANCE = 0.02
    MIN_BI_BARS = 4
    DIF_TOLERANCE = 0.01

    CFG = {
        'b1_allow_consol_diverge': True,
        'b3s3_first_pullback_only': False,
        'b2s2_anchor_to_first': False,
        'b2_macd_zero_pullback': False,
        'drop_upgraded_pivots': False,
        # L88-90 中阴阶段MACD精确运用: 中阴判定除BOLL收口外, 加入MACD特征
        #   'off'  = 维持原判定(仅BOLL收口+末笔未离开中枢)
        #   'and'  = 须同时满足"黄白线绕0轴缠绕"(更严格, 减少误判中阴而拦截的好买点)
        #   'or'   = 满足其一即算中阴(更宽松, 拦截更多)
        'zhongyin_macd_mode': 'or',   # 实测'or'最优: 累计收益+35pp(383%→418%), 胜率45.3%→46.8%
    }

    def __init__(self, df: pd.DataFrame):
        self.df_raw = df.reset_index(drop=True)
        self.close = self.df_raw['close']
        self.dif, self.dea, self.macd_bar = calc_macd(self.close)
        self.merged = merge_klines(self.df_raw)
        self.fractals = find_fractals(self.merged)
        self.bis = find_bis(self.fractals, min_k=self.MIN_BI_BARS)
        self._bi_index = {id(b): k for k, b in enumerate(self.bis)}
        self.segs = find_segs(self.bis)
        self.pivots_all = find_pivots(self.bis, self.segs)
        if self.CFG.get('drop_upgraded_pivots'):
            last_upg_idx = -1
            for k, p in enumerate(self.pivots_all):
                if p.upgraded_level:
                    last_upg_idx = k
            operative = [p for k, p in enumerate(self.pivots_all)
                         if k > last_upg_idx and not p.upgraded_level]
            self.pivots = operative
        else:
            self.pivots = self.pivots_all
        self.trend = classify_trend(self.pivots)

    @property
    def n_bis(self): return len(self.bis)
    @property
    def n_segs(self): return len(self.segs)
    @property
    def n_pivots(self): return len(self.pivots)
    @property
    def n_pivots_all(self): return len(self.pivots_all)
    @property
    def n_trend_pivots(self): return count_trend_pivots(self.pivots)
    @property
    def has_upgraded_pivot(self):
        return any(p.upgraded_level for p in self.pivots_all)

    @staticmethod
    def _ds(ts):
        ts = pd.Timestamp(ts)
        if ts.hour == 0 and ts.minute == 0:
            return ts.strftime('%Y-%m-%d')
        return ts.strftime('%Y-%m-%d %H:%M')

    def _validate_abc(self, direction: str):
        if self.n_pivots < 2:
            return None
        last_piv = self.pivots[-1]
        prev_piv = self.pivots[-2]
        ratio = len(prev_piv.bis) / max(len(last_piv.bis), 1)
        if not (1 / 3 <= ratio <= 3):
            return None
        a_start_idx = prev_piv.bis[0]
        a_end_idx = last_piv.bis[0]
        if a_end_idx <= a_start_idx:
            return None
        c_start_idx = last_piv.bis[-1] + 1
        if c_start_idx >= self.n_bis:
            return None
        return {'a_start_idx': a_start_idx, 'a_end_idx': a_end_idx,
                'b_pivot': last_piv, 'c_start_idx': c_start_idx}

    def _validate_abc_consol(self, direction: str):
        if self.n_pivots < 1:
            return None
        piv = self.pivots[-1]
        a_end_idx = piv.bis[0]
        if a_end_idx <= 0:
            return None
        c_start_idx = piv.bis[-1] + 1
        if c_start_idx >= self.n_bis:
            return None
        want = 'down' if direction == 'down' else 'up'
        a_start_idx = a_end_idx
        for k in range(a_end_idx - 1, -1, -1):
            a_start_idx = k
            if k >= 1 and self.bis[k].direction != want and self.bis[k-1].direction != want:
                a_start_idx = k + 1
                break
        if a_start_idx >= a_end_idx:
            return None
        return {'a_start_idx': a_start_idx, 'a_end_idx': a_end_idx,
                'b_pivot': piv, 'c_start_idx': c_start_idx}

    def _check_c_new_extreme(self, c_start_idx: int, direction: str):
        want = 'up' if direction == 'up' else 'down'
        c_bis = [self.bis[k] for k in range(c_start_idx, self.n_bis)
                 if self.bis[k].direction == want]
        if not c_bis:
            return False, None
        last_piv = self.pivots[-1] if self.pivots else None
        if direction == 'up':
            c_ext = max(b.end.price for b in c_bis)
            prior = (last_piv.gg if last_piv is not None
                     else max((b.high for b in self.bis[:c_start_idx]), default=0.0))
            return c_ext > prior * (1 - 0.001), c_ext
        else:
            c_ext = min(b.end.price for b in c_bis)
            prior = (last_piv.dd if last_piv is not None
                     else min((b.low for b in self.bis[:c_start_idx]), default=1e18))
            return c_ext < prior * (1 + 0.001), c_ext

    def _b_returns_to_zero(self, pivot) -> bool:
        dates = self.df_raw['date']
        mask = (dates >= pivot.start_date) & (dates <= pivot.end_date)
        dif_b = self.dif[mask]
        dea_b = self.dea[mask]
        if len(dif_b) == 0 or len(dea_b) == 0:
            return False
        def near_zero(x):
            if x.min() <= 0 <= x.max():
                return True
            return x.abs().min() < max(float(x.abs().max()), 1e-9) * 0.25
        return near_zero(dif_b) and near_zero(dea_b)

    def detect_double_pullback_to_zero(self, window: int = 40) -> bool:
        n = len(self.dif)
        if n < 10:
            return False
        dif = self.dif.iloc[-min(window, n):].reset_index(drop=True)
        dea = self.dea.iloc[-min(window, n):].reset_index(drop=True)
        hist = self.macd_bar.iloc[-min(window, n):].reset_index(drop=True)
        scale = max(float(dif.abs().max()), float(dea.abs().max()), 1e-9)
        near = scale * 0.25
        zero_pulls = [i for i in range(len(dif))
                      if abs(float(dif.iloc[i])) <= near and abs(float(dea.iloc[i])) <= near]
        if len(zero_pulls) < 2:
            return False
        def peak_between(left, right):
            vals = dif.iloc[left + 1:right]
            if len(vals) < 2:
                return None
            rel = int(vals.idxmax())
            return float(dif.iloc[rel]), float(hist.iloc[max(left + 1, rel - 3):rel + 1].clip(lower=0).sum())
        def trough_between(left, right):
            vals = dif.iloc[left + 1:right]
            if len(vals) < 2:
                return None
            rel = int(vals.idxmin())
            return float(dif.iloc[rel]), float(hist.iloc[max(left + 1, rel - 3):rel + 1].clip(upper=0).abs().sum())
        first_pull, second_pull = zero_pulls[-2], zero_pulls[-1]
        p1, p2 = peak_between(first_pull, second_pull), peak_between(second_pull, len(dif))
        if p1 and p2 and p1[0] > 0 and p2[0] > 0 and p2[0] < p1[0] and p2[1] <= p1[1]:
            return True
        t1, t2 = trough_between(first_pull, second_pull), trough_between(second_pull, len(dif))
        if t1 and t2 and t1[0] < 0 and t2[0] < 0 and t2[0] > t1[0] and t2[1] <= t1[1]:
            return True
        return False

    def divergence_strength_by_position(self) -> str:
        if len(self.dif) < 5:
            return 'strong_pullback'
        dif_now = float(self.dif.iloc[-1])
        dif_abs_max = float(self.dif.abs().max())
        if dif_abs_max <= 1e-9:
            return 'strong_pullback'
        if abs(dif_now) >= dif_abs_max * 0.85:
            return 'weak_pullback'
        return 'strong_pullback'

    def classify_post_divergence(self, direction: str) -> dict:
        if not self.pivots or self.n_bis < 2:
            return {'evolution': 'unknown', 'reason': '无中枢或笔不足'}
        last_piv = self.pivots[-1]
        after_idx = last_piv.bis[-1] + 1
        def first_reversal_seg(want_dir):
            for s in self.segs:
                if not getattr(s, 'confirmed', True) or s.direction != want_dir:
                    continue
                try:
                    first_idx = self._bi_index[id(s.bis[0])]
                except KeyError:
                    continue
                if first_idx >= after_idx:
                    return s
            return None
        if direction == 'down':
            rebound = first_reversal_seg('up')
            if rebound is None:
                rebound = None
                for b in self.bis[after_idx:]:
                    if b.direction == 'up':
                        rebound = b; break
            if rebound is None:
                return {'evolution': 'unknown', 'reason': '无反弹笔'}
            if rebound.high < last_piv.zd:
                return {'evolution': 'case1_extend',
                        'reason': f'反弹高{rebound.high:.3f}<最后中枢ZD{last_piv.zd:.3f} → 第29课情况①未回中枢(最弱,宜尽快撤)'}
            if rebound.high >= last_piv.zd:
                return {'evolution': 'case2_3_turn',
                        'reason': f'反弹回到中枢(高{rebound.high:.3f}≥ZD{last_piv.zd:.3f}) → 第29课情况②③转折(可持有等三买)'}
            return {'evolution': 'case1_extend',
                    'reason': f'反弹未回中枢(高{rebound.high:.3f}<ZD{last_piv.zd:.3f}) → 偏向中枢扩展'}
        else:
            pullback = first_reversal_seg('down')
            if pullback is None:
                pullback = None
                for b in self.bis[after_idx:]:
                    if b.direction == 'down':
                        pullback = b; break
            if pullback is None:
                return {'evolution': 'unknown', 'reason': '无回落笔'}
            if pullback.low > last_piv.zg:
                return {'evolution': 'case1_extend',
                        'reason': f'回落低{pullback.low:.3f}>最后中枢ZG{last_piv.zg:.3f} → 第29课情况①未回中枢(最弱)'}
            if pullback.low <= last_piv.zg:
                return {'evolution': 'case2_3_turn',
                        'reason': f'回落回到中枢(低{pullback.low:.3f}≤ZG{last_piv.zg:.3f}) → 第29课情况②③转折'}
            return {'evolution': 'case1_extend',
                    'reason': f'回落未回中枢 → 偏向中枢扩展'}

    def macd_wrap_zero(self, window: int = 15) -> bool:
        """L88-90: 中阴阶段的MACD特征 —— 黄白线(DIF/DEA)绕0轴缠绕。
        近window根K线中, DIF与DEA的绝对值大多压在历史摆幅的25%以内即视为缠绕。"""
        n = len(self.dif)
        if n < window + 5:
            return False
        dif = self.dif.iloc[-window:]
        dea = self.dea.iloc[-window:]
        scale = max(float(self.dif.abs().tail(120).max()),
                    float(self.dea.abs().tail(120).max()), 1e-9)
        near = scale * 0.25
        frac = float(((dif.abs() <= near) & (dea.abs() <= near)).mean())
        return frac >= 0.6

    def macd_clarity(self, window: int = 60) -> dict:
        """L50: 本级别MACD的"清晰度" —— 柱子面积幅度 + 黄白线分离度, 归一化打分。
        清晰度高的级别其背驰判定更可靠; 多级别联立时应优先采信清晰级别的MACD结论。"""
        n = len(self.dif)
        if n < 10:
            return {'score': 0.0, 'label': '数据不足'}
        w = min(window, n)
        dif = self.dif.iloc[-w:]
        dea = self.dea.iloc[-w:]
        hist = self.macd_bar.iloc[-w:]
        px = max(float(self.close.iloc[-1]), 1e-9)
        bar_amp = float(hist.abs().mean()) / px          # 柱子相对幅度
        sep     = float((dif - dea).abs().mean()) / px   # 黄白线分离度
        # 黄白线贴着0轴乱绕 → 不清晰
        wrap_penalty = 0.5 if self.macd_wrap_zero() else 1.0
        score = (bar_amp * 0.6 + sep * 0.4) * 1e3 * wrap_penalty
        label = '清晰' if score >= 1.0 else ('一般' if score >= 0.4 else '模糊(黄白线/柱子贴0轴)')
        return {'score': round(score, 3), 'label': label,
                'bar_amp': round(bar_amp * 1e3, 3), 'sep': round(sep * 1e3, 3)}

    def in_zhongyin(self) -> dict:
        n = len(self.close)
        if n < 20 or not self.pivots:
            return {'in_zhongyin': False, 'boll_squeeze': False, 'reason': '数据不足'}
        ma = self.close.rolling(20).mean()
        sd = self.close.rolling(20).std()
        if ma.iloc[-1] and ma.iloc[-1] > 0:
            width = float((4 * sd.iloc[-1]) / ma.iloc[-1])
        else:
            width = 0.0
        wseries = (4 * sd / ma).dropna().tail(60)
        squeeze = bool(len(wseries) >= 20 and width <= wseries.quantile(0.30))
        osc = self.zhongshu_oscillation_monitor()
        macd_wrap = self.macd_wrap_zero()
        dbl_pull = self.detect_double_pullback_to_zero()
        mode = self.CFG.get('zhongyin_macd_mode', 'off')
        if mode == 'and':
            in_zy = (not osc.get('alert', False)) and squeeze and macd_wrap
        elif mode == 'or':
            in_zy = (not osc.get('alert', False)) and (squeeze or macd_wrap)
        else:
            in_zy = (not osc.get('alert', False)) and squeeze
        reason = (f"BOLL带宽{width:.3f}{'(收口→中阴)' if squeeze else '(开口)'}; "
                  f"MACD黄白线{'绕0轴缠绕(L88-90中阴特征)' if macd_wrap else '已展开'}"
                  f"{'; 双回拉0轴(L89: 中阴结束转折预备)' if dbl_pull else ''}; {osc.get('reason','')}")
        return {'in_zhongyin': in_zy, 'boll_squeeze': squeeze,
                'macd_wrap_zero': macd_wrap, 'double_pullback_zero': dbl_pull,
                'reason': reason}

    def zhongshu_oscillation_monitor(self) -> dict:
        if not self.pivots or self.n_bis < 1:
            return {'alert': False, 'direction': '', 'reason': '无中枢'}
        last_piv = self.pivots[-1]
        cur = self.bis[-1]
        if cur.low > last_piv.zg:
            return {'alert': True, 'direction': 'up',
                    'reason': f'第92课: 末笔({cur.low:.3f}~{cur.high:.3f})已离开中枢上沿ZG{last_piv.zg:.3f} → 向上变盘预警'}
        if cur.high < last_piv.zd:
            return {'alert': True, 'direction': 'down',
                    'reason': f'第92课: 末笔({cur.low:.3f}~{cur.high:.3f})已离开中枢下沿ZD{last_piv.zd:.3f} → 向下变盘预警'}
        return {'alert': False, 'direction': '', 'reason': '末笔仍在中枢区间内, 中枢震荡延续'}

    def bottom_construction_state(self) -> str:
        has_b1 = self.detect_b1() is not None
        has_b3 = self.detect_b3() is not None
        has_s3 = self.detect_s3() is not None
        if has_s3:
            return 'failed'
        if has_b3:
            return 'completed'
        if has_b1:
            return 'constructing'
        return 'none'

    def _seg_index_range(self, seg):
        m = self._bi_index
        try:
            return m[id(seg.bis[0])], m[id(seg.bis[-1])]
        except KeyError:
            return None

    def _bi_exit_pullback_fallback(self, pivot, exit_dir: str, pull_dir: str):
        pe = pivot.bis[-1]
        leave = pull = None
        for k in range(pe + 1, self.n_bis):
            b = self.bis[k]
            if leave is None:
                if b.direction == exit_dir:
                    leave = b
                continue
            if b.direction == pull_dir:
                pull = b
                break
        if leave is None or pull is None:
            return None
        return leave, pull

    def _last_exit_pullback_segments(self, pivot, exit_dir: str, pull_dir: str):
        pivot_end = pivot.bis[-1]
        seq = []
        for s in (self.segs or []):
            if not getattr(s, 'confirmed', True):
                continue
            rng = self._seg_index_range(s)
            if rng is None:
                continue
            first_idx, last_idx = rng
            if last_idx < pivot_end:
                continue
            seq.append((s, first_idx, last_idx))
        first_only = self.CFG.get('b3s3_first_pullback_only')
        if first_only:
            for i in range(len(seq) - 1):
                leave, lf, _ = seq[i]
                pull = seq[i + 1][0]
                if leave.direction == exit_dir and pull.direction == pull_dir and lf >= pivot_end:
                    return leave, pull
        else:
            for i in range(len(seq) - 1, 0, -1):
                pull, _, _ = seq[i]
                leave, _, _ = seq[i - 1]
                if leave.direction == exit_dir and pull.direction == pull_dir:
                    return leave, pull
        return self._bi_exit_pullback_fallback(pivot, exit_dir, pull_dir)

    def assess_divergence(self, a_start, a_end, c_start, c_end, direction: str) -> DivergenceGrade:
        dates = self.df_raw['date']
        n_tp = self.n_trend_pivots
        is_trend_div = n_tp >= 2
        a_area = macd_area_between(a_start, a_end, self.macd_bar, dates, direction)
        c_area = macd_area_between(c_start, c_end, self.macd_bar, dates, direction)
        if a_area <= 1e-9:
            return DivergenceGrade('NONE', False, False, 0.0, a_area, c_area, 0.0, 0.0,
                                   direction, 'A段MACD面积为0,无可比基准',
                                   is_trend_divergence=is_trend_div, n_trend_pivots=n_tp)
        ratio = c_area / a_area
        area_ok = ratio < self.DIVERGE_RATIO
        TOL = self.DIF_TOLERANCE
        if direction == 'up':
            a_dif = dif_extreme_in(a_start, a_end, self.dif, dates, 'peak')
            c_dif = dif_extreme_in(c_start, c_end, self.dif, dates, 'peak')
            dif_ok = c_dif < a_dif * (1 - TOL)
        else:
            a_dif = dif_extreme_in(a_start, a_end, self.dif, dates, 'trough')
            c_dif = dif_extreme_in(c_start, c_end, self.dif, dates, 'trough')
            dif_ok = c_dif > a_dif * (1 - TOL)
        ext_label = 'DIF峰' if direction == 'up' else 'DIF谷'
        div_kind = '趋势背驰' if is_trend_div else '盘整背驰'
        if area_ok and dif_ok:
            grade = 'STRONG'
            reason = f'标准{div_kind}(面积+DIF均满足)|A面积:{a_area:.4f} C面积:{c_area:.4f}(比值{ratio:.1%}<{self.DIVERGE_RATIO:.0%})|{ext_label} A:{a_dif:.4f} C:{c_dif:.4f}|{n_tp}中枢'
        elif area_ok and not dif_ok:
            grade = 'WEAK'
            reason = f'{div_kind}信号(面积触发)|A面积:{a_area:.4f} C面积:{c_area:.4f}(比值{ratio:.1%}<{self.DIVERGE_RATIO:.0%})|{n_tp}中枢'
        elif dif_ok and not area_ok:
            grade = 'WEAK'
            reason = f'{div_kind}信号(DIF触发)|{ext_label} A:{a_dif:.4f} C:{c_dif:.4f}|{n_tp}中枢'
        else:
            grade = 'NONE'
            reason = f'无背驰(两判据均不满足)|面积比{ratio:.1%}>={self.DIVERGE_RATIO:.0%}|{ext_label} A:{a_dif:.4f} C:{c_dif:.4f}'
        return DivergenceGrade(grade, area_ok, dif_ok, ratio, a_area, c_area, a_dif, c_dif,
                               direction, reason, is_trend_divergence=is_trend_div, n_trend_pivots=n_tp)

    def _find_prev_b1(self) -> tuple:
        if not self.pivots:
            return (None, '')
        last_piv = self.pivots[-1]
        pivot_first_bi_idx = last_piv.bis[0]
        if pivot_first_bi_idx <= 0:
            return (None, '')
        downs = []
        for k in range(pivot_first_bi_idx - 1, -1, -1):
            b = self.bis[k]
            downs.append(b)
            if b.direction == 'up' and k - 1 >= 0 and self.bis[k-1].direction == 'down':
                if len(downs) >= 4:
                    break
        down_bis = [b for b in downs if b.direction == 'down']
        if not down_bis:
            return (None, '')
        b1 = min(down_bis, key=lambda b: b.end.price)
        return (b1.end.price, self._ds(b1.end.date))

    def _find_prev_b2(self) -> tuple:
        b1_price, b1_date_str = self._find_prev_b1()
        if b1_price is None or not self.pivots:
            return (None, '')
        b1_date = pd.Timestamp(b1_date_str)
        last_piv = self.pivots[-1]
        right_bi_idx = last_piv.bis[-1] + 1
        for b in self.bis[:right_bi_idx]:
            if b.direction != 'down':
                continue
            if b.end.date <= b1_date:
                continue
            if b.end.price > b1_price + 1e-9:
                return (b.end.price, self._ds(b.end.date))
        return (None, '')

    def _find_prev_s1(self) -> tuple:
        if not self.pivots:
            return (None, '')
        last_piv = self.pivots[-1]
        pivot_first_bi_idx = last_piv.bis[0]
        if pivot_first_bi_idx <= 0:
            return (None, '')
        ups = []
        for k in range(pivot_first_bi_idx - 1, -1, -1):
            b = self.bis[k]
            ups.append(b)
            if b.direction == 'down' and k - 1 >= 0 and self.bis[k-1].direction == 'up':
                if len(ups) >= 4:
                    break
        up_bis = [b for b in ups if b.direction == 'up']
        if not up_bis:
            return (None, '')
        s1 = max(up_bis, key=lambda b: b.end.price)
        return (s1.end.price, self._ds(s1.end.date))

    def _find_prev_s2(self) -> tuple:
        s1_price, s1_date_str = self._find_prev_s1()
        if s1_price is None or not self.pivots:
            return (None, '')
        s1_date = pd.Timestamp(s1_date_str)
        last_piv = self.pivots[-1]
        right_bi_idx = last_piv.bis[-1] + 1
        for b in self.bis[:right_bi_idx]:
            if b.direction != 'up':
                continue
            if b.end.date <= s1_date:
                continue
            if b.end.price < s1_price - 1e-9:
                return (b.end.price, self._ds(b.end.date))
        return (None, '')

    def detect_b1(self) -> Optional[Signal]:
        if self.n_bis < 5:
            return None
        cur = self.bis[-1]
        if cur.direction != 'down':
            return None
        dif_now = float(self.dif.iloc[-1])
        if dif_now >= 0:
            return None
        trend_ok = (self.n_pivots >= 2 and self.trend == 'down_trend')
        consol_ok = (self.CFG.get('b1_allow_consol_diverge') and self.n_pivots >= 1)
        if not (trend_ok or consol_ok):
            return None
        abc = self._validate_abc('down')
        if abc is None and consol_ok:
            abc = self._validate_abc_consol('down')
        if abc is None:
            return None
        c_ok, c_low = self._check_c_new_extreme(abc['c_start_idx'], 'down')
        if not c_ok:
            return None
        last_piv = self.pivots[-1]
        a_down = [self.bis[k] for k in range(abc['a_start_idx'], abc['a_end_idx'])
                  if self.bis[k].direction == 'down']
        c_down = [self.bis[k] for k in range(abc['c_start_idx'], self.n_bis)
                  if self.bis[k].direction == 'down']
        if not a_down or not c_down:
            return None
        dg = self.assess_divergence(a_down[0].start.date, a_down[-1].end.date,
                                    c_down[0].start.date, c_down[-1].end.date, 'down')
        if dg.grade == 'NONE':
            return None
        div_kind = '趋势底背驰' if dg.is_trend_divergence else '盘整底背驰(第27课)'
        b_zero = self._b_returns_to_zero(last_piv)
        dbl_pull = self.detect_double_pullback_to_zero()
        pos_strength = self.divergence_strength_by_position()
        post_evo = self.classify_post_divergence('down')
        a_low_bi = min(a_down, key=lambda b: b.end.price)
        c_low_bi = min(c_down, key=lambda b: b.end.price)
        return Signal(kind='B1', date=self.df_raw['date'].iloc[-1], price=float(self.close.iloc[-1]),
                      reason=f'{div_kind}一买|{self.n_pivots}中枢|ABC三段{"+B回0轴" if b_zero else ""}|{dg.reason}|DIF={dif_now:.4f}<0',
                      pivot_zg=last_piv.zg, pivot_zd=last_piv.zd, macd_ratio=dg.area_ratio, dif_value=dif_now,
                      n_pivots=self.n_pivots, trend=self.trend,
                      extras={'a_seg': (a_down[0].start.date, a_down[-1].end.date, a_down[0].start.price, a_down[-1].end.price),
                              'diverge_grade': dg.grade,
                              'a_low': float(a_low_bi.end.price),
                              'a_low_date': self._ds(a_low_bi.end.date),
                              'b1_price': float(cur.end.price),
                              'b1_date': self._ds(cur.end.date),
                              'macd_grade': dg.grade,
                              'macd_area_ratio': dg.area_ratio,
                              'dif_ok': dg.dif_ok,
                              'area_ok': dg.area_ok,
                              'b_returns_zero': b_zero,
                              'double_pullback': dbl_pull,
                              'pos_strength': pos_strength,
                              'post_evolution': post_evo['evolution'],
                              'c_new_low': c_low,
                              'c_new_low_date': self._ds(c_low_bi.end.date),
                              'n_trend_pivots': self.n_trend_pivots,
                              'price_date': self._ds(cur.end.date),
                              'pivot_zg_date': self._ds(last_piv.zg_date) if last_piv.zg_date else '',
                              'pivot_zd_date': self._ds(last_piv.zd_date) if last_piv.zd_date else '',
                              'pivot_start_date': self._ds(last_piv.start_date),
                              'pivot_end_date': self._ds(last_piv.end_date)},
                      diverge_grade=dg)

    def detect_b2(self) -> Optional[Signal]:
        if self.n_bis < 4:
            return None
        cur = self.bis[-1]
        if cur.direction != 'down':
            return None
        prev_downs = [b for b in self.bis[:-1] if b.direction == 'down']
        if not prev_downs:
            return None
        prev = prev_downs[-1]
        if not (cur.low >= prev.low and cur.end.price > prev.end.price):
            return None
        cur_price = float(self.close.iloc[-1])
        if cur_price < cur.end.price * (1 - self.PIVOT_TOLERANCE):
            return None
        if self.CFG.get('b2s2_anchor_to_first'):
            b1_anchor, _ = self._find_prev_b1()
            if b1_anchor is None:
                return None
            if cur.end.price < b1_anchor - 1e-9:
                return None
        dif_now = float(self.dif.iloc[-1])
        if self.CFG.get('b2_macd_zero_pullback'):
            look = self.dif.tail(12)
            crossed_up = bool((look > 0).any())
            if not crossed_up:
                return None
            if dif_now < -self.DIF_TOLERANCE:
                return None
        last_piv = self.pivots[-1] if self.pivots else None
        b1_price, b1_date = prev.end.price, self._ds(prev.end.date)
        return Signal(kind='B2', date=self.df_raw['date'].iloc[-1], price=cur_price,
                      reason=f'二买:一买后回踩不破|当前低{cur.end.price:.3f}>一买{b1_price:.3f}|二买不以背驰为成立条件(第21课)',
                      pivot_zg=None, pivot_zd=None,
                      macd_ratio=None, dif_value=dif_now, n_pivots=self.n_pivots, trend=self.trend,
                      extras={'prev_low': prev.end.price, 'cur_low': cur.end.price,
                              'cur_low_date': self._ds(cur.end.date),
                              'prev_low_date': self._ds(prev.end.date),
                              'b1_price': b1_price,
                              'b1_date': b1_date,
                              'price_date': self._ds(cur.end.date),
                              'context_pivot_zg': last_piv.zg if last_piv else None,
                              'context_pivot_zd': last_piv.zd if last_piv else None,
                              'context_pivot_zg_date': self._ds(last_piv.zg_date) if last_piv and last_piv.zg_date else '',
                              'context_pivot_zd_date': self._ds(last_piv.zd_date) if last_piv and last_piv.zd_date else '',
                              'context_pivot_start_date': self._ds(last_piv.start_date) if last_piv else '',
                              'context_pivot_end_date': self._ds(last_piv.end_date) if last_piv else ''},
                      diverge_grade=None)

    def detect_b3(self) -> Optional[Signal]:
        if self.n_bis < 5 or self.n_pivots < 1:
            return None
        late_trend_b3 = self.n_trend_pivots >= 2
        last_piv = self.pivots[-1]
        zg, zd = last_piv.zg, last_piv.zd
        piv_height = zg - zd
        cur = self.bis[-1]
        if cur.direction != 'up':
            return None
        pair = self._last_exit_pullback_segments(last_piv, 'up', 'down')
        if pair is None:
            return None
        exit_seg, pull_seg = pair
        if not (exit_seg.low <= zg * (1 + self.PIVOT_TOLERANCE) and exit_seg.high > zg):
            return None
        if pull_seg.low < zg * (1 - self.PIVOT_TOLERANCE):
            return None
        leaves_pivot = pull_seg.low >= zg
        cur_price = float(self.close.iloc[-1])
        if cur_price <= zg:
            return None
        ex_amp = exit_seg.high - exit_seg.low
        if piv_height > 0 and ex_amp < piv_height * 0.5:
            return None
        if len(last_piv.bis) < 3:
            return None
        confirm_txt = '回踩离枢确认(新中枢生成)' if leaves_pivot else '回踩贴ZG(容差内,新中枢待确认)'
        b1_price, b1_date = self._find_prev_b1()
        b2_price, b2_date = self._find_prev_b2()
        warn_txt = '|第二个以上同向中枢,实盘宜改用低级别一买' if late_trend_b3 else ''
        return Signal(kind='B3', date=self.df_raw['date'].iloc[-1], price=cur_price,
                      reason=f'标准三买|ZG={zg:.3f},ZD={zd:.3f}|线段离枢:{exit_seg.low:.3f}→{exit_seg.high:.3f}(幅度{ex_amp:.3f})|线段回试低{pull_seg.low:.3f}|{confirm_txt}|当前{cur_price:.3f}>ZG{warn_txt}',
                      pivot_zg=zg, pivot_zd=zd, macd_ratio=None, dif_value=float(self.dif.iloc[-1]),
                      n_pivots=self.n_pivots, trend=self.trend,
                      extras={'exit_seg': (exit_seg.low, exit_seg.high),
                              'pull_low': pull_seg.low,
                              'pull_low_date': self._ds(pull_seg.end.date),
                              'exit_start_date': self._ds(exit_seg.start.date),
                              'exit_end_date': self._ds(exit_seg.end.date),
                              'b1_price': b1_price,
                              'b1_date': b1_date,
                              'b2_price': b2_price,
                              'b2_date': b2_date,
                              'piv_bi_count': len(last_piv.bis), 'leaves_pivot': leaves_pivot,
                              'late_trend_b3': late_trend_b3,
                              'price_date': self._ds(self.df_raw['date'].iloc[-1]),
                              'pivot_zg_date': self._ds(last_piv.zg_date) if last_piv.zg_date else '',
                              'pivot_zd_date': self._ds(last_piv.zd_date) if last_piv.zd_date else '',
                              'pivot_start_date': self._ds(last_piv.start_date),
                              'pivot_end_date': self._ds(last_piv.end_date)},
                      diverge_grade=None)

    def detect_s1(self) -> Optional[Signal]:
        if self.n_bis < 5:
            return None
        cur = self.bis[-1]
        if cur.direction != 'up':
            return None
        trend_ok = (self.n_pivots >= 2 and self.trend == 'up_trend')
        consol_ok = (self.CFG.get('b1_allow_consol_diverge') and self.n_pivots >= 1)
        if not (trend_ok or consol_ok):
            return None
        abc = self._validate_abc('up')
        if abc is None and consol_ok:
            abc = self._validate_abc_consol('up')
        if abc is None:
            return None
        last_piv = self.pivots[-1]
        a_start_idx = abc['a_start_idx']; a_end_idx = abc['a_end_idx']
        if a_end_idx <= a_start_idx:
            return None
        a_up_bis = [self.bis[k] for k in range(a_start_idx, a_end_idx) if self.bis[k].direction == 'up']
        if not a_up_bis:
            return None
        a_high = max(b.end.price for b in a_up_bis)
        c_start_idx = last_piv.bis[-1] + 1
        c_up_bis = [self.bis[k] for k in range(c_start_idx, self.n_bis) if self.bis[k].direction == 'up']
        if not c_up_bis:
            return None
        c_high = max(b.end.price for b in c_up_bis)
        if c_high <= a_high:
            return None
        a_high_bi = max(a_up_bis, key=lambda b: b.end.price)
        dg = self.assess_divergence(a_up_bis[0].start.date, a_up_bis[-1].end.date,
                                    c_up_bis[0].start.date, c_up_bis[-1].end.date, 'up')
        if dg.grade == 'NONE':
            return None
        b_zero = self._b_returns_to_zero(last_piv)
        dbl_pull = self.detect_double_pullback_to_zero()
        pos_strength = self.divergence_strength_by_position()
        post_evo = self.classify_post_divergence('up')
        dif_now = float(self.dif.iloc[-1])
        c_high_bi = max(c_up_bis, key=lambda b: b.end.price)
        return Signal(kind='S1', date=self.df_raw['date'].iloc[-1], price=float(self.close.iloc[-1]),
                      reason=f'一卖|上涨趋势{self.n_pivots}中枢|价创新高C{c_high:.3f}>A段高{a_high:.3f}{"+B回0轴" if b_zero else ""}|{dg.reason}',
                      pivot_zg=last_piv.zg, pivot_zd=last_piv.zd, macd_ratio=dg.area_ratio, dif_value=dif_now,
                      n_pivots=self.n_pivots, trend=self.trend,
                      extras={'a_high': a_high, 'c_high': c_high, 'a_area': dg.a_area, 'c_area': dg.c_area,
                              'diverge_grade': dg.grade,
                              'a_high_date': self._ds(a_high_bi.end.date),
                              'b_returns_zero': b_zero,
                              'double_pullback': dbl_pull,
                              'pos_strength': pos_strength,
                              'post_evolution': post_evo['evolution'],
                              'c_high_date': self._ds(c_high_bi.end.date),
                              'price_date': self._ds(cur.end.date),
                              'pivot_zg_date': self._ds(last_piv.zg_date) if last_piv.zg_date else '',
                              'pivot_zd_date': self._ds(last_piv.zd_date) if last_piv.zd_date else '',
                              'pivot_start_date': self._ds(last_piv.start_date),
                              'pivot_end_date': self._ds(last_piv.end_date)},
                      diverge_grade=dg)

    def detect_s2(self) -> Optional[Signal]:
        if self.n_bis < 4:
            return None
        cur = self.bis[-1]
        if cur.direction != 'up':
            return None
        prev_ups = [b for b in self.bis[:-1] if b.direction == 'up']
        if not prev_ups:
            return None
        prev = prev_ups[-1]
        if cur.high < prev.high and cur.end.price < prev.end.price:
            cur_price = float(self.close.iloc[-1])
            if cur_price > cur.end.price * (1 + self.PIVOT_TOLERANCE):
                return None
            if self.CFG.get('b2s2_anchor_to_first'):
                s1_anchor, _ = self._find_prev_s1()
                if s1_anchor is None:
                    return None
                if cur.end.price > s1_anchor + 1e-9:
                    return None
            dif_now = float(self.dif.iloc[-1])
            last_piv = self.pivots[-1] if self.pivots else None
            s1_price, s1_date = prev.end.price, self._ds(prev.end.date)
            return Signal(kind='S2', date=self.df_raw['date'].iloc[-1], price=cur_price,
                          reason=f'二卖:一卖后反弹不破|当前高{cur.end.price:.3f}<一卖{s1_price:.3f}|二卖不以背驰为成立条件(第21课)',
                          pivot_zg=None, pivot_zd=None,
                          dif_value=dif_now, n_pivots=self.n_pivots, trend=self.trend,
                          extras={'prev_high': prev.end.price, 'cur_high': cur.end.price,
                                  'cur_high_date': self._ds(cur.end.date),
                                  'prev_high_date': self._ds(prev.end.date),
                                  's1_price': s1_price, 's1_date': s1_date,
                                  'price_date': self._ds(cur.end.date),
                                  'context_pivot_zg': last_piv.zg if last_piv else None,
                                  'context_pivot_zd': last_piv.zd if last_piv else None,
                                  'context_pivot_zg_date': self._ds(last_piv.zg_date) if last_piv and last_piv.zg_date else '',
                                  'context_pivot_zd_date': self._ds(last_piv.zd_date) if last_piv and last_piv.zd_date else '',
                                  'context_pivot_start_date': self._ds(last_piv.start_date) if last_piv else '',
                                  'context_pivot_end_date': self._ds(last_piv.end_date) if last_piv else ''},
                          diverge_grade=None)
        return None

    def detect_s3(self) -> Optional[Signal]:
        if self.n_bis < 5 or self.n_pivots < 1:
            return None
        last_piv = self.pivots[-1]
        zg, zd = last_piv.zg, last_piv.zd
        if len(self.bis) < 3:
            return None
        cur = self.bis[-1]
        if cur.direction != 'down':
            return None
        s1_price, s1_date = self._find_prev_s1()
        s2_price, s2_date = self._find_prev_s2()
        base_dates = {'price_date': self._ds(self.df_raw['date'].iloc[-1]),
                      's1_price': s1_price, 's1_date': s1_date,
                      's2_price': s2_price, 's2_date': s2_date,
                      'pivot_zg_date': self._ds(last_piv.zg_date) if last_piv.zg_date else '',
                      'pivot_zd_date': self._ds(last_piv.zd_date) if last_piv.zd_date else '',
                      'pivot_start_date': self._ds(last_piv.start_date),
                      'pivot_end_date': self._ds(last_piv.end_date)}
        pair = self._last_exit_pullback_segments(last_piv, 'down', 'up')
        if pair is None:
            return None
        exit_seg, pull_seg = pair
        if not (exit_seg.high >= zd * (1 - self.PIVOT_TOLERANCE) and exit_seg.low < zd):
            return None
        if pull_seg.high > zd * (1 + self.PIVOT_TOLERANCE):
            return None
        if float(self.close.iloc[-1]) >= zd:
            return None
        return Signal(kind='S3', date=self.df_raw['date'].iloc[-1], price=float(self.close.iloc[-1]),
                      reason=f'标准三卖|ZD={zd:.3f}|线段离枢低{exit_seg.low:.3f}<ZD|线段回抽高{pull_seg.high:.3f}未过ZD',
                      pivot_zg=zg, pivot_zd=zd, dif_value=float(self.dif.iloc[-1]),
                      n_pivots=self.n_pivots, trend=self.trend, extras=base_dates, diverge_grade=None)

    def get_signal(self) -> Optional[Signal]:
        for fn in [self.detect_s1, self.detect_s2, self.detect_s3]:
            sig = fn()
            if sig is not None:
                return sig
        for fn in [self.detect_b3, self.detect_b2, self.detect_b1]:
            sig = fn()
            if sig is not None:
                return sig
        return None

    def get_all_signals(self) -> list:
        all_sigs = []
        for fn in [self.detect_b1, self.detect_b2, self.detect_b3,
                   self.detect_s1, self.detect_s2, self.detect_s3]:
            sig = fn()
            if sig is not None:
                all_sigs.append(sig)
        return all_sigs

    def l36_segment_note(self) -> str:
        """L36 结合律: 走势分解的唯一性靠结合律保证 —— a+A+b+B+c 的划分中, 同一段
        K线不能既归前段又归后段。本引擎线段划分采用特征序列分型(标准缠论)处理
        包含关系, 分型一旦确认即锁定段的归属, 等价于结合律的程序化执行。
        该函数对当前末端给出"是否存在划分歧义"的提示。"""
        if len(self.segs) < 2:
            return '第36课: 线段不足2段, 无划分歧义问题'
        last = self.segs[-1]
        n_last = len(getattr(last, 'bis', []) or [])
        if n_last < 3:
            return (f'第36课: 末段仅{n_last}笔(<3), 末端划分尚未唯一确认 —— '
                    f'当下操作应按两种归属做完全分类预案, 等待特征序列分型锁定')
        return '第36课: 末段≥3笔且特征序列分型已锁定, 当前划分唯一, 无歧义'

    def diagnose(self) -> dict:
        cur_price = float(self.close.iloc[-1]) if len(self.close) else 0.0
        last = self.bis[-1] if self.bis else None
        out = {k: [] for k in ('B1', 'B2', 'B3', 'S1', 'S2', 'S3')}
        def add(k, ok, msg):
            out[k].append(('✓' if ok else '✗') + ' ' + msg)
        add('B1', self.n_bis >= 5, f'笔数 {self.n_bis} >= 5')
        add('B1', self.n_pivots >= 2, f'中枢数 {self.n_pivots} >= 2')
        add('B1', self.trend == 'down_trend', f'当前走势={self.trend}, B1要求下跌趋势')
        add('B1', bool(last and last.direction == 'down'), f'最后一笔方向={last.direction if last else ""}, B1要求向下')
        add('B1', float(self.dif.iloc[-1]) < 0 if len(self.dif) else False, f'DIF={float(self.dif.iloc[-1]):.4f}, B1要求DIF<0')
        abc_down = self._validate_abc('down')
        add('B1', abc_down is not None, 'A/B/C三段背驰结构成立')
        if abc_down is not None:
            c_ok, c_low = self._check_c_new_extreme(abc_down['c_start_idx'], 'down')
            add('B1', c_ok, f'C段创新低{"" if c_low is None else f"({c_low:.3f})"}')
        add('B2', self.n_bis >= 4, f'笔数 {self.n_bis} >= 4')
        add('B2', bool(last and last.direction == 'down'), f'最后一笔方向={last.direction if last else ""}, B2要求回踩向下')
        prev_downs = [b for b in self.bis[:-1] if b.direction == 'down'] if last else []
        add('B2', bool(prev_downs), '存在同一轮前一个下跌低点作为一买锚')
        if last and prev_downs:
            prev = prev_downs[-1]
            add('B2', last.low >= prev.low and last.end.price > prev.end.price,
                f'回踩不创新低: 本次低{last.end.price:.3f} > 一买/前低{prev.end.price:.3f}')
            add('B2', cur_price >= last.end.price * (1 - self.PIVOT_TOLERANCE),
                f'现价{cur_price:.3f}未跌破B2回踩锚{last.end.price:.3f}; 跌破则二买失效')
        add('B3', self.n_pivots >= 1, f'中枢数 {self.n_pivots} >= 1')
        add('B3', bool(last and last.direction == 'up'), f'最后一笔方向={last.direction if last else ""}, B3要求向上确认')
        if self.pivots:
            p = self.pivots[-1]
            pair = self._last_exit_pullback_segments(p, 'up', 'down')
            add('B3', pair is not None, '存在已确认线段级别的向上离枢 + 向下回试')
            if pair is not None:
                exit_seg, pull_seg = pair
                add('B3', exit_seg.low <= p.zg * (1 + self.PIVOT_TOLERANCE) and exit_seg.high > p.zg,
                    f'离枢线段突破ZG: {exit_seg.low:.3f}~{exit_seg.high:.3f}, ZG={p.zg:.3f}')
                add('B3', pull_seg.low >= p.zg * (1 - self.PIVOT_TOLERANCE),
                    f'回试低点{pull_seg.low:.3f}不破ZG={p.zg:.3f}')
                add('B3', cur_price > p.zg, f'现价{cur_price:.3f}站上ZG={p.zg:.3f}')
        add('S1', self.n_bis >= 5, f'笔数 {self.n_bis} >= 5')
        add('S1', self.n_pivots >= 2, f'中枢数 {self.n_pivots} >= 2')
        add('S1', self.trend == 'up_trend', f'当前走势={self.trend}, S1要求上涨趋势')
        add('S1', bool(last and last.direction == 'up'), f'最后一笔方向={last.direction if last else ""}, S1要求向上')
        abc_up = self._validate_abc('up')
        add('S1', abc_up is not None, 'A/B/C三段顶背驰结构成立')
        if abc_up is not None:
            c_ok, c_high = self._check_c_new_extreme(abc_up['c_start_idx'], 'up')
            add('S1', c_ok, f'C段创新高{"" if c_high is None else f"({c_high:.3f})"}')
        add('S2', self.n_bis >= 4, f'笔数 {self.n_bis} >= 4')
        add('S2', bool(last and last.direction == 'up'), f'最后一笔方向={last.direction if last else ""}, S2要求反弹向上')
        prev_ups = [b for b in self.bis[:-1] if b.direction == 'up'] if last else []
        add('S2', bool(prev_ups), '存在同一轮前一个上涨高点作为一卖锚')
        if last and prev_ups:
            prev = prev_ups[-1]
            add('S2', last.high < prev.high and last.end.price < prev.end.price,
                f'反弹不创新高: 本次高{last.end.price:.3f} < 一卖/前高{prev.end.price:.3f}')
            add('S2', cur_price <= last.end.price * (1 + self.PIVOT_TOLERANCE),
                f'现价{cur_price:.3f}未重新升破S2反弹锚{last.end.price:.3f}; 升破则二卖失效')
        add('S3', self.n_pivots >= 1, f'中枢数 {self.n_pivots} >= 1')
        add('S3', bool(last and last.direction == 'down'), f'最后一笔方向={last.direction if last else ""}, S3要求向下确认')
        if self.pivots:
            p = self.pivots[-1]
            pair = self._last_exit_pullback_segments(p, 'down', 'up')
            add('S3', pair is not None, '存在已确认线段级别的向下离枢 + 向上回抽')
            if pair is not None:
                exit_seg, pull_seg = pair
                add('S3', exit_seg.high >= p.zd * (1 - self.PIVOT_TOLERANCE) and exit_seg.low < p.zd,
                    f'离枢线段跌破ZD: {exit_seg.low:.3f}~{exit_seg.high:.3f}, ZD={p.zd:.3f}')
                add('S3', pull_seg.high <= p.zd * (1 + self.PIVOT_TOLERANCE),
                    f'回抽高点{pull_seg.high:.3f}不破ZD={p.zd:.3f}')
                add('S3', cur_price < p.zd, f'现价{cur_price:.3f}跌破ZD={p.zd:.3f}')
        return out


class SameLevelDecomposition:
    def __init__(self, analyzer: 'ChanAnalyzer'):
        self.an = analyzer
        self.segs = analyzer.segs

    def current_phase(self) -> dict:
        if len(self.segs) < 2:
            return {'seg_dir': '', 'stage': 'unknown', 'action': 'WATCH',
                    'reason': '线段不足, 无法做同级别分解'}
        last = self.segs[-1]
        prev = self.segs[-2]
        seg_dir = last.direction
        if seg_dir == 'up':
            stage = 'up_run'
            prev_up = None
            for s in reversed(self.segs[:-1]):
                if s.direction == 'up':
                    prev_up = s; break
            if prev_up is None:
                return {'seg_dir': seg_dir, 'stage': stage, 'action': 'HOLD',
                        'reason': '向上段运作中(无前向上段可比), 持有'}
            if last.high <= prev_up.high:
                return {'seg_dir': seg_dir, 'stage': stage, 'action': 'SELL',
                        'reason': f'向上段不创新高({last.high:.3f}≤前高{prev_up.high:.3f}) → 先卖(第38课)'}
            dg = self.an.assess_divergence(prev_up.start.date, prev_up.end.date,
                                           last.start.date, last.end.date, 'up')
            if dg.grade != 'NONE':
                return {'seg_dir': seg_dir, 'stage': stage, 'action': 'SELL',
                        'reason': f'向上段创新高但盘整背驰({dg.grade}) → 卖(第38课)'}
            return {'seg_dir': seg_dir, 'stage': stage, 'action': 'HOLD',
                    'reason': '向上段创新高且不背驰 → 持有(第38课)'}
        else:
            stage = 'down_run'
            prev_down = None
            for s in reversed(self.segs[:-1]):
                if s.direction == 'down':
                    prev_down = s; break
            if prev_down is None:
                return {'seg_dir': seg_dir, 'stage': stage, 'action': 'WATCH',
                        'reason': '向下段运作中(无前向下段可比), 观望等买点'}
            if last.low >= prev_down.low:
                return {'seg_dir': seg_dir, 'stage': stage, 'action': 'BUY',
                        'reason': f'向下段不创新低({last.low:.3f}≥前低{prev_down.low:.3f}) → 买入(第38课)'}
            dg = self.an.assess_divergence(prev_down.start.date, prev_down.end.date,
                                           last.start.date, last.end.date, 'down')
            if dg.grade != 'NONE':
                return {'seg_dir': seg_dir, 'stage': stage, 'action': 'BUY',
                        'reason': f'向下段创新低但盘整背驰({dg.grade}) → 买入(第38课)'}
            return {'seg_dir': seg_dir, 'stage': stage, 'action': 'WATCH',
                    'reason': '向下段创新低且不背驰 → 观望等下跌背驰(第38课)'}


class BottomTracker:
    def __init__(self):
        self.state = 'none'

    def update(self, analyzer: 'ChanAnalyzer') -> str:
        snap = analyzer.bottom_construction_state()
        if self.state in ('none', 'failed', 'completed'):
            if snap == 'constructing':
                self.state = 'constructing'
            elif snap == 'completed':
                self.state = 'completed'
            elif snap == 'failed':
                self.state = 'failed'
            else:
                self.state = 'none'
        elif self.state == 'constructing':
            if snap == 'completed':
                self.state = 'completed'
            elif snap == 'failed':
                self.state = 'failed'
        return self.state