diff --git a/research/HANDOFF.md b/research/HANDOFF.md index b7fb3e3..928f51c 100644 --- a/research/HANDOFF.md +++ b/research/HANDOFF.md @@ -1036,6 +1036,56 @@ B3 当年的解法不是调参，是**重新定义成实时判据**（B4：收 > 末端，也就是 `mom60` 最高的那个区间——而那正是我们实测最差的四分位** > （Q4 发现期余量只剩 1.44bp）。真要做实时一类，先验就不利。 +### 3.394 实时版一类（fast B1/S1）：滞后不是病根，一类没有边（step56） + +§3.393 判定引擎 B1/S1 不可用后，用户要求照 B3→B4 的路子做实时版。做了， +**没救回来**，但这次实验把病因锁死了。 + +`research/lib/fast_bsp1.py` 把引擎 B1 的三个条件全换成当根可判的代理： + +| 引擎条件 | 滞后源 | 实时代理 | +|---|---|---| +| 中枢已成 `zs.is_sure` | 无 | `available_ts`（与 B4 同口径） | +| 一笔向下离开中枢 | 笔端点 | **收盘** < zd | +| 背驰 `check_bi_div` | 笔端点 | 进入段面积在中枢确认时已是历史；离开段面积从突破根逐根累加 | +| 触发 `leave_bi.sure_time` | **主要滞后源** | 收盘反向越过前一根极值（沿用 B4 的转强判据） | + +注意一类和三类**方向相反**：中枢向下突破后 B4 做空，fast B1 是在同一段下跌里 +找力竭然后做多。 + +**结果（5 币，5m/15m，出场 2/3/8/2/48）** + +| 信号 | 5m 笔数 | 5m 胜率 | 5m PF | 5m t值 | 15m PF | 15m t值 | +|---|---|---|---|---|---|---| +| fastB1 一买 | 482 | 22.4% | 0.38 | −9.4 | 0.37 | −10.6 | +| fastB1 一卖 | 528 | 20.3% | 0.36 | −12.5 | 0.41 | −9.6 | +| **fastB3 三买**（同管线对照） | 972 | **55.0%** | **1.83** | **+10.6** | 1.84 | +8.5 | +| **fastB3 三卖**（同管线对照） | 927 | **54.6%** | **1.93** | **+10.5** | 2.06 | +10.7 | + +**对照组是这次实验的全部价值所在**：同一份数据、同一套出场、同一条管线， +B4 跑出 PF 1.83~2.06 / t +8.5~+10.7。所以 fast B1 的负数不是管线接错了。 + +**病根锁定：滞后不是主因。** 引擎版 PF 0.22 → 实时版 0.38，确实有改善， +但离 1.0 还差得远。更关键的是滞后分档： + +| lag（入场根 − 极值根） | 5m 笔数 | 5m PF | +|---|---|---| +| ≤1 根（≈买在极值上） | 472 | 0.40 | +| 2-3 根 | 390 | 0.36 | +| 4-6 根 | 125 | 0.29 | + +**即使入场就在极值那一根，PF 仍只有 0.40。** §3.393 里「等 8 根导致几何劣势」 +的解释因此只对了一半：滞后确实在扣分，但**把滞后清零也救不活**——一类在这个 +市场/周期/出场结构下本身就是负边缘的。 + +背驰强度有单调性但不够：15m 上最强四分位 PF 0.99、余量 −0.24bp、t −2.0， +毛收益刚够打平、净收益仍为负；且这是四选一的样本内挑选，不构成可交易的结论。 + +**判定：一类到此为止，不再投入。** 这与 §3.4 消融、§3.391 的 `mom60` 结论 +互相印证——这一族信号的 alpha 在**顺势延续**上，不在反转上。fast_bsp 的文档 +头早就写了「B4 不只是 B3 提前几根，两者统计性质符号相反」，一类的实验说明 +反过来也成立：**把反转信号提前，它还是反转信号。** + ### 3.4 alpha 的来源（step32 消融） 逐条拆掉 `fast_bsp3` 的条件后发现：**alpha 完全来自缠论中枢的上下文定位， diff --git a/research/lib/fast_bsp1.py b/research/lib/fast_bsp1.py new file mode 100644 index 0000000..1bd8ae4 --- /dev/null +++ b/research/lib/fast_bsp1.py @@ -0,0 +1,218 @@ +"""快速一类买卖点（fast B1/S1）——把引擎的 B1/S1 改写成当根可判的形式。 + +**动机**：step55 实测引擎 `find_all_bsp` 的 B1/S1 在 5m/15m 上是重亏的 +（PF 0.20~0.24、胜率 15~21%、t −17~−22），而且原因是几何性的：引擎在 +`leave_bi.sure_time` 才发信号，中位滞后 8~9 根，此时价格已朝反弹方向跑了 +1.68 ATR。逆势信号上，这个滞后是加倍惩罚——新入场价往下 2 ATR 的止损落在 +比原始低点还低 0.3 ATR 处，几乎贴着极值。 + +B3 当年也是同样的病（PF 0.66），解法不是调参而是**重新定义成实时判据** +（B4，见 `chanlun/analysis/fast_bsp.py`）。本模块对一类做同样的事。 + +引擎 B1 是三个条件的合取（`bsp.py: find_all_bsp` + `check_bi_div`）： + + 中枢已成 -> 一笔向下离开中枢 -> 该笔 MACD 面积 < 进入笔面积（底背驰） + +三条都有当根可判的代理，滞后来源只有一处——`leave_bi.is_sure`： + + 中枢已成 zs.is_sure -> zones.available_ts（与 B4 同口径，已解决） + 向下离开 leave_bi 端点在 zd 下 -> **收盘** < zd，当根可判 + 背驰 整笔面积对比 -> 进入笔面积在中枢确认时已是历史， + 离开段面积从突破根起逐根累加，当根可判 + 触发 等笔确认（滞后之源） -> 收盘反向越过前一根极值（沿用 B4 的转强判据） + +**与三类的方向关系相反**：三类顺着突破方向做，一类逆着做。中枢向下突破后， +B4 会做空，而 fast B1 是在同一段下跌里找力竭然后**做多**。 + +⚠️ 先验不利，落地前请先看数：§3.391 实测 `mom60` 最高的四分位是最差的 +（余量只剩 1.44bp），而一类按定义就住在一段已走完的行情末端。本模块是用来 +证伪这个先验的，不是用来假定它不成立的。 +""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from chanlun.analysis.fast_bsp import _resolve_avail_bi, timestamps_ms + + +def zones_with_enter_area(zs_list, src: pd.DataFrame, + avail_bi: int | None = None) -> pd.DataFrame: + """区间表 + 进入笔的 MACD 面积与方向。 + + 这两列**不引入滞后**：进入笔是中枢第一笔的前一笔，中枢确认时它早已收尾。 + 背驰判据里唯一需要等待的是离开段，而那一段可以逐根累加。 + + 面积口径必须和 `ChanBI.cal_macd_hist` 一致：只累加顺方向那一侧的 hist + 并取绝对值（上升笔累加正值，下降笔累加负值的绝对值），故恒非负。 + """ + ts_of = dict(zip(src["date"].dt.strftime("%Y-%m-%d %H:%M:%S"), + timestamps_ms(src))) + i = _resolve_avail_bi(avail_bi) + + rows = [] + for zs in zs_list: + bis = getattr(zs, "bi_list", []) + if not bis: + continue + key_bi = bis[i] if (i == -1 or len(bis) > i) else bis[-1] + avail = (ts_of.get(str(getattr(key_bi, "sure_time", "") or "")) + or ts_of.get(str(getattr(key_bi, "end_time", "") or ""))) + if avail is None: + continue + enter_bi = getattr(bis[0], "pre", None) + if enter_bi is None: + continue + # dir 用 +1/−1 表示，避免把引擎枚举漏进研究侧 + e_dir = 1 if str(enter_bi.dir).endswith("UP") else -1 + rows.append({ + "zg": float(zs.zg), "zd": float(zs.zd), + "start_ts": ts_of.get(str(bis[0].start_time), avail), + "available_ts": int(avail), + "enter_area": abs(float(enter_bi.macd_hist)), + "enter_dir": e_dir, + }) + out = pd.DataFrame(rows) + if out.empty: + return out + return out.sort_values("available_ts").reset_index(drop=True) + + +def find_fast_bsp1( + df: pd.DataFrame, + zones: pd.DataFrame, + scan: int = 200, + leave_win: int = 60, + min_leave_bars: int = 3, + max_div: float = 1.0, + max_per_zone: int = 1, + diag: dict | None = None, +) -> pd.DataFrame: + """扫描每个中枢，找「离开中枢后力度背驰、随即转向」的入场点。 + + zones 需含 zg / zd / available_ts / enter_area / enter_dir， + 即 `zones_with_enter_area` 的输出。 + + `min_leave_bars` 要求离开段至少走这么多根才允许触发。没有它的话，转强判据 + 会在突破后第一根小反弹就打进去，那时离开段的面积还没累起来，背驰条件几乎 + 自动成立——会退化成「随便一个反弹就抄底」。 + + `max_div` 是背驰的松紧：离开段面积 / 进入段面积 < max_div 才算背驰。 + 1.0 等于引擎的 `macdhist_div < 0`；调小则只认力度衰减更明显的。 + + 返回列： + entry_idx 实时可下单的K线 + direction +1 一买 / −1 一卖（**与突破方向相反**） + bo_idx 离开中枢的突破根 + ext_idx 离开段的极值根 + lag entry_idx − ext_idx，即相对理想入场点的滞后 + div 离开段面积 / 进入段面积，越小背驰越强 + depth 离开段极值越过中枢边界的幅度（相对边界） + """ + need = {"zg", "zd", "available_ts", "enter_area", "enter_dir"} + if zones.empty or not need <= set(zones.columns): + return pd.DataFrame() + + ts = df["timestamp"].to_numpy() + close = df["close"].to_numpy(float) + high = df["high"].to_numpy(float) + low = df["low"].to_numpy(float) + hist = df["macdhist"].to_numpy(float) + n = len(df) + rows = [] + + def note(key: str) -> None: + if diag is not None: + diag[key] = diag.get(key, 0) + 1 + + for zone_i, (_, z) in enumerate(zones.iterrows()): + note("中枢总数") + zg, zd = float(z["zg"]), float(z["zd"]) + e_area, e_dir = float(z["enter_area"]), int(z["enter_dir"]) + if zg <= zd or not np.isfinite(e_area) or e_area <= 0: + note("×无效中枢或进入段无面积") + continue + start = int(np.searchsorted(ts, z["available_ts"], side="left")) + if start >= n - 2: + note("×中枢太靠后") + continue + scan_end = min(start + scan * max_per_zone, n) + cursor = start + + for occ in range(1, max_per_zone + 1): + if cursor >= n - 2: + break + was_inside = False + bo_idx, d = None, 0 + for j in range(cursor, scan_end): + 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: + if occ == 1: + note("×窗口内未离开中枢") + break + + # 引擎要求 enter_bi.dir == leave_bi.dir 才比面积，否则背驰无从谈起 + if e_dir != d: + note("×进入段与离开段不同向") + cursor = bo_idx + 1 + continue + + area = 0.0 + ext = low[bo_idx] if d == -1 else high[bo_idx] + ext_idx = bo_idx + entry_idx, div_at = None, np.nan + for j in range(bo_idx, min(bo_idx + leave_win + 1, n)): + h = hist[j] + # 与 ChanBI.cal_macd_hist 同口径：只取顺方向一侧，取绝对值 + if d == -1 and h < 0: + area -= h + elif d == 1 and h > 0: + area += h + if (low[j] < ext) if d == -1 else (high[j] > ext): + ext, ext_idx = (low[j] if d == -1 else high[j]), j + if j < bo_idx + min_leave_bars: + continue + # 收盘回到中枢内 -> 这不是一次有效的离开，弃掉 + if zd <= close[j] <= zg: + note("×离开段收盘回到中枢内") + break + if area >= e_area * max_div: + continue # 力度未衰减，不是背驰 + go = close[j] > high[j - 1] if d == -1 else close[j] < low[j - 1] + if go: + entry_idx, div_at = j, area / e_area + break + if entry_idx is None: + if occ == 1: + note("×未在窗口内背驰转向") + cursor = bo_idx + 1 + continue + if occ == 1: + note("√成交") + + edge = zd if d == -1 else zg + rows.append({ + "entry_idx": entry_idx, "direction": -d, + "bo_idx": bo_idx, "ext_idx": ext_idx, + "lag": entry_idx - ext_idx, + "div": div_at, + "depth": abs(ext - edge) / edge, + "zg": zg, "zd": zd, + "width_pct": (zg - zd) / close[bo_idx], + "occ": occ, "zone_i": zone_i, + }) + cursor = entry_idx + 1 + + out = pd.DataFrame(rows) + if out.empty: + return out + return (out.sort_values(["entry_idx", "occ", "zone_i"]) + .drop_duplicates("entry_idx", keep="first") + .reset_index(drop=True)) diff --git a/research/step56_fast_bsp1.py b/research/step56_fast_bsp1.py new file mode 100644 index 0000000..965d204 --- /dev/null +++ b/research/step56_fast_bsp1.py @@ -0,0 +1,194 @@ +"""实时版一类买卖点（fast B1/S1）能否翻正。 + +step55 已判定引擎的 B1/S1 不可用（5m PF 0.24/0.20、胜率 18.6%/14.8%、 +t −17/−22），且病根是滞后 8~9 根带来的几何劣势，不是参数。B3 当年同病 +（PF 0.66），靠重新定义成实时判据翻到 1.59（B4）。本脚本对一类做同样的尝试。 + +对照组三条，缺一不可： + 引擎 B1/S1 step55 的数，说明「不改判据」是什么下场 + fast B1/S1 本次 + fast B3/S3 同一份数据、同一套出场跑 B4，确认管线本身能跑出正数 + —— 少了它，fast B1 若为负就分不清是判据不行还是管线接错了 +""" +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" / "step56_fast_bsp1.feather" +SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48 + + +def collect(sym: str, rows: int, tf: str) -> pd.DataFrame | None: + from chanlun import TF_DF + from chanlun.analysis.fast_bsp import ( + ensure_timestamp, + find_fast_bsp3, + zones_from_zs_list, + ) + from lib.data import fetch_ohlcv + from lib.exit_model import cfg_name, walk_exits + from lib.fast_bsp1 import find_fast_bsp1, zones_with_enter_area + + 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) + cdf = ensure_timestamp(chan.dataframe) + zs_list = chan.cal_bi_zs_list_pure(chan.bi_list) + if not zs_list: + return None + + atr = cdf["atr"].to_numpy(float) + cl = cdf["close"].to_numpy(float) + cfg = cfg_name(SL, RUNNER, MAXB, RSTOP) + parts = [] + + d1 = {} + s1 = find_fast_bsp1(cdf, zones_with_enter_area(zs_list, cdf), diag=d1) + if not s1.empty: + s1["kind"] = "fastB1" + parts.append(s1) + # 同数据同出场跑一遍 B4，作为「管线能出正数」的存在性证明 + s3 = find_fast_bsp3(cdf, zones_from_zs_list(zs_list, cdf)) + if not s3.empty: + s3["kind"] = "fastB3" + parts.append(s3) + if not parts: + return None + + r = pd.concat(parts, ignore_index=True) + r = r[(r.entry_idx < len(cdf) - 2) + & np.isfinite(atr[r.entry_idx.values]) + & (atr[r.entry_idx.values] > 0)].reset_index(drop=True) + if r.empty: + return None + res = walk_exits(cdf, r[["entry_idx", "direction"]], [SL], [RUNNER], + [MAXB], scale_at=SCALE_AT, runners=(RUNNER,), + runner_stops=(RSTOP,)) + if len(res) != len(r): + return None + for k in ("g", "r", "c", "b"): + r[k] = res[f"{cfg}_{k}"].to_numpy() + r["sym"], r["tf"] = sym, tf + r["atr_pct"] = atr[r.entry_idx.values] / cl[r.entry_idx.values] + r["date"] = cdf["date"].to_numpy()[r.entry_idx.values] + r["diag"] = str(d1) + return r + except Exception as e: # noqa: BLE001 + print(f" {sym} {tf} 失败: {type(e).__name__}: {e}", flush=True) + return None + + +def stats(g: pd.DataFrame) -> dict: + from lib.exit_model import fee_of, taker_notional + + net = g.g.values - fee_of(g.r.values, g.c.values) + gR = g.g.values / (SL * g.atr_pct.values) + R = net / (SL * g.atr_pct.values) + tn = taker_notional(g.r.values, g.c.values) + w, o = net[net > 0].sum(), -net[net <= 0].sum() + return { + "笔数": len(g), "胜率": f"{(net > 0).mean()*100:.1f}%", + "毛R": round(gR.mean(), 3), "净均R": round(R.mean(), 3), + "PF": round(w / o, 2) if o > 0 else np.inf, + "R夏普": round(R.mean() / R.std(ddof=1), 3), + "余量bp": round(net.mean() / tn.mean() * 1e4, 2), + "中位持仓": int(np.median(g.b.values)), + "t值": round(gR.mean() / (gR.std(ddof=1) / np.sqrt(len(g))), 2), + } + + +def report(d: pd.DataFrame) -> None: + for tf, x in d.groupby("tf"): + span = (pd.to_datetime(x.date).max() + - pd.to_datetime(x.date).min()).total_seconds() / 86400 + print("\n" + "#" * 92) + print(f"########## {tf} · {x.sym.nunique()} 币 · 跨 {span:.0f} 天" + f" ##########") + rows = [] + for (kind, dirn), g in x.groupby(["kind", "direction"]): + if len(g) < 40: + continue + nm = {"fastB1": ("一买", "一卖"), "fastB3": ("三买", "三卖")}[kind] + rows.append({"信号": f"{kind} {nm[0 if dirn == 1 else 1]}", + "每币每天": round(len(g) / span / x.sym.nunique(), 3), + **stats(g)}) + if rows: + print(pd.DataFrame(rows).to_string(index=False)) + print("\n对照 · step55 引擎原生（同周期同出场）：") + print(" 5m B1 PF 0.24 胜率 18.6% t −17.3 ｜ S1 PF 0.20 胜率 14.8% " + "t −22.3") + print(" 15m B1 PF 0.21 t −18.9 ｜ S1 PF 0.23 t −15.9") + + f1 = x[x.kind == "fastB1"] + if len(f1) >= 60: + print("\n背驰强度分档（div = 离开段面积/进入段面积，越小背驰越强）：") + q = pd.qcut(f1["div"], 4, labels=["Q1最强", "Q2", "Q3", "Q4最弱"], + duplicates="drop") + print(pd.DataFrame([{"档": k, **stats(v)} + for k, v in f1.groupby(q, observed=True) + if len(v) >= 20]).to_string(index=False)) + print("\n滞后分档（lag = 入场根 − 离开段极值根）：") + b = pd.cut(f1["lag"], [-1, 1, 3, 6, 12, 1e9], + labels=["≤1根", "2-3根", "4-6根", "7-12根", ">12根"]) + print(pd.DataFrame([{"档": k, **stats(v)} + for k, v in f1.groupby(b, observed=True) + if len(v) >= 20]).to_string(index=False)) + + dg = d[d.kind == "fastB1"].diag.dropna() + if len(dg): + print("\n" + "=" * 92) + print("fast B1 漏斗（首个中枢样本）：", dg.iloc[0]) + + +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(): + d = pd.read_feather(OUT) + else: + syms = [s.strip() for s in args.symbols.split(",")] + tfs = [t.strip() for t in args.tfs.split(",")] + print(f"[实时版一类] {len(syms)} 币 × {tfs} × {args.rows} 根\n", + flush=True) + parts = [] + with ProcessPoolExecutor(max_workers=args.workers) as ex: + fut = {ex.submit(collect, s, args.rows, t): (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()