jackyu66gitandCursor
081a57a90e
独立 wyckoff 引擎 + 按需 include_wyckoff；主站 Lightweight 绘制区间/阶段/事件/VP。 Co-authored-by: Cursor <cursoragent@cursor.com>
109 lines
2.9 KiB
Python
109 lines
2.9 KiB
Python
"""交易区间检测：ATR 容差下的近期震荡箱。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, Optional
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
|
|
def _atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
|
|
high = df["high"].astype(float)
|
|
low = df["low"].astype(float)
|
|
close = df["close"].astype(float)
|
|
prev_close = close.shift(1)
|
|
tr = pd.concat(
|
|
[
|
|
(high - low).abs(),
|
|
(high - prev_close).abs(),
|
|
(low - prev_close).abs(),
|
|
],
|
|
axis=1,
|
|
).max(axis=1)
|
|
return tr.rolling(period, min_periods=max(3, period // 2)).mean()
|
|
|
|
|
|
def detect_trading_range(
|
|
df: pd.DataFrame,
|
|
lookback: int = 120,
|
|
min_bars: int = 24,
|
|
atr_mult: float = 1.2,
|
|
tail_reserve: int = 12,
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
在最近 lookback 根内寻找高低点波动受控的连续段作为交易区间。
|
|
尾部预留 tail_reserve 根用于事件（Spring/SOS），不参与箱体边界计算。
|
|
"""
|
|
if df is None or len(df) < min_bars + 5:
|
|
return None
|
|
work = df.tail(lookback).reset_index(drop=True)
|
|
n = len(work)
|
|
reserve = min(tail_reserve, max(0, n - min_bars - 2))
|
|
core_end = n - reserve if reserve > 0 else n
|
|
core = work.iloc[:core_end]
|
|
if len(core) < min_bars:
|
|
core = work
|
|
core_end = n
|
|
reserve = 0
|
|
|
|
atr = _atr(work)
|
|
last_atr = float(atr.iloc[core_end - 1]) if atr.notna().iloc[:core_end].any() else float(
|
|
(core["high"] - core["low"]).mean()
|
|
)
|
|
if not np.isfinite(last_atr) or last_atr <= 0:
|
|
last_atr = float(core["close"].iloc[-1]) * 0.01
|
|
|
|
best = None
|
|
cn = len(core)
|
|
for length in range(min(cn, lookback), min_bars - 1, -4):
|
|
seg = core.iloc[-length:]
|
|
hi = float(seg["high"].max())
|
|
lo = float(seg["low"].min())
|
|
width = hi - lo
|
|
if width <= 0 or width > last_atr * atr_mult * 3.5:
|
|
continue
|
|
tol = last_atr * atr_mult * 0.35
|
|
near_hi = int((seg["high"] >= hi - tol).sum())
|
|
near_lo = int((seg["low"] <= lo + tol).sum())
|
|
if near_hi < 2 or near_lo < 2:
|
|
continue
|
|
inside = ((seg["close"] >= lo - tol) & (seg["close"] <= hi + tol)).mean()
|
|
if inside < 0.75:
|
|
continue
|
|
start_i = cn - length
|
|
end_i = cn - 1
|
|
mid = (hi + lo) / 2.0
|
|
last_c = float(work["close"].iloc[-1])
|
|
active = (lo - tol * 1.5) <= last_c <= (hi + tol * 1.5)
|
|
best = {
|
|
"start_idx": int(start_i),
|
|
"end_idx": int(end_i),
|
|
"high": hi,
|
|
"low": lo,
|
|
"mid": mid,
|
|
"active": bool(active),
|
|
"atr": last_atr,
|
|
"tol": tol,
|
|
"bars": int(length),
|
|
}
|
|
break
|
|
|
|
if best is None:
|
|
return None
|
|
|
|
def _ts(row) -> Any:
|
|
if "date" in work.columns and pd.notna(row["date"]):
|
|
return row["date"]
|
|
if "timestamp" in work.columns:
|
|
return row["timestamp"]
|
|
return None
|
|
|
|
best["start_time"] = _ts(work.iloc[best["start_idx"]])
|
|
# 区间时间结束取 core 末，事件可落在其后
|
|
best["end_time"] = _ts(work.iloc[best["end_idx"]])
|
|
offset = len(df) - len(work)
|
|
best["abs_start_idx"] = offset + best["start_idx"]
|
|
best["abs_end_idx"] = offset + best["end_idx"]
|
|
best["abs_scan_end_idx"] = offset + n - 1
|
|
return best
|