jackyu66gitandCursor
7292ef5dd4
把 step62 的 ext_run 质量过滤接到 step56 的低滞后版上——两者各解决一半约束， 是唯一同时处理滞后和识别精度的路径。 失败，且失败方式本身是判据：过滤器符号反了。5m 上延伸度分档在滞后版是 Q1 0.12 / Q4 0.46（越延伸越好），在低滞后版是 Q1 0.58 / Q4 0.33（越短越好）。 叠加后 PF 从 0.37 掉到 0.30~0.33，比不过滤更差。div 同样翻转。 同一批信号、同一个特征，换个入场时点最优方向就反过来，说明这些过滤效果是 入场时点的交互产物而非信号的稳定属性，继续挑阈值就是拟合噪声。 一类线六次独立进攻全部止步 1.0 以下：引擎原生 0.24、反手 0.92（因果回放后）、 实时重写 0.41、结构止损 0.36、用未来函数选样 0.83、质量过滤 0.54、 低滞后+过滤 0.33。第五项尤其说明问题——即使选样做到完美也只到 0.83。 判定关闭。病根是入场价已在结构底上方 2.9 ATR、实盘再晚 15 根，来自「等笔确认」 机制本身而非参数。识别可以优化（精度能翻倍），但识别从来不是瓶颈。 fast_bsp1 顺带补 ext_atr 字段，用 ATR 归一才与 step62 的 ext_run 同口径。 Co-authored-by: Cursor <cursoragent@cursor.com>
226 lines
9.6 KiB
Python
226 lines
9.6 KiB
Python
"""快速一类买卖点（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)
|
||
atr = df["atr"].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
|
||
# 延伸度用 ATR 归一，才和 step62 的 ext_run 同口径 —— 那里实测它是
|
||
# 唯一单调区分「趋势末端 vs 趋势中途」的特征（命中率 12%→44%）。
|
||
# 用相对边界的百分比会混入价格水平，不同币之间不可比。
|
||
a_ext = atr[ext_idx]
|
||
rows.append({
|
||
"entry_idx": entry_idx, "direction": -d,
|
||
"bo_idx": bo_idx, "ext_idx": ext_idx,
|
||
"lag": entry_idx - ext_idx,
|
||
"div": div_at,
|
||
"ext_atr": (abs(ext - edge) / a_ext
|
||
if np.isfinite(a_ext) and a_ext > 0 else np.nan),
|
||
"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))
|