research: 低滞后信号口径定稿与实盘前偏差审计
fast_bsp3 改用 tol=-1 + require_touch=False，信号滞后从 5.8 根降到 2.2 根。 滞后与收益严格单调（年化 370% -> 906%，同一份数据同一套成本）， 这是本轮提升的主因，也意味着实盘延迟会直接侵蚀收益。 新增 step31~39 验证策略能否落地： - 跨品种样本外——8 个未参与调参的币，PF 2.73 / t 28.5，无一为负 - 时点重建——只喂到信号那一根重算，同根命中 100%，确认无未来函数； 1m 在 2000 根窗口即饱和，计算耗时 0.20s - 偏差审计——多空对称、中枢生效时刻零回退、滑点稳健至 30bp、持仓几乎不重叠 - 消融——alpha 来自缠论中枢的上下文定位，而非「收盘转强」这个触发动作 补 research/HANDOFF.md：记录确切口径与参数、已排除的偏差、 已验证无效因而不必重做的方向，以及下一步用影子交易器实测执行滑点的方案。 清理 step1~20 的输出：早期方法论已被推翻（存在未来函数偏差）， 其结论不再被引用；脚本保留，需要时可重跑。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
@@ -0,0 +1,241 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user