From a70fc91f96b8d618b3cacf8c2d8576aed8cde1cd Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Thu, 16 Jul 2026 14:23:17 -0400 Subject: [PATCH] =?UTF-8?q?Add=20chart-pattern=20entry=20archetype=20(XAU?= =?UTF-8?q?=5FCP)=20=E2=80=94=20disabled=20by=20default?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_015kby7M9JwpYi8NQT2yoxAi --- backtest/gpu_backtest.py | 16 +- chart_patterns.py | 287 +++++++++++++++++++++++++++++++++++ config.py | 28 ++++ signal_aggregator.py | 153 +++++++++++++++++++ test_all.py | 318 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 801 insertions(+), 1 deletion(-) create mode 100644 chart_patterns.py diff --git a/backtest/gpu_backtest.py b/backtest/gpu_backtest.py index 9eaa89e..ccf180c 100644 --- a/backtest/gpu_backtest.py +++ b/backtest/gpu_backtest.py @@ -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 = ( diff --git a/chart_patterns.py b/chart_patterns.py new file mode 100644 index 0000000..84895db --- /dev/null +++ b/chart_patterns.py @@ -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 diff --git a/config.py b/config.py index 9d860c8..16c6347 100755 --- a/config.py +++ b/config.py @@ -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: diff --git a/signal_aggregator.py b/signal_aggregator.py index 326e9bf..d654df1 100755 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -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 ────────────────────────────────────────────────── diff --git a/test_all.py b/test_all.py index 46da68d..2a917fa 100644 --- a/test_all.py +++ b/test_all.py @@ -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 # ══════════════════════════════════════════════════════════════════════