Spaces:
Running
Running
| """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"]<row["ma_slow"] | |
| chk("1_ema20_50", ema_dir_ok, f"EMA20({'>' 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_OB if htf_bias==1 else 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) | |