jackyu66gitandCursor
7f393b93ed
删除根目录旧 Chan 模块、策略、配置、文档及 wyckoff 相关代码；更新缠论 pipeline 与笔中枢计算；补充 research 研究与 web 测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
89 lines
3.7 KiB
Python
89 lines
3.7 KiB
Python
"""真正的区间套：大级别分型定位 + 小级别三类买卖点入场。
|
|
|
|
结构（缠论正统）：
|
|
1. 大级别（1h / 4h）出顶底分型 —— 确认滞后仅 1 根，负责「预设转折点」与方向
|
|
2. 小级别（15m）在该位置形成中枢并被突破
|
|
3. 突破后回抽不回中枢 -> 15m 的第三类买卖点 —— 负责精确入场
|
|
|
|
关键：中枢与三买都在小级别，大级别只出分型。
|
|
"""
|
|
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]))
|
|
|
|
|
|
def htf_fx_timeline(sig_htf: pd.DataFrame, df_htf: pd.DataFrame) -> pd.DataFrame:
|
|
"""把大级别分型压成一条按确认时间排序的时间线。
|
|
|
|
confirm_ts 是该分型最早可被使用的时刻，用它做对齐可避免未来函数。
|
|
注意 timestamp 是K线开盘时刻，而分型要等这根K线收盘才算数，
|
|
所以整体后移一个大级别周期；否则小级别会提前一整根大级别K线拿到信号。
|
|
"""
|
|
ts = df_htf["timestamp"].to_numpy()
|
|
period = int(np.median(np.diff(ts))) if len(ts) > 1 else 0
|
|
out = pd.DataFrame({
|
|
"confirm_ts": ts[sig_htf["confirm_idx"].to_numpy().astype(int)] + period,
|
|
"fx_ts": ts[sig_htf["fx_idx"].to_numpy().astype(int)],
|
|
"direction": sig_htf["direction"].to_numpy(),
|
|
"price": sig_htf["price"].to_numpy(),
|
|
"is_divergence": sig_htf["is_divergence"].to_numpy(),
|
|
"ratio": sig_htf["ratio"].to_numpy(),
|
|
})
|
|
return out.sort_values("confirm_ts").reset_index(drop=True)
|
|
|
|
|
|
def attach_htf_context(
|
|
ev: pd.DataFrame, df_ltf: pd.DataFrame, tl: pd.DataFrame, prefix: str
|
|
) -> pd.DataFrame:
|
|
"""给每个小级别买卖点挂上「入场时刻之前最近的大级别分型」。
|
|
|
|
ev 需含 entry_idx。产出列：
|
|
{prefix}_dir 最近大级别分型方向（+1 底 / -1 顶）
|
|
{prefix}_age_bars 距该分型确认过了多少根小级别K线
|
|
{prefix}_agree 大级别分型方向与本信号方向是否一致
|
|
{prefix}_div 该大级别分型是否背驰
|
|
{prefix}_dist 入场价相对该分型极值的距离（相对值）
|
|
"""
|
|
out = ev.copy()
|
|
if tl.empty or ev.empty:
|
|
for c in ("dir", "age_bars", "agree", "div", "dist"):
|
|
out[f"{prefix}_{c}"] = np.nan
|
|
return out
|
|
|
|
ts_ltf = df_ltf["timestamp"].to_numpy()
|
|
close = df_ltf["close"].to_numpy(dtype=float)
|
|
entry_idx = ev["entry_idx"].to_numpy().astype(int)
|
|
entry_ts = ts_ltf[entry_idx]
|
|
|
|
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)
|
|
fx_ts = tl["confirm_ts"].to_numpy()[k_safe]
|
|
fx_px = tl["price"].to_numpy()[k_safe].astype(float)
|
|
fx_div = tl["is_divergence"].to_numpy()[k_safe].astype(float)
|
|
|
|
# 用小级别K线间隔把时间差换算成根数
|
|
step = np.median(np.diff(ts_ltf)) if len(ts_ltf) > 1 else 1
|
|
age = (entry_ts - fx_ts) / max(step, 1)
|
|
|
|
out[f"{prefix}_dir"] = np.where(valid, fx_dir, np.nan)
|
|
out[f"{prefix}_age_bars"] = np.where(valid, age, np.nan)
|
|
# 分型确认时刻当作该分型的唯一标识，用来判断多个小级别信号是否同源
|
|
out[f"{prefix}_fx_ts"] = np.where(valid, fx_ts, np.nan)
|
|
out[f"{prefix}_div"] = np.where(valid, fx_div, np.nan)
|
|
out[f"{prefix}_dist"] = np.where(
|
|
valid, (close[entry_idx] - fx_px) / np.clip(np.abs(fx_px), 1e-9, None), np.nan
|
|
)
|
|
out[f"{prefix}_agree"] = np.where(
|
|
valid, (fx_dir == out["direction"].to_numpy()).astype(float), np.nan
|
|
)
|
|
return out
|