Chan/research/step12_zs_breakout.py

"""Step 12：中枢突破深挖 —— Step 11 里唯一正期望的信号。

Step 11 中 4h 中枢突破 21 笔 PF 1.90，但样本太小无法定论。
本步把样本扩到多周期（15m/1h/4h，最长 2524 天），并区分：

  A 突破即入场
  B 突破 + 回抽不回中枢（三类买卖点的本质，但用突破确认而非等笔）
  C 假突破反向（突破后重回中枢 -> 反向做）

同时统计假突破率，这决定了 B 相对 A 的价值。
"""
from __future__ import annotations

import argparse
import sys
import warnings
from pathlib import Path

import numpy as np
import pandas as pd

warnings.filterwarnings("ignore")
sys.path.insert(0, str(Path(__file__).resolve().parent))

from lib.breakout import run_trades, summarize_trades
from lib.data import fetch_ohlcv
from lib.nested_level import build_htf_zones

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from chanlun import TF_DF

pd.set_option("display.width", 260)


def collect_breakouts(
    df: pd.DataFrame, zones: pd.DataFrame, scan: int = 300, pullback: int = 20
) -> pd.DataFrame:
    """找出每个中枢的首次有效突破，并记录后续是否假突破 / 是否回抽确认。"""
    if zones.empty:
        return pd.DataFrame()
    ts = df["timestamp"].to_numpy()
    close = df["close"].to_numpy(dtype=float)
    low = df["low"].to_numpy(dtype=float)
    high = df["high"].to_numpy(dtype=float)
    n = len(df)
    rows = []

    for _, z in zones.iterrows():
        zg, zd = float(z["zg"]), float(z["zd"])
        width = zg - zd
        if width <= 0:
            continue
        start = int(np.searchsorted(ts, z["available_ts"], side="left"))
        was_inside = False
        bo_idx = None
        d = 0
        for j in range(start, min(start + scan, n)):
            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:
            continue

        # 突破后是否重回中枢（假突破）
        fake_idx = None
        for j in range(bo_idx + 1, min(bo_idx + pullback + 1, n)):
            if zd <= close[j] <= zg:
                fake_idx = j
                break

        # 回抽确认：回踩到边界附近但收盘未回中枢，随后再度顺势
        edge = zg if d == 1 else zd
        conf_idx = None
        touched = False
        for j in range(bo_idx + 1, min(bo_idx + pullback + 1, n)):
            near = (low[j] <= edge * 1.002) if d == 1 else (high[j] >= edge * 0.998)
            inside = zd <= close[j] <= zg
            if inside:
                break
            if near:
                touched = True
                continue
            if touched and ((close[j] > close[j - 1]) if d == 1 else (close[j] < close[j - 1])):
                conf_idx = j
                break

        rows.append({
            "zs_start_ts": z["start_ts"], "zg": zg, "zd": zd,
            "width_pct": width / close[bo_idx],
            "bo_idx": bo_idx, "dir": d,
            "is_fake": fake_idx is not None,
            "conf_idx": conf_idx,
        })
    return pd.DataFrame(rows)


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--symbol", default="BTC/USDT:USDT")
    ap.add_argument("--tfs", default="15m,1h,4h")
    args = ap.parse_args()

    all_rows = []
    for tf in args.tfs.split(","):
        df = fetch_ohlcv(args.symbol, tf, 10**9)
        chan = TF_DF(df, 1, tf)
        cdf = chan.dataframe
        zones = build_htf_zones(df, tf)
        bo = collect_breakouts(cdf, zones)
        if bo.empty:
            continue

        days = (cdf["date"].iloc[-1] - cdf["date"].iloc[0]).days
        print(f"\n{'=' * 92}")
        print(f"周期 {tf}  K线 {len(cdf)}  跨度 {days} 天  中枢 {len(zones)}  "
              f"突破 {len(bo)}  假突破率 {bo['is_fake'].mean() * 100:.1f}%")
        print(f"  回抽确认成功 {bo['conf_idx'].notna().sum()} / {len(bo)}")

        entries_a = list(zip(bo["bo_idx"].astype(int), bo["dir"].astype(int)))
        real = bo[~bo["is_fake"]]
        entries_real = list(zip(real["bo_idx"].astype(int), real["dir"].astype(int)))
        cf = bo.dropna(subset=["conf_idx"])
        entries_b = list(zip(cf["conf_idx"].astype(int), cf["dir"].astype(int)))
        fake = bo[bo["is_fake"]]
        entries_c = list(zip(fake["bo_idx"].astype(int), -fake["dir"].astype(int)))

        rows = [
            summarize_trades(run_trades(cdf, entries_a, 1.5, 3.0, 48), "A 突破即入"),
            summarize_trades(run_trades(cdf, entries_b, 1.5, 3.0, 48), "B 回抽确认"),
            summarize_trades(run_trades(cdf, entries_c, 1.5, 3.0, 48), "C 假突破反向"),
            summarize_trades(run_trades(cdf, entries_real, 1.5, 3.0, 48), "D 事后真突破*"),
        ]
        print(pd.DataFrame(rows).to_string(index=False))
        print("  * D 用了事后信息，仅作上界参考，不可交易。")

        for _, r in bo.iterrows():
            all_rows.append({**r.to_dict(), "tf": tf})

        # 中枢宽度分层：窄中枢突破是否更有效
        bo2 = bo.copy()
        bo2["w_bin"] = pd.qcut(bo2["width_pct"], 3, labels=["窄", "中", "宽"], duplicates="drop")
        rows = []
        for name, g in bo2.groupby("w_bin", observed=True):
            e = list(zip(g["bo_idx"].astype(int), g["dir"].astype(int)))
            if len(e) >= 20:
                rows.append(summarize_trades(run_trades(cdf, e, 1.5, 3.0, 48), f"宽度-{name}"))
        if rows:
            print(pd.DataFrame(rows).to_string(index=False))

        # 多空拆分
        rows = []
        for d, nm in [(1, "向上突破"), (-1, "向下突破")]:
            e = [(i, dd) for i, dd in entries_a if dd == d]
            if len(e) >= 20:
                rows.append(summarize_trades(run_trades(cdf, e, 1.5, 3.0, 48), nm))
        if rows:
            print(pd.DataFrame(rows).to_string(index=False))

    if all_rows:
        out = Path(__file__).parent / "out" / "step12_zs_breakouts.csv"
        out.parent.mkdir(exist_ok=True)
        pd.DataFrame(all_rows).to_csv(out, index=False)
        print(f"\n明细已写入 {out}")


if __name__ == "__main__":
    main()