"""快速三类买卖点（引擎内称第四类，B4/S4）。

原本长在 `research/lib/fast_bsp3.py`，现移入引擎作为唯一实现，研究脚本改为转发导入，
这样回测口径与 web 图表永远一致。

命名说明：缠论原文里没有「第四类买卖点」，但 B4/S4 也不只是「B3/S3 提前几根」——
两者的选样口径不同，统计性质符号相反，故单独立类。形态条件是实时可判的：

    中枢已成 -> 收盘突破 zg -> 收盘未跌回中枢 -> 重新上行

最后一步发生的当根就能下单，滞后约 2 根，而引擎 B3/S3 要等 pullback_bi.sure_time，
滞后 9~10 根。但差别不止滞后：

  判据   引擎用笔端点事后判（回拉笔低点 >= zg），本函数用收盘价实时判
  方向   引擎由离开笔方向决定，本函数由收盘从哪一侧突破决定
  口径   627 个中枢里引擎发 625 个信号（几乎不筛），本函数只认 212 个（34%）；
         被拒的多数是「中枢确认时价格早已离开、此后再没回来」的历史区间

step30 同条件对拍（同一套 pure 笔中枢、同一组过滤器、同样的 1.5/3.0/48 出场）：

                 原始 PF   +大级别同向   +同向+阶梯   胜率    t值
    引擎 B3/S3     0.66      0.66         0.71      27.4%  -18.76
    本函数 B4/S4   1.59      1.85         2.26      47.1%  +10.09

同一组过滤器对 B4 有效、对 B3 无效，滞后差解释不了这一点（入场后移 1~4 根只是
PF 3.18->2.53 的平滑衰减）。且 SL1.5/TP3.0 下随机入场胜率约 33%，引擎那 27.4%
低于随机——它选中的是一批系统性反向的样本，不是「晚了所以差」。

require_touch 控制是否强求回抽碰到中枢边界。step32/33 的实测表明强求反而更差：
这等于排除掉「突破后一去不回头」的强势段，而那正是缠论里最强的趋势形态。
故默认 False（笔数 +21%、滞后 -1 根、PF 2.26→2.37）。

判据本身（收盘越过前一根极值）没有独立预测力：裸用 15 万笔样本 PF 0.95，
把中枢换成「近20根高点」这类伪阻力位后 PF 0.90。alpha 全部来自中枢结构，
判据只负责在这个已知价位上确认动能恢复。它也不能用于识别笔端点——
入场前的回抽极值命中笔端点±2根的比例 19.3%，低于随机基准 22%。

全部判定只使用当根及之前的数据，无未来函数。
"""
from __future__ import annotations

import numpy as np
import pandas as pd

from chanlun.core.ChanEnum import Chan_FX_TYPE

# 中枢的「可用时刻」取第几笔的确认时间。
#   -1  = 最后一笔（历史默认）。中枢每吸收一根K线，最后一笔就可能后移，
#         于是 available_ts 跟着漂——这是右边缘重画的根因（见 HANDOFF §5.41）。
#    2  = 第三笔。三笔重叠即中枢成立，此后不再变，理论上更早也更稳。
# ⛔ step47 已判定 bis[2] 作废（毛 R 0.933 → −0.000，见 HANDOFF §5.42）。
# 开关保留只为可复现那次 A/B，**不要改默认值**。
import os as _os

AVAIL_BI_INDEX = -1


def _resolve_avail_bi(avail_bi: int | None) -> int:
    """优先级：显式入参 > 环境变量 CHAN_AVAIL_BI > 模块默认。

    环境变量在调用时读取而非 import 时——ProcessPoolExecutor 在 fork 启动方式下
    子进程会继承已 import 的模块，import 时读就固化成父进程的值了。
    """
    if avail_bi is not None:
        return avail_bi
    return int(_os.environ.get("CHAN_AVAIL_BI", AVAIL_BI_INDEX))


def timestamps_ms(src: pd.DataFrame) -> np.ndarray:
    """取毫秒时间戳。研究侧的 df 自带 timestamp，web 侧的不一定，故按 date 回退。

    回退写法不能用 `date.astype("int64") // 10**6`：该值单位取决于列精度，
    对毫秒精度的列会把时间戳砸平。
    """
    if "timestamp" in src.columns:
        return src["timestamp"].to_numpy()
    d = pd.to_datetime(src["date"])
    if getattr(d.dt, "tz", None) is None:
        d = d.dt.tz_localize("UTC")
    return (d.dt.tz_convert("UTC").dt.tz_localize(None)
             .astype("datetime64[ms]").astype("int64").to_numpy())


def ensure_timestamp(df: pd.DataFrame) -> pd.DataFrame:
    """保证 df 带 timestamp 列，缺失时补一份副本，不改动调用方的对象。"""
    if "timestamp" in df.columns:
        return df
    out = df.copy()
    out["timestamp"] = timestamps_ms(out)
    return out


def build_htf_zones(df_htf: pd.DataFrame, tf: str, chan=None, avail_bi: int | None = None) -> pd.DataFrame:
    """算 pure 笔中枢，返回带生效时间的区间表。

    available_ts —— 该中枢最早可被使用的时间戳（其确认时刻）。
    传入已构建好的 chan 可避免重复跑一遍 pipeline（大数据集上省一半时间）。
    """
    if chan is None:
        from chanlun import TF_DF

        chan = TF_DF(df_htf, 1, tf)
    zs_list = chan.cal_bi_zs_list_pure(chan.bi_list)
    # 用引擎自己的 dataframe 对齐，避免调用方传入的 df 与引擎内部行数不一致
    src = chan.dataframe if getattr(chan, "dataframe", None) is not None else df_htf
    return zones_from_zs_list(zs_list, src, avail_bi=avail_bi)


def zones_from_zs_list(zs_list, src: pd.DataFrame, avail_bi: int | None = None) -> pd.DataFrame:
    """把已算好的 pure 笔中枢转成区间表。

    调用方手工跑过 cal_bi_zs_list_pure 时走这里，免得再算一遍（web 的 analyze_chan
    就是这种用法）。
    """
    ts_of = dict(zip(src["date"].dt.strftime("%Y-%m-%d %H:%M:%S"), timestamps_ms(src)))
    # 中枢可用时刻取第几笔。在循环外解析一次，别让每个中枢都去读一遍环境变量。
    i = _resolve_avail_bi(avail_bi)

    rows = []
    for zs in zs_list:
        bis = getattr(zs, "bi_list", [])
        if not bis:
            continue
        # 笔数不够时退回最后一笔（i=2 需要至少 3 笔才成立）
        key_bi = bis[i] if (i == -1 or len(bis) > i) else bis[-1]
        sure_key = str(getattr(key_bi, "sure_time", "") or "")
        end_key = str(getattr(key_bi, "end_time", "") or "")
        avail = ts_of.get(sure_key) or ts_of.get(end_key)
        if avail is None:
            continue
        start_key = str(bis[0].start_time)
        rows.append({
            "zg": float(zs.zg), "zd": float(zs.zd),
            "gg": float(getattr(zs, "gg", zs.zg)), "dd": float(getattr(zs, "dd", zs.zd)),
            "start_ts": ts_of.get(start_key, avail),
            "available_ts": int(avail),
        })
    out = pd.DataFrame(rows)
    return out.sort_values("available_ts").reset_index(drop=True) if not out.empty else out


def add_zone_ladder(zones: pd.DataFrame) -> pd.DataFrame:
    """标注每个中枢相对前一个中枢是否同向推进（缠论「趋势 vs 盘整」）。

    z_above / z_below 分别对应向上、向下推进。买信号要求 z_above、卖信号要求 z_below
    时，30m/2h 的 PF 从 2.72 升到 3.41。
    """
    out = zones.copy()
    if out.empty:
        out["z_above"], out["z_below"] = pd.Series(dtype=bool), pd.Series(dtype=bool)
        return out
    pg, pdn = out["zg"].shift(), out["zd"].shift()
    out["z_above"] = (out["zd"] > pg).fillna(False)
    out["z_below"] = (out["zg"] < pdn).fillna(False)
    return out


def htf_fx_timeline(chan_htf, df_htf: pd.DataFrame | None = None) -> pd.DataFrame:
    """把大级别分型压成一条按确认时间排序的时间线。

    confirm_ts 是该分型最早可被使用的时刻。timestamp 是K线开盘时刻，而分型要等这根K线
    收盘才算数，所以整体后移一个大级别周期；否则小级别会提前一整根大级别K线拿到信号。

    只取方向与确认时刻——同向过滤用不到背驰强度，省掉 MACD 面积计算。
    """
    src = chan_htf.dataframe if getattr(chan_htf, "dataframe", None) is not None else df_htf
    if src is None or len(src) == 0:
        return pd.DataFrame(columns=["confirm_ts", "fx_ts", "direction", "price"])
    ts = timestamps_ms(src)
    idx_of = {t: i for i, t in enumerate(src["date"].dt.strftime("%Y-%m-%d %H:%M:%S"))}
    period = int(np.median(np.diff(ts))) if len(ts) > 1 else 0

    rows = []
    for klc in getattr(chan_htf, "klc_list", []):
        if klc.fx not in (Chan_FX_TYPE.TOP, Chan_FX_TYPE.BOTTOM):
            continue
        if klc.next is None or klc.next.end_klu is None:
            continue
        e_key, c_key = str(klc.end_time), str(klc.next.end_klu.time)
        if e_key not in idx_of or c_key not in idx_of:
            continue
        fx_idx, confirm_idx = idx_of[e_key], idx_of[c_key]
        if confirm_idx <= fx_idx:
            continue
        d = 1 if klc.fx == Chan_FX_TYPE.BOTTOM else -1
        rows.append({
            "confirm_ts": int(ts[confirm_idx]) + period,
            "fx_ts": int(ts[fx_idx]),
            "direction": d,
            "price": float(klc.low if d == 1 else klc.high),
        })
    out = pd.DataFrame(rows)
    if out.empty:
        return pd.DataFrame(columns=["confirm_ts", "fx_ts", "direction", "price"])
    return out.sort_values("confirm_ts").reset_index(drop=True)


def attach_htf_agree(sig: pd.DataFrame, df_ltf: pd.DataFrame, tl: pd.DataFrame) -> pd.DataFrame:
    """给每个小级别信号挂上「入场时刻之前最近的大级别分型」是否同向。

    产出 htf_dir（+1 底 / -1 顶）与 htf_agree（1 同向 / 0 反向 / NaN 无可用分型）。
    """
    out = sig.copy()
    if sig.empty or tl.empty:
        out["htf_dir"] = np.nan
        out["htf_agree"] = np.nan
        return out

    ts_ltf = timestamps_ms(df_ltf)
    entry_ts = ts_ltf[out["entry_idx"].to_numpy().astype(int)]
    k = np.searchsorted(tl["confirm_ts"].to_numpy(), entry_ts, side="right") - 1
    valid = k >= 0
    k_safe = np.clip(k, 0, len(tl) - 1)

    fx_dir = tl["direction"].to_numpy()[k_safe].astype(float)
    out["htf_dir"] = np.where(valid, fx_dir, np.nan)
    out["htf_agree"] = np.where(
        valid, (fx_dir == out["direction"].to_numpy()).astype(float), np.nan
    )
    return out


def attach_zone_ladder(sig: pd.DataFrame, zones: pd.DataFrame) -> pd.DataFrame:
    """按信号方向取该中枢的阶梯标记：买看 z_above、卖看 z_below。

    zone_i 是 find_fast_bsp3 内 enumerate 出的位置序号，故用 iloc 定位。
    """
    out = sig.copy()
    if sig.empty:
        out["ladder_ok"] = pd.Series(dtype=bool)
        return out
    z = zones if "z_above" in zones.columns else add_zone_ladder(zones)
    above = z["z_above"].to_numpy()
    below = z["z_below"].to_numpy()
    zi = out["zone_i"].to_numpy().astype(int)
    ok = np.where(out["direction"].to_numpy() == 1, above[zi], below[zi])
    out["ladder_ok"] = ok.astype(bool)
    return out


def find_fast_bsp3(
    df: pd.DataFrame,
    zones: pd.DataFrame,
    scan: int = 200,
    pullback_win: int = 30,
    tol: float = -1.0,
    max_per_zone: int = 1,
    diag: dict | None = None,
    require_touch: bool = False,
) -> pd.DataFrame:
    """扫描每个中枢，找突破后回抽不回中枢的入场点。

    zones 需含 zg / zd / available_ts，且 available_ts 已是可用时刻。
    max_per_zone > 1 时，同一中枢在首次入场后继续往后找二次、三次突破回抽，
    用来检验「趋势里同一中枢反复给机会」是否值得做。

    tol 是「算作回抽中」的边界容差。require_touch=False 时它不再是入场门槛，
    仅决定哪些K线被视为回抽中而跳过转强判定，故 tol 越大入场越晚。
    默认负值 = 完全禁用该跳过，突破后每根都检查转强，滞后压到 2.2 根。
    step34 实测滞后与收益严格单调：9.9根 PF1.88 / 5.8根 2.37 / 2.2根 2.86。

    返回列：
      entry_idx   实时可下单的K线
      direction   +1 三买 / -1 三卖
      bo_idx      突破根
      pb_idx      回抽极值根
      lag         entry_idx - bo_idx
      depth       回抽深度（相对中枢边界，负值表示曾插入中枢）
      occ         这是该中枢的第几次入场
    """
    if zones.empty:
        return pd.DataFrame()

    ts = df["timestamp"].to_numpy()
    close = df["close"].to_numpy(dtype=float)
    high = df["high"].to_numpy(dtype=float)
    low = df["low"].to_numpy(dtype=float)
    n = len(df)
    rows = []

    def note(key: str) -> None:
        if diag is not None:
            diag[key] = diag.get(key, 0) + 1

    for zone_i, (_, z) in enumerate(zones.iterrows()):
        note("中枢总数")
        zg, zd = float(z["zg"]), float(z["zd"])
        if zg <= zd:
            note("×无效中枢")
            continue
        start = int(np.searchsorted(ts, z["available_ts"], side="left"))
        if start >= n - 2:
            note("×中枢太靠后")
            continue
        # 允许多次入场时按比例放宽扫描窗口，否则后几次机会会被窗口截断
        scan_end = min(start + scan * max_per_zone, n)
        cursor = start

        for occ in range(1, max_per_zone + 1):
            if cursor >= n - 2:
                break
            # 第一步：找突破。要求突破前确实待在中枢内，避免把远处的价格当突破。
            was_inside = False
            bo_idx, d = None, 0
            for j in range(cursor, scan_end):
                c = close[j]
                if zd <= c <= zg:
                    was_inside = True
                    continue
                if not was_inside:
                    continue
                bo_idx, d = j, (1 if c > zg else -1)
                break
            if bo_idx is None:
                if occ == 1:
                    note("×窗口内未突破")
                break

            edge = zg if d == 1 else zd
            # 第二步：突破后监控回抽，回抽不跌回中枢且重新顺势 -> 入场
            touched = False
            pb_idx = None
            pb_ext = None
            entry_idx = None
            fell_back = False
            for j in range(bo_idx + 1, min(bo_idx + pullback_win + 1, n)):
                # 收盘跌回中枢 -> 突破失效
                if zd <= close[j] <= zg:
                    fell_back = True
                    break
                # 回抽触及边界附近（允许 tol 的毛刺）
                near = (low[j] <= edge * (1 + tol)) if d == 1 else (high[j] >= edge * (1 - tol))
                if near:
                    touched = True
                    ext = low[j] if d == 1 else high[j]
                    if pb_ext is None or ((ext < pb_ext) if d == 1 else (ext > pb_ext)):
                        pb_ext, pb_idx = ext, j
                    continue
                # 回抽后重新顺势：收盘创出前一根之上（三买）/ 之下（三卖）
                # require_touch=False 时不强求回抽碰到中枢边界，
                # 这样「突破后一去不回头」的强势段也能收进来。
                if (touched and pb_idx is not None) or not require_touch:
                    go = close[j] > high[j - 1] if d == 1 else close[j] < low[j - 1]
                    if go:
                        entry_idx = j
                        break
            if entry_idx is None:
                if occ == 1:
                    note("×突破后跌回中枢" if fell_back
                         else "×回抽未触及边界" if not touched
                         else "×触及边界但未转强")
                # 这次突破没走成，从突破点之后继续找下一次
                cursor = bo_idx + 1
                continue
            if pb_ext is None:
                # 未触及边界就转强（require_touch=False）：取突破至入场间的实际极值
                seg = slice(bo_idx + 1, entry_idx + 1)
                pb_ext = float(low[seg].min() if d == 1 else high[seg].max())
                pb_idx = int((low[seg].argmin() if d == 1 else high[seg].argmax())
                             + bo_idx + 1)
            if occ == 1:
                note("√成交")

            # 回抽深度：>0 表示未插入中枢，越大表示回抽越浅
            depth = (pb_ext - zg) / zg if d == 1 else (zd - pb_ext) / zd
            rows.append({
                "entry_idx": entry_idx, "direction": d,
                "bo_idx": bo_idx, "pb_idx": pb_idx,
                "lag": entry_idx - bo_idx,
                "depth": depth,
                "zg": zg, "zd": zd,
                "width_pct": (zg - zd) / close[bo_idx],
                "occ": occ,
                "zone_i": zone_i,
            })
            cursor = entry_idx + 1

    out = pd.DataFrame(rows)
    if out.empty:
        return out
    # 相邻中枢可能突破到同一根K线，同一时刻只能有一个仓位，保留最早成型的那个
    return (out.sort_values(["entry_idx", "occ", "zone_i"])
               .drop_duplicates("entry_idx", keep="first")
               .reset_index(drop=True))