"""Step 19：区间套在不同级别对上的一致性。 15m/1h+4h 只有 57 笔核心样本，无法定论。但区间套是级别无关的结构， 把整套上移一级（1h 小级别有 2524 天，是 15m 的两倍）既能扩样本， 又能验证它究竟是普适结构还是只在某一个级别对上凑巧成立。 级别对： 5m / 15m + 1h 15m / 1h + 4h （step16~17 已测） 1h / 4h + 1d """ 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"] DEFAULT_PAIRS = "5m:15m:1h,15m:1h:4h,1h:4h:1d" SL, TP, MAXB = 1.5, 3.0, 48 def E(g): return list(zip(g["entry_idx"].astype(int), g["direction"].astype(int))) # 同一 (品种, 周期) 的 pipeline 在多个级别对之间反复出现，缓存后能省掉大部分耗时 _engine_cache: dict = {} _timeline_cache: dict = {} def _engine(symbol: str, tf: str, min_rows: int): key = (symbol, tf) if key not in _engine_cache: try: df = fetch_ohlcv(symbol, tf, 10**9) except Exception: df = None if df is None or len(df) < min_rows: _engine_cache[key] = None else: _engine_cache[key] = TF_DF(df, 1, tf) return _engine_cache[key] def _timeline(symbol: str, tf: str): key = (symbol, tf) if key not in _timeline_cache: chan = _engine(symbol, tf, 300) if chan is None: _timeline_cache[key] = None else: s = signals_to_frame(extract_fx_signals(chan, chan.dataframe)) _timeline_cache[key] = htf_fx_timeline(s, chan.dataframe) return _timeline_cache[key] def build(symbol: str, ltf: str, htf1: str, htf2: str): chan_l = _engine(symbol, ltf, 3000) if chan_l is None: return None, None cdf_l = chan_l.dataframe sig = find_fast_bsp3(cdf_l, build_htf_zones(cdf_l, ltf, chan=chan_l)) if sig.empty: return cdf_l, None for tf, pref in ((htf1, "h1"), (htf2, "h2")): tl = _timeline(symbol, tf) if tl is None or tl.empty: continue sig = attach_htf_context(sig, cdf_l, tl, pref) return cdf_l, sig def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--core-only", action="store_true", help="只输出核心组合（大级别同向 + 浅回抽）") ap.add_argument("--pairs", default=DEFAULT_PAIRS, help="级别对，格式 小级别:大级别1:大级别2，逗号分隔") ap.add_argument("--symbols", default="BTC,ETH,SOL") args = ap.parse_args() global SYMBOLS SYMBOLS = [f"{s.strip()}/USDT:USDT" for s in args.symbols.split(",") if s.strip()] pairs = [tuple(p.split(":")) for p in args.pairs.split(",") if p] all_rows, pool = [], [] for ltf, h1, h2 in pairs: print(f"\n{'=' * 100}") print(f"级别对：小级别 {ltf} 大级别 {h1} + {h2}") rows = [] pair_pool = [] for sym in SYMBOLS: cdf, sig = build(sym, ltf, h1, h2) if sig is None or sig.empty or len(sig) < 10: continue tr = run_trades(cdf, E(sig), SL, TP, MAXB) if tr.empty: continue s = summarize_trades(tr, f"{sym.split('/')[0]:>4} 全部") s["滞后"] = f"{sig['lag'].median():.0f}" rows.append(s) m = sig.set_index("entry_idx") tr = tr.copy() tr["symbol"] = sym.split("/")[0] tr["pair"] = f"{ltf}/{h1}" tr["date"] = cdf["date"].to_numpy()[tr["entry_idx"].to_numpy()] tr["h1_agree"] = tr["entry_idx"].map(m["h1_agree"]) if "h1_agree" in m else np.nan tr["depth"] = tr["entry_idx"].map(m["depth"]) pair_pool.append(tr) if not pair_pool: print(" 样本不足") continue if rows and not args.core_only: print(pd.DataFrame(rows).to_string(index=False)) pt = pd.concat(pair_pool, ignore_index=True) pool.append(pt) a1 = pt["h1_agree"] == 1 dq = pt["depth"].quantile(0.66) core = pt[a1 & (pt["depth"] >= dq)] sub = [summarize_trades(pt, f"{ltf} 全部")] if len(pt[a1]) >= 20: sub.append(summarize_trades(pt[a1], f"{ltf} +{h1}同向")) if len(core) >= 15: sub.append(summarize_trades(core, f"{ltf} 核心(同向+浅回抽)")) if len(pt[~a1]) >= 15: sub.append(summarize_trades(pt[~a1], f"{ltf} 反向(对照)")) print(pd.DataFrame(sub).to_string(index=False)) if len(core) >= 15: r = core["ret"].to_numpy() all_rows.append({ "级别对": f"{ltf} / {h1}+{h2}", "核心笔数": len(r), "胜率": f"{(r > 0).mean() * 100:.1f}%", "均收益": f"{r.mean() * 100:+.3f}%", "中位数": f"{np.median(r) * 100:+.3f}%", "PF": f"{r[r > 0].sum() / abs(r[r <= 0].sum()):.2f}", "偏度": f"{pd.Series(r).skew():.2f}", "t值": f"{r.mean() / (r.std(ddof=1) / np.sqrt(len(r))):+.2f}", }) print(f"\n{'=' * 100}") print("########## 跨级别对：核心组合汇总 ##########") if all_rows: print(pd.DataFrame(all_rows).to_string(index=False)) if pool: allp = pd.concat(pool, ignore_index=True) a1 = allp["h1_agree"] == 1 cores = [] for p, g in allp.groupby("pair"): dq = g["depth"].quantile(0.66) cores.append(g[(g["h1_agree"] == 1) & (g["depth"] >= dq)]) core_all = pd.concat(cores, ignore_index=True) r = core_all["ret"].to_numpy() print(f"\n 三个级别对合并核心样本 {len(r)} 笔：") print(f" 胜率 {(r > 0).mean() * 100:.1f}% 均收益 {r.mean() * 100:+.3f}% " f"中位数 {np.median(r) * 100:+.3f}%") print(f" PF {r[r > 0].sum() / abs(r[r <= 0].sum()):.2f} " f"偏度 {pd.Series(r).skew():.2f} " f"t值 {r.mean() / (r.std(ddof=1) / np.sqrt(len(r))):+.2f}") for k in (2, 5, 10): v = r[r <= np.quantile(r, 1 - k / 100)] print(f" 剔除最赚{k:>2}%: PF {v[v > 0].sum() / abs(v[v <= 0].sum()):.2f} " f"t {v.mean() / (v.std(ddof=1) / np.sqrt(len(v))):+.2f} " f"中位 {np.median(v) * 100:+.3f}%") core_all["year"] = pd.to_datetime(core_all["date"]).dt.year print("\n 核心样本分年：") rows = [{"年份": y, "笔数": len(g), "胜率": f"{(g.ret > 0).mean() * 100:.0f}%", "均收益": f"{g.ret.mean() * 100:+.3f}%", "PF": f"{g.ret[g.ret > 0].sum() / abs(g.ret[g.ret <= 0].sum()):.2f}" if (g.ret <= 0).any() else "inf"} for y, g in core_all.groupby("year") if len(g) >= 8] print(pd.DataFrame(rows).to_string(index=False)) out = Path(__file__).parent / "out" / "step19_level_pairs.csv" out.parent.mkdir(exist_ok=True) allp.to_csv(out, index=False) print(f"\n明细已写入 {out}") if __name__ == "__main__": main()