feat(ECR-007): Wyckoff Live Structure with Confirmed/Live isolation

Add live.py lifecycle and event candidates; assemble confirmed vs live
in engine; Summary partition; execution_signal source=confirmed only.
Keep strategies untouched; do not lower Confirmed thresholds for Live.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
2026-08-07 03:14:19 +08:00
co-authored by Cursor
parent 1e60ab3bfa
commit 276481e02c
33 changed files with 2527 additions and 401 deletions
+3 -2
View File
@@ -1,6 +1,7 @@
"""威科夫分析(启发式):交易区间 / 阶段 / 事件 / Volume Profile。"""
"""威科夫分析(启发式):交易区间 / 阶段 / 事件 / Volume Profile / Live"""
from __future__ import annotations
from .engine import analyze_wyckoff
from .live import execution_signal_from_wyckoff
__all__ = ["analyze_wyckoff"]
__all__ = ["analyze_wyckoff", "execution_signal_from_wyckoff"]
+145 -29
View File
@@ -1,12 +1,18 @@
"""威科夫分析入口"""
"""威科夫分析入口Cycle → Phase → Event → VP + LiveMULTI-CYCLE / LIVE-STRUCTURE)。
range.py 只产 TradingRangeConfirmed 走 events.pyLive 走 live.py。
cycles[0]=ACTIVE;禁止 cycles[-1] 取 active。
Execution 只消费 Confirmed(见 live.execution_signal_from_wyckoff)。
"""
from __future__ import annotations
from typing import Any, Dict, Optional
from typing import Any, Dict, List, Optional
import pandas as pd
from .events import build_phases, detect_bias_and_events
from .range import detect_trading_range
from .live import analyze_live_structure
from .range import detect_trading_ranges
from .volume_profile import compute_volume_profile
@@ -21,31 +27,55 @@ def _fmt_time(v) -> Optional[str]:
return str(v)
def analyze_wyckoff(df: pd.DataFrame, lookback: int = 120, vp_bins: int = 50) -> Dict[str, Any]:
"""
对主周期 OHLCV DataFrame 做威科夫启发式分析。
需要列: open, high, low, close, volume;建议有 date 或 timestamp。
"""
empty = {
def _empty(vp_bins: int) -> Dict[str, Any]:
return {
"cycles": [],
"trading_range": None,
"bias": "unknown",
"phases": [],
"events": [],
"volume_profile": {"bins": [], "poc": None, "vah": None, "val": None, "bin_count": vp_bins},
"volume_confirm": {"avg_volume": 0.0, "event_checks": {}},
"live": None,
}
if df is None or len(df) < 30:
return empty
if not all(c in df.columns for c in ("open", "high", "low", "close")):
return empty
work = df.copy()
if "volume" not in work.columns:
work["volume"] = 1.0
tr = detect_trading_range(work, lookback=lookback)
if tr is None:
return empty
def _confidence_for_confirmed(
tr: Dict[str, Any],
phases: List[Dict[str, Any]],
events: List[Dict[str, Any]],
) -> Dict[str, float]:
range_c = float(tr.get("range_confidence") or 0.5)
labels = {p.get("phase") for p in phases}
phase_c = 0.35
if "A" in labels and "B" in labels:
phase_c += 0.15
if "C" in labels:
phase_c += 0.2
if "D" in labels or "E" in labels:
phase_c += 0.15
phase_c = min(0.95, phase_c)
types = {e.get("type") for e in events}
event_c = 0.25
for t in ("Spring", "UTAD", "SOS", "SOW", "LPS", "LPSY"):
if t in types:
event_c += 0.12
event_c = min(0.95, event_c)
overall = 0.4 * range_c + 0.3 * phase_c + 0.3 * event_c
return {
"range": round(range_c, 3),
"phase": round(phase_c, 3),
"event": round(event_c, 3),
"overall": round(overall, 3),
}
def _build_cycle(
work: pd.DataFrame,
tr: Dict[str, Any],
cycle_id: int,
vp_bins: int,
) -> Dict[str, Any]:
bias, events, volume_confirm = detect_bias_and_events(work, tr)
phases = build_phases(work, tr, bias, events)
vp = compute_volume_profile(
@@ -54,27 +84,113 @@ def analyze_wyckoff(df: pd.DataFrame, lookback: int = 120, vp_bins: int = 50) ->
int(tr["abs_end_idx"]),
bin_count=vp_bins,
)
trading_range = {
"start_time": _fmt_time(tr.get("start_time")),
"end_time": _fmt_time(tr.get("end_time")),
"high": float(tr["high"]),
"low": float(tr["low"]),
"mid": float(tr["mid"]),
"active": bool(tr.get("active", True)),
"bars": int(tr.get("bars", 0)),
}
for ev in events:
ev["time"] = _fmt_time(ev.get("time"))
for ph in phases:
ph["start_time"] = _fmt_time(ph.get("start_time"))
ph["end_time"] = _fmt_time(ph.get("end_time"))
is_active = cycle_id == 0
trading_range = {
"start_time": _fmt_time(tr.get("start_time")),
"end_time": _fmt_time(tr.get("end_time")),
"high": float(tr["high"]),
"low": float(tr["low"]),
"mid": float(tr["mid"]),
"active": bool(is_active),
"bars": int(tr.get("bars", 0)),
}
conf = _confidence_for_confirmed(tr, phases, events)
# Live 层:仅 ACTIVE 周期做推演;历史周期归档为 COMPLETED
if is_active:
live = analyze_live_structure(
work, tr, confirmed_events=events, confirmed_phases=phases, bias=bias,
)
lifecycle = live.get("lifecycle") or "FORMING"
else:
live = None
lifecycle = "COMPLETED"
return {
"id": int(cycle_id),
"role": "latest" if is_active else "historical",
# MULTI-CYCLE:时间线角色
"status": "ACTIVE" if is_active else "HISTORICAL",
# LIVE-STRUCTURE:生命周期
"lifecycle": lifecycle,
"direction": "latest" if is_active else "historical",
"period": {
"start_time": _fmt_time(tr.get("start_time")),
"end_time": _fmt_time(tr.get("end_time")),
"bars": int(tr.get("bars", 0)),
},
"confidence": conf,
"trading_range": trading_range,
"bias": bias,
# 兼容旧读法:顶层 phases/events = confirmed
"phases": phases,
"events": events,
"confirmed": {
"phases": phases,
"events": events,
"volume_confirm": volume_confirm,
},
"live": live,
"volume_profile": vp,
"volume_confirm": volume_confirm,
}
def analyze_wyckoff(
df: pd.DataFrame,
lookback: int = 120,
vp_bins: int = 50,
min_bars: int = 24,
atr_mult: float = 1.2,
range_start_time=None,
prefer_start_time=None,
max_cycles: int = 8,
) -> Dict[str, Any]:
"""
多周期威科夫分析。
cycles[0] = ACTIVE;顶层 phases/events 只镜像 Confirmed。
顶层 live 镜像 cycles[0].live。
"""
empty = _empty(vp_bins)
if df is None or len(df) < 30:
return empty
if not all(c in df.columns for c in ("open", "high", "low", "close")):
return empty
work = df.copy()
if "volume" not in work.columns:
work["volume"] = 1.0
trs = detect_trading_ranges(
work,
lookback=lookback,
min_bars=max(8, int(min_bars)),
atr_mult=atr_mult,
max_cycles=max(1, min(8, int(max_cycles))),
prefer_start_time=prefer_start_time,
range_start_time=range_start_time,
)
if not trs:
return empty
cycles: List[Dict[str, Any]] = []
for i, tr in enumerate(trs):
cycles.append(_build_cycle(work, tr, cycle_id=i, vp_bins=vp_bins))
active = cycles[0]
return {
"cycles": cycles,
"trading_range": active["trading_range"],
"bias": active["bias"],
"phases": active["confirmed"]["phases"],
"events": active["confirmed"]["events"],
"volume_profile": active["volume_profile"],
"volume_confirm": active["volume_confirm"],
"live": active.get("live"),
"lifecycle": active.get("lifecycle"),
}
+157 -35
View File
@@ -1,7 +1,7 @@
"""威科夫阶段与事件(启发式)。"""
from __future__ import annotations
from typing import Any, Dict, List, Tuple
from typing import Any, Dict, List, Optional, Tuple
import numpy as np
import pandas as pd
@@ -29,6 +29,9 @@ def detect_bias_and_events(
) -> Tuple[str, List[Dict[str, Any]], Dict[str, Any]]:
"""
返回 bias、events、volume_confirm。
Spring/UTAD 相对「结构高低」判定:取区间内次低/次高(剔除单根极值),
避免箱体把假破低点吃进 lo 后永远刺不破、从而无 C 阶段。
"""
hi = float(tr["high"])
lo = float(tr["low"])
@@ -38,6 +41,24 @@ def detect_bias_and_events(
e = int(tr["abs_end_idx"])
events: List[Dict[str, Any]] = []
# 结构边界:用次低/次高作假破参照(至少 8 根才启用)
seg = df.iloc[s : e + 1]
event_lo, event_hi = lo, hi
if len(seg) >= 8:
lows = seg["low"].astype(float)
highs = seg["high"].astype(float)
# nsmallest(2) 的较大者 = 次低;nlargest(2) 的较小者 = 次高
event_lo = float(lows.nsmallest(min(2, len(lows))).iloc[-1])
event_hi = float(highs.nlargest(min(2, len(highs))).iloc[-1])
# 勿比公布箱沿更「松」:结构带应在箱内
event_lo = max(event_lo, lo)
event_hi = min(event_hi, hi)
# 若次低仍等于极值(多根同价),略抬参照便于识别收回
if abs(event_lo - lo) < 1e-12:
event_lo = lo + max(tol * 0.35, (hi - lo) * 0.02)
if abs(event_hi - hi) < 1e-12:
event_hi = hi - max(tol * 0.35, (hi - lo) * 0.02)
# 扫描区间内及之后(含 tail_reserve
scan_end = int(tr.get("abs_scan_end_idx", min(len(df) - 1, e + 15)))
scan_end = min(len(df) - 1, max(scan_end, e))
@@ -57,8 +78,8 @@ def detect_bias_and_events(
avg_v = _avg_vol(df, i)
ratio = vol / avg_v if avg_v else 0.0
# Spring: pierce below low then close back above low
if spring is None and low < lo - tol * 0.5 and close >= lo - tol * 0.2:
# Spring: pierce below structural support then close back
if spring is None and low < event_lo - tol * 0.35 and close >= event_lo - tol * 0.35:
vol_ok = ratio <= 1.35 or (i + 1 <= scan_end and float(df.iloc[min(i + 1, scan_end)]["volume"]) / avg_v < 1.2)
spring = {
"type": "Spring",
@@ -70,8 +91,8 @@ def detect_bias_and_events(
"idx": i,
}
# UTAD: pierce above high then close back below
if utad is None and high > hi + tol * 0.5 and close <= hi + tol * 0.2:
# UTAD: pierce above structural resistance then close back
if utad is None and high > event_hi + tol * 0.35 and close <= event_hi + tol * 0.35:
vol_ok = ratio >= 0.8
utad = {
"type": "UTAD",
@@ -154,11 +175,15 @@ def detect_bias_and_events(
}
break
# 冲突清理:已判定吸筹且有 SOS 时,丢弃更早的 UTAD(避免阶段/图面误导)
# 派发且有 SOW 时,丢弃更晚才合理的 Spring 假信号同理在偏置后再滤
keep = []
for ev in (spring, sos, lps, utad, sod, lpsy):
if ev:
events.append({k: v for k, v in ev.items() if k != "idx"})
if not ev:
continue
keep.append(ev)
# bias
# bias(先算)
last_c = float(df["close"].iloc[-1])
bias = "unknown"
if sos and (not sod or int(sos.get("idx", 0)) >= int(sod.get("idx", 0))):
@@ -174,6 +199,16 @@ def detect_bias_and_events(
else:
bias = "distribution"
filtered = []
for ev in keep:
if bias == "accumulation" and ev["type"] == "UTAD" and sos and int(ev["idx"]) <= int(sos["idx"]):
continue
if bias == "distribution" and ev["type"] == "Spring" and sod and int(ev["idx"]) <= int(sod["idx"]):
continue
filtered.append(ev)
events = [{k: v for k, v in ev.items() if k != "idx"} for ev in filtered]
avg_volume = float(df["volume"].astype(float).iloc[max(0, e - 20) : e + 1].mean()) if "volume" in df.columns else 0.0
volume_confirm = {
"avg_volume": avg_volume,
@@ -189,59 +224,146 @@ def build_phases(
events: List[Dict[str, Any]],
min_bars: int = 3,
) -> List[Dict[str, Any]]:
"""按时间切分 A–E 粗阶段;保证非重叠且每段至少 min_bars 根(空间不足则截断尾部阶段)。"""
"""
按威科夫事件锚点切分 A–E(启发式)。
吸筹:A停止 → B筑底 → C测试(Spring) → D拉升(SOS…LPS) → E离开
派发:A停止 → B筑顶 → C测试(UTAD) → D派发(SOW…LPSY) → E离开
无 Spring/UTAD 时:若已有 SOS/SOW,用突破前末次沿带测试补 C;仍无则省略 C。
"""
s = int(tr["abs_start_idx"])
e = int(tr["abs_end_idx"])
hi = float(tr["high"])
lo = float(tr["low"])
n_last = len(df) - 1
min_span = max(2, min_bars - 1)
range_len = max(1, e - s)
event_idx = {}
for ev in events:
t = ev.get("time")
for i in range(s, min(len(df), e + 20)):
def _match_idx(t) -> Optional[int]:
if t is None:
return None
lo = max(0, s - 2)
hi = min(len(df), e + 40)
for i in range(lo, hi):
if _bar_time(df, i) == t:
event_idx[ev["type"]] = i
return i
try:
tt = pd.Timestamp(t)
sample = None
if "date" in df.columns and len(df):
sample = df["date"].iloc[min(s, n_last)]
if sample is not None and getattr(sample, "tzinfo", None) is not None and tt.tzinfo is None:
tt = tt.tz_localize(sample.tzinfo)
for i in range(lo, hi):
bt = _bar_time(df, i)
try:
if abs((pd.Timestamp(bt) - tt).total_seconds()) <= 1:
return i
except Exception:
continue
except Exception:
pass
return None
event_idx: Dict[str, int] = {}
for ev in events:
idx = _match_idx(ev.get("time"))
if idx is not None:
event_idx[str(ev.get("type"))] = idx
accum = bias != "distribution"
if accum:
c_ev = event_idx.get("Spring")
d_ev = event_idx.get("SOS")
d_tail = event_idx.get("LPS") or d_ev
else:
c_ev = event_idx.get("UTAD")
d_ev = event_idx.get("SOW")
d_tail = event_idx.get("LPSY") or d_ev
# 有 D 无明确测试事件时:用突破前最后一次触及下/上沿作为 C(次级测试)
if c_ev is None and d_ev is not None:
band = lo + (hi - lo) * 0.28 if accum else hi - (hi - lo) * 0.28
for i in range(int(d_ev) - 1, s + 1, -1):
row = df.iloc[i]
if accum and float(row["low"]) <= band:
c_ev = i
break
if not accum and float(row["high"]) >= band:
c_ev = i
break
a_end = s + max(min_bars, (e - s) // 5)
c_anchor = event_idx.get("Spring") or event_idx.get("UTAD") or (s + (e - s) // 2)
d_anchor = event_idx.get("SOS") or event_idx.get("SOW") or e
def _lab(phase: str) -> str:
if bias == "distribution":
m = {"A": "A停止上涨", "B": "B筑顶", "C": "C测试", "D": "D派发", "E": "E下跌"}
else:
if accum:
m = {"A": "A停止下跌", "B": "B筑底", "C": "C测试", "D": "D拉升", "E": "E离开"}
else:
m = {"A": "A停止上涨", "B": "B筑顶", "C": "C测试", "D": "D派发", "E": "E离开"}
return m.get(phase, phase)
# 理想切点(随后再强制非重叠 + 最小跨度)
raw = [
("A", s, a_end),
("B", a_end, c_anchor),
("C", c_anchor, d_anchor),
("D", d_anchor, min(n_last, d_anchor + max(min_bars, (e - s) // 6))),
("E", min(n_last, d_anchor + max(min_bars, (e - s) // 6)), min(n_last, max(e, d_anchor + max(min_bars * 2, 8)))),
]
a_end = s + max(min_bars, range_len // 5)
c_start = c_end = None
if c_ev is not None:
c_start = max(s, int(c_ev) - 1)
c_end = min(n_last, int(c_ev) + 1)
if d_ev is not None:
d_start = int(d_ev)
d_end = min(n_last, max(int(d_tail or d_ev), d_start) + max(min_bars, range_len // 8))
if d_tail is not None:
d_end = max(d_end, min(n_last, int(d_tail) + 1))
else:
d_start = d_end = None
if c_start is not None:
b_end = max(a_end + 1, c_start)
elif d_start is not None:
b_end = max(a_end + 1, d_start)
else:
b_end = max(a_end + 1, e)
if d_end is not None:
e_start = min(n_last, d_end)
e_end = n_last
else:
e_start = e_end = None
raw = [("A", s, a_end), ("B", a_end, b_end)]
if c_start is not None and c_end is not None:
raw.append(("C", c_start, c_end))
if d_start is not None and d_end is not None:
raw.append(("D", d_start, d_end))
if e_start is not None and e_end is not None and e_end > e_start:
raw.append(("E", e_start, e_end))
phases: List[Dict[str, Any]] = []
cursor = s
for phase, _a, _b in raw:
if cursor >= n_last:
break
a = max(int(_a), cursor)
b = int(max(_b, a + min_span))
b = int(max(int(_b), a))
need = 1 if phase == "C" else min_span
if b < a + need:
b = min(n_last, a + need)
b = int(np.clip(b, a, n_last))
if b - a < min_span:
# 尾部空间不足:并入上一段终点并停止新增
if phases:
phases[-1]["end_time"] = _bar_time(df, n_last)
break
if b < a:
continue
if phases and phases[-1].get("_a") == a and phases[-1].get("_b") == b:
continue
phases.append(
{
"phase": phase,
"label": _lab(phase),
"start_time": _bar_time(df, a),
"end_time": _bar_time(df, b),
"_a": a,
"_b": b,
}
)
cursor = b
for p in phases:
p.pop("_a", None)
p.pop("_b", None)
return phases
+258
View File
@@ -0,0 +1,258 @@
"""威科夫 Live / Developing 层(WYCKOFF-LIVE-STRUCTURE-001)。
独立于 Confirmed Engine:不修改 events 确认条件,不写入 confirmed.events。
Execution 不得消费本模块输出。
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional, Set
import numpy as np
import pandas as pd
def _avg_vol(df: pd.DataFrame, i: int, win: int = 20) -> float:
a = max(0, i - win + 1)
v = df["volume"].astype(float).iloc[a : i + 1]
m = float(v.mean()) if len(v) else 0.0
return m if m > 0 else 1.0
def _empty_live() -> Dict[str, Any]:
return {
"lifecycle": "UNKNOWN",
"range_formation": None,
"phase_candidate": None,
"event_candidates": [],
"next_expected": None,
"confidence": {
"cycle": 0.0,
"phase": 0.0,
"event": 0.0,
"structure": 0.0,
"volume": 0.0,
"overall": 0.0,
},
"note": "",
}
def analyze_live_structure(
df: pd.DataFrame,
tr: Optional[Dict[str, Any]],
confirmed_events: Optional[List[Dict[str, Any]]] = None,
confirmed_phases: Optional[List[Dict[str, Any]]] = None,
bias: str = "unknown",
) -> Dict[str, Any]:
"""
基于当前 TradingRange 与已确认事件,推演 Live candidates。
confirmed_* 只读,用于避免重复提示已确认事件,不修改之。
"""
out = _empty_live()
if df is None or len(df) < 20 or tr is None:
out["note"] = "insufficient structure"
return out
confirmed_events = confirmed_events or []
confirmed_phases = confirmed_phases or []
confirmed_types: Set[str] = {str(e.get("type")) for e in confirmed_events if e.get("type")}
s = int(tr["abs_start_idx"])
e = int(tr["abs_end_idx"])
scan_end = int(tr.get("abs_scan_end_idx", len(df) - 1))
scan_end = min(len(df) - 1, max(scan_end, e))
hi = float(tr["high"])
lo = float(tr["low"])
mid = float(tr["mid"])
tol = float(tr.get("tol") or (hi - lo) * 0.05)
atr = float(tr.get("atr") or max((hi - lo) * 0.2, 1e-9))
seg = df.iloc[s : e + 1]
if len(seg) < 8:
out["note"] = "range too short"
return out
# —— Range Formation(横盘 / 波动收敛)——
closes = seg["close"].astype(float)
highs = seg["high"].astype(float)
lows = seg["low"].astype(float)
vols = seg["volume"].astype(float) if "volume" in seg.columns else pd.Series([1.0] * len(seg))
half = max(4, len(seg) // 2)
vol_early = float(np.std(closes.iloc[:half])) if half > 1 else 0.0
vol_late = float(np.std(closes.iloc[-half:])) if half > 1 else 0.0
width = hi - lo
width_atr = width / atr if atr > 0 else 99.0
converging = vol_early > 1e-12 and vol_late < vol_early * 0.85
range_ok = 1.2 <= width_atr <= 10.0 and len(seg) >= 16
structure_score = 0.35
if range_ok:
structure_score += 0.25
if converging:
structure_score += 0.2
if width_atr <= 6.0:
structure_score += 0.1
structure_score = float(min(0.95, structure_score))
out["range_formation"] = {
"potential_trading_range": bool(range_ok),
"converging": bool(converging),
"width_atr": round(width_atr, 3),
"bars": int(len(seg)),
}
# —— 最近 K 形态(Phase C / Event candidates)——
i = scan_end
row = df.iloc[i]
o = float(row["open"])
h = float(row["high"])
l = float(row["low"])
c = float(row["close"])
rng = max(h - l, 1e-9)
lower_wick = min(o, c) - l
upper_wick = h - max(o, c)
avg_v = _avg_vol(df, i)
vol = float(row["volume"]) if "volume" in df.columns else avg_v
vol_ratio = vol / avg_v if avg_v else 1.0
volume_score = float(np.clip(1.1 - abs(vol_ratio - 1.0) * 0.35, 0.2, 0.95))
phase_candidate = None
phase_conf = 0.0
# Phase C:测低 + 下影 + 缩量(吸筹语境)
near_lo = l <= lo + tol * 1.2
test_low = l < mid and lower_wick >= rng * 0.35
vol_contract = vol_ratio <= 1.05
if bias != "distribution" and near_lo and test_low and vol_contract:
phase_candidate = "C"
phase_conf = 0.55 + (0.1 if lower_wick >= rng * 0.5 else 0) + (0.08 if vol_ratio < 0.9 else 0)
# Phase D 候选:价格在箱上半、有上破意图但未确认 SOS
elif c >= mid and (h >= hi - tol or c > hi - tol * 0.5):
phase_candidate = "D"
phase_conf = 0.5 + (0.1 if c > mid else 0)
elif c < mid and (l <= lo + tol):
phase_candidate = "B"
phase_conf = 0.45
# 已有 confirmed phase 时,candidate 取「下一阶段」提示,不覆盖事实
confirmed_phase_set = {str(p.get("phase")) for p in confirmed_phases}
if "E" in confirmed_phase_set:
phase_candidate = phase_candidate or "E"
phase_conf = max(phase_conf, 0.7)
elif "D" in confirmed_phase_set and phase_candidate is None:
phase_candidate = "D"
phase_conf = max(phase_conf, 0.65)
out["phase_candidate"] = phase_candidate
phase_conf = float(min(0.92, phase_conf))
# —— Event candidates(仅 Spring / SOS / LPS / UTAD)——
candidates: List[Dict[str, Any]] = []
def _add(typ: str, conf: float, note: str) -> None:
if typ in confirmed_types:
return # 已确认则不再作为 candidate
candidates.append(
{
"type": typ,
"confidence": round(float(min(0.9, conf)), 3),
"confirmed": False,
"note": note,
}
)
# Spring candidate:刺破或贴近下沿,收盘收回,但未达 Confirmed 规则(或不在 confirmed
pierce_lo = l < lo - tol * 0.15
close_back = c >= lo - tol * 0.5
if pierce_lo and close_back:
_add("Spring", 0.5 + (0.12 if vol_ratio <= 1.2 else 0) + (0.08 if close_back else 0), "假破下沿收回(未确认)")
elif l <= lo + tol * 0.35 and close_back and lower_wick >= rng * 0.4:
_add("Spring", 0.45 + (0.1 if vol_contract else 0), "测下沿长下影(未确认)")
# UTAD candidate
pierce_hi = h > hi + tol * 0.15
close_back_dn = c <= hi + tol * 0.5
if pierce_hi and close_back_dn:
_add("UTAD", 0.5 + (0.1 if vol_ratio >= 0.9 else 0), "假破上沿跌回(未确认)")
# SOS candidate:接近/轻破上沿,量能一般,未确认
if c > hi - tol * 0.4 or h >= hi:
sos_conf = 0.48 + (0.12 if c > hi else 0) + (0.1 if vol_ratio >= 1.05 else 0)
_add("SOS", sos_conf, "上破/逼近箱顶(未确认)")
# LPS candidate:站上 mid/上沿带后回踩
if c >= mid and l >= mid - tol * 1.5 and l > lo + (hi - lo) * 0.25:
_add("LPS", 0.46 + (0.1 if vol_ratio <= 1.0 else 0), "箱内上沿带回踩(未确认)")
candidates.sort(key=lambda x: x["confidence"], reverse=True)
out["event_candidates"] = candidates[:4]
event_score = float(candidates[0]["confidence"]) if candidates else 0.25
# next_expected(简规则)
next_exp = None
if "Spring" in confirmed_types and "SOS" not in confirmed_types:
next_exp = "SOS"
elif "SOS" in confirmed_types and "LPS" not in confirmed_types:
next_exp = "LPS"
elif "UTAD" in confirmed_types and "SOW" not in confirmed_types:
next_exp = "SOW"
elif any(c["type"] == "Spring" for c in candidates):
next_exp = "Test"
elif any(c["type"] == "SOS" for c in candidates):
next_exp = "LPS"
out["next_expected"] = next_exp
# —— lifecycle ——
key_confirmed = confirmed_types & {"Spring", "SOS", "UTAD", "SOW", "LPS", "LPSY"}
if key_confirmed:
lifecycle = "CONFIRMED"
elif range_ok or phase_candidate or candidates:
lifecycle = "FORMING"
else:
lifecycle = "UNKNOWN"
out["lifecycle"] = lifecycle
cycle_c = structure_score
overall = 0.35 * cycle_c + 0.25 * phase_conf + 0.25 * event_score + 0.15 * volume_score
out["confidence"] = {
"cycle": round(cycle_c, 3),
"phase": round(phase_conf, 3),
"event": round(event_score, 3),
"structure": round(structure_score, 3),
"volume": round(volume_score, 3),
"overall": round(float(overall), 3),
}
parts = []
if out["range_formation"]["potential_trading_range"]:
parts.append("Potential TR")
if phase_candidate:
parts.append(f"Phase {phase_candidate} candidate")
if candidates:
parts.append(f"{candidates[0]['type']} candidate")
out["note"] = "; ".join(parts) if parts else "observing"
return out
def execution_signal_from_wyckoff(payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""
Execution 边界:只允许 Confirmed。
返回 source='confirmed' 的信号描述;Live-only 时返回 None。
"""
if not payload:
return None
cycles = payload.get("cycles") or []
active = cycles[0] if cycles else None
events = []
if active and isinstance(active.get("confirmed"), dict):
events = list(active["confirmed"].get("events") or [])
if not events:
# 兼容旧顶层 events(均为 confirmed 镜像)
events = list(payload.get("events") or [])
if not events:
return None
last = events[-1]
return {
"source": "confirmed",
"type": last.get("type"),
"time": last.get("time"),
"lifecycle": (active or {}).get("lifecycle") or "CONFIRMED",
}
+371 -57
View File
@@ -1,11 +1,18 @@
"""交易区间检测:ATR 容差下按评分选取近期震荡箱。"""
"""交易区间检测:仅负责 TradingRange(起止/高低/结构分)。
WYCKOFF-MULTI-CYCLE-001Phase/Event/VP 不得进入本模块。
过滤顺序固定:detect → quality → trend → overlap(<0.2) → accept → mask。
"""
from __future__ import annotations
from typing import Any, Dict, Optional
from typing import Any, Dict, List, Optional, Tuple
import numpy as np
import pandas as pd
MAX_CYCLES = 8
OVERLAP_RATIO_MAX = 0.2
def _atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
high = df["high"].astype(float)
@@ -23,6 +30,15 @@ def _atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
return tr.rolling(period, min_periods=max(3, period // 2)).mean()
def _robust_width(seg: pd.DataFrame) -> float:
"""用 90/10 分位估宽,避免单根影线把长窗卡死。"""
h = seg["high"].astype(float)
l = seg["low"].astype(float)
if len(seg) < 6:
return float(h.max() - l.min())
return float(np.nanpercentile(h, 90) - np.nanpercentile(l, 10))
def _score_segment(
length: int,
near_hi: int,
@@ -31,27 +47,161 @@ def _score_segment(
width: float,
atr: float,
) -> float:
"""触边密度 + 箱内比例 − 相对宽度;弱奖励长度以免只追最长"""
touch_density = (near_hi + near_lo) / float(max(length, 1))
"""结构质量分(非 Phase/Event"""
touch = min(near_hi, 6) + min(near_lo, 6)
width_pen = (width / atr) if atr > 0 else width
return touch_density * 50.0 + float(inside) * 30.0 - width_pen * 3.0 + min(length / 40.0, 2.0)
return float(touch) * 4.0 + float(inside) * 25.0 - width_pen * 3.0 + min(length / 40.0, 2.0)
def detect_trading_range(
def _time_col(df: pd.DataFrame) -> Optional[str]:
if "date" in df.columns:
return "date"
if "timestamp" in df.columns:
return "timestamp"
return None
def _bar_index_at_or_after(work: pd.DataFrame, ts: Any) -> Optional[int]:
col = _time_col(work)
if col is None or ts is None:
return None
try:
target = pd.Timestamp(ts)
except Exception:
return None
series = pd.to_datetime(work[col], utc=True, errors="coerce")
if target.tzinfo is None:
target = target.tz_localize("UTC")
else:
target = target.tz_convert("UTC")
if series.isna().all():
return None
ge = series >= target
if ge.any():
return int(np.flatnonzero(ge.to_numpy())[0])
return 0
def _pack_range(
work: pd.DataFrame,
df: pd.DataFrame,
lookback: int = 120,
start_i: int,
end_i: int,
hi: float,
lo: float,
tol: float,
last_atr: float,
score: float,
n: int,
window_offset: int = 0,
) -> Dict[str, Any]:
"""组装 TradingRange(仅结构字段)。"""
mid = (hi + lo) / 2.0
last_c = float(work["close"].iloc[min(end_i, len(work) - 1)])
price_in_box = (lo - tol * 1.5) <= last_c <= (hi + tol * 1.5)
bars = int(end_i - start_i + 1)
# 结构置信:归一化 score(启发式)
range_conf = float(np.clip(score / 55.0, 0.05, 0.99))
best = {
"start_idx": int(start_i),
"end_idx": int(end_i),
"high": float(hi),
"low": float(lo),
"mid": float(mid),
"active": bool(price_in_box),
"atr": float(last_atr),
"tol": float(tol),
"bars": bars,
"score": float(score),
"quality": float(score),
"range_confidence": range_conf,
}
def _ts(row) -> Any:
col = _time_col(work)
if col and pd.notna(row[col]):
return row[col]
return None
best["start_time"] = _ts(work.iloc[best["start_idx"]])
best["end_time"] = _ts(work.iloc[best["end_idx"]])
# window_offsetslice 相对父 DataFrame 的起点;勿用 len(df)-len(work)
offset = int(window_offset)
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
def _overlap_ratio(a0: int, a1: int, b0: int, b1: int) -> float:
"""两闭区间重叠长度 / 较短区间长度。"""
lo = max(a0, b0)
hi = min(a1, b1)
if hi < lo:
return 0.0
overlap = hi - lo + 1
shorter = min(a1 - a0 + 1, b1 - b0 + 1)
if shorter <= 0:
return 0.0
return float(overlap) / float(shorter)
def _passes_quality(tr: Dict[str, Any], min_bars: int) -> bool:
if tr is None:
return False
if int(tr.get("bars") or 0) < max(8, min_bars // 2):
return False
if float(tr.get("score") or 0) < 12.0:
return False
hi = float(tr["high"])
lo = float(tr["low"])
atr = float(tr.get("atr") or 0) or 1.0
if (hi - lo) / atr > 12.0:
return False
return True
def _passes_trend_filter(work: pd.DataFrame, tr: Dict[str, Any]) -> bool:
"""趋势污染:定向位移过大则非震荡箱。"""
s = int(tr["start_idx"])
e = int(tr["end_idx"])
seg = work.iloc[s : e + 1]
if len(seg) < 8:
return False
c0 = float(seg["close"].iloc[0])
c1 = float(seg["close"].iloc[-1])
atr = float(tr.get("atr") or 0) or 1.0
drift = abs(c1 - c0) / atr
# 相对箱宽:漂移占箱宽过大 → 趋势
width = max(float(tr["high"]) - float(tr["low"]), atr)
drift_frac = abs(c1 - c0) / width
if drift > 6.0 and drift_frac > 0.55:
return False
return True
def _detect_in_window(
df: pd.DataFrame,
win_start: int,
win_end: int,
min_bars: int = 24,
atr_mult: float = 1.2,
tail_reserve: int = 12,
prefer_start_time: Any = None,
range_start_time: Any = None,
) -> Optional[Dict[str, Any]]:
"""
最近 lookback 根内寻找高低点波动受控的连续段作为交易区间
尾部预留 tail_reserve 根用于事件(Spring/SOS),不参与箱体边界计算
在硬门槛之上按评分取最优段(非仅最长窗口)。
df[win_start:win_end+1] 内检测单个 TradingRange
只返回箱体结构,不含 Phase/Event/VP
"""
if df is None or len(df) < min_bars + 5:
if df is None or win_end < win_start:
return None
work = df.tail(lookback).reset_index(drop=True)
slice_df = df.iloc[win_start : win_end + 1].reset_index(drop=True)
lookback = len(slice_df)
if lookback < min_bars + 5:
return None
work = slice_df
n = len(work)
reserve = min(tail_reserve, max(0, n - min_bars - 2))
core_end = n - reserve if reserve > 0 else n
@@ -68,61 +218,225 @@ def detect_trading_range(
if not np.isfinite(last_atr) or last_atr <= 0:
last_atr = float(core["close"].iloc[-1]) * 0.01
best = None
best_score = float("-inf")
eff_atr_mult = float(atr_mult)
if lookback >= 280:
eff_atr_mult = atr_mult * 1.7
elif lookback >= 160:
eff_atr_mult = atr_mult * 1.3
width_factor = 3.8 + min(2.2, max(0.0, (lookback - 80) / 100.0))
max_width = last_atr * eff_atr_mult * width_factor
tol = last_atr * eff_atr_mult * 0.35
prefer_i = None
if prefer_start_time is not None:
prefer_i = _bar_index_at_or_after(work, prefer_start_time)
if range_start_time is not None:
start_i = _bar_index_at_or_after(work, range_start_time)
if start_i is not None and start_i <= core_end - 8:
seg = work.iloc[start_i:core_end]
hi = float(seg["high"].max())
lo = float(seg["low"].min())
rw = _robust_width(seg)
if 0 < rw <= max_width * 1.15:
near_hi = int((seg["high"] >= hi - tol).sum())
near_lo = int((seg["low"] <= lo + tol).sum())
inside = float(((seg["close"] >= lo - tol) & (seg["close"] <= hi + tol)).mean())
if near_hi >= 2 and near_lo >= 2 and inside >= 0.70:
score = _score_segment(len(seg), near_hi, near_lo, inside, rw, last_atr)
return _pack_range(
work, df, start_i, core_end - 1, hi, lo, tol, last_atr, score, n,
window_offset=win_start,
)
eff_min_bars = max(8, int(min_bars))
cn = len(core)
for length in range(min(cn, lookback), min_bars - 1, -4):
seg = core.iloc[-length:]
max_bars = min(cn, max(eff_min_bars * 2, min(96, max(eff_min_bars + 8, int(cn * 0.5)))))
cands: List[Tuple[float, int, int, int, float, float, float]] = []
def _try_seg(start_i: int, end_i: int, prefer_boost: float = 0.0) -> None:
if end_i - start_i + 1 < eff_min_bars:
return
if start_i < 0 or end_i >= cn or start_i > end_i:
return
seg = work.iloc[start_i : end_i + 1]
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
rw = _robust_width(seg)
if rw <= 0 or rw > max_width:
return
raw_w = hi - lo
if raw_w > max_width * 1.35:
return
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
return
inside = float(((seg["close"] >= lo - tol) & (seg["close"] <= hi + tol)).mean())
if inside < 0.75:
continue
score = _score_segment(length, near_hi, near_lo, inside, width, last_atr)
if score <= best_score:
continue
if inside < 0.72:
return
length = end_i - start_i + 1
score = _score_segment(length, near_hi, near_lo, inside, rw, last_atr) + prefer_boost
cands.append((score, length, start_i, end_i, hi, lo, rw))
for length in range(min(cn, max_bars), eff_min_bars - 1, -4):
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_score = score
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),
"score": float(score),
}
boost = 0.0
if prefer_i is not None:
dist = abs(start_i - int(prefer_i))
if dist <= 6:
boost = 10.0
elif dist <= 14:
boost = 4.0
elif start_i > int(prefer_i) + 16:
boost = -10.0
_try_seg(start_i, cn - 1, boost)
if best is None:
if prefer_i is not None:
pi = int(prefer_i)
if 0 <= pi < cn:
align_max = min(cn, max(max_bars, int(cn * 0.65)))
alen = cn - pi
if eff_min_bars <= alen <= align_max:
_try_seg(pi, cn - 1, prefer_boost=18.0)
elif alen > align_max:
start_i = max(0, cn - align_max)
if start_i > pi:
start_i = pi
end_i = min(cn - 1, pi + align_max - 1)
else:
end_i = cn - 1
_try_seg(start_i, end_i, prefer_boost=12.0)
if not cands:
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
cands.sort(key=lambda x: x[0], reverse=True)
best_score = cands[0][0]
band = max(4.0, abs(best_score) * 0.10)
near = [c for c in cands if c[0] >= best_score - band]
chosen = max(near, key=lambda x: (x[1], x[0]))
score, _length, start_i, end_i, hi, lo, _rw = chosen
return _pack_range(work, df, start_i, end_i, hi, lo, tol, last_atr, score, n, window_offset=win_start)
best["start_time"] = _ts(work.iloc[best["start_idx"]])
# 区间时间结束取 core 末,事件可落在其后
best["end_time"] = _ts(work.iloc[best["end_idx"]])
def detect_trading_ranges(
df: pd.DataFrame,
lookback: Optional[int] = None,
min_bars: int = 24,
atr_mult: float = 1.2,
tail_reserve: int = 12,
max_cycles: int = MAX_CYCLES,
prefer_start_time: Any = None,
range_start_time: Any = None,
) -> List[Dict[str, Any]]:
"""
倒序切多段 TradingRange(近→远)。
过滤顺序:detect → quality → trend → overlap → accept → mask。
返回列表已按时间倒序,调用方将 [0] 标为 ACTIVE。
"""
if df is None or len(df) < min_bars + 5:
return []
lb = int(lookback) if lookback is not None else len(df)
work = df.tail(lb).reset_index(drop=True)
n = len(work)
occupied: List[Dict[str, Any]] = []
accepted: List[Dict[str, Any]] = []
# 搜索右端从 n-1 往左收缩;每接受一段后右端移到该段 start 之前
search_end = n - 1
prefer = prefer_start_time
hard_start = range_start_time
while len(accepted) < max(1, int(max_cycles)) and search_end >= min_bars + 4:
# 在剩余历史内从右往左试多个右边界,避免历史箱必须贴住 search_end
# (否则中间趋势会挡住更早的真实箱)
cand = None
step = max(4, min(12, (search_end - min_bars) // 10 or 4))
for end_try in range(search_end, min_bars + 4, -step):
trial = _detect_in_window(
work,
0,
end_try,
min_bars=min_bars,
atr_mult=atr_mult,
tail_reserve=tail_reserve,
prefer_start_time=prefer if len(accepted) == 0 and end_try == search_end else None,
range_start_time=hard_start if len(accepted) == 0 and end_try == search_end else None,
)
# 1) detect
if trial is None:
continue
# 2) quality
if not _passes_quality(trial, min_bars):
continue
# 3) trend contamination
if not _passes_trend_filter(work, trial):
continue
# 4) overlap with accepted
a0, a1 = int(trial["abs_start_idx"]), int(trial["abs_end_idx"])
overlap_bad = False
for occ in occupied:
ratio = _overlap_ratio(a0, a1, int(occ["start"]), int(occ["end"]))
if ratio >= OVERLAP_RATIO_MAX:
overlap_bad = True
break
if overlap_bad:
continue
# 取最靠右的合格箱(倒序第一段)
cand = trial
break
if cand is None:
break
# 5) accept
accepted.append(cand)
a0, a1 = int(cand["abs_start_idx"]), int(cand["abs_end_idx"])
# 6) mask
occupied.append(
{
"start": a0,
"end": max(a1, int(cand.get("abs_scan_end_idx", a1))),
"quality": float(cand.get("quality") or 0),
"high": float(cand["high"]),
"low": float(cand["low"]),
}
)
# 下一轮只在更早窗口搜
search_end = int(cand["abs_start_idx"]) - 1
hard_start = None
prefer = None
# abs_* 目前相对 work;若 df 比 work 长需加 offset
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
if offset:
for tr in accepted:
tr["abs_start_idx"] = int(tr["abs_start_idx"]) + offset
tr["abs_end_idx"] = int(tr["abs_end_idx"]) + offset
tr["abs_scan_end_idx"] = int(tr["abs_scan_end_idx"]) + offset
return accepted
def detect_trading_range(
df: pd.DataFrame,
lookback: int = 120,
min_bars: int = 24,
atr_mult: float = 1.2,
tail_reserve: int = 12,
range_start_time: Any = None,
prefer_start_time: Any = None,
) -> Optional[Dict[str, Any]]:
"""兼容旧接口:返回倒序列表中的第一段(ACTIVE 候选)。"""
ranges = detect_trading_ranges(
df,
lookback=lookback,
min_bars=min_bars,
atr_mult=atr_mult,
tail_reserve=tail_reserve,
max_cycles=1,
prefer_start_time=prefer_start_time,
range_start_time=range_start_time,
)
return ranges[0] if ranges else None