jackyu66gitandCursor
9b72173285
研究侧的 fast_bsp3 一直只活在 research/lib/ 里，web 端看不到，回测与目视 两条线对不上。这次把它搬进引擎，作为独立的第四类买卖点。 之所以单独立类而不是当作 B3/S3 的低滞后版：step30/31 显示引擎原生的 B3/S3 统计上呈逆势、显著亏损（胜率 27.4%、PF 0.66、t −18.76），而同一组 过滤器把 B4 从 PF 1.59 提到 2.26 却对它无效（0.66→0.71）。两者选的是 不同的交易群体，不是同一信号的早晚两版。 - chanlun/analysis/fast_bsp.py 原样搬入 find_fast_bsp3 与 build_htf_zones， 另加 add_zone_ladder / htf_fx_timeline / attach_htf_agree - research/lib/ 两个模块改为转发，所有 step 脚本导入不变，信号逐条比对一致 - 大级别上下文用 resample 从同一份 df 构建，不额外拉数据，因此与界面上选的 周期和时间范围无关 - 前端三个复选框 + 过滤模式下拉；未过滤的原始信号用浅色，避免与主口径混淆 Co-authored-by: Cursor <cursoragent@cursor.com>
72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
"""区间套：用大级别中枢边界给小级别信号定位。
|
|
|
|
思路（缠论正统做法）：
|
|
大级别中枢的 zg/zd 是支撑压力位 -> 小级别在这些位置附近出现的分型+背驰，
|
|
才是高质量的预设转折点。位置本身就是过滤器，不需要等笔/中枢确认。
|
|
|
|
严格性：只使用在信号时刻之前就已经确认（sure_time 已过）的大级别中枢，
|
|
避免用到当时尚不可知的结构。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
|
|
# build_htf_zones 已移入引擎，与 web 共用同一份实现；annotate_position 仍是研究专用
|
|
from chanlun.analysis.fast_bsp import build_htf_zones # noqa: F401
|
|
|
|
|
|
def annotate_position(
|
|
sig: pd.DataFrame, df_ltf: pd.DataFrame, zones: pd.DataFrame, tol: float = 0.01
|
|
) -> pd.DataFrame:
|
|
"""给每个小级别信号标注它相对大级别中枢的位置。
|
|
|
|
tol —— 判定"贴近"边界的相对距离阈值（默认 1%）。
|
|
"""
|
|
if zones.empty:
|
|
out = sig.copy()
|
|
for c in ("near_support", "near_resistance", "inside_zone", "outside_zone", "zone_pos"):
|
|
out[c] = False if c != "zone_pos" else np.nan
|
|
return out
|
|
|
|
ts = df_ltf["timestamp"].to_numpy()
|
|
zone_avail = zones["available_ts"].to_numpy()
|
|
zg = zones["zg"].to_numpy()
|
|
zd = zones["zd"].to_numpy()
|
|
|
|
near_sup, near_res, inside, outside, zpos = [], [], [], [], []
|
|
for _, r in sig.iterrows():
|
|
i = int(r["confirm_idx"])
|
|
now_ts = ts[i]
|
|
price = float(r["price"])
|
|
# 最近一个在此刻之前已可用的大级别中枢
|
|
k = np.searchsorted(zone_avail, now_ts, side="right") - 1
|
|
if k < 0:
|
|
near_sup.append(False); near_res.append(False)
|
|
inside.append(False); outside.append(False); zpos.append(np.nan)
|
|
continue
|
|
z_g, z_d = zg[k], zd[k]
|
|
width = z_g - z_d
|
|
near_sup.append(abs(price - z_d) / price <= tol)
|
|
near_res.append(abs(price - z_g) / price <= tol)
|
|
inside.append(z_d <= price <= z_g)
|
|
outside.append(price > z_g or price < z_d)
|
|
zpos.append((price - z_d) / width if width > 0 else np.nan)
|
|
|
|
out = sig.copy()
|
|
out["near_support"] = near_sup
|
|
out["near_resistance"] = near_res
|
|
out["inside_zone"] = inside
|
|
out["outside_zone"] = outside
|
|
out["zone_pos"] = zpos # 0=中枢下沿, 1=中枢上沿
|
|
# 顺位：买信号贴支撑 / 卖信号贴压力，才算"位置正确"
|
|
out["position_ok"] = np.where(
|
|
out["direction"] == 1, out["near_support"], out["near_resistance"]
|
|
)
|
|
return out
|