#!/usr/bin/env python3 """V5 range-market strategy scanner/reviewer for XAUUSD. Tests programmable non-ICT/SMC range strategies: - Bollinger Band Mean Reversion - RSI Range Reversion - VWAP Mean Reversion (session VWAP proxy) - Donchian Range Boundary Fade - Keltner Channel Mean Reversion - Market Regime Switch Combo Fair rules: candidate + entry touched => outcome TP/SL first. Losses included. Research outputs only; no production journal writes. """ from __future__ import annotations import json, math, statistics, argparse from collections import Counter, defaultdict from datetime import datetime, timezone from pathlib import Path ROOT=Path(__file__).resolve().parents[1] OUT_C=ROOT/'candidates/candidates_v5_range_strategies.json' OUT_J=ROOT/'reviews/fair_journal_preview_v5_range_strategies.json' OUT_S=ROOT/'reviews/shadow_v5_range_strategies.json' OUT_R=ROOT/'reports/fair_v5_range_stats_summary.json' def parse_dt(v): dt=datetime.fromisoformat(str(v).replace('Z','+00:00')) if dt.tzinfo is None: dt=dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) def iso(dt): return dt.replace(microsecond=0).isoformat().replace('+00:00','Z') def load_m5(): obj=json.loads((ROOT/'data/ohlcv_M5.json').read_text()); rows=obj.get('bars',obj) if isinstance(obj,dict) else obj; out=[] for r in rows: t=r.get('timestamp',r.get('time',r.get('date'))) if isinstance(r,dict) else r[0] vals=r if isinstance(r,dict) else {'open':r[1],'high':r[2],'low':r[3],'close':r[4]} out.append({'dt':parse_dt(t),'timestamp':iso(parse_dt(t)),'open':float(vals['open']),'high':float(vals['high']),'low':float(vals['low']),'close':float(vals['close'])}) return sorted(out,key=lambda x:x['dt']) def sma(vals,n): return sum(vals[-n:])/n if len(vals)>=n else None def stdev(vals,n): return statistics.pstdev(vals[-n:]) if len(vals)>=n else None def ema(prev,val,n): a=2/(n+1); return val if prev is None else prev+a*(val-prev) def rsi(closes,n=14): if len(closes) use typical price average from session start if not hist: return None cur_date=hist[-1]['dt'].date(); cur_session=session(hist[-1]['dt']) ss=[b for b in hist if b['dt'].date()==cur_date and session(b['dt'])==cur_session] if not ss: ss=hist[-24:] return statistics.mean([(b['high']+b['low']+b['close'])/3 for b in ss]) def range_regime(hist): closes=[b['close'] for b in hist] if len(hist)<50: return False, {} adx=adx_proxy(hist,14); ma=sma(closes,20); sd=stdev(closes,20); bbw=(4*sd/ma*100) if ma and sd else 999 # Range if low directional efficiency and not too expanded. is_range=(adx is not None and adx<28 and bbw<1.2) return is_range, {'adx_proxy':round(adx,2) if adx is not None else None,'bb_width_pct':round(bbw,3)} def hit(direction,entry,stop,target,forward): for b in forward: if direction=='Long': sl=b['low']<=stop; tp=b['high']>=target else: sl=b['high']>=stop; tp=b['low']<=target if sl and tp: return 'ambiguous_same_bar',b['timestamp'] if sl: return 'Loss',b['timestamp'] if tp: return 'Win',b['timestamp'] return 'Review',None def mk(ts,strategy,direction,entry,stop,target,planned,flags,regime,notes): dt=parse_dt(ts) return {'id':f"candidate_v5_{ts.replace(':','-')}_{direction.lower()}_{strategy.lower().replace(' ','_').replace('/','_')}",'date':dt.date().isoformat(),'timestamp':ts,'symbol':'XAUUSD','strategy':strategy,'direction':direction,'session':session(dt),'market_regime':'range','entry_timeframe':'M5','entry':round(entry,3),'stop':round(stop,3),'target':round(target,3),'planned_rr':round(planned,3),'scanner_flags':flags,'regime':regime,'notes':notes,'isFinalBacktest':False} def gen_candidates(bars): c=[]; closes=[]; ema20=None; atrs=[] for i,b in enumerate(bars): closes.append(b['close']); ema20=ema(ema20,b['close'],20); hist=bars[:i+1] is_range,reg=range_regime(hist) if not is_range or i<60: continue ma=sma(closes,20); sd=stdev(closes,20); a=atr(hist,14); rs=rsi(closes,14); vw=vwap_session(hist) if not (ma and sd and a and rs is not None and vw): continue upper=ma+2*sd; lower=ma-2*sd; k_up=ema20+1.5*a; k_lo=ema20-1.5*a # planned 2R by target distance; stop placed half target-distance beyond extreme. # Bollinger if b['low']<=lower and b['close']>lower: entry=b['close']; target=ma; reward=target-entry if reward>0: c.append(mk(b['timestamp'],'Bollinger Band Mean Reversion','Long',entry,entry-reward/2,target,2.0,['bb_lower_reclaim'],reg,'Lower band reclaim to mean.')) if b['high']>=upper and b['close']0: c.append(mk(b['timestamp'],'Bollinger Band Mean Reversion','Short',entry,entry+reward/2,target,2.0,['bb_upper_reject'],reg,'Upper band rejection to mean.')) # RSI if rs<32 and b['close']>b['open']: entry=b['close']; target=ma; reward=target-entry if reward>0: c.append(mk(b['timestamp'],'RSI Range Reversion','Long',entry,entry-reward/2,target,2.0,['rsi_oversold_reclaim'],reg,'RSI oversold reclaim toward mean.')) if rs>68 and b['close']0: c.append(mk(b['timestamp'],'RSI Range Reversion','Short',entry,entry+reward/2,target,2.0,['rsi_overbought_reject'],reg,'RSI overbought rejection toward mean.')) # VWAP deviation dev=abs(b['close']-vw) if dev>1.2*a: if b['close']b['open']: entry=b['close']; target=vw; reward=target-entry if reward>0: c.append(mk(b['timestamp'],'VWAP Mean Reversion','Long',entry,entry-reward/2,target,2.0,['vwap_lower_deviation_reclaim'],reg,'VWAP lower deviation reclaim.')) if b['close']>vw and b['close']0: c.append(mk(b['timestamp'],'VWAP Mean Reversion','Short',entry,entry+reward/2,target,2.0,['vwap_upper_deviation_reject'],reg,'VWAP upper deviation reject.')) # Donchian 40 range boundary fade look=hist[-40:]; hi=max(x['high'] for x in look[:-1]); lo=min(x['low'] for x in look[:-1]); mid=(hi+lo)/2 if b['low']<=lo+0.15*(hi-lo) and b['close']>b['open']: entry=b['close']; target=mid; reward=target-entry if reward>0: c.append(mk(b['timestamp'],'Donchian Range Boundary Fade','Long',entry,entry-reward/2,target,2.0,['range_low_reject'],reg,'Donchian range low rejection to mid.')) if b['high']>=hi-0.15*(hi-lo) and b['close']0: c.append(mk(b['timestamp'],'Donchian Range Boundary Fade','Short',entry,entry+reward/2,target,2.0,['range_high_reject'],reg,'Donchian range high rejection to mid.')) # Keltner if b['low']<=k_lo and b['close']>k_lo: entry=b['close']; target=ema20; reward=target-entry if reward>0: c.append(mk(b['timestamp'],'Keltner Channel Mean Reversion','Long',entry,entry-reward/2,target,2.0,['keltner_lower_reclaim'],reg,'Keltner lower reclaim to EMA.')) if b['high']>=k_up and b['close']0: c.append(mk(b['timestamp'],'Keltner Channel Mean Reversion','Short',entry,entry+reward/2,target,2.0,['keltner_upper_reject'],reg,'Keltner upper reject to EMA.')) # dedupe by timestamp/strategy/direction/entry seen=set(); out=[] for x in c: k=(x['timestamp'],x['strategy'],x['direction'],x['entry']) if k not in seen: seen.add(k); out.append(x) return out def review(cands,bars): import bisect times=[b['dt'] for b in bars]; trades=[]; shadows=[] for idx,c in enumerate(cands,1): i=bisect.bisect_right(times,parse_dt(c['timestamp'])) fwd=bars[i:min(len(bars),i+144)] outcome,out_ts=hit(c['direction'],c['entry'],c['stop'],c['target'],fwd) if outcome=='Review': shadows.append({**c,'type':'Range shadow unresolved','outcome':'Review','r':None}) continue if outcome=='ambiguous_same_bar': shadows.append({**c,'type':'Range shadow ambiguous','outcome':'Review','r':None}) continue r=c['planned_rr'] if outcome=='Win' else -1.0 trades.append({'id':f"fair_v5_range_{idx:05d}",'date':c['date'],'timestamp':c['timestamp'],'symbol':'XAUUSD','strategy':c['strategy'],'direction':c['direction'],'session':c['session'],'market_regime':'range','narrative':'Range/sideways fair strategy backtest','htf_context':f"Range regime: ADX proxy {c['regime'].get('adx_proxy')}, BB width {c['regime'].get('bb_width_pct')}%",'ltf_trigger':c['notes'],'entry':c['entry'],'stop':c['stop'],'target':c['target'],'planned_rr':c['planned_rr'],'outcome':outcome,'r':r,'confidence':'V5 range deterministic review','notes':f"{c['notes']} outcome={outcome} at {out_ts}",'source_candidate_id':c['id'],'source_review':'hybrid/reviews/fair_journal_preview_v5_range_strategies.json','is_confirmed_trade':True}) return trades,shadows def maxdd(trades): eq=10000; peak=eq; md=0; pct=0; when=None for t in sorted(trades,key=lambda x:x['timestamp']): eq+=float(t['r'])*100; peak=max(peak,eq); d=peak-eq; p=d/peak*100 if peak else 0 if p>pct: pct=p; md=d; when=t['timestamp'] return round(md/100,2),round(pct,2),when,round(eq,2) def stats(arr): c=Counter(t['outcome'] for t in arr); rs=[float(t['r']) for t in arr]; ddr,ddp,when,eq=maxdd(arr) return {'trades':len(arr),'wins':c['Win'],'losses':c['Loss'],'winrate':round(c['Win']/len(arr)*100,2) if arr else 0,'total_R':round(sum(rs),2) if arr else 0,'avg_R':round(statistics.mean(rs),3) if rs else 0,'maxDD_R':ddr,'maxDD_pct_peak':ddp,'maxDD_when':when,'final_equity_100risk':eq} def main(): bars=load_m5(); cands=gen_candidates(bars); trades,shadows=review(cands,bars) OUT_C.write_text(json.dumps({'metadata':{'purpose':'v5_range_candidates','count':len(cands),'isFinalBacktest':False},'candidates':cands},indent=2,ensure_ascii=False)+'\n') OUT_J.write_text(json.dumps({'metadata':{'purpose':'fair_v5_range_journal_preview','trade_count':len(trades),'isFinalBacktest':False},'trades':trades},indent=2,ensure_ascii=False)+'\n') OUT_S.write_text(json.dumps({'metadata':{'purpose':'v5_range_shadow','count':len(shadows),'isFinalBacktest':False},'shadowTrades':shadows},indent=2,ensure_ascii=False)+'\n') by=defaultdict(list) for t in trades: by[t['strategy']].append(t) summary={'generated_at':iso(datetime.now(timezone.utc)),'candidate_count':len(cands),'shadow_count':len(shadows),'overall':stats(trades),'strategy_stats':{k:stats(v) for k,v in sorted(by.items())},'candidate_strategy_counts':dict(Counter(c['strategy'] for c in cands))} OUT_R.write_text(json.dumps(summary,indent=2,ensure_ascii=False)+'\n') print(json.dumps(summary,indent=2)) if __name__=='__main__': main()