Add chart-pattern entry archetype (XAU_CP) — disabled by default
Third entry archetype: completion of classical chart patterns (double bottom/top, H&S top / inverse H&S, bull/bear flag) — the tight canon with the strongest documented stats. New chart_patterns.py detector module (pure functions, ATR-unit geometry, closed-bar slices only, first-close-beyond trigger discipline so a breakout fires exactly once and never repaints). Pipeline step 10 in analyze_from_dataframes: fills only empty setup slots, allows weak-composite F grades (normal at reversal completion) but inherits session/day blocks, mirrors MTF-opposition + SELL vol/hour + M5 protections. Geometry: SL = pattern anchor -/+ 0.5 ATR (min $8), TP1 = measured move, $9 spread gate, trailing exit does the real work. Strategy tag XAU_CP. cp_enabled defaults to False everywhere — per-TF adoption happens in pair_signals/xauusd.py only after the standard validation rule passes. Backtester: --cp / --cp-patterns flags to test the candidate, XAU_CP tag recognition. Suite: 183/183 (21 new tests: detectors, trigger discipline, invalidation, gates, pipeline, default-off safety). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015kby7M9JwpYi8NQT2yoxAi
This commit is contained in:
Harwinder Sekhon
@@ -0,0 +1,287 @@
|
||||
"""
|
||||
Classical chart-pattern detection for the XAU_CP entry archetype.
|
||||
================================================================
|
||||
|
||||
Third entry archetype (after the composite-breakout generators and the
|
||||
XAU_PB pullback): trade the completion of a proven multi-bar price
|
||||
pattern. Pattern set is deliberately tight — the classical patterns
|
||||
with the strongest documented statistics (Bulkowski, *Encyclopedia of
|
||||
Chart Patterns*) that are unambiguous from OHLC at bar close:
|
||||
|
||||
double_bottom bullish reversal two matching swing lows, break of
|
||||
the interim peak (neckline)
|
||||
double_top bearish reversal mirrored
|
||||
inv_hs bullish reversal inverse head & shoulders, break of
|
||||
the neckline (max of the two peaks
|
||||
flanking the head — conservative
|
||||
horizontal line, no sloped fits)
|
||||
hs_top bearish reversal head & shoulders top, mirrored
|
||||
bull_flag bullish continuation strong pole, shallow ordered
|
||||
retrace, close at a new high
|
||||
bear_flag bearish continuation mirrored
|
||||
|
||||
Triangles / wedges / cup-and-handle are excluded on purpose: weaker
|
||||
historical stats and apex-timing ambiguity — more parameters for less
|
||||
edge.
|
||||
|
||||
Design contract (mirrors the rest of the signal stack):
|
||||
- Detectors see ONLY the visible closed-bar slice the aggregator was
|
||||
given; the last row is the just-closed signal bar. No forming candle,
|
||||
no future bars → identical behavior live and in the bar-replay
|
||||
backtester, and signals never repaint.
|
||||
- All geometric thresholds are in ATR units so one parameter set is
|
||||
scale-free across timeframes. ATR comes from the engine (Wilder 14).
|
||||
- Trigger discipline: a pattern fires only on the bar that COMPLETES it
|
||||
— the first close beyond the neckline/boundary since the pattern
|
||||
formed. One breakout → at most one signal.
|
||||
- Swing points use the same local-extremum definition as the engine's
|
||||
structure S/R (window w bars each side), so a "swing" here is the
|
||||
same object a chartist sees on the engine's key levels.
|
||||
|
||||
Every detector returns None or a dict:
|
||||
{"pattern": str, "direction": "BUY"|"SELL",
|
||||
"anchor": float, # SL reference (pattern invalidation price)
|
||||
"trigger": float, # neckline / boundary that was broken
|
||||
"height": float} # measured-move distance in price units
|
||||
|
||||
Defaults below are the textbook definitions, not tuned values. The
|
||||
validation record for which patterns/timeframes are enabled lives in
|
||||
pair_signals/xauusd.py (per repo rules, per-TF adoption happens in the
|
||||
factory, never in global defaults).
|
||||
"""
|
||||
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
__all__ = ["detect_chart_pattern", "PATTERN_DETECTORS"]
|
||||
|
||||
|
||||
def _swing_indices(values: np.ndarray, w: int, find_high: bool) -> List[int]:
|
||||
"""Indices of local extrema with `w` bars on each side.
|
||||
|
||||
A bar i is a swing high if values[i] is the strict maximum of
|
||||
values[i-w : i+w+1] (ties reject — a flat top is not a clean swing).
|
||||
The last `w` bars can never qualify, which is exactly right: a swing
|
||||
isn't confirmed until `w` bars close after it.
|
||||
"""
|
||||
n = len(values)
|
||||
out: List[int] = []
|
||||
for i in range(w, n - w):
|
||||
window = values[i - w : i + w + 1]
|
||||
if np.isnan(window).any():
|
||||
continue
|
||||
center = values[i]
|
||||
rest = np.delete(window, w)
|
||||
if find_high:
|
||||
if center > rest.max():
|
||||
out.append(i)
|
||||
else:
|
||||
if center < rest.min():
|
||||
out.append(i)
|
||||
return out
|
||||
|
||||
|
||||
def _first_close_beyond(closes: np.ndarray, start: int, level: float,
|
||||
above: bool) -> bool:
|
||||
"""True iff the LAST close is the first one beyond `level` since
|
||||
`start` (exclusive of the last bar)."""
|
||||
last = closes[-1]
|
||||
prior = closes[start:-1]
|
||||
if above:
|
||||
return last > level and (len(prior) == 0 or prior.max() <= level)
|
||||
return last < level and (len(prior) == 0 or prior.min() >= level)
|
||||
|
||||
|
||||
# ── Double bottom / top ──────────────────────────────────────────────
|
||||
|
||||
def _detect_double(o, h, l, c, atr, cfg, bullish: bool) -> Optional[Dict]:
|
||||
w = cfg.cp_swing_window
|
||||
ext = l if bullish else h
|
||||
idx = _swing_indices(ext, w, find_high=not bullish)
|
||||
if len(idx) < 2:
|
||||
return None
|
||||
i1, i2 = idx[-2], idx[-1]
|
||||
n = len(c)
|
||||
if i2 - i1 < cfg.cp_min_sep_bars:
|
||||
return None
|
||||
if (n - 1) - i2 > cfg.cp_recency_bars:
|
||||
return None
|
||||
e1, e2 = ext[i1], ext[i2]
|
||||
if abs(e1 - e2) > cfg.cp_tolerance_atr * atr:
|
||||
return None
|
||||
|
||||
if bullish:
|
||||
neckline = float(h[i1 : i2 + 1].max())
|
||||
floor = min(e1, e2)
|
||||
height = neckline - floor
|
||||
else:
|
||||
neckline = float(l[i1 : i2 + 1].min())
|
||||
ceil_ = max(e1, e2)
|
||||
height = ceil_ - neckline
|
||||
|
||||
if not (cfg.cp_min_height_atr * atr <= height <= cfg.cp_max_height_atr * atr):
|
||||
return None
|
||||
|
||||
# Pattern must not have been invalidated after the second extreme:
|
||||
# a lower low (bullish) / higher high (bearish) breaks the pattern.
|
||||
if bullish:
|
||||
if i2 + 1 < n and l[i2 + 1 :].min() < min(e1, e2) - 0.1 * atr:
|
||||
return None
|
||||
if not _first_close_beyond(c, i2, neckline, above=True):
|
||||
return None
|
||||
return {"pattern": "double_bottom", "direction": "BUY",
|
||||
"anchor": float(min(e1, e2)), "trigger": neckline,
|
||||
"height": float(height)}
|
||||
else:
|
||||
if i2 + 1 < n and h[i2 + 1 :].max() > max(e1, e2) + 0.1 * atr:
|
||||
return None
|
||||
if not _first_close_beyond(c, i2, neckline, above=False):
|
||||
return None
|
||||
return {"pattern": "double_top", "direction": "SELL",
|
||||
"anchor": float(max(e1, e2)), "trigger": neckline,
|
||||
"height": float(height)}
|
||||
|
||||
|
||||
# ── Head & shoulders (top and inverse) ───────────────────────────────
|
||||
|
||||
def _detect_hs(o, h, l, c, atr, cfg, bullish: bool) -> Optional[Dict]:
|
||||
"""bullish=True → inverse H&S (three swing LOWS, head deepest)."""
|
||||
w = cfg.cp_swing_window
|
||||
ext = l if bullish else h
|
||||
idx = _swing_indices(ext, w, find_high=not bullish)
|
||||
if len(idx) < 3:
|
||||
return None
|
||||
i1, i2, i3 = idx[-3], idx[-2], idx[-1]
|
||||
n = len(c)
|
||||
if (n - 1) - i3 > cfg.cp_recency_bars:
|
||||
return None
|
||||
s1, head, s2 = ext[i1], ext[i2], ext[i3]
|
||||
|
||||
if bullish:
|
||||
# Head strictly the deepest low; shoulders roughly level.
|
||||
if not (head < s1 - cfg.cp_hs_head_min_atr * atr
|
||||
and head < s2 - cfg.cp_hs_head_min_atr * atr):
|
||||
return None
|
||||
if abs(s1 - s2) > cfg.cp_hs_shoulder_tol_atr * atr:
|
||||
return None
|
||||
# Neckline: the LOWER of the two interim peaks — the price must
|
||||
# clear both, so this is the conservative horizontal trigger.
|
||||
p1 = float(h[i1 : i2 + 1].max())
|
||||
p2 = float(h[i2 : i3 + 1].max())
|
||||
neckline = min(p1, p2)
|
||||
height = neckline - head
|
||||
else:
|
||||
if not (head > s1 + cfg.cp_hs_head_min_atr * atr
|
||||
and head > s2 + cfg.cp_hs_head_min_atr * atr):
|
||||
return None
|
||||
if abs(s1 - s2) > cfg.cp_hs_shoulder_tol_atr * atr:
|
||||
return None
|
||||
t1 = float(l[i1 : i2 + 1].min())
|
||||
t2 = float(l[i2 : i3 + 1].min())
|
||||
neckline = max(t1, t2)
|
||||
height = head - neckline
|
||||
|
||||
if not (cfg.cp_min_height_atr * atr <= height <= cfg.cp_max_height_atr * atr):
|
||||
return None
|
||||
|
||||
if bullish:
|
||||
# Invalidated if price undercut the head after the right shoulder.
|
||||
if i3 + 1 < n and l[i3 + 1 :].min() < head - 0.1 * atr:
|
||||
return None
|
||||
if not _first_close_beyond(c, i3, neckline, above=True):
|
||||
return None
|
||||
return {"pattern": "inv_hs", "direction": "BUY",
|
||||
"anchor": float(s2), "trigger": neckline,
|
||||
"height": float(height)}
|
||||
else:
|
||||
if i3 + 1 < n and h[i3 + 1 :].max() > head + 0.1 * atr:
|
||||
return None
|
||||
if not _first_close_beyond(c, i3, neckline, above=False):
|
||||
return None
|
||||
return {"pattern": "hs_top", "direction": "SELL",
|
||||
"anchor": float(s2), "trigger": neckline,
|
||||
"height": float(height)}
|
||||
|
||||
|
||||
# ── Flags (bull and bear) ────────────────────────────────────────────
|
||||
|
||||
def _detect_flag(o, h, l, c, atr, cfg, bullish: bool) -> Optional[Dict]:
|
||||
n = len(c)
|
||||
lo_r, hi_r = cfg.cp_flag_retrace
|
||||
if bullish:
|
||||
# Pole top = highest high BEFORE the trigger bar.
|
||||
ip = int(np.argmax(h[:-1]))
|
||||
flag_len = (n - 1) - ip
|
||||
if not (cfg.cp_flag_min_bars <= flag_len <= cfg.cp_flag_max_bars):
|
||||
return None
|
||||
pole_start = max(0, ip - cfg.cp_flag_pole_bars + 1)
|
||||
pole_low = float(l[pole_start : ip + 1].min())
|
||||
pole = float(h[ip]) - pole_low
|
||||
if pole < cfg.cp_flag_pole_atr * atr:
|
||||
return None
|
||||
flag_low = float(l[ip + 1 : n].min())
|
||||
retrace = (float(h[ip]) - flag_low) / pole
|
||||
if not (lo_r <= retrace <= hi_r):
|
||||
return None
|
||||
if not _first_close_beyond(c, ip, float(h[ip]), above=True):
|
||||
return None
|
||||
return {"pattern": "bull_flag", "direction": "BUY",
|
||||
"anchor": flag_low, "trigger": float(h[ip]),
|
||||
"height": float(pole)}
|
||||
else:
|
||||
ip = int(np.argmin(l[:-1]))
|
||||
flag_len = (n - 1) - ip
|
||||
if not (cfg.cp_flag_min_bars <= flag_len <= cfg.cp_flag_max_bars):
|
||||
return None
|
||||
pole_start = max(0, ip - cfg.cp_flag_pole_bars + 1)
|
||||
pole_high = float(h[pole_start : ip + 1].max())
|
||||
pole = pole_high - float(l[ip])
|
||||
if pole < cfg.cp_flag_pole_atr * atr:
|
||||
return None
|
||||
flag_high = float(h[ip + 1 : n].max())
|
||||
retrace = (flag_high - float(l[ip])) / pole
|
||||
if not (lo_r <= retrace <= hi_r):
|
||||
return None
|
||||
if not _first_close_beyond(c, ip, float(l[ip]), above=False):
|
||||
return None
|
||||
return {"pattern": "bear_flag", "direction": "SELL",
|
||||
"anchor": flag_high, "trigger": float(l[ip]),
|
||||
"height": float(pole)}
|
||||
|
||||
|
||||
# Priority order: most structured / rarest first — if two patterns
|
||||
# complete on the same bar, the more specific one names the trade.
|
||||
PATTERN_DETECTORS = {
|
||||
"inv_hs": lambda o, h, l, c, atr, cfg: _detect_hs(o, h, l, c, atr, cfg, True),
|
||||
"hs_top": lambda o, h, l, c, atr, cfg: _detect_hs(o, h, l, c, atr, cfg, False),
|
||||
"double_bottom": lambda o, h, l, c, atr, cfg: _detect_double(o, h, l, c, atr, cfg, True),
|
||||
"double_top": lambda o, h, l, c, atr, cfg: _detect_double(o, h, l, c, atr, cfg, False),
|
||||
"bull_flag": lambda o, h, l, c, atr, cfg: _detect_flag(o, h, l, c, atr, cfg, True),
|
||||
"bear_flag": lambda o, h, l, c, atr, cfg: _detect_flag(o, h, l, c, atr, cfg, False),
|
||||
}
|
||||
|
||||
|
||||
def detect_chart_pattern(df: pd.DataFrame, atr: float, cfg) -> Optional[Dict]:
|
||||
"""Run enabled detectors over the last cp_lookback(+1) bars.
|
||||
|
||||
`df` is the visible history at decision time (last row = signal
|
||||
bar). Returns the first match in PATTERN_DETECTORS priority order,
|
||||
or None.
|
||||
"""
|
||||
if atr <= 0 or len(df) < cfg.cp_lookback + 1:
|
||||
return None
|
||||
tail = df.iloc[-(cfg.cp_lookback + 1):]
|
||||
o = tail["Open"].to_numpy(dtype=np.float64)
|
||||
h = tail["High"].to_numpy(dtype=np.float64)
|
||||
l = tail["Low"].to_numpy(dtype=np.float64)
|
||||
c = tail["Close"].to_numpy(dtype=np.float64)
|
||||
enabled = set(cfg.cp_patterns)
|
||||
for name, det in PATTERN_DETECTORS.items():
|
||||
if name not in enabled:
|
||||
continue
|
||||
hit = det(o, h, l, c, atr, cfg)
|
||||
if hit is not None:
|
||||
return hit
|
||||
return None
|
||||
Reference in New Issue
Block a user