"""analysis/timeframes.py — HTF/MTF/LTF/WTF timeframe analysis functions.""" import numpy as np import pandas as pd from config import (RSI_OB, RSI_OS, RSI_MA_PERIOD, ADX_MIN, MTF_MIN_SCORE) from analysis.structure import detect_swings, find_sr, analyse_structure, analyse_volume, analyse_sr, analyse_momentum from data.oanda import pip, digs def analyse_htf(df): """ D1 bias using EMA20 / EMA50 / SMA200 stack. Bullish: EMA20 > EMA50 > SMA200 AND price above EMA20 Bearish: EMA20 < EMA50 < SMA200 AND price below EMA20 Neutral: mixed / conflicting 3-point scoring: +1 EMA20 vs EMA50 direction +1 Full EMA stack aligned (EMA20/50/SMA200) +1 Price on correct side of EMA20 """ row = df.iloc[-1] ema_bull = row["ma_fast"] > row["ma_slow"] # EMA20 > EMA50 ema_bear = row["ma_fast"] < row["ma_slow"] # Full stack check (EMA20 > EMA50 > SMA200) has_trend = "ma_trend" in df.columns and not pd.isna(row.get("ma_trend", float("nan"))) stack_bull = bool(row.get("ema_stack_bull", False)) if has_trend else ema_bull stack_bear = bool(row.get("ema_stack_bear", False)) if has_trend else ema_bear # Price vs EMA20 price_bull = row["close"] > row["ma_fast"] price_bear = row["close"] < row["ma_fast"] # SMA200 slope (compare to 5 bars ago) if has_trend: sma200_now = row["ma_trend"] sma200_prev = df["ma_trend"].iloc[-6] sma_up = sma200_now > sma200_prev sma_down = sma200_now < sma200_prev sma_label = "rising" if sma_up else "falling" if sma_down else "flat" sma_val = f"{sma200_now:.5f}" else: sma_up = sma_down = False; sma_label = "N/A"; sma_val = "N/A" bull = sum([ema_bull, stack_bull, price_bull]) bear = sum([ema_bear, stack_bear, price_bear]) if bull >= 2 and bull > bear: bias = 1 elif bear >= 2 and bear > bull: bias = -1 else: bias = 0 return { "ema_fast": float(row["ema_fast"]), "ema_mid": float(row["ema_mid"]), "ema_slow": float(row["ema_slow"]), "bias": bias, "rsi": float(row["rsi"]), "atr": float(row["atr"]), "sr_levels": find_sr(df, instrument, n=8), } def detect_rsi_ma_bounce(df, bias): """ RSI SMA-9 Bounce — 4H bonus confluence check. Bullish bounce (bias=1): RSI dipped below its SMA9 recently (last 5 bars) RSI is now back above SMA9 and rising → Momentum pulled back to average and is resuming upward Bearish bounce (bias=-1): RSI rallied above its SMA9 recently (last 5 bars) RSI is now back below SMA9 and falling → Momentum bounced to average and is resuming downward This is a bonus +0.5 — not a required check. Most valuable in trending markets (ADX > 20). Returns: (detected: bool, note: str, rsi_val: float, rsi_ma_val: float) """ if "rsi_ma" not in df.columns or len(df) < RSI_MA_PERIOD + 3: return False, "RSI MA not available", 0, 0 row = df.iloc[-1] prev = df.iloc[-2] recent = df.iloc[-6:-1] # last 5 bars before current rsi_now = row["rsi"] rsi_ma_now = row["rsi_ma"] rsi_prev = prev["rsi"] if pd.isna(rsi_ma_now): return False, "RSI MA not ready", rsi_now, 0 if bias == 1: # Bullish — RSI bouncing up off SMA9 # Recently dipped below SMA9 dipped_below = any(recent["rsi"] < recent["rsi_ma"]) # Now back above SMA9 back_above = rsi_now > rsi_ma_now # RSI rising rsi_rising = rsi_now > rsi_prev # SMA9 itself is rising or flat (not falling steeply) ma_direction = rsi_ma_now >= df["rsi_ma"].iloc[-4] detected = dipped_below and back_above and rsi_rising note = (f"RSI({rsi_now:.1f}) bounced off SMA9({rsi_ma_now:.1f}) ↑ " f"{'✅' if detected else '❌'} " f"| dip:{'✅' if dipped_below else '❌'} " f"above:{'✅' if back_above else '❌'} " f"rising:{'✅' if rsi_rising else '❌'}") else: # Bearish — RSI bouncing down off SMA9 # Recently rallied above SMA9 rallied_above = any(recent["rsi"] > recent["rsi_ma"]) # Now back below SMA9 back_below = rsi_now < rsi_ma_now # RSI falling rsi_falling = rsi_now < rsi_prev # SMA9 itself is falling or flat ma_direction = rsi_ma_now <= df["rsi_ma"].iloc[-4] detected = rallied_above and back_below and rsi_falling note = (f"RSI({rsi_now:.1f}) rejected at SMA9({rsi_ma_now:.1f}) ↓ " f"{'✅' if detected else '❌'} " f"| rally:{'✅' if rallied_above else '❌'} " f"below:{'✅' if back_below else '❌'} " f"falling:{'✅' if rsi_falling else '❌'}") return detected, note, round(rsi_now, 1), round(rsi_ma_now, 1) def find_weekly_sr(df_wtf, instrument, n=6): """ Weekly S&R levels — highest quality levels because they represent multi-week institutional accumulation/distribution zones. Uses swing highs/lows from W1 chart with larger clustering zone. """ if df_wtf.empty or len(df_wtf) < 5: return [] df_sw = detect_swings(df_wtf, n=2) highs = df_sw["swing_high"].dropna() lows = df_sw["swing_low"].dropna() raw = pd.concat([highs, lows]).sort_values().values if not len(raw): return [] # Weekly levels cluster in a wider zone (20 pips) zone = 20 * pip(instrument) clusters = []; cur = [raw[0]] for p_val in raw[1:]: if p_val - cur[-1] <= zone: cur.append(p_val) else: clusters.append(cur); cur = [p_val] clusters.append(cur) clusters.sort(key=len, reverse=True) return [round(np.mean(c), digs(instrument)) for c in clusters[:n]] def check_round_number(price, instrument, pips=20): """ Round number proximity check. Round numbers (1.3000, 150.00, 2.0000) act as magnets — price gravitates toward them and often reacts strongly. Checks if price is within `pips` of a round number. Returns (near: bool, nearest_round: float, distance_pips: float) """ p = pip(instrument) d = digs(instrument) # Determine round number step based on instrument if "_JPY" in instrument: # JPY pairs: round at 0.5 and whole numbers (e.g. 150.00, 150.50) step = 0.5 elif instrument == "XAU_USD": # Gold: round at 10s (1800, 1810, 1900) step = 10.0 else: # All others: round at 0.0100 and 0.0050 (e.g. 1.3000, 1.3050) step = 0.01 # Find nearest round number nearest = round(round(price / step) * step, d) distance_pips = abs(price - nearest) / p near = distance_pips <= pips return near, nearest, round(distance_pips, 1) def check_multitf_sr(df_htf, df_mtf, instrument, price): """ Multi-timeframe S&R confluence. A level that appears on BOTH D1 and 4H is far stronger than one that only shows on 4H alone. Returns: confluence: bool — a level is confirmed on both TFs level: float — the confluent level note: str — description """ d1_levels = find_sr(df_htf, instrument, n=8) h4_levels = find_sr(df_mtf, instrument, n=8) zone = 15 * pip(instrument) # 15 pip matching zone between TFs d = digs(instrument) confluent = [] for d1_l in d1_levels: for h4_l in h4_levels: if abs(d1_l - h4_l) <= zone: avg = (d1_l + h4_l) / 2 dist = abs(price - avg) / pip(instrument) confluent.append((avg, dist)) if not confluent: return False, None, "No D1+4H confluent levels" # Find nearest confluent level to current price confluent.sort(key=lambda x: x[1]) best_level, best_dist = confluent[0] near = best_dist <= 30 # within 30 pips return near, round(best_level, d), f"D1+4H level @ {best_level:.{d}f} ({best_dist:.0f} pips away)" def analyse_wtf(df): """ Weekly (W1) macro bias — top of the stack. Uses same EMA20/EMA50/SMA200 logic as D1 but on weekly candles. Weekly bias is the most important layer — sets the macro direction. Scoring (3 points): +1 EMA20 vs EMA50 direction +1 Full stack aligned (EMA20/50/SMA200) +1 Price on correct side of weekly EMA20 Returns bias: 1 (bullish), -1 (bearish), 0 (neutral) Plus extra context: key weekly levels, trend strength """ if df.empty or len(df) < 10: return {"bias": 0, "bull_pts": 0, "bear_pts": 0, "ema20": None, "ema50": None, "sma200": "N/A", "sma_slope": "N/A", "rsi": 50.0, "adx": 0.0, "stack_bull": False, "stack_bear": False, "note": "Insufficient weekly data"} row = df.iloc[-1] ema_bull = row["ma_fast"] > row["ma_slow"] ema_bear = row["ma_fast"] < row["ma_slow"] has_trend = "ma_trend" in df.columns and not pd.isna(row.get("ma_trend", float("nan"))) stack_bull = bool(row.get("ema_stack_bull", False)) if has_trend else ema_bull stack_bear = bool(row.get("ema_stack_bear", False)) if has_trend else ema_bear price_bull = row["close"] > row["ma_fast"] price_bear = row["close"] < row["ma_fast"] if has_trend: sma200_now = row["ma_trend"] sma200_prev = df["ma_trend"].iloc[-4] # 4 weeks ago sma_label = "rising" if sma200_now > sma200_prev else "falling" if sma200_now < sma200_prev else "flat" sma_val = f"{sma200_now:.5f}" else: sma_label = "N/A"; sma_val = "N/A"; sma200_now = None bull = sum([ema_bull, stack_bull, price_bull]) bear = sum([ema_bear, stack_bear, price_bear]) if bull >= 2 and bull > bear: bias = 1 elif bear >= 2 and bear > bull: bias = -1 else: bias = 0 # Weekly key levels — swing highs/lows as major S/R w_sw = detect_swings(df, n=2) recent_w = w_sw.iloc[-20:] w_highs = recent_w["swing_high"].dropna().values[-3:] if len(recent_w["swing_high"].dropna()) >= 1 else [] w_lows = recent_w["swing_low"].dropna().values[-3:] if len(recent_w["swing_low"].dropna()) >= 1 else [] return { "bias": bias, "bull_pts": bull, "bear_pts": bear, "ema20": round(row["ma_fast"], 5), "ema50": round(row["ma_slow"], 5), "sma200": sma_val, "sma200_raw": sma200_now, "sma_slope": sma_label, "rsi": round(row["rsi"], 1), "adx": round(row["adx"], 1), "stack_bull": stack_bull, "stack_bear": stack_bear, "zone": row["price_zone"], "w_highs": [round(h, 5) for h in w_highs], "w_lows": [round(l, 5) for l in w_lows], "w_sr": [], "note": "", } def analyse_mtf(df,htf_bias,instrument): if htf_bias==0: return {"confirmed":False,"score":0,"max":7,"checks":{}} row=df.iloc[-1]; checks={}; score=0 def chk(key,ok,note): nonlocal score; checks[key]=(ok,note); score+=int(ok) # EMA20 vs EMA50 direction ema_dir_ok = row["ma_fast"]>row["ma_slow"] if htf_bias==1 else row["ma_fast"]' if htf_bias==1 else '<'})EMA50: {row['ma_fast']:.5f} vs {row['ma_slow']:.5f}") # Full EMA stack: EMA20/EMA50/SMA200 all aligned has_stack = "ema_stack_bull" in df.columns if has_stack: stack_ok = bool(row.get("ema_stack_bull",False)) if htf_bias==1 else bool(row.get("ema_stack_bear",False)) sma200 = f"{row['ma_trend']:.5f}" if "ma_trend" in df.columns else "N/A" chk("1b_ema_stack", stack_ok, f"EMA20/50/SMA200 stack {'✅' if stack_ok else '❌'} | SMA200:{sma200}") zone_ok=row["price_zone"] in (("above_both","between_pb_up") if htf_bias==1 else ("below_both","between_pb_dn")) zone_labels = { "above_both": "above both EMAs — strong trend ✅", "between_pb_up": "between EMAs — healthy pullback ✅", "below_both": "below both EMAs — strong downtrend ✅", "between_pb_dn": "between EMAs — bearish pullback ✅", "mixed": "mixed — price not clearly positioned ❌", } zone_note = zone_labels.get(row["price_zone"], row["price_zone"]) chk("2_zone", zone_ok, f"Zone: {row['price_zone']} → {zone_note}") rsi_ok=row["rsi"]RSI_OS chk("3_rsi",rsi_ok,f"RSI:{row['rsi']:.1f}") ok,note=analyse_structure(df,htf_bias); chk("4_struct",ok,note) ok,note=analyse_volume(df,htf_bias); chk("5_vol",ok,note) ok,note=analyse_sr(df,htf_bias,instrument); chk("6_sr",ok,note) ok,note=analyse_momentum(df,htf_bias); chk("7_mom",ok,note) mtf_max = 8 if any("ema_stack" in k for k in checks) else 7 # ── RSI MA Bounce bonus (+0.5, not counted in max score) ── rsi_bounce_ok, rsi_bounce_note, rsi_val, rsi_ma_val = detect_rsi_ma_bounce(df, htf_bias) rsi_bonus = 0.5 if rsi_bounce_ok else 0.0 return { "zone": row["price_zone"], "sr_levels": find_sr(df, instrument, n=8), } def analyse_ltf(df, htf_bias): """ LTF (1H or 30M) entry trigger. Two conditions must pass: 1. EMA cross — EMA20 crossed EMA50 in bias direction within last 5 bars OR EMA20 is currently on correct side (crossed more than 5 bars ago but still valid — strong trend continuation) 2. RSI confirmation — two modes: PULLBACK mode (normal): RSI reversed from extreme → RSI dipped near OS (sell: spiked near OB) and now reversing TREND mode (strong trend, ADX > 30 on LTF): → RSI just needs to be on correct side and falling/rising → Used when price is in a straight trend without RSI spikes → e.g. NZD/USD strong downtrend: RSI at 30 and falling = valid SELL """ if htf_bias == 0: return {"triggered":False,"signal":0,"cross_ok":False,"rsi_ok":False, "rsi_mode":"none"} row = df.iloc[-1] prev = df.iloc[-2] recent = df.iloc[-10:] # look back 10 bars for RSI extreme # ── EMA Cross check ────────────────────────────────── # Primary: crossover bar in last 5 bars # Secondary: EMA20 already on correct side + crossed in last 10 bars # Tertiary: EMA20 aligned + strong trend (ADX > 25) — cross happened earlier but trend intact if htf_bias == 1: cross_recent = any(df["cross"].iloc[-5:] == 1) cross_aligned = row["ma_fast"] > row["ma_slow"] adx_strong_ltf = row["adx"] > 25 # require momentum for trend entries cross_extended = any(df["cross"].iloc[-10:] == 1) and adx_strong_ltf # cross_trend: EMA aligned + ADX > 25 — prevents false entries in ranging markets cross_trend = cross_aligned and adx_strong_ltf and row["price_zone"] in ("above_both", "between_pb_up") else: cross_recent = any(df["cross"].iloc[-5:] == -1) cross_aligned = row["ma_fast"] < row["ma_slow"] adx_strong_ltf = row["adx"] > 25 cross_extended = any(df["cross"].iloc[-10:] == -1) and adx_strong_ltf # cross_trend: EMA aligned + ADX > 25 cross_trend = cross_aligned and adx_strong_ltf and row["price_zone"] in ("below_both", "between_pb_dn") cross_ok = cross_recent or cross_extended or cross_trend # ── RSI check — two modes ──────────────────────────── adx_strong = row["adx"] > 30 # strong trend mode if htf_bias == 1: # PULLBACK mode: RSI dipped near oversold then recovering rsi_pullback = ((recent["rsi"] < RSI_OS + 10).any() and row["rsi"] > prev["rsi"] and row["rsi"] < RSI_OB) # TREND mode: RSI below 55 and rising (strong bull, ADX > 30) rsi_trend = adx_strong and row["rsi"] < 55 and row["rsi"] > prev["rsi"] # MOMENTUM mode: RSI above 55 and rising regardless of ADX # → Bullish counterpart: RSI in bullish territory and rising = valid rsi_momentum = (row["rsi"] > 55 and row["rsi"] > prev["rsi"] and row["rsi"] < RSI_OB) rsi_ok = rsi_pullback or rsi_trend or rsi_momentum rsi_mode = ("trend" if rsi_trend and not rsi_pullback else "momentum" if rsi_momentum and not rsi_pullback else "pullback") else: # PULLBACK mode: RSI spiked near overbought then falling rsi_pullback = ((recent["rsi"] > RSI_OB - 10).any() and row["rsi"] < prev["rsi"] and row["rsi"] > RSI_OS) # TREND mode: RSI above 50 and falling (strong bear, ADX > 30) rsi_trend = adx_strong and row["rsi"] > 45 and row["rsi"] < prev["rsi"] # CONTINUATION mode: RSI deeply oversold + falling (strong bear, ADX > 30) rsi_oversold = adx_strong and row["rsi"] < RSI_OS + 5 # MOMENTUM mode: RSI below 45 and falling regardless of ADX # → NZD/USD: RSI 38.3 falling, W1+D1 bearish, EMA crossed — valid signal # → No spike needed when RSI is already in bearish territory and moving down # → D1 ADX must be strong (>30) to compensate for low H1 ADX rsi_momentum = (row["rsi"] < 45 and row["rsi"] < prev["rsi"] and row["rsi"] > RSI_OS) # not yet oversold (room to run) rsi_ok = rsi_pullback or rsi_trend or rsi_oversold or rsi_momentum rsi_mode = ("oversold_cont" if rsi_oversold and not rsi_pullback else "trend" if rsi_trend and not rsi_pullback else "momentum" if rsi_momentum and not rsi_pullback else "pullback") triggered = cross_ok and rsi_ok return { "triggered": triggered, "signal": htf_bias if triggered else 0, "cross_ok": cross_ok, "rsi_ok": rsi_ok, "rsi": round(row["rsi"], 1), "zone": row["price_zone"], "rsi_mode": rsi_mode, "adx": round(row["adx"], 1), } # Compatibility wrappers for newer scorer imports def analyse_wtf_v2(df): return analyse_wtf(df) def analyse_htf_v2(df): return analyse_htf(df) def analyse_mtf_v2(df, htf_bias, instrument): return analyse_mtf(df, htf_bias, instrument) def analyse_ltf_v2(df, htf_bias): return analyse_ltf(df, htf_bias)