Chan/research/step17_fast_bsp3_validate.py

"""Step 17：快速三买的多品种验证与参数敏感性。

Step 16 在 BTC 15m 上把三买 PF 从 0.50 提到 1.40，但只有 138 笔，t=1.77 不够。
本步扩样本、查参数，回答它是不是真信号。

  关一  多品种   BTC / ETH / SOL
  关二  参数面   回抽窗口 × 容差
  关三  组合     浅回抽 + 大级别同向 是否叠加增益
  关四  分年     是否依赖某段行情
"""
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.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

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

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

SYMBOLS = ["BTC/USDT:USDT", "ETH/USDT:USDT", "SOL/USDT:USDT"]
SL, TP, MAXB = 1.5, 3.0, 48

_cache: dict = {}


def prep(symbol: str, ltf: str, htf1: str, htf2: str):
    """小级别引擎 + 大级别分型时间线，缓存复用。"""
    key = (symbol, ltf)
    if key in _cache:
        return _cache[key]
    df_l = fetch_ohlcv(symbol, ltf, 10**9)
    if df_l is None or len(df_l) < 3000:
        _cache[key] = None
        return None
    chan_l = TF_DF(df_l, 1, ltf)
    cdf_l = chan_l.dataframe
    zones = build_htf_zones(df_l, ltf)
    tls = {}
    for tf, pref in ((htf1, "h1"), (htf2, "h2")):
        try:
            df_h = fetch_ohlcv(symbol, tf, 10**9)
        except Exception:
            continue
        if df_h is None or len(df_h) < 500:
            continue
        chan_h = TF_DF(df_h, 1, tf)
        s = signals_to_frame(extract_fx_signals(chan_h, chan_h.dataframe))
        tls[pref] = htf_fx_timeline(s, chan_h.dataframe)
    _cache[key] = (cdf_l, zones, tls)
    return _cache[key]


def signals_for(symbol, ltf, htf1, htf2, pb_win=30, tol=0.003):
    got = prep(symbol, ltf, htf1, htf2)
    if got is None:
        return None, None
    cdf_l, zones, tls = got
    sig = find_fast_bsp3(cdf_l, zones, pullback_win=pb_win, tol=tol)
    if sig.empty:
        return cdf_l, None
    for pref, tl in tls.items():
        sig = attach_htf_context(sig, cdf_l, tl, pref)
    return cdf_l, sig


def E(g):
    return list(zip(g["entry_idx"].astype(int), g["direction"].astype(int)))


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--ltf", default="15m")
    ap.add_argument("--htf1", default="1h")
    ap.add_argument("--htf2", default="4h")
    args = ap.parse_args()

    print("########## 关一：多品种（快速三类，全部信号）##########")
    rows, pool = [], []
    for sym in SYMBOLS:
        cdf, sig = signals_for(sym, args.ltf, args.htf1, args.htf2)
        if sig is None or sig.empty:
            continue
        tr = run_trades(cdf, E(sig), SL, TP, MAXB)
        if tr.empty:
            continue
        s = summarize_trades(tr, f"{sym.split('/')[0]:>4} {args.ltf}")
        s["滞后中位"] = f"{sig['lag'].median():.0f}"
        rows.append(s)
        tr = tr.copy()
        tr["symbol"] = sym.split("/")[0]
        tr["date"] = cdf["date"].to_numpy()[tr["entry_idx"].to_numpy()]
        # 把过滤标记挂到交易上，供后续组合分析
        m = sig.set_index("entry_idx")
        tr["h1_agree"] = tr["entry_idx"].map(m["h1_agree"]) if "h1_agree" in m else np.nan
        tr["h2_agree"] = tr["entry_idx"].map(m["h2_agree"]) if "h2_agree" in m else np.nan
        tr["depth"] = tr["entry_idx"].map(m["depth"])
        pool.append(tr)
    print(pd.DataFrame(rows).to_string(index=False))

    if not pool:
        print("无样本")
        return
    allt = pd.concat(pool, ignore_index=True)
    r = allt["ret"].to_numpy()
    win, loss = r[r > 0], r[r <= 0]
    sd = r.std(ddof=1)
    print(f"\n  合并 {len(r)} 笔：胜率 {(r > 0).mean() * 100:.1f}%  "
          f"均收益 {r.mean() * 100:+.3f}%  PF {win.sum() / abs(loss.sum()):.2f}  "
          f"t值 {r.mean() / (sd / np.sqrt(len(r))):+.2f}")

    print("\n########## 关三：过滤组合（合并全部品种）##########")
    a1 = allt["h1_agree"] == 1
    dq = allt["depth"].quantile(0.66)
    shallow = allt["depth"] >= dq
    rows = []
    for m, nm in [
        (pd.Series(True, index=allt.index), "全部"),
        (a1, f"+{args.htf1}同向"),
        (shallow, "+浅回抽"),
        (a1 & shallow, f"+{args.htf1}同向 且 浅回抽"),
        (~a1, f"{args.htf1}反向(对照)"),
    ]:
        g = allt[m]
        if len(g) < 20:
            continue
        rows.append(summarize_trades(g, nm))
    print(pd.DataFrame(rows).to_string(index=False))

    print("\n########## 关四：分年（合并全部品种）##########")
    allt["year"] = pd.to_datetime(allt["date"]).dt.year
    rows = [summarize_trades(g, str(y)) for y, g in allt.groupby("year") if len(g) >= 20]
    print(pd.DataFrame(rows).to_string(index=False))
    pos = sum(1 for _, g in allt.groupby("year") if len(g) >= 20 and g["ret"].mean() > 0)
    tot = sum(1 for _, g in allt.groupby("year") if len(g) >= 20)
    print(f"\n  盈利年份 {pos}/{tot}")

    print("\n########## 关二：参数敏感性（回抽窗口 × 容差，合并全部品种）##########")
    rows = []
    for pb in (15, 30, 50):
        for tol in (0.001, 0.003, 0.006):
            sub = []
            for sym in SYMBOLS:
                cdf, sig = signals_for(sym, args.ltf, args.htf1, args.htf2, pb, tol)
                if sig is None or sig.empty:
                    continue
                t = run_trades(cdf, E(sig), SL, TP, MAXB)
                if not t.empty:
                    sub.append(t)
            if sub:
                rows.append(summarize_trades(pd.concat(sub, ignore_index=True),
                                             f"窗口{pb} 容差{tol}"))
    grid = pd.DataFrame(rows)
    print(grid.to_string(index=False))
    pos = sum(1 for v in grid["均收益"] if v.startswith("+"))
    print(f"\n  正收益参数点 {pos}/{len(grid)}")

    print("\n########## 多空拆分（合并全部品种）##########")
    rows = [summarize_trades(allt[allt.direction == d], nm)
            for d, nm in ((1, "三买做多"), (-1, "三卖做空"))
            if len(allt[allt.direction == d]) >= 20]
    print(pd.DataFrame(rows).to_string(index=False))

    out = Path(__file__).parent / "out" / "step17_fast_bsp3_trades.csv"
    out.parent.mkdir(exist_ok=True)
    allt.to_csv(out, index=False)
    print(f"\n明细已写入 {out}")


if __name__ == "__main__":
    main()