jackyu66gitandCursor
7f393b93ed
删除根目录旧 Chan 模块、策略、配置、文档及 wyckoff 相关代码；更新缠论 pipeline 与笔中枢计算；补充 research 研究与 web 测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
107 lines
4.1 KiB
Python
107 lines
4.1 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]))
|
|
|
|
from chanlun import TF_DF
|
|
|
|
|
|
def build_htf_zones(df_htf: pd.DataFrame, tf: str, chan: TF_DF | None = None) -> pd.DataFrame:
|
|
"""算 pure 笔中枢，返回带生效时间的区间表。
|
|
|
|
available_ts —— 该中枢最早可被使用的时间戳（其确认时刻）。
|
|
传入已构建好的 chan 可避免重复跑一遍 pipeline（大数据集上省一半时间）。
|
|
"""
|
|
if chan is None:
|
|
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
|
|
ts_of = dict(zip(src["date"].dt.strftime("%Y-%m-%d %H:%M:%S"), src["timestamp"]))
|
|
|
|
rows = []
|
|
for zs in zs_list:
|
|
bis = getattr(zs, "bi_list", [])
|
|
if not bis:
|
|
continue
|
|
# 中枢可用时刻：构成它的最后一笔被确认之时
|
|
last_bi = bis[-1]
|
|
sure_key = str(getattr(last_bi, "sure_time", "") or "")
|
|
end_key = str(getattr(last_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 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
|