"""Step 32：「收盘重新转强」这个判据，到底是判据有用还是上下文有用。 质疑：如果 close > high[-1] 就能抓住启动，那不就能判断任意笔的端点了？ 本步用消融实验回答，逐层剥掉前置条件： E 完整 中枢存在 + 突破 + 回抽触边界 + 未跌回 + 收盘转强 D 去掉中枢 用近20根高点冒充「阻力位」，其余照旧 C 去掉回抽 突破后不要求回抽触及边界，转强即入 B 只要回调 近10根内创新低后收盘转强（无任何中枢/突破概念） A 裸判据 close > high[-1] 就买，别的都不管 另外直接检验：这些入场点前的回抽极值，有多少真的是引擎认定的笔端点。 """ from __future__ import annotations 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", 320) SL, TP, MAXB = 1.5, 3.0, 48 FEE, SLIP = 0.0004, 0.0001 def variant_signals(cdf: pd.DataFrame, zones: pd.DataFrame, mode: str) -> pd.DataFrame: """按消融档位产生信号。方向统一为做多/做空对称处理。""" close = cdf["close"].to_numpy(dtype=float) high = cdf["high"].to_numpy(dtype=float) low = cdf["low"].to_numpy(dtype=float) n = len(cdf) rows = [] if mode == "A": # 裸判据：收盘高于前一根最高价 -> 做多；低于前一根最低价 -> 做空 for j in range(1, n - 1): if close[j] > high[j - 1]: rows.append((j, 1)) elif close[j] < low[j - 1]: rows.append((j, -1)) return pd.DataFrame(rows, columns=["entry_idx", "direction"]) if mode == "B": # 近10根创新低后转强（有回调概念，无中枢） for j in range(11, n - 1): if low[j - 1] == low[j - 11:j].min() and close[j] > high[j - 1]: rows.append((j, 1)) elif high[j - 1] == high[j - 11:j].max() and close[j] < low[j - 1]: rows.append((j, -1)) return pd.DataFrame(rows, columns=["entry_idx", "direction"]) if mode == "D": # 用近20根极值冒充阻力位，走完整流程 for j in range(21, n - 1): edge = high[j - 21:j - 1].max() broke = close[j - 1] > edge if broke and low[j] <= edge * 1.003 and close[j] > high[j - 1]: rows.append((j, 1)) edge2 = low[j - 21:j - 1].min() if close[j - 1] < edge2 and high[j] >= edge2 * 0.997 and close[j] < low[j - 1]: rows.append((j, -1)) return pd.DataFrame(rows, columns=["entry_idx", "direction"]) # C / E 都要用真中枢 ts = cdf["timestamp"].to_numpy() for _, z in zones.iterrows(): zg, zd = float(z["zg"]), float(z["zd"]) if zg <= zd: continue start = int(np.searchsorted(ts, z["available_ts"], side="left")) if start >= n - 2: continue was_inside = False bo_idx, d = None, 0 for j in range(start, min(start + 200, 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 edge = zg if d == 1 else zd touched = False entry_idx = None for j in range(bo_idx + 1, min(bo_idx + 31, n)): if zd <= close[j] <= zg: break near = (low[j] <= edge * 1.003) if d == 1 else (high[j] >= edge * 0.997) if near: touched = True continue need = touched if mode == "E" else True # C 不要求回抽触及 if need: go = close[j] > high[j - 1] if d == 1 else close[j] < low[j - 1] if go: entry_idx = j break if entry_idx is not None: rows.append((entry_idx, d)) return pd.DataFrame(rows, columns=["entry_idx", "direction"]) def run_one(sym: str) -> 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.nested_level import build_htf_zones ltf = "30m" try: df_l = fetch_ohlcv(f"{sym}/USDT:USDT", ltf, 10**9) if df_l is None: return None chan = TF_DF(df_l, 1, ltf) cdf = chan.dataframe zones = build_htf_zones(cdf, ltf, chan=chan).reset_index(drop=True) # 引擎认定的笔端点，用于检验「能否判断端点」 idx_map = {k: i for i, k in enumerate(cdf["date"].dt.strftime("%Y-%m-%d %H:%M:%S"))} bi_ends = set() for bi in chan.bi_list: for attr in ("end_time",): k = str(getattr(bi, attr, "") or "") if k in idx_map: bi_ends.add(idx_map[k]) out, hits = [], [] for mode in ("A", "B", "D", "C", "E"): sig = variant_signals(cdf, zones, mode) if sig.empty or len(sig) < 30: continue sig = sig.drop_duplicates("entry_idx") 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: continue tr["mode"], tr["symbol"] = mode, sym out.append(tr) # 入场点前一根（回抽极值处）是否命中笔端点 e = sig["entry_idx"].to_numpy() hit = np.mean([any((x - 1 + k) in bi_ends for k in (-2, -1, 0, 1, 2)) for x in e]) hits.append({"symbol": sym, "mode": mode, "笔数": len(e), "命中笔端点±2根": hit * 100}) if not out: return None return {"sym": sym, "trades": pd.concat(out, ignore_index=True), "hits": pd.DataFrame(hits), "n_bi": len(bi_ends), "n_bar": len(cdf)} except Exception as e: return {"sym": sym, "error": repr(e)[:250]} def stat(g: pd.DataFrame, label: str) -> dict: r = g["gross"].to_numpy() - FEE - SLIP if len(r) < 20: 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", "t值": f"{r.mean() / (sd / np.sqrt(len(r))):+.2f}"} NAMES = {"A": "A 裸判据 close>high[-1]", "B": "B 创新低后转强", "D": "D 近20根高点当阻力", "C": "C 真中枢突破但不要求回抽", "E": "E 完整（现用版本）"} def main() -> None: res = [] with ProcessPoolExecutor(max_workers=3) as ex: futs = {ex.submit(run_one, s): s for s in ("BTC", "ETH", "SOL")} for f in as_completed(futs): r = f.result() if r is None or "error" in (r or {}): print(f" {futs[f]} 跳过 {(r or {}).get('error', '')}", flush=True) continue res.append(r) print(f" {r['sym']} ok（{r['n_bar']} 根K线，{r['n_bi']} 个笔端点）", flush=True) if not res: return allt = pd.concat([r["trades"] for r in res], ignore_index=True) hits = pd.concat([r["hits"] for r in res], ignore_index=True) print("\n" + "=" * 100) print("########## 1. 逐层剥掉前置条件（30m，无大级别过滤）##########") rows = [stat(allt[allt["mode"] == m], NAMES[m]) for m in ("A", "B", "D", "C", "E")] print(pd.DataFrame([r for r in rows if r]).to_string(index=False)) print(" 判据不变，只改上下文。PF 的落差就是上下文的贡献。") print("\n########## 2. 这些点是不是笔端点 ##########") g = hits.groupby("mode").agg(笔数=("笔数", "sum"), 命中率=("命中笔端点±2根", "mean")).reset_index() g["档位"] = g["mode"].map(NAMES) g["命中率"] = g["命中率"].round(1).astype(str) + "%" print(g[["档位", "笔数", "命中率"]].to_string(index=False)) print(" 命中率高不代表能预测端点——笔端点在K线里本就密集，需与随机基准比较。") print("\n########## 3. 随机基准：同样数量的随机点能命中多少 ##########") rng = np.random.default_rng(42) base = [] for r in res: n_bar, n_bi = r["n_bar"], r["n_bi"] # 笔端点±2根覆盖的K线占全样本比例，即随机命中概率上界 cover = min(1.0, n_bi * 5 / n_bar) base.append({"品种": r["sym"], "K线数": n_bar, "笔端点数": n_bi, "±2根覆盖占比": f"{cover * 100:.1f}%"}) print(pd.DataFrame(base).to_string(index=False)) print(" 若各档命中率都接近覆盖占比，说明判据对端点没有任何识别能力。") if __name__ == "__main__": main()