Chan/research/step26_zone_sequence.py

"""Step 26：用中枢序列检验缠论的趋势/盘整定义。

缠论原义：一个中枢是盘整，两个以上同向且不重叠的中枢才构成趋势。
对应到三买，同样是「突破中枢」，但所处位置的含义完全不同：

  分型后第 1 个中枢突破  —— 转折刚确立，后面空间最大
  第 2 个中枢且不重叠推高 —— 趋势延续段
  第 2 个中枢但与前重叠  —— 原地震荡，突破多是假的

本步给每个中枢标上「分型段内序号」和「与前一中枢的位置关系」，
分组比较三买质量，看理论能不能被数据支持。
"""
from __future__ import annotations

import argparse
import os
import sys
import warnings
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path

import numpy as np
import pandas as pd

warnings.filterwarnings("ignore")
for v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"):
    os.environ.setdefault(v, "1")

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
sys.path.insert(0, str(HERE.parent))
pd.set_option("display.width", 300)

SL, TP, MAXB = 1.5, 3.0, 48
FEE, SLIP = 0.0004, 0.0001


def run_one(task: tuple) -> dict | None:
    import warnings as _w
    _w.filterwarnings("ignore")
    sys.path.insert(0, str(HERE))
    sys.path.insert(0, str(HERE.parent))

    from chanlun import TF_DF
    from lib.breakout import run_trades
    from lib.data import fetch_ohlcv
    from lib.fast_bsp3 import find_fast_bsp3
    from lib.fx_signal import extract_fx_signals, signals_to_frame
    from lib.nested_bsp import attach_htf_context, htf_fx_timeline
    from lib.nested_level import build_htf_zones

    sym, ltf, h1, h2 = task
    try:
        df_l = fetch_ohlcv(f"{sym}/USDT:USDT", ltf, 10**9)
        if df_l is None or len(df_l) < 3000:
            return None
        chan_l = TF_DF(df_l, 1, ltf)
        cdf = chan_l.dataframe
        zones = build_htf_zones(cdf, ltf, chan=chan_l).reset_index(drop=True)
        if zones.empty:
            return None
        sig = find_fast_bsp3(cdf, zones)
        if sig.empty or len(sig) < 10:
            return None

        # 中枢之间的位置关系：完全在上方/下方=推进，否则=重叠
        z = zones.copy()
        pg, pd_ = z["zg"].shift(), z["zd"].shift()
        z["z_above"] = z["zd"] > pg
        z["z_below"] = z["zg"] < pd_
        z["z_overlap"] = ~(z["z_above"] | z["z_below"]) & pg.notna()
        z["z_first"] = pg.isna()
        z["zone_i"] = np.arange(len(z))

        # 大级别分型：h1 的结构既要用来切分型段，也要挂到信号上，只算一次
        timelines = {}
        for tf, pref in ((h1, "h1"), (h2, "h2")):
            df_h = fetch_ohlcv(f"{sym}/USDT:USDT", tf, 10**9)
            if df_h is None or len(df_h) < 300:
                continue
            chan_h = TF_DF(df_h, 1, tf)
            s = signals_to_frame(extract_fx_signals(chan_h, chan_h.dataframe))
            timelines[pref] = htf_fx_timeline(s, chan_h.dataframe)
        if "h1" not in timelines:
            return None

        k = np.searchsorted(timelines["h1"]["confirm_ts"].to_numpy(),
                            z["available_ts"].to_numpy(), side="right") - 1
        z["fx_id"] = k
        z["seq_in_fx"] = z.groupby("fx_id").cumcount() + 1

        sig = sig.merge(
            z[["zone_i", "z_above", "z_below", "z_overlap", "z_first", "seq_in_fx"]],
            on="zone_i", how="left")

        for pref, tl in timelines.items():
            sig = attach_htf_context(sig, cdf, tl, pref)

        entries = list(zip(sig["entry_idx"].astype(int), sig["direction"].astype(int)))
        tr = run_trades(cdf, entries, SL, TP, MAXB, fee=0.0, entry_delay=1)
        if tr.empty:
            return None
        m = sig.drop_duplicates("entry_idx").set_index("entry_idx")
        tr["symbol"], tr["ltf"] = sym, ltf
        tr["date"] = cdf["date"].to_numpy()[tr["entry_idx"].to_numpy()]
        for c in ("h1_agree", "h2_agree", "seq_in_fx", "z_above", "z_below",
                  "z_overlap", "z_first", "width_pct", "depth"):
            tr[c] = tr["entry_idx"].map(m[c]) if c in m.columns else np.nan
        return {"task": f"{sym} {ltf}", "trades": tr}
    except Exception as e:
        return {"task": f"{sym} {ltf}", "error": repr(e)[:200]}


def desc(r: np.ndarray, label: str, minn: int = 20) -> dict:
    if len(r) < minn:
        return {}
    w, o = r[r > 0], r[r <= 0]
    sd = r.std(ddof=1)
    return {
        "分组": label, "笔数": len(r),
        "胜率": f"{(r > 0).mean() * 100:.1f}%",
        "均收益": f"{r.mean() * 100:+.3f}%",
        "中位": f"{np.median(r) * 100:+.3f}%",
        "PF": f"{w.sum() / abs(o.sum()):.2f}" if len(o) else "inf",
        "偏度": f"{pd.Series(r).skew():.2f}",
        "t值": f"{r.mean() / (sd / np.sqrt(len(r))):+.2f}",
    }


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--pairs", default="5m:15m:1h,15m:1h:4h,30m:2h:4h")
    ap.add_argument("--symbols", default="BTC,ETH,SOL")
    ap.add_argument("--workers", type=int, default=6)
    ap.add_argument("--reuse", action="store_true")
    ap.add_argument("--tag", default="near", help="不同级别对分开存盘，避免互相覆盖")
    args = ap.parse_args()

    cache = HERE / "out" / f"step26_zoneseq_{args.tag}.csv"
    if args.reuse and cache.exists():
        allt = pd.read_csv(cache, parse_dates=["date"])
        print(f"[复用] {len(allt)} 笔\n")
    else:
        tasks = [(s, *tuple(p.split(":")))
                 for p in args.pairs.split(",") if p
                 for s in args.symbols.split(",")]
        print(f"[中枢序列] {len(tasks)} 个任务\n", flush=True)
        res = []
        with ProcessPoolExecutor(max_workers=args.workers) as ex:
            futs = {ex.submit(run_one, t): t for t in tasks}
            for i, f in enumerate(as_completed(futs), 1):
                r = f.result()
                if r is None or "error" in (r or {}):
                    print(f"  [{i}] 跳过 {(r or {}).get('error', '')}", flush=True)
                    continue
                res.append(r)
                print(f"  [{i}/{len(tasks)}] {r['task']} — {len(r['trades'])} 笔", flush=True)
        if not res:
            return
        allt = pd.concat([r["trades"] for r in res], ignore_index=True)
        allt.to_csv(cache, index=False)

    allt["date"] = pd.to_datetime(allt["date"])
    allt["ret_net"] = allt["gross"] - FEE - SLIP
    a = allt[allt["h1_agree"] == 1].copy()
    # 顺着信号方向推进才算趋势：三买要新中枢在上方，三卖要在下方
    a["trend_push"] = np.where(a["direction"] == 1, a["z_above"], a["z_below"]).astype(bool)

    print("=" * 115)
    print("########## 1. 分型段内第几个中枢（你的核心命题）##########")
    for tf, g in a.groupby("ltf"):
        rows = [desc(g[g.seq_in_fx == 1]["ret_net"].to_numpy(), f"{tf} 第1个中枢"),
                desc(g[g.seq_in_fx == 2]["ret_net"].to_numpy(), f"{tf} 第2个中枢"),
                desc(g[g.seq_in_fx >= 3]["ret_net"].to_numpy(), f"{tf} 第3个+")]
        rows = [r for r in rows if r]
        if rows:
            print(pd.DataFrame(rows).to_string(index=False))

    print("\n########## 2. 与前一中枢的位置关系（趋势 vs 盘整）##########")
    for tf, g in a.groupby("ltf"):
        rows = [desc(g[g.trend_push]["ret_net"].to_numpy(), f"{tf} 顺向推进(趋势)"),
                desc(g[g.z_overlap == True]["ret_net"].to_numpy(), f"{tf} 与前重叠(盘整)"),
                desc(g[(~g.trend_push) & (g.z_overlap != True)]["ret_net"].to_numpy(),
                     f"{tf} 逆向推进")]
        rows = [r for r in rows if r]
        if rows:
            print(pd.DataFrame(rows).to_string(index=False))
    print("  缠论预期：顺向推进 > 重叠。若成立，重叠中枢的三买应当直接放弃。")

    print("\n########## 3. 二维交叉：中枢序号 × 位置关系（合并全级别）##########")
    rows = []
    for seq, sl in ((1, "第1个"), (2, "第2个"), (99, "第3个+")):
        g = a[a.seq_in_fx == seq] if seq < 99 else a[a.seq_in_fx >= 3]
        rows.append(desc(g[g.trend_push]["ret_net"].to_numpy(), f"{sl}+顺向推进"))
        rows.append(desc(g[g.z_overlap == True]["ret_net"].to_numpy(), f"{sl}+重叠"))
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 4. 若只做「顺向推进」，各级别提升多少 ##########")
    rows = []
    for tf, g in a.groupby("ltf"):
        rows.append(desc(g["ret_net"].to_numpy(), f"{tf} 现状(全要)"))
        rows.append(desc(g[g.trend_push]["ret_net"].to_numpy(), f"{tf} 仅顺向推进"))
        rows.append(desc(g[g.trend_push | g.z_first]["ret_net"].to_numpy(),
                         f"{tf} 顺向推进+首个中枢"))
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 5. 组合资金曲线对比 ##########")
    def build(mode):
        parts = []
        for tf in ("5m", "15m", "30m"):
            g = a[a.ltf == tf]
            if g.empty:
                continue
            if tf == "5m":
                q = g["risk_pct"].quantile(0.67)
                g = g[(g["h2_agree"] == 1) & (g["risk_pct"] > q)]
            if mode == "push":
                g = g[g.trend_push]
            elif mode == "push_first":
                g = g[g.trend_push | g.z_first]
            parts.append(g)
        return pd.concat(parts) if parts else pd.DataFrame()

    for mode, name in (("all", "现状(全要)"), ("push", "仅顺向推进"),
                       ("push_first", "顺向推进+首个中枢")):
        g = build(mode).sort_values("date")
        if len(g) < 30:
            continue
        r = g["ret_net"].to_numpy()
        lev = np.clip(0.01 / np.clip(g["risk_pct"].to_numpy(), 0.002, None), 0, 20)
        pnl = r * lev
        eq = np.cumprod(1 + pnl)
        yrs = (g["date"].max() - g["date"].min()).days / 365.25
        dd = (1 - eq / np.maximum.accumulate(eq)).max()
        print(f"  {name:>18}: n={len(r):>4} 年{len(r) / yrs:>3.0f}笔  "
              f"年化 {(eq[-1] ** (1 / yrs) - 1) * 100:+6.1f}%  回撤 {dd * 100:4.1f}%  "
              f"Sharpe {pnl.mean() / pnl.std(ddof=1) * np.sqrt(len(pnl) / yrs):4.2f}  "
              f"中位 {np.median(r) * 100:+.3f}%")


if __name__ == "__main__":
    main()