jackyu66gitandCursor
d2fcb27d01
§5.41 发现 available_ts 取「中枢最后一笔」是右边缘重画的根因，改取「第三笔」 能把重画率从 6.5% 压到 1.2%，当时据此判断它是「唯一可能同时改善收益与稳定性」 的改动。那个判断只测了稳定性，过早了。 8 个样本外币 × 30 万根 1m，两组共用同一个 TF_DF，只切 available_ts 的取法。 实盘口径（深色 ∧ ATR≥8bp，955 vs 989 笔）： 毛 R 0.933 → -0.000 净均 R 0.798 → -0.147 PF 3.22 → 0.80 滑点余量 15.07 → -2.20 bp 判决依据是毛 R 那一行：扣任何费用之前 edge 就没了，所以不是成本、门控或出场 参数的问题，是信号本身不再有预测力。逐币 8/8 全部变差。滞后确实降了 （2.16 → 2.01），但换来的是另一批交易——两组重合度只有约 30%。 原因是中枢没发育完就下注，支撑/压力还没立住。「等中枢最后一笔」那段等待不是 可以优化掉的延迟，它就是 alpha 本身。由此得一条一般规则：任何以「让信号更早 确定」为目标的改动，先测毛 R，不能只看重画率和滞后。 开关 AVAIL_BI_INDEX 保留只为可复现该 A/B，默认 -1 维持现行口径。环境变量在 调用时解析而非 import 时——fork 启动的子进程会继承已 import 的模块，import 时读会固化成父进程的值。 顺带交叉验证：现行口径本次算出滑点余量 15.07bp，与用优化前代码算的同组同期 15.19bp 吻合。 Co-authored-by: Cursor <cursoragent@cursor.com>
385 lines
17 KiB
Python
385 lines
17 KiB
Python
"""快速三类买卖点（引擎内称第四类，B4/S4）。
|
||
|
||
原本长在 `research/lib/fast_bsp3.py`，现移入引擎作为唯一实现，研究脚本改为转发导入，
|
||
这样回测口径与 web 图表永远一致。
|
||
|
||
命名说明：缠论原文里没有「第四类买卖点」，但 B4/S4 也不只是「B3/S3 提前几根」——
|
||
两者的选样口径不同，统计性质符号相反，故单独立类。形态条件是实时可判的：
|
||
|
||
中枢已成 -> 收盘突破 zg -> 收盘未跌回中枢 -> 重新上行
|
||
|
||
最后一步发生的当根就能下单，滞后约 2 根，而引擎 B3/S3 要等 pullback_bi.sure_time，
|
||
滞后 9~10 根。但差别不止滞后：
|
||
|
||
判据 引擎用笔端点事后判（回拉笔低点 >= zg），本函数用收盘价实时判
|
||
方向 引擎由离开笔方向决定，本函数由收盘从哪一侧突破决定
|
||
口径 627 个中枢里引擎发 625 个信号（几乎不筛），本函数只认 212 个（34%）；
|
||
被拒的多数是「中枢确认时价格早已离开、此后再没回来」的历史区间
|
||
|
||
step30 同条件对拍（同一套 pure 笔中枢、同一组过滤器、同样的 1.5/3.0/48 出场）：
|
||
|
||
原始 PF +大级别同向 +同向+阶梯 胜率 t值
|
||
引擎 B3/S3 0.66 0.66 0.71 27.4% -18.76
|
||
本函数 B4/S4 1.59 1.85 2.26 47.1% +10.09
|
||
|
||
同一组过滤器对 B4 有效、对 B3 无效，滞后差解释不了这一点（入场后移 1~4 根只是
|
||
PF 3.18->2.53 的平滑衰减）。且 SL1.5/TP3.0 下随机入场胜率约 33%，引擎那 27.4%
|
||
低于随机——它选中的是一批系统性反向的样本，不是「晚了所以差」。
|
||
|
||
require_touch 控制是否强求回抽碰到中枢边界。step32/33 的实测表明强求反而更差：
|
||
这等于排除掉「突破后一去不回头」的强势段，而那正是缠论里最强的趋势形态。
|
||
故默认 False（笔数 +21%、滞后 -1 根、PF 2.26→2.37）。
|
||
|
||
判据本身（收盘越过前一根极值）没有独立预测力：裸用 15 万笔样本 PF 0.95，
|
||
把中枢换成「近20根高点」这类伪阻力位后 PF 0.90。alpha 全部来自中枢结构，
|
||
判据只负责在这个已知价位上确认动能恢复。它也不能用于识别笔端点——
|
||
入场前的回抽极值命中笔端点±2根的比例 19.3%，低于随机基准 22%。
|
||
|
||
全部判定只使用当根及之前的数据，无未来函数。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
from chanlun.core.ChanEnum import Chan_FX_TYPE
|
||
|
||
# 中枢的「可用时刻」取第几笔的确认时间。
|
||
# -1 = 最后一笔（历史默认）。中枢每吸收一根K线，最后一笔就可能后移，
|
||
# 于是 available_ts 跟着漂——这是右边缘重画的根因（见 HANDOFF §5.41）。
|
||
# 2 = 第三笔。三笔重叠即中枢成立，此后不再变，理论上更早也更稳。
|
||
# ⛔ step47 已判定 bis[2] 作废（毛 R 0.933 → −0.000，见 HANDOFF §5.42）。
|
||
# 开关保留只为可复现那次 A/B，**不要改默认值**。
|
||
import os as _os
|
||
|
||
AVAIL_BI_INDEX = -1
|
||
|
||
|
||
def _resolve_avail_bi(avail_bi: int | None) -> int:
|
||
"""优先级：显式入参 > 环境变量 CHAN_AVAIL_BI > 模块默认。
|
||
|
||
环境变量在调用时读取而非 import 时——ProcessPoolExecutor 在 fork 启动方式下
|
||
子进程会继承已 import 的模块，import 时读就固化成父进程的值了。
|
||
"""
|
||
if avail_bi is not None:
|
||
return avail_bi
|
||
return int(_os.environ.get("CHAN_AVAIL_BI", AVAIL_BI_INDEX))
|
||
|
||
|
||
def timestamps_ms(src: pd.DataFrame) -> np.ndarray:
|
||
"""取毫秒时间戳。研究侧的 df 自带 timestamp，web 侧的不一定，故按 date 回退。
|
||
|
||
回退写法不能用 `date.astype("int64") // 10**6`：该值单位取决于列精度，
|
||
对毫秒精度的列会把时间戳砸平。
|
||
"""
|
||
if "timestamp" in src.columns:
|
||
return src["timestamp"].to_numpy()
|
||
d = pd.to_datetime(src["date"])
|
||
if getattr(d.dt, "tz", None) is None:
|
||
d = d.dt.tz_localize("UTC")
|
||
return (d.dt.tz_convert("UTC").dt.tz_localize(None)
|
||
.astype("datetime64[ms]").astype("int64").to_numpy())
|
||
|
||
|
||
def ensure_timestamp(df: pd.DataFrame) -> pd.DataFrame:
|
||
"""保证 df 带 timestamp 列，缺失时补一份副本，不改动调用方的对象。"""
|
||
if "timestamp" in df.columns:
|
||
return df
|
||
out = df.copy()
|
||
out["timestamp"] = timestamps_ms(out)
|
||
return out
|
||
|
||
|
||
def build_htf_zones(df_htf: pd.DataFrame, tf: str, chan=None, avail_bi: int | None = None) -> pd.DataFrame:
|
||
"""算 pure 笔中枢，返回带生效时间的区间表。
|
||
|
||
available_ts —— 该中枢最早可被使用的时间戳（其确认时刻）。
|
||
传入已构建好的 chan 可避免重复跑一遍 pipeline（大数据集上省一半时间）。
|
||
"""
|
||
if chan is None:
|
||
from chanlun import TF_DF
|
||
|
||
chan = TF_DF(df_htf, 1, tf)
|
||
zs_list = chan.cal_bi_zs_list_pure(chan.bi_list)
|
||
# 用引擎自己的 dataframe 对齐，避免调用方传入的 df 与引擎内部行数不一致
|
||
src = chan.dataframe if getattr(chan, "dataframe", None) is not None else df_htf
|
||
return zones_from_zs_list(zs_list, src, avail_bi=avail_bi)
|
||
|
||
|
||
def zones_from_zs_list(zs_list, src: pd.DataFrame, avail_bi: int | None = None) -> pd.DataFrame:
|
||
"""把已算好的 pure 笔中枢转成区间表。
|
||
|
||
调用方手工跑过 cal_bi_zs_list_pure 时走这里，免得再算一遍（web 的 analyze_chan
|
||
就是这种用法）。
|
||
"""
|
||
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
|
||
# 笔数不够时退回最后一笔（i=2 需要至少 3 笔才成立）
|
||
key_bi = bis[i] if (i == -1 or len(bis) > i) else bis[-1]
|
||
sure_key = str(getattr(key_bi, "sure_time", "") or "")
|
||
end_key = str(getattr(key_bi, "end_time", "") or "")
|
||
avail = ts_of.get(sure_key) or ts_of.get(end_key)
|
||
if avail is None:
|
||
continue
|
||
start_key = str(bis[0].start_time)
|
||
rows.append({
|
||
"zg": float(zs.zg), "zd": float(zs.zd),
|
||
"gg": float(getattr(zs, "gg", zs.zg)), "dd": float(getattr(zs, "dd", zs.zd)),
|
||
"start_ts": ts_of.get(start_key, avail),
|
||
"available_ts": int(avail),
|
||
})
|
||
out = pd.DataFrame(rows)
|
||
return out.sort_values("available_ts").reset_index(drop=True) if not out.empty else out
|
||
|
||
|
||
def add_zone_ladder(zones: pd.DataFrame) -> pd.DataFrame:
|
||
"""标注每个中枢相对前一个中枢是否同向推进（缠论「趋势 vs 盘整」）。
|
||
|
||
z_above / z_below 分别对应向上、向下推进。买信号要求 z_above、卖信号要求 z_below
|
||
时，30m/2h 的 PF 从 2.72 升到 3.41。
|
||
"""
|
||
out = zones.copy()
|
||
if out.empty:
|
||
out["z_above"], out["z_below"] = pd.Series(dtype=bool), pd.Series(dtype=bool)
|
||
return out
|
||
pg, pdn = out["zg"].shift(), out["zd"].shift()
|
||
out["z_above"] = (out["zd"] > pg).fillna(False)
|
||
out["z_below"] = (out["zg"] < pdn).fillna(False)
|
||
return out
|
||
|
||
|
||
def htf_fx_timeline(chan_htf, df_htf: pd.DataFrame | None = None) -> pd.DataFrame:
|
||
"""把大级别分型压成一条按确认时间排序的时间线。
|
||
|
||
confirm_ts 是该分型最早可被使用的时刻。timestamp 是K线开盘时刻，而分型要等这根K线
|
||
收盘才算数，所以整体后移一个大级别周期；否则小级别会提前一整根大级别K线拿到信号。
|
||
|
||
只取方向与确认时刻——同向过滤用不到背驰强度，省掉 MACD 面积计算。
|
||
"""
|
||
src = chan_htf.dataframe if getattr(chan_htf, "dataframe", None) is not None else df_htf
|
||
if src is None or len(src) == 0:
|
||
return pd.DataFrame(columns=["confirm_ts", "fx_ts", "direction", "price"])
|
||
ts = timestamps_ms(src)
|
||
idx_of = {t: i for i, t in enumerate(src["date"].dt.strftime("%Y-%m-%d %H:%M:%S"))}
|
||
period = int(np.median(np.diff(ts))) if len(ts) > 1 else 0
|
||
|
||
rows = []
|
||
for klc in getattr(chan_htf, "klc_list", []):
|
||
if klc.fx not in (Chan_FX_TYPE.TOP, Chan_FX_TYPE.BOTTOM):
|
||
continue
|
||
if klc.next is None or klc.next.end_klu is None:
|
||
continue
|
||
e_key, c_key = str(klc.end_time), str(klc.next.end_klu.time)
|
||
if e_key not in idx_of or c_key not in idx_of:
|
||
continue
|
||
fx_idx, confirm_idx = idx_of[e_key], idx_of[c_key]
|
||
if confirm_idx <= fx_idx:
|
||
continue
|
||
d = 1 if klc.fx == Chan_FX_TYPE.BOTTOM else -1
|
||
rows.append({
|
||
"confirm_ts": int(ts[confirm_idx]) + period,
|
||
"fx_ts": int(ts[fx_idx]),
|
||
"direction": d,
|
||
"price": float(klc.low if d == 1 else klc.high),
|
||
})
|
||
out = pd.DataFrame(rows)
|
||
if out.empty:
|
||
return pd.DataFrame(columns=["confirm_ts", "fx_ts", "direction", "price"])
|
||
return out.sort_values("confirm_ts").reset_index(drop=True)
|
||
|
||
|
||
def attach_htf_agree(sig: pd.DataFrame, df_ltf: pd.DataFrame, tl: pd.DataFrame) -> pd.DataFrame:
|
||
"""给每个小级别信号挂上「入场时刻之前最近的大级别分型」是否同向。
|
||
|
||
产出 htf_dir（+1 底 / -1 顶）与 htf_agree（1 同向 / 0 反向 / NaN 无可用分型）。
|
||
"""
|
||
out = sig.copy()
|
||
if sig.empty or tl.empty:
|
||
out["htf_dir"] = np.nan
|
||
out["htf_agree"] = np.nan
|
||
return out
|
||
|
||
ts_ltf = timestamps_ms(df_ltf)
|
||
entry_ts = ts_ltf[out["entry_idx"].to_numpy().astype(int)]
|
||
k = np.searchsorted(tl["confirm_ts"].to_numpy(), entry_ts, side="right") - 1
|
||
valid = k >= 0
|
||
k_safe = np.clip(k, 0, len(tl) - 1)
|
||
|
||
fx_dir = tl["direction"].to_numpy()[k_safe].astype(float)
|
||
out["htf_dir"] = np.where(valid, fx_dir, np.nan)
|
||
out["htf_agree"] = np.where(
|
||
valid, (fx_dir == out["direction"].to_numpy()).astype(float), np.nan
|
||
)
|
||
return out
|
||
|
||
|
||
def attach_zone_ladder(sig: pd.DataFrame, zones: pd.DataFrame) -> pd.DataFrame:
|
||
"""按信号方向取该中枢的阶梯标记：买看 z_above、卖看 z_below。
|
||
|
||
zone_i 是 find_fast_bsp3 内 enumerate 出的位置序号，故用 iloc 定位。
|
||
"""
|
||
out = sig.copy()
|
||
if sig.empty:
|
||
out["ladder_ok"] = pd.Series(dtype=bool)
|
||
return out
|
||
z = zones if "z_above" in zones.columns else add_zone_ladder(zones)
|
||
above = z["z_above"].to_numpy()
|
||
below = z["z_below"].to_numpy()
|
||
zi = out["zone_i"].to_numpy().astype(int)
|
||
ok = np.where(out["direction"].to_numpy() == 1, above[zi], below[zi])
|
||
out["ladder_ok"] = ok.astype(bool)
|
||
return out
|
||
|
||
|
||
def find_fast_bsp3(
|
||
df: pd.DataFrame,
|
||
zones: pd.DataFrame,
|
||
scan: int = 200,
|
||
pullback_win: int = 30,
|
||
tol: float = -1.0,
|
||
max_per_zone: int = 1,
|
||
diag: dict | None = None,
|
||
require_touch: bool = False,
|
||
) -> pd.DataFrame:
|
||
"""扫描每个中枢，找突破后回抽不回中枢的入场点。
|
||
|
||
zones 需含 zg / zd / available_ts，且 available_ts 已是可用时刻。
|
||
max_per_zone > 1 时，同一中枢在首次入场后继续往后找二次、三次突破回抽，
|
||
用来检验「趋势里同一中枢反复给机会」是否值得做。
|
||
|
||
tol 是「算作回抽中」的边界容差。require_touch=False 时它不再是入场门槛，
|
||
仅决定哪些K线被视为回抽中而跳过转强判定，故 tol 越大入场越晚。
|
||
默认负值 = 完全禁用该跳过，突破后每根都检查转强，滞后压到 2.2 根。
|
||
step34 实测滞后与收益严格单调：9.9根 PF1.88 / 5.8根 2.37 / 2.2根 2.86。
|
||
|
||
返回列：
|
||
entry_idx 实时可下单的K线
|
||
direction +1 三买 / -1 三卖
|
||
bo_idx 突破根
|
||
pb_idx 回抽极值根
|
||
lag entry_idx - bo_idx
|
||
depth 回抽深度（相对中枢边界，负值表示曾插入中枢）
|
||
occ 这是该中枢的第几次入场
|
||
"""
|
||
if zones.empty:
|
||
return pd.DataFrame()
|
||
|
||
ts = df["timestamp"].to_numpy()
|
||
close = df["close"].to_numpy(dtype=float)
|
||
high = df["high"].to_numpy(dtype=float)
|
||
low = df["low"].to_numpy(dtype=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"])
|
||
if zg <= zd:
|
||
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
|
||
|
||
edge = zg if d == 1 else zd
|
||
# 第二步：突破后监控回抽，回抽不跌回中枢且重新顺势 -> 入场
|
||
touched = False
|
||
pb_idx = None
|
||
pb_ext = None
|
||
entry_idx = None
|
||
fell_back = False
|
||
for j in range(bo_idx + 1, min(bo_idx + pullback_win + 1, n)):
|
||
# 收盘跌回中枢 -> 突破失效
|
||
if zd <= close[j] <= zg:
|
||
fell_back = True
|
||
break
|
||
# 回抽触及边界附近（允许 tol 的毛刺）
|
||
near = (low[j] <= edge * (1 + tol)) if d == 1 else (high[j] >= edge * (1 - tol))
|
||
if near:
|
||
touched = True
|
||
ext = low[j] if d == 1 else high[j]
|
||
if pb_ext is None or ((ext < pb_ext) if d == 1 else (ext > pb_ext)):
|
||
pb_ext, pb_idx = ext, j
|
||
continue
|
||
# 回抽后重新顺势：收盘创出前一根之上（三买）/ 之下（三卖）
|
||
# require_touch=False 时不强求回抽碰到中枢边界，
|
||
# 这样「突破后一去不回头」的强势段也能收进来。
|
||
if (touched and pb_idx is not None) or not require_touch:
|
||
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 None:
|
||
if occ == 1:
|
||
note("×突破后跌回中枢" if fell_back
|
||
else "×回抽未触及边界" if not touched
|
||
else "×触及边界但未转强")
|
||
# 这次突破没走成，从突破点之后继续找下一次
|
||
cursor = bo_idx + 1
|
||
continue
|
||
if pb_ext is None:
|
||
# 未触及边界就转强（require_touch=False）：取突破至入场间的实际极值
|
||
seg = slice(bo_idx + 1, entry_idx + 1)
|
||
pb_ext = float(low[seg].min() if d == 1 else high[seg].max())
|
||
pb_idx = int((low[seg].argmin() if d == 1 else high[seg].argmax())
|
||
+ bo_idx + 1)
|
||
if occ == 1:
|
||
note("√成交")
|
||
|
||
# 回抽深度：>0 表示未插入中枢，越大表示回抽越浅
|
||
depth = (pb_ext - zg) / zg if d == 1 else (zd - pb_ext) / zd
|
||
rows.append({
|
||
"entry_idx": entry_idx, "direction": d,
|
||
"bo_idx": bo_idx, "pb_idx": pb_idx,
|
||
"lag": entry_idx - bo_idx,
|
||
"depth": depth,
|
||
"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
|
||
# 相邻中枢可能突破到同一根K线，同一时刻只能有一个仓位，保留最早成型的那个
|
||
return (out.sort_values(["entry_idx", "occ", "zone_i"])
|
||
.drop_duplicates("entry_idx", keep="first")
|
||
.reset_index(drop=True))
|