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). Claude-Session: https://claude.ai/code/session_015kby7M9JwpYi8NQT2yoxAi
This commit is contained in:
Harwinder Sekhon
@@ -1541,6 +1541,8 @@ def _replay_span(span: Tuple[int, int]):
|
||||
first_bear = report.bearish_reasons[0] if report.bearish_reasons else ""
|
||||
if "XAU PULLBACK" in first_bull or "XAU PULLBACK" in first_bear:
|
||||
tag = "XAU_PB"
|
||||
elif "XAU CHART PATTERN" in first_bull or "XAU CHART PATTERN" in first_bear:
|
||||
tag = "XAU_CP"
|
||||
elif "XAU STRATEGY" in first_bull:
|
||||
tag = "XAU"
|
||||
|
||||
@@ -2141,7 +2143,13 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
|
||||
help="Override TechnicalConfig.swing_sr_lookback (0 = use config default)")
|
||||
p.add_argument("--quality", default="A+,A,B,C", help="Comma-separated allowed grades")
|
||||
p.add_argument("--strategies", default="",
|
||||
help="Comma-separated strategy tags to include (XAU,XAU_PB,GENERIC). Empty = all.")
|
||||
help="Comma-separated strategy tags to include (XAU,XAU_PB,XAU_CP,GENERIC). Empty = all.")
|
||||
p.add_argument("--cp", action="store_true",
|
||||
help="Force-enable the chart-pattern entry (XAU_CP) regardless "
|
||||
"of the factory default — for validating the candidate.")
|
||||
p.add_argument("--cp-patterns", default=None,
|
||||
help="Comma-separated pattern subset for --cp (e.g. "
|
||||
"double_bottom,bull_flag). Default: all detectors.")
|
||||
p.add_argument("--no-cache", action="store_true", help="Skip parquet cache (force API fetch)")
|
||||
p.add_argument("--spread", type=float, default=None,
|
||||
help="Round-trip cost in price units per trade (default: pair's "
|
||||
@@ -2208,6 +2216,12 @@ def main(argv: Optional[List[str]] = None) -> int:
|
||||
)
|
||||
if args.swing_sr_lookback > 0:
|
||||
platform_cfg.technical.swing_sr_lookback = args.swing_sr_lookback
|
||||
if args.cp:
|
||||
platform_cfg.xau_strategy.cp_enabled = True
|
||||
if args.cp_patterns:
|
||||
platform_cfg.xau_strategy.cp_patterns = [
|
||||
s.strip() for s in args.cp_patterns.split(",") if s.strip()
|
||||
]
|
||||
confirm_tfs = list(platform_cfg.confirmation_timeframes)
|
||||
|
||||
end = (
|
||||
|
||||
@@ -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
|
||||
@@ -274,6 +274,34 @@ class XauStrategyConfig:
|
||||
# Price must have touched EMA21/KC-mid within this many bars.
|
||||
m5_pullback_touch_bars: int = 3
|
||||
|
||||
# ── Chart-pattern entry (XAU_CP) ──────────────────────────────
|
||||
# Third entry archetype: completion of a classical chart pattern
|
||||
# (see chart_patterns.py for the pattern set and design contract).
|
||||
# Defaults are textbook geometry, NOT tuned values. Disabled by
|
||||
# default — per-TF adoption happens in pair_signals/xauusd.py only
|
||||
# after the pattern book passes the standard validation rule
|
||||
# (positive R both halves of ≥12mo, PF ≥ 1.3, $0.30 spread,
|
||||
# trailing exits, next-open fills).
|
||||
cp_enabled: bool = False
|
||||
cp_patterns: List[str] = field(default_factory=lambda: [
|
||||
"inv_hs", "hs_top", "double_bottom", "double_top",
|
||||
"bull_flag", "bear_flag",
|
||||
])
|
||||
cp_lookback: int = 72 # bars scanned for pattern structure
|
||||
cp_swing_window: int = 3 # bars each side to confirm a swing
|
||||
cp_recency_bars: int = 20 # last extreme must be this fresh
|
||||
cp_min_sep_bars: int = 8 # min bars between double extremes
|
||||
cp_tolerance_atr: float = 0.5 # double top/bottom level match tol
|
||||
cp_min_height_atr: float = 1.5 # pattern height floor (noise gate)
|
||||
cp_max_height_atr: float = 8.0 # pattern height ceiling
|
||||
cp_hs_head_min_atr: float = 0.5 # head must exceed shoulders by this
|
||||
cp_hs_shoulder_tol_atr: float = 1.0 # shoulder level match tol
|
||||
cp_flag_pole_atr: float = 3.0 # min pole size
|
||||
cp_flag_pole_bars: int = 12 # pole measured over ≤ this many bars
|
||||
cp_flag_min_bars: int = 3 # flag consolidation length bounds
|
||||
cp_flag_max_bars: int = 12
|
||||
cp_flag_retrace: tuple = (0.1, 0.5) # flag depth as fraction of pole
|
||||
|
||||
|
||||
@dataclass
|
||||
class FilterConfig:
|
||||
|
||||
@@ -13,6 +13,7 @@ from typing import Dict, List, Optional
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from chart_patterns import detect_chart_pattern
|
||||
from config import PlatformConfig, SignalStrength, Timeframe
|
||||
from data_provider import ForexDataProvider
|
||||
from engine import TechnicalEngine, TechnicalAnalysis
|
||||
@@ -98,6 +99,8 @@ class SignalAggregator:
|
||||
# Diagnostic funnel for the M5 pullback route (shared across
|
||||
# instances; read by backtest tooling, harmless in live use).
|
||||
pb_funnel: Dict[str, int] = {}
|
||||
# Same, for the chart-pattern route (XAU_CP).
|
||||
cp_funnel: Dict[str, int] = {}
|
||||
|
||||
def __init__(self, config: PlatformConfig):
|
||||
self.config = config
|
||||
@@ -563,6 +566,138 @@ class SignalAggregator:
|
||||
target = report.bullish_reasons if is_buy else report.bearish_reasons
|
||||
target.insert(0, marker)
|
||||
|
||||
# ── Chart-pattern entry (XAU_CP) ────────────────────────────────
|
||||
|
||||
def _try_chart_pattern(
|
||||
self, report: SignalReport, primary_df: pd.DataFrame, pip_value: float
|
||||
) -> None:
|
||||
"""Build an XAU_CP trade setup if a completed chart pattern
|
||||
passes the pair-level protections.
|
||||
|
||||
Unlike the composite paths there is no minimum-score gate: a
|
||||
reversal pattern completes exactly where the composite is still
|
||||
pointing the old way. Protection comes from the pattern's own
|
||||
structural gates plus the validated directional filters below.
|
||||
"""
|
||||
cfg = self.config.xau_strategy
|
||||
funnel = SignalAggregator.cp_funnel
|
||||
|
||||
def _reject(reason: str) -> None:
|
||||
funnel[reason] = funnel.get(reason, 0) + 1
|
||||
|
||||
pa = report.primary_analysis
|
||||
if pa is None:
|
||||
return
|
||||
atr = pa.key_levels.get("atr", 0.0)
|
||||
# Same $-ATR floor the swing generator applies (dead tape).
|
||||
if atr < cfg.min_atr_threshold:
|
||||
return
|
||||
|
||||
pat = detect_chart_pattern(primary_df, atr, cfg)
|
||||
if pat is None:
|
||||
return
|
||||
_reject("detected")
|
||||
direction = pat["direction"]
|
||||
is_buy = direction == "BUY"
|
||||
|
||||
# Never against multi-TF consensus (mirror of gate 6f).
|
||||
sigs = list(report.mtf_signals.values())
|
||||
bull_n = sum(1 for s in sigs if s in (SignalStrength.BUY, SignalStrength.STRONG_BUY))
|
||||
bear_n = sum(1 for s in sigs if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL))
|
||||
thr = self.config.signals.mtf_agreement_threshold
|
||||
if (is_buy and bear_n >= thr) or (not is_buy and bull_n >= thr):
|
||||
_reject("mtf_opposition")
|
||||
return
|
||||
|
||||
# Directional protections — gold's macro-uptrend asymmetry is
|
||||
# structural, so the validated SELL gates apply to pattern
|
||||
# shorts too (quiet-vol shorts fade, blocked SELL hours lose).
|
||||
if not is_buy:
|
||||
if (report.volatility_regime
|
||||
and report.volatility_regime not in cfg.sell_vol_regimes):
|
||||
_reject("sell_vol")
|
||||
return
|
||||
if cfg.sell_blocked_hours_utc:
|
||||
try:
|
||||
close_h = (
|
||||
pd.Timestamp(report.timestamp)
|
||||
+ pd.Timedelta(
|
||||
seconds=ForexDataProvider.CANDLE_SECONDS[
|
||||
self.config.primary_timeframe
|
||||
]
|
||||
)
|
||||
).hour
|
||||
except Exception:
|
||||
close_h = None
|
||||
if close_h in cfg.sell_blocked_hours_utc:
|
||||
_reject("sell_hour")
|
||||
return
|
||||
if self.config.primary_timeframe == Timeframe.M5:
|
||||
rsi_ind = pa.indicators.get("rsi")
|
||||
if rsi_ind is not None and rsi_ind.value < cfg.m5_sell_min_rsi:
|
||||
_reject("m5_sell_rsi")
|
||||
return
|
||||
|
||||
# M5 chop-transition volatility block (validated, both ways).
|
||||
if (self.config.primary_timeframe == Timeframe.M5
|
||||
and report.volatility_regime in cfg.m5_blocked_vol_regimes):
|
||||
_reject("vol_regime")
|
||||
return
|
||||
|
||||
entry = report.current_price
|
||||
min_sl_price = self.config.signals.min_sl_pips * pip_value
|
||||
height = pat["height"]
|
||||
if is_buy:
|
||||
sl_dist = max(entry - (pat["anchor"] - 0.5 * atr), min_sl_price)
|
||||
sl = entry - sl_dist
|
||||
tp1 = entry + height
|
||||
tp2 = entry + 1.5 * height
|
||||
else:
|
||||
sl_dist = max((pat["anchor"] + 0.5 * atr) - entry, min_sl_price)
|
||||
sl = entry + sl_dist
|
||||
tp1 = entry - height
|
||||
tp2 = entry - 1.5 * height
|
||||
|
||||
# The measured move IS the pattern target — if it can't clear
|
||||
# the RR floor or the spread gate, the pattern is untradeable.
|
||||
rr = height / sl_dist if sl_dist > 0 else 0.0
|
||||
if rr < self.config.signals.min_rr_ratio - 1e-9:
|
||||
_reject("rr")
|
||||
return
|
||||
min_tp = (self.config.signals.round_trip_cost
|
||||
* self.config.signals.min_tp_cost_multiple)
|
||||
if height < min_tp:
|
||||
_reject("spread_gate")
|
||||
return
|
||||
|
||||
report.trade_setup = TradeSetup(
|
||||
pair=report.symbol,
|
||||
display_name=report.display_name,
|
||||
direction=direction,
|
||||
entry_price=entry,
|
||||
stop_loss=sl,
|
||||
take_profit_1=tp1,
|
||||
take_profit_2=tp2,
|
||||
risk_reward=round(rr, 2),
|
||||
risk_pips=sl_dist / pip_value,
|
||||
reward_pips=height / pip_value,
|
||||
pip_value=pip_value,
|
||||
strategy="XAU_CP",
|
||||
)
|
||||
if report.signal_quality not in ("A+", "A", "B"):
|
||||
report.signal_quality = "B"
|
||||
report.final_signal = (SignalStrength.BUY if is_buy
|
||||
else SignalStrength.SELL)
|
||||
report.action_recommendation = (
|
||||
f"{direction} — {pat['pattern']} pattern completion"
|
||||
)
|
||||
marker = (f"XAU CHART PATTERN: {pat['pattern']} "
|
||||
f"({direction}) break of {pat['trigger']:.2f}, "
|
||||
f"height {height / atr:.1f} ATR")
|
||||
target = report.bullish_reasons if is_buy else report.bearish_reasons
|
||||
target.insert(0, marker)
|
||||
_reject("accepted")
|
||||
|
||||
# ── Public: bar-replayable entry point ─────────────────────────
|
||||
|
||||
def analyze_from_dataframes(
|
||||
@@ -825,6 +960,24 @@ class SignalAggregator:
|
||||
and report.signal_quality != "F"):
|
||||
self._try_m5_pullback(report, primary_df, pip_value)
|
||||
|
||||
# ── 10. Chart-pattern entry (XAU_CP) ────────────────────────
|
||||
# Third archetype: completion of a classical chart pattern
|
||||
# (double top/bottom, H&S, flag — see chart_patterns.py). Only
|
||||
# fills empty slots (core + pullback setups take precedence).
|
||||
# Unlike XAU_PB this must NOT require quality != "F": a weak
|
||||
# composite (grade F on |score| < 15) is the normal state at a
|
||||
# reversal-pattern completion. Session/day blocks still bind —
|
||||
# they are detected by their warnings (and the live wrapper +
|
||||
# backtest replay apply the bar-close session predicate again
|
||||
# on top, so a pattern setup can never trade a blocked hour).
|
||||
if (report.display_name == "XAU/USD"
|
||||
and self.config.xau_strategy.cp_enabled
|
||||
and report.trade_setup is None
|
||||
and not any(w.startswith(("SESSION FILTER", "DAY FILTER",
|
||||
"WEEKEND FILTER"))
|
||||
for w in report.warnings)):
|
||||
self._try_chart_pattern(report, primary_df, pip_value)
|
||||
|
||||
return report
|
||||
|
||||
# ── Internals ──────────────────────────────────────────────────
|
||||
|
||||
+318
@@ -2170,6 +2170,324 @@ def _():
|
||||
assert t2.entry_time == idx[2] and t2.entry_price == 110.0 and t2.sl == 102.0
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
# MODULE 12: Chart-pattern entry (XAU_CP)
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
|
||||
print("\n═══ Module 12: chart patterns (XAU_CP) ═══")
|
||||
|
||||
from chart_patterns import detect_chart_pattern
|
||||
from config import XauStrategyConfig
|
||||
|
||||
|
||||
def make_cp_df(closes, end="2026-07-08 14:00", freq="1h", wick=1.5):
|
||||
"""Deterministic OHLCV from an explicit close path (fixed end time:
|
||||
Wed, label hour 14, close hour 15 — clear of every session block)."""
|
||||
idx = pd.date_range(end=end, periods=len(closes), freq=freq, tz="UTC")
|
||||
c = np.asarray(closes, dtype=float)
|
||||
o = np.concatenate([[c[0]], c[:-1]])
|
||||
return pd.DataFrame({"Open": o, "High": c + wick, "Low": c - wick,
|
||||
"Close": c, "Volume": 1000.0}, index=idx)
|
||||
|
||||
|
||||
def mirror_cp_df(df, pivot=6000.0):
|
||||
"""Price-mirror a fixture: bullish pattern → exact bearish twin."""
|
||||
return pd.DataFrame({
|
||||
"Open": pivot - df["Open"], "High": pivot - df["Low"],
|
||||
"Low": pivot - df["High"], "Close": pivot - df["Close"],
|
||||
"Volume": df["Volume"]}, index=df.index)
|
||||
|
||||
|
||||
def double_bottom_closes():
|
||||
"""L1@30 (2984) and L2@65 (2984.5), neckline peak 3000 @45,
|
||||
below-neckline drift, breakout close 3004 on the last bar."""
|
||||
return (list(np.linspace(2988, 3013, 15))
|
||||
+ list(np.linspace(3011, 2984, 16))
|
||||
+ list(np.linspace(2986, 3000, 15))
|
||||
+ list(np.linspace(2999, 2984.5, 20))
|
||||
+ list(np.linspace(2986, 2998, 13))
|
||||
+ [3004.0])
|
||||
|
||||
|
||||
def hs_top_closes():
|
||||
"""S1@20 (3000), head@38 (3010), S2@62 (2999), neckline ~2989.5,
|
||||
breakdown close 2988 on the last bar."""
|
||||
return (list(np.linspace(2982, 3000, 21))
|
||||
+ list(np.linspace(2998.5, 2990, 7))
|
||||
+ list(np.linspace(2992, 3010, 11))
|
||||
+ list(np.linspace(3008, 2991, 8))
|
||||
+ list(np.linspace(2992, 2999, 16))
|
||||
+ list(np.linspace(2998.5, 2991, 16))
|
||||
+ [2988.0])
|
||||
|
||||
|
||||
def bull_flag_closes():
|
||||
"""Pole 60→69 (3000→3025), flag drift 70→78, breakout close 3028.5."""
|
||||
return (list(np.linspace(3000, 2996, 60))
|
||||
+ list(np.linspace(2999, 3025, 10))
|
||||
+ list(np.linspace(3023, 3018, 9))
|
||||
+ [3028.5])
|
||||
|
||||
|
||||
_CP_CFG = XauStrategyConfig()
|
||||
|
||||
|
||||
@test("CP: double bottom detected on breakout bar with correct geometry")
|
||||
def _():
|
||||
df = make_cp_df(double_bottom_closes())
|
||||
hit = detect_chart_pattern(df, atr=5.0, cfg=_CP_CFG)
|
||||
assert hit is not None, "double bottom not detected"
|
||||
assert hit["pattern"] == "double_bottom" and hit["direction"] == "BUY"
|
||||
assert abs(hit["anchor"] - 2982.5) < 1e-9, f"anchor {hit['anchor']}"
|
||||
assert abs(hit["trigger"] - 3001.5) < 1e-9, f"trigger {hit['trigger']}"
|
||||
assert abs(hit["height"] - 19.0) < 1e-9, f"height {hit['height']}"
|
||||
|
||||
|
||||
@test("CP: double top detected on mirrored fixture")
|
||||
def _():
|
||||
df = mirror_cp_df(make_cp_df(double_bottom_closes()))
|
||||
hit = detect_chart_pattern(df, atr=5.0, cfg=_CP_CFG)
|
||||
assert hit is not None and hit["pattern"] == "double_top"
|
||||
assert hit["direction"] == "SELL"
|
||||
assert abs(hit["anchor"] - (6000 - 2982.5)) < 1e-9
|
||||
assert abs(hit["trigger"] - (6000 - 3001.5)) < 1e-9
|
||||
|
||||
|
||||
@test("CP: trigger discipline — no signal before breakout close")
|
||||
def _():
|
||||
df = make_cp_df(double_bottom_closes()).iloc[:-1]
|
||||
assert detect_chart_pattern(df, atr=5.0, cfg=_CP_CFG) is None
|
||||
|
||||
|
||||
@test("CP: trigger discipline — no re-fire on the bar after breakout")
|
||||
def _():
|
||||
df = make_cp_df(double_bottom_closes() + [3006.0])
|
||||
assert detect_chart_pattern(df, atr=5.0, cfg=_CP_CFG) is None
|
||||
|
||||
|
||||
@test("CP: double bottom rejected when lows differ beyond tolerance")
|
||||
def _():
|
||||
closes = double_bottom_closes()
|
||||
closes[46:66] = list(np.linspace(2999, 2979, 20)) # L2 5.5 below L1
|
||||
df = make_cp_df(closes)
|
||||
assert detect_chart_pattern(df, atr=5.0, cfg=_CP_CFG) is None
|
||||
|
||||
|
||||
@test("CP: double bottom invalidated by a lower low after L2")
|
||||
def _():
|
||||
closes = double_bottom_closes()
|
||||
closes[77] = 2978.0 # undercuts both bottoms before the breakout
|
||||
df = make_cp_df(closes)
|
||||
assert detect_chart_pattern(df, atr=5.0, cfg=_CP_CFG) is None
|
||||
|
||||
|
||||
@test("CP: head & shoulders top detected (SELL, right-shoulder anchor)")
|
||||
def _():
|
||||
df = make_cp_df(hs_top_closes())
|
||||
hit = detect_chart_pattern(df, atr=5.0, cfg=_CP_CFG)
|
||||
assert hit is not None, "H&S top not detected"
|
||||
assert hit["pattern"] == "hs_top" and hit["direction"] == "SELL"
|
||||
assert abs(hit["anchor"] - 3000.5) < 1e-9, f"anchor {hit['anchor']}"
|
||||
assert abs(hit["trigger"] - 2989.5) < 1e-9, f"trigger {hit['trigger']}"
|
||||
|
||||
|
||||
@test("CP: inverse head & shoulders detected on mirrored fixture")
|
||||
def _():
|
||||
df = mirror_cp_df(make_cp_df(hs_top_closes()))
|
||||
hit = detect_chart_pattern(df, atr=5.0, cfg=_CP_CFG)
|
||||
assert hit is not None and hit["pattern"] == "inv_hs"
|
||||
assert hit["direction"] == "BUY"
|
||||
|
||||
|
||||
@test("CP: bull flag detected (pole, shallow retrace, new-high close)")
|
||||
def _():
|
||||
df = make_cp_df(bull_flag_closes())
|
||||
hit = detect_chart_pattern(df, atr=5.0, cfg=_CP_CFG)
|
||||
assert hit is not None, "bull flag not detected"
|
||||
assert hit["pattern"] == "bull_flag" and hit["direction"] == "BUY"
|
||||
assert abs(hit["trigger"] - 3026.5) < 1e-9
|
||||
assert hit["height"] > 15.0 # pole ≥ cp_flag_pole_atr × atr
|
||||
|
||||
|
||||
@test("CP: bear flag detected on mirrored fixture")
|
||||
def _():
|
||||
df = mirror_cp_df(make_cp_df(bull_flag_closes()))
|
||||
hit = detect_chart_pattern(df, atr=5.0, cfg=_CP_CFG)
|
||||
assert hit is not None and hit["pattern"] == "bear_flag"
|
||||
assert hit["direction"] == "SELL"
|
||||
|
||||
|
||||
@test("CP: cp_patterns subset filter excludes disabled detectors")
|
||||
def _():
|
||||
from dataclasses import replace
|
||||
df = make_cp_df(double_bottom_closes())
|
||||
cfg = XauStrategyConfig(cp_patterns=["bull_flag"])
|
||||
assert detect_chart_pattern(df, atr=5.0, cfg=cfg) is None
|
||||
|
||||
|
||||
def _cp_report(df, config, vol_regime="normal", mtf_signals=None):
|
||||
"""SignalReport shell around a real engine analysis of `df`."""
|
||||
from engine import TechnicalEngine
|
||||
eng = TechnicalEngine(config.technical)
|
||||
pa = eng.analyze(df, "XAU/USD", config.primary_timeframe.value)
|
||||
rep = SignalReport(symbol="XAU/USD", display_name="XAU/USD",
|
||||
timestamp=str(df.index[-1]))
|
||||
rep.primary_analysis = pa
|
||||
rep.current_price = float(df["Close"].iloc[-1])
|
||||
rep.volatility_regime = vol_regime
|
||||
rep.mtf_signals = mtf_signals or {}
|
||||
rep.signal_quality = "D"
|
||||
return rep
|
||||
|
||||
|
||||
def _cp_config(**xau_overrides):
|
||||
cfg = PlatformConfig()
|
||||
cfg.primary_timeframe = Timeframe.H1
|
||||
cfg.xau_strategy.cp_enabled = True
|
||||
for k, v in xau_overrides.items():
|
||||
setattr(cfg.xau_strategy, k, v)
|
||||
return cfg
|
||||
|
||||
|
||||
@test("CP gate: BUY pattern builds an XAU_CP setup with measured-move TP")
|
||||
def _():
|
||||
cfg = _cp_config()
|
||||
df = make_cp_df(double_bottom_closes())
|
||||
rep = _cp_report(df, cfg)
|
||||
SignalAggregator(cfg)._try_chart_pattern(rep, df, 0.10)
|
||||
ts = rep.trade_setup
|
||||
assert ts is not None, "pattern setup not generated"
|
||||
assert ts.strategy == "XAU_CP" and ts.direction == "BUY"
|
||||
assert ts.stop_loss < ts.entry_price
|
||||
assert abs((ts.take_profit_1 - ts.entry_price) - 19.0) < 1e-6, \
|
||||
"TP1 must be the measured move"
|
||||
assert rep.signal_quality == "B"
|
||||
assert rep.bullish_reasons[0].startswith("XAU CHART PATTERN"), \
|
||||
"tag marker must sit at reasons[0]"
|
||||
|
||||
|
||||
@test("CP gate: MTF opposition blocks the pattern trade")
|
||||
def _():
|
||||
cfg = _cp_config()
|
||||
df = make_cp_df(double_bottom_closes())
|
||||
rep = _cp_report(df, cfg, mtf_signals={
|
||||
"15min": SignalStrength.SELL, "4h": SignalStrength.STRONG_SELL})
|
||||
SignalAggregator(cfg)._try_chart_pattern(rep, df, 0.10)
|
||||
assert rep.trade_setup is None, "BUY pattern must not fire against consensus"
|
||||
|
||||
|
||||
@test("CP gate: SELL pattern blocked in quiet volatility, allowed in high")
|
||||
def _():
|
||||
cfg = _cp_config()
|
||||
df = mirror_cp_df(make_cp_df(double_bottom_closes()))
|
||||
rep = _cp_report(df, cfg, vol_regime="normal")
|
||||
SignalAggregator(cfg)._try_chart_pattern(rep, df, 0.10)
|
||||
assert rep.trade_setup is None, "quiet-vol SELL must be blocked"
|
||||
rep2 = _cp_report(df, cfg, vol_regime="high")
|
||||
SignalAggregator(cfg)._try_chart_pattern(rep2, df, 0.10)
|
||||
assert rep2.trade_setup is not None and rep2.trade_setup.direction == "SELL"
|
||||
|
||||
|
||||
@test("CP gate: SELL blocked-hour (bar close) stops pattern shorts only")
|
||||
def _():
|
||||
cfg = _cp_config(sell_blocked_hours_utc=[15]) # fixture closes 15:00
|
||||
df = mirror_cp_df(make_cp_df(double_bottom_closes()))
|
||||
rep = _cp_report(df, cfg, vol_regime="high")
|
||||
SignalAggregator(cfg)._try_chart_pattern(rep, df, 0.10)
|
||||
assert rep.trade_setup is None, "SELL at blocked close hour must not fire"
|
||||
# BUY side unaffected at the same hour.
|
||||
df_buy = make_cp_df(double_bottom_closes())
|
||||
rep_buy = _cp_report(df_buy, cfg)
|
||||
SignalAggregator(cfg)._try_chart_pattern(rep_buy, df_buy, 0.10)
|
||||
assert rep_buy.trade_setup is not None
|
||||
|
||||
|
||||
@test("CP gate: M5 chop-transition volatility blocks both directions")
|
||||
def _():
|
||||
cfg = _cp_config()
|
||||
cfg.primary_timeframe = Timeframe.M5
|
||||
df = make_cp_df(double_bottom_closes())
|
||||
rep = _cp_report(df, cfg, vol_regime="high") # 'high' blocked on M5
|
||||
SignalAggregator(cfg)._try_chart_pattern(rep, df, 0.10)
|
||||
assert rep.trade_setup is None
|
||||
|
||||
|
||||
@test("CP gate: spread gate rejects patterns smaller than TP minimum")
|
||||
def _():
|
||||
cfg = _cp_config()
|
||||
cfg.signals.min_tp_cost_multiple = 1000.0 # $300 minimum target
|
||||
df = make_cp_df(double_bottom_closes())
|
||||
rep = _cp_report(df, cfg)
|
||||
SignalAggregator(cfg)._try_chart_pattern(rep, df, 0.10)
|
||||
assert rep.trade_setup is None
|
||||
|
||||
|
||||
@test("CP pipeline: fires end-to-end when core generators pass")
|
||||
def _():
|
||||
cfg = _cp_config()
|
||||
cfg.confirmation_timeframes = [Timeframe.M15, Timeframe.H4]
|
||||
df = make_cp_df(double_bottom_closes())
|
||||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||||
agg = SignalAggregator(cfg)
|
||||
with patch.object(SignalAggregator, "_generate_trade_setup",
|
||||
return_value=None), \
|
||||
patch.object(SignalAggregator, "_generate_xau_atr_trade_setup",
|
||||
return_value=None):
|
||||
rep = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||||
assert rep.trade_setup is not None, \
|
||||
f"pipeline did not produce a setup (warnings: {rep.warnings})"
|
||||
assert rep.trade_setup.strategy == "XAU_CP"
|
||||
|
||||
|
||||
@test("CP pipeline: disabled by default — no XAU_CP setups anywhere")
|
||||
def _():
|
||||
assert XauStrategyConfig().cp_enabled is False
|
||||
assert PlatformConfig().xau_strategy.cp_enabled is False
|
||||
from pair_signals import PAIR_PRODUCERS
|
||||
for tf in (Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4):
|
||||
assert PAIR_PRODUCERS[XAUUSD](tf, None).xau_strategy.cp_enabled is False
|
||||
cfg = PlatformConfig()
|
||||
cfg.primary_timeframe = Timeframe.H1
|
||||
cfg.confirmation_timeframes = [Timeframe.M15, Timeframe.H4]
|
||||
df = make_cp_df(double_bottom_closes())
|
||||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||||
with patch.object(SignalAggregator, "_generate_trade_setup",
|
||||
return_value=None), \
|
||||
patch.object(SignalAggregator, "_generate_xau_atr_trade_setup",
|
||||
return_value=None):
|
||||
rep = SignalAggregator(cfg).analyze_from_dataframes(
|
||||
"XAU/USD", dfs, use_news=False)
|
||||
assert rep.trade_setup is None
|
||||
|
||||
|
||||
@test("CP pipeline: session-blocked bar never hosts a pattern trade")
|
||||
def _():
|
||||
cfg = _cp_config()
|
||||
cfg.confirmation_timeframes = [Timeframe.M15, Timeframe.H4]
|
||||
# Label hour 13 is in blocked_hours_utc → 6b sets F + SESSION FILTER.
|
||||
df = make_cp_df(double_bottom_closes(), end="2026-07-08 13:00")
|
||||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||||
with patch.object(SignalAggregator, "_generate_trade_setup",
|
||||
return_value=None), \
|
||||
patch.object(SignalAggregator, "_generate_xau_atr_trade_setup",
|
||||
return_value=None):
|
||||
rep = SignalAggregator(cfg).analyze_from_dataframes(
|
||||
"XAU/USD", dfs, use_news=False)
|
||||
assert rep.trade_setup is None
|
||||
assert rep.signal_quality == "F"
|
||||
|
||||
|
||||
@test("CP backtester: --cp flag enables patterns, tag filter knows XAU_CP")
|
||||
def _():
|
||||
from gpu_backtest import _parse_args
|
||||
args = _parse_args(["--pair", "XAUUSD", "--cp",
|
||||
"--cp-patterns", "double_bottom,bull_flag"])
|
||||
assert args.cp is True
|
||||
assert args.cp_patterns == "double_bottom,bull_flag"
|
||||
args2 = _parse_args(["--pair", "XAUUSD"])
|
||||
assert args2.cp is False
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
# SUMMARY
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
|
||||
Reference in New Issue
Block a user