#!/usr/bin/env python3 """XAUUSD hybrid scanner v2: multi-strategy candidate finder. Candidate finder only — not final backtest proof. It scans historical OHLCV for four strategy families that were not covered by the phase-1 Liquidity Sweep Reversal scaffold: - Aryy HTF Narrative Fib 50/62 - ICT Sweep + MSS + FVG - Order Block Reclaim - Breaker Continuation The output is strictly candidate/research data. Final validity still requires chronological manual TradingView replay review before journal promotion. """ from __future__ import annotations import argparse import json from collections import Counter from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] DEFAULT_OUTPUT = ROOT / 'candidates' / 'candidates_v2_multistrategy.json' SYMBOL = 'XAUUSD' VERSION = '0.2.0-multistrategy-scaffold' STRATEGIES = { 'fib_50_62': 'Aryy HTF Narrative Fib 50/62', 'sweep_mss_fvg': 'ICT Sweep + MSS + FVG', 'order_block_reclaim': 'Order Block Reclaim', 'breaker_continuation': 'Breaker Continuation', } SESSION_WINDOWS_UTC = [ ('Asia', 0, 7), ('London', 7, 10), ('NY AM', 12, 17), ('London Close', 15, 17), ('NY PM', 17, 21) ] @dataclass(frozen=True) class Bar: timestamp: str open: float high: float low: float close: float volume: float | None = None @property def dt(self) -> datetime: return parse_dt(self.timestamp) @property def body(self) -> float: return abs(self.close - self.open) @property def range(self) -> float: return self.high - self.low @property def bullish(self) -> bool: return self.close > self.open @property def bearish(self) -> bool: return self.close < self.open def now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace('+00:00','Z') def parse_dt(value: Any) -> datetime: if isinstance(value, (int, float)): seconds = value/1000 if value > 10_000_000_000 else value return datetime.fromtimestamp(seconds, tz=timezone.utc) text = str(value).strip().replace('Z', '+00:00') dt = datetime.fromisoformat(text) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) def norm_ts(value: Any) -> str: return parse_dt(value).replace(microsecond=0).isoformat().replace('+00:00','Z') def load_json(path: Path) -> Any: return json.loads(path.read_text(encoding='utf-8')) def rows(payload: Any) -> list[Any]: if isinstance(payload, list): return payload if isinstance(payload, dict): for k in ['bars','ohlcv','data','candles','values']: if isinstance(payload.get(k), list): return payload[k] raise ValueError('OHLCV JSON missing bars array') def row_to_bar(row: Any) -> Bar: if isinstance(row, dict): t = row.get('timestamp', row.get('time', row.get('date'))) return Bar(norm_ts(t), float(row['open']), float(row['high']), float(row['low']), float(row['close']), float(row['volume']) if row.get('volume') is not None else None) if isinstance(row, (list, tuple)) and len(row) >= 5: return Bar(norm_ts(row[0]), float(row[1]), float(row[2]), float(row[3]), float(row[4]), float(row[5]) if len(row) > 5 and row[5] is not None else None) raise ValueError(f'Unsupported row {row!r}') def load_bars(tf: str) -> list[Bar]: path = ROOT / 'data' / f'ohlcv_{tf}.json' return sorted([row_to_bar(r) for r in rows(load_json(path))], key=lambda b:b.dt) def session(dt: datetime) -> str: h = dt.hour for name,start,end in SESSION_WINDOWS_UTC: if start <= h < end: return name return 'Other' def slug_ts(ts: str) -> str: return ts.replace(':','-').replace('+00:00','Z') def mk_candidate(*, ts: str, strategy_key: str, direction: str, timeframe: str, entry_timeframe: str, score: float, flags: list[str], notes: str, levels: dict[str, Any]) -> dict[str, Any]: dt=parse_dt(ts) strategy=STRATEGIES[strategy_key] sid=strategy.lower().replace(' + ','_').replace(' ','_').replace('/','_') return { 'id': f"candidate_v2_{slug_ts(ts)}_{direction.lower()}_{sid}", 'date': dt.date().isoformat(), 'timestamp': ts, 'symbol': SYMBOL, 'strategy': strategy, 'strategy_family': strategy_key, 'direction': direction, 'session': session(dt), 'timeframe': timeframe, 'entry_timeframe': entry_timeframe, 'narrative': 'TBD manual replay', 'htf_context': 'TBD manual replay', 'ltf_trigger': 'Mechanical v2 candidate; manual TradingView replay required', 'entry': None, 'stop': None, 'target': None, 'fib': 'TBD', 'pd': 'TBD', 'candidate_score': round(score,2), 'confidence': 'Unreviewed', 'status': 'unreviewed', 'requires_manual_replay': True, 'scanner_flags': flags, 'levels': levels, 'notes': notes, } def avg_body(bars: list[Bar]) -> float: return sum(b.body for b in bars)/max(len(bars),1) def fvg_bull(prev: Bar, nxt: Bar) -> bool: return nxt.low > prev.high def fvg_bear(prev: Bar, nxt: Bar) -> bool: return nxt.high < prev.low def htf_bias(h1: list[Bar], dt: datetime) -> str: # Simple mechanical H1 trend proxy: close vs 20-bar midpoint/slope. hist=[b for b in h1 if b.dt <= dt] if len(hist) < 25: return 'unknown' win=hist[-20:] first=sum(b.close for b in win[:5])/5; last=sum(b.close for b in win[-5:])/5 if last > first and hist[-1].close > sum(b.close for b in win)/len(win): return 'bullish' if last < first and hist[-1].close < sum(b.close for b in win)/len(win): return 'bearish' return 'range' def scan_sweep_mss_fvg(m15: list[Bar], m5: list[Bar]) -> list[dict[str,Any]]: out=[] for i in range(12, len(m15)-2): b=m15[i]; prev=m15[i-1]; nxt=m15[i+1] look=m15[max(0,i-10):i] disp=b.body > avg_body(look)*1.4 if look else False bull_sweep=b.low < min(x.low for x in look[-5:]) and b.close > prev.low bear_sweep=b.high > max(x.high for x in look[-5:]) and b.close < prev.high bull_fvg = disp and b.bullish and fvg_bull(m15[i-1], m15[i+1]) bear_fvg = disp and b.bearish and fvg_bear(m15[i-1], m15[i+1]) if bull_sweep and bull_fvg: out.append(mk_candidate(ts=b.timestamp, strategy_key='sweep_mss_fvg', direction='Long', timeframe='M15', entry_timeframe='M5', score=62, flags=['sellside_sweep','bullish_displacement','bullish_fvg_candidate'], notes='Sweep + displacement + M15 bullish FVG candidate; confirm MSS/FVG replay manually.', levels={'sweep_low':b.low,'fvg_low':m15[i-1].high,'fvg_high':m15[i+1].low})) if bear_sweep and bear_fvg: out.append(mk_candidate(ts=b.timestamp, strategy_key='sweep_mss_fvg', direction='Short', timeframe='M15', entry_timeframe='M5', score=62, flags=['buyside_sweep','bearish_displacement','bearish_fvg_candidate'], notes='Sweep + displacement + M15 bearish FVG candidate; confirm MSS/FVG replay manually.', levels={'sweep_high':b.high,'fvg_low':m15[i+1].high,'fvg_high':m15[i-1].low})) return out def scan_fib_50_62(m30: list[Bar], m15: list[Bar], h1: list[Bar]) -> list[dict[str,Any]]: out=[] for i in range(20, len(m30)-8): b=m30[i]; look=m30[i-10:i] disp=b.body > avg_body(look)*1.7 if look else False if not disp: continue bias=htf_bias(h1, b.dt) # Look for retrace to 50-62 of displacement range within next 8 M30 bars. if b.bullish and bias in ['bullish','range']: hi=b.high; lo=b.low; z50=hi-(hi-lo)*0.5; z62=hi-(hi-lo)*0.62 for j in range(i+1, min(len(m30), i+9)): r=m30[j] if r.low <= max(z50,z62) and r.high >= min(z50,z62): out.append(mk_candidate(ts=r.timestamp, strategy_key='fib_50_62', direction='Long', timeframe='M30', entry_timeframe='M15/M5', score=58 if bias=='range' else 68, flags=['bullish_impulse','fib_50_62_retrace','htf_'+bias], notes='HTF/Fib 50-62 retracement candidate after bullish displacement; validate PD array overlap manually.', levels={'impulse_low':lo,'impulse_high':hi,'fib50':z50,'fib62':z62})) break if b.bearish and bias in ['bearish','range']: hi=b.high; lo=b.low; z50=lo+(hi-lo)*0.5; z62=lo+(hi-lo)*0.62 for j in range(i+1, min(len(m30), i+9)): r=m30[j] if r.high >= min(z50,z62) and r.low <= max(z50,z62): out.append(mk_candidate(ts=r.timestamp, strategy_key='fib_50_62', direction='Short', timeframe='M30', entry_timeframe='M15/M5', score=58 if bias=='range' else 68, flags=['bearish_impulse','fib_50_62_retrace','htf_'+bias], notes='HTF/Fib 50-62 retracement candidate after bearish displacement; validate PD array overlap manually.', levels={'impulse_low':lo,'impulse_high':hi,'fib50':z50,'fib62':z62})) break return out def scan_order_block_reclaim(m15: list[Bar]) -> list[dict[str,Any]]: out=[] for i in range(15, len(m15)-10): prev=m15[i-1]; impulse=m15[i] look=m15[max(0,i-10):i] disp=impulse.body > avg_body(look)*1.8 if look else False if not disp: continue if prev.bearish and impulse.bullish and impulse.close > max(x.high for x in look[-5:]): zone_low,zone_high=prev.low,prev.high for j in range(i+1, min(len(m15), i+12)): r=m15[j] if r.low <= zone_high and r.high >= zone_low and r.close > zone_low: out.append(mk_candidate(ts=r.timestamp, strategy_key='order_block_reclaim', direction='Long', timeframe='M15', entry_timeframe='M5', score=60, flags=['bullish_ob_candidate','displacement_break','ob_retest'], notes='Bullish OB reclaim/retest candidate; validate OB quality and liquidity context manually.', levels={'ob_low':zone_low,'ob_high':zone_high,'displacement_high':impulse.high})) break if prev.bullish and impulse.bearish and impulse.close < min(x.low for x in look[-5:]): zone_low,zone_high=prev.low,prev.high for j in range(i+1, min(len(m15), i+12)): r=m15[j] if r.high >= zone_low and r.low <= zone_high and r.close < zone_high: out.append(mk_candidate(ts=r.timestamp, strategy_key='order_block_reclaim', direction='Short', timeframe='M15', entry_timeframe='M5', score=60, flags=['bearish_ob_candidate','displacement_break','ob_retest'], notes='Bearish OB reclaim/retest candidate; validate OB quality and liquidity context manually.', levels={'ob_low':zone_low,'ob_high':zone_high,'displacement_low':impulse.low})) break return out def scan_breaker_continuation(m15: list[Bar]) -> list[dict[str,Any]]: out=[] # Approx breaker: sweep one side, break opposite structure, then retest failed prior candle zone. for i in range(20, len(m15)-16): pre=m15[i-12:i] b=m15[i] # bullish breaker: lows swept then later close above pre high, retest midpoint zone if b.low < min(x.low for x in pre): broken_idx=None pre_high=max(x.high for x in pre) for j in range(i+1, min(len(m15), i+8)): if m15[j].close > pre_high: broken_idx=j; break if broken_idx: zone_low,zone_high=b.open,b.high for k in range(broken_idx+1, min(len(m15), broken_idx+10)): r=m15[k] if r.low <= max(zone_low,zone_high) and r.close > min(zone_low,zone_high): out.append(mk_candidate(ts=r.timestamp, strategy_key='breaker_continuation', direction='Long', timeframe='M15', entry_timeframe='M5', score=57, flags=['sellside_sweep','bullish_breaker_candidate','breaker_retest'], notes='Bullish breaker continuation candidate; validate failed OB/breaker manually.', levels={'sweep_low':b.low,'breaker_low':min(zone_low,zone_high),'breaker_high':max(zone_low,zone_high),'break_high':pre_high})) break if b.high > max(x.high for x in pre): broken_idx=None pre_low=min(x.low for x in pre) for j in range(i+1, min(len(m15), i+8)): if m15[j].close < pre_low: broken_idx=j; break if broken_idx: zone_low,zone_high=b.low,b.open for k in range(broken_idx+1, min(len(m15), broken_idx+10)): r=m15[k] if r.high >= min(zone_low,zone_high) and r.close < max(zone_low,zone_high): out.append(mk_candidate(ts=r.timestamp, strategy_key='breaker_continuation', direction='Short', timeframe='M15', entry_timeframe='M5', score=57, flags=['buyside_sweep','bearish_breaker_candidate','breaker_retest'], notes='Bearish breaker continuation candidate; validate failed OB/breaker manually.', levels={'sweep_high':b.high,'breaker_low':min(zone_low,zone_high),'breaker_high':max(zone_low,zone_high),'break_low':pre_low})) break return out def dedupe(cands: list[dict[str,Any]]) -> list[dict[str,Any]]: seen=set(); out=[] for c in sorted(cands, key=lambda x:(x['timestamp'], x['strategy_family'], x['direction'])): key=(c['timestamp'], c['strategy_family'], c['direction']) if key in seen: continue seen.add(key); out.append(c) return out def main() -> int: ap=argparse.ArgumentParser() ap.add_argument('--output', type=Path, default=DEFAULT_OUTPUT) args=ap.parse_args() m5=load_bars('M5'); m15=load_bars('M15'); m30=load_bars('M30'); h1=load_bars('H1') candidates=[] candidates += scan_sweep_mss_fvg(m15,m5) candidates += scan_fib_50_62(m30,m15,h1) candidates += scan_order_block_reclaim(m15) candidates += scan_breaker_continuation(m15) candidates=dedupe(candidates) counts=Counter(c['strategy'] for c in candidates) obj={'metadata':{'symbol':SYMBOL,'mode':'hybrid_candidate_scanner_v2_multistrategy','period':'2026','source':'historical_ohlcv_mechanical_scan_v2','isFinalBacktest':False,'generated_at':now_iso(),'scanner_version':VERSION,'candidate_count':len(candidates),'strategy_counts':dict(counts),'notes':'Candidate finder only for phase-2 multi-strategy coverage; final validity requires manual TradingView Bar Replay.'},'candidates':candidates} args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(obj, indent=2, ensure_ascii=False)+'\n', encoding='utf-8') print(f'wrote {len(candidates)} candidates -> {args.output}') print('strategy_counts', dict(counts)) return 0 if __name__ == '__main__': raise SystemExit(main())