"""极值点的波动率：它是不是既拖慢了确认，又让目标够不着？ 用户的观察：一二类买卖点长在极值点上，那个区域波动天然很大，这可能正是确认 慢的原因。 这个假设若成立会同时解释两件事，而且机制不同： 确认慢 波动大 -> 分型/笔要更多根才稳定下来 -> sure_time 更晚 赚不到 ATR 被造成极值的那根插针抬高 -> 止损 2×虚高ATR 其实宽松， 但目标 3/8 ATR 变得够不着，因为入场后波动率会均值回复下来 第二条尤其要紧：它意味着交易不是「被打掉」，而是「永远走不到目标」， 与 §3.397 判定的「趋势继续」是**不同的失败模式**，应对办法也不同 （该换波动率口径，而不是换方向）。 三个量： atr_z 极值处 ATR / 该点之前 200 根的中位 ATR —— 是否真的偏高 atr_fwd 入场后 48 根的平均 ATR / 入场时 ATR —— 是否均值回复 与滞后相关 atr_z 高的信号，lag_bars 是否更长 """ from __future__ import annotations import argparse 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") HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) sys.path.insert(0, str(HERE.parent)) OUT = HERE / "out" / "step61_vol.feather" SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48 BASE_WIN = 200 def collect(sym: str, tf: str, rows: int) -> pd.DataFrame | None: from chanlun import TF_DF from chanlun.core.ChanEnum import Chan_BSP_TYPE from lib.data import fetch_ohlcv from lib.exit_model import cfg_name, walk_exits try: df = fetch_ohlcv(f"{sym}/USDT:USDT", tf, rows) if df is None or len(df) < 5_000: return None chan = TF_DF(df, 1, tf, lean=False) cdf = chan.dataframe bz = chan.cal_bi_zs_list_pure(chan.bi_list) if not bz: return None bsp = chan.find_all_bsp(chan.bi_list, bz) or [] dser = pd.to_datetime(cdf["date"]) if dser.dt.tz is not None: dser = dser.dt.tz_localize(None) didx = pd.DatetimeIndex(dser) n = len(cdf) def to_i(ts) -> int: t = pd.Timestamp(ts) return int(didx.searchsorted(t.tz_localize(None) if t.tz else t)) atr = cdf["atr"].to_numpy(float) cl = cdf["close"].to_numpy(float) # 基准 ATR 只用**该点之前**的窗口，含当根会把要检验的那根插针算进去 base = (pd.Series(atr).rolling(BASE_WIN, min_periods=50) .median().shift(1).to_numpy()) rec = [] # 对照组：所有笔端点，与 §3.397 同源 for bi in chan.bi_list: if not getattr(bi, "is_sure", False) or bi.end_klc is None: continue i = to_i(bi.end_klc.end_time) if not (0 <= i < n) or not np.isfinite(base[i]) or base[i] <= 0: continue rec.append({"kind": "笔端点", "dir": 0, "i_ext": i, "i_sure": -1, "atr_z": atr[i] / base[i], "lag_bars": -1}) want = {Chan_BSP_TYPE.B1: ("B1", 1), Chan_BSP_TYPE.S1: ("S1", -1), Chan_BSP_TYPE.B2: ("B2", 1), Chan_BSP_TYPE.S2: ("S2", -1), Chan_BSP_TYPE.B3: ("B3", 1), Chan_BSP_TYPE.S3: ("S3", -1)} for b in bsp: tag = want.get(b.type) if tag is None or b.sure_time is None: continue name, d = tag i_ext, i_sure = to_i(b.klc.end_time), to_i(b.sure_time) if not (0 <= i_ext < n and 0 <= i_sure < n): continue if not np.isfinite(base[i_ext]) or base[i_ext] <= 0: continue rec.append({"kind": name, "dir": d, "i_ext": i_ext, "i_sure": i_sure, "atr_z": atr[i_ext] / base[i_ext], "lag_bars": i_sure - i_ext}) r = pd.DataFrame(rec) # 入场后的实现波动率：目标够不够得着，取决于入场**之后**的 ATR ok = r.i_sure >= 0 fwd = np.full(len(r), np.nan) for k, i in zip(np.where(ok)[0], r.i_sure[ok].values): j = min(int(i) + 1 + MAXB, n) if j > int(i) + 1 and atr[int(i)] > 0: fwd[k] = np.nanmean(atr[int(i) + 1:j]) / atr[int(i)] r["atr_fwd"] = fwd sig = r[ok & (r.i_sure < n - 2)].copy() sig = sig[np.isfinite(atr[sig.i_sure.values]) & (atr[sig.i_sure.values] > 0)] if not sig.empty: cfg = cfg_name(SL, RUNNER, MAXB, RSTOP) res = walk_exits(cdf, pd.DataFrame({ "entry_idx": sig.i_sure.values, "direction": sig.dir.values}), [SL], [RUNNER], [MAXB], scale_at=SCALE_AT, runners=(RUNNER,), runner_stops=(RSTOP,)) if len(res) == len(sig): for c in ("g", "r", "c", "b"): sig[c] = res[f"{cfg}_{c}"].to_numpy() sig["atr_pct"] = (atr[sig.i_sure.values] / cl[sig.i_sure.values]) r = r.merge(sig[["i_ext", "kind", "g", "r", "c", "b", "atr_pct"]], on=["i_ext", "kind"], how="left") r["sym"], r["tf"] = sym, tf return r except Exception as e: # noqa: BLE001 print(f" {sym} {tf} 失败: {type(e).__name__}: {e}", flush=True) return None def report(d: pd.DataFrame) -> None: from lib.exit_model import fee_of, taker_notional for tf, x in d.groupby("tf"): print("\n" + "#" * 96) print(f"########## {tf} · 极值点的波动率 ##########") print("\n【一】极值处 ATR 是不是真的偏高（atr_z = 当点ATR / 前200根中位ATR）") rows = [] for kind in ["笔端点", "B1", "S1", "B2", "S2", "B3", "S3"]: g = x[x.kind == kind] if len(g) < 30: continue rows.append({ "信号": kind, "样本": len(g), "atr_z中位": round(g.atr_z.median(), 3), "P75": round(g.atr_z.quantile(.75), 3), "P90": round(g.atr_z.quantile(.90), 3), "高于基准占比": f"{(g.atr_z > 1).mean()*100:.0f}%", }) print(pd.DataFrame(rows).to_string(index=False)) print("\n【二】波动大是不是确认更慢（按 atr_z 四分位看 lag_bars）") y = x[x.kind.isin(["B1", "S1"]) & (x.lag_bars >= 0)].copy() if len(y) >= 100: q = pd.qcut(y.atr_z, 4, labels=["Q1低", "Q2", "Q3", "Q4高"], duplicates="drop") print(pd.DataFrame([{ "atr_z档": k, "笔数": len(g), "atr_z中位": round(g.atr_z.median(), 2), "滞后中位": round(g.lag_bars.median(), 1), "滞后均值": round(g.lag_bars.mean(), 1), } for k, g in y.groupby(q, observed=True)]).to_string(index=False)) c = np.corrcoef(y.atr_z, y.lag_bars)[0, 1] print(f" 相关系数 corr(atr_z, lag_bars) = {c:+.3f}") print(" 正相关支持「波动大 -> 确认慢」；接近 0 则该假设不成立。") print("\n【三】入场后波动率是否回落（atr_fwd = 后48根平均ATR / 入场ATR）") rows = [] for kind in ["B1", "S1", "B3", "S3"]: g = x[(x.kind == kind)].dropna(subset=["atr_fwd"]) if len(g) < 30: continue rows.append({ "信号": kind, "样本": len(g), "atr_fwd中位": round(g.atr_fwd.median(), 3), "回落占比(<1)": f"{(g.atr_fwd < 1).mean()*100:.0f}%", }) print(pd.DataFrame(rows).to_string(index=False)) print(" <1 = 入场后波动率比入场那刻低，则按入场ATR定的 3/8 ATR 目标" "\n 在绝对价格上被高估，会系统性够不着。") if "g" not in x.columns: continue print("\n【四】目标够不着的证据：出场原因分布（一类，按原方向）") z = x[x.kind.isin(["B1", "S1"])].dropna(subset=["g"]) if len(z) >= 60: print((z.r.value_counts(normalize=True) * 100).round(1) .to_frame("占比%").to_string()) print(f" 中位持仓 {z.b.median():.0f} 根 / 上限 {MAXB} 根") print("\n【五】按 atr_z 分档看一类表现（原方向）") q = pd.qcut(z.atr_z, 4, labels=["Q1低", "Q2", "Q3", "Q4高"], duplicates="drop") rows = [] for k, g in z.groupby(q, observed=True): net = g.g.values - fee_of(g.r.values, g.c.values) gR = g.g.values / (SL * g.atr_pct.values) tn = taker_notional(g.r.values, g.c.values) w, o = net[net > 0].sum(), -net[net <= 0].sum() rows.append({ "atr_z档": k, "笔数": len(g), "atr_z中位": round(g.atr_z.median(), 2), "胜率": f"{(net > 0).mean()*100:.1f}%", "毛R": round(gR.mean(), 3), "PF": round(w / o, 2) if o > 0 else np.inf, "余量bp": round(net.mean() / tn.mean() * 1e4, 2), }) print(pd.DataFrame(rows).to_string(index=False)) def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--symbols", default="BTC,ETH,SOL,LINK,DOGE") ap.add_argument("--tfs", default="5m,15m") ap.add_argument("--rows", type=int, default=200_000) ap.add_argument("--workers", type=int, default=3) ap.add_argument("--reuse", action="store_true") args = ap.parse_args() if args.reuse and OUT.exists(): report(pd.read_feather(OUT)) return syms = [s.strip() for s in args.symbols.split(",")] tfs = [t.strip() for t in args.tfs.split(",")] parts = [] with ProcessPoolExecutor(max_workers=args.workers) as ex: fut = {ex.submit(collect, s, t, args.rows): (s, t) for s in syms for t in tfs} for i, f in enumerate(as_completed(fut), 1): r = f.result() s, t = fut[f] print(f" [{i}/{len(fut)}] {s} {t} " f"{0 if r is None else len(r)}", flush=True) if r is not None: parts.append(r) if not parts: print("无结果") return d = pd.concat(parts, ignore_index=True) OUT.parent.mkdir(exist_ok=True) d.to_feather(OUT) report(d) if __name__ == "__main__": main()