Spaces:
Running
Running
| """analysis/structure.py — swing detection, S&R, volume, momentum.""" | |
| import numpy as np | |
| import pandas as pd | |
| from config import SR_LOOKBACK, SR_ZONE_PIPS, STRUCT_LOOKBACK, ADX_MIN | |
| from data.oanda import pip, digs | |
| def detect_swings(df, n=3): | |
| df = df.copy() | |
| df["swing_high"] = df["high"][(df["high"] == df["high"].rolling(2*n+1, center=True).max())] | |
| df["swing_low"] = df["low"] [(df["low"] == df["low"].rolling(2*n+1, center=True).min())] | |
| return df | |
| def analyse_structure(df, bias): | |
| df2 = detect_swings(df, n=2) | |
| r = df2.iloc[-STRUCT_LOOKBACK:] | |
| sh = r["swing_high"].dropna() | |
| sl = r["swing_low"].dropna() | |
| if len(sh) < 2 or len(sl) < 2: | |
| df1 = detect_swings(df, n=1) | |
| r = df1.iloc[-STRUCT_LOOKBACK:] | |
| sh = r["swing_high"].dropna() | |
| sl = r["swing_low"].dropna() | |
| if len(sh) < 2 or len(sl) < 2: | |
| if "adx" in df.columns and df.iloc[-1]["adx"] > 30: | |
| return True, "Strong trend (ADX>30) — structure implied ✅" | |
| return False, f"Insufficient swings ({len(sh)} highs, {len(sl)} lows in {STRUCT_LOOKBACK} bars)" | |
| lh = sh.iloc[-2:].values | |
| ll = sl.iloc[-2:].values | |
| if bias == 1: | |
| hh = lh[-1] > lh[-2]; hl = ll[-1] > ll[-2] | |
| return hh and hl, f"HH:{'✅' if hh else '❌'} HL:{'✅' if hl else '❌'} ({lh[-2]:.5f}→{lh[-1]:.5f} | {ll[-2]:.5f}→{ll[-1]:.5f})" | |
| lhh = lh[-1] < lh[-2]; lll = ll[-1] < ll[-2] | |
| return lhh and lll, f"LH:{'✅' if lhh else '❌'} LL:{'✅' if lll else '❌'} ({lh[-2]:.5f}→{lh[-1]:.5f} | {ll[-2]:.5f}→{ll[-1]:.5f})" | |
| def find_sr(df, instrument, n=8): | |
| df = detect_swings(df) | |
| raw = pd.concat([df.iloc[-SR_LOOKBACK:]["swing_high"].dropna(), | |
| df.iloc[-SR_LOOKBACK:]["swing_low"].dropna()]).sort_values().values | |
| if not len(raw): return [] | |
| zone = SR_ZONE_PIPS * pip(instrument); clusters = []; cur = [raw[0]] | |
| for p in raw[1:]: | |
| if p - cur[-1] <= zone: cur.append(p) | |
| else: clusters.append(cur); cur = [p] | |
| clusters.append(cur); clusters.sort(key=len, reverse=True) | |
| return [np.mean(c) for c in clusters[:n]] | |
| def analyse_sr(df, bias, instrument): | |
| levels = find_sr(df, instrument) | |
| if not levels: return False, "No S&R levels" | |
| row = df.iloc[-1]; prev = df.iloc[-2] | |
| price = row["close"]; p = pip(instrument); d = digs(instrument) | |
| zone = SR_ZONE_PIPS * p | |
| near = [l for l in levels if abs(price - l) <= zone * 3] | |
| if not near: | |
| adx_strong = row["adx"] > 35 | |
| recent_high = df["high"].iloc[-20:].max() | |
| recent_low = df["low"].iloc[-20:].min() | |
| if bias == 1 and adx_strong and price >= recent_high * 0.999: | |
| return True, f"Breakout to new highs — SR not required (ADX:{row['adx']:.1f}) ✅" | |
| if bias == -1 and adx_strong and price <= recent_low * 1.001: | |
| return True, f"Breakout to new lows — SR not required (ADX:{row['adx']:.1f}) ✅" | |
| if bias == 1: | |
| below = sorted([l for l in levels if l < price], reverse=True) | |
| if below and (price - below[0]) <= zone * 5: | |
| return True, f"Prior resistance now support @ {below[0]:.{d}f} ✅" | |
| else: | |
| above = sorted([l for l in levels if l > price]) | |
| if above and (above[0] - price) <= zone * 5: | |
| return True, f"Prior support now resistance @ {above[0]:.{d}f} ✅" | |
| return False, f"Not near key level (nearest: {min(levels, key=lambda l: abs(price-l)):.{d}f})" | |
| nearest = min(near, key=lambda l: abs(price - l)) | |
| if bias == 1: | |
| at_support = price >= nearest | |
| approaching = price < nearest and price > prev["close"] and (nearest - price) <= zone * 2 | |
| ok = at_support or approaching | |
| label = f"Support {nearest:.{d}f}" if at_support else f"Approaching {nearest:.{d}f}" | |
| return ok, f"{label} {'✅' if ok else '❌'}" | |
| at_resistance = price <= nearest | |
| approaching = price > nearest and price < prev["close"] and (price - nearest) <= zone * 2 | |
| ok = at_resistance or approaching | |
| label = f"Resistance {nearest:.{d}f}" if at_resistance else f"Approaching {nearest:.{d}f}" | |
| return ok, f"{label} {'✅' if ok else '❌'}" | |
| def analyse_volume(df, bias): | |
| row = df.iloc[-1] | |
| lookback = 30 if row["adx"] > 35 else 15 | |
| r = df.iloc[-lookback:].copy() | |
| imp = r[r["close"] > r["open"]] if bias == 1 else r[r["close"] < r["open"]] | |
| pb = r[r["close"] < r["open"]] if bias == 1 else r[r["close"] > r["open"]] | |
| if imp.empty or pb.empty: return False, "Insufficient data" | |
| vol_ok = imp["tick_volume"].mean() > pb["tick_volume"].mean() | |
| adx_strong = row["adx"] > 35 | |
| ratio_thresh = 0.4 if adx_strong else 0.8 | |
| cur_ok = row["vol_ratio"] >= ratio_thresh | |
| trend_note = " [relaxed — strong trend]" if adx_strong else "" | |
| return (vol_ok and cur_ok, | |
| f"ImpVol>PbVol:{'✅' if vol_ok else '❌'} " | |
| f"Ratio:{row['vol_ratio']:.2f}{'✅' if cur_ok else '❌'}" | |
| f"(min {ratio_thresh}){trend_note}") | |
| def analyse_momentum(df, bias): | |
| row = df.iloc[-1]; prev = df.iloc[-2] | |
| adx_ok = row["adx"] > ADX_MIN | |
| adx_strong = row["adx"] > 35 | |
| if bias == 1: | |
| macd_side = row["macd"] > row["macd_sig"] | |
| macd_accel = row["macd_hist"] > prev["macd_hist"] | |
| macd_ok = macd_side if adx_strong else (macd_side and macd_accel) | |
| di_ok = row["plus_di"] > row["minus_di"] | |
| else: | |
| macd_side = row["macd"] < row["macd_sig"] | |
| macd_accel = row["macd_hist"] < prev["macd_hist"] | |
| macd_ok = macd_side if adx_strong else (macd_side and macd_accel) | |
| di_ok = row["minus_di"] > row["plus_di"] | |
| ok = macd_ok and adx_ok and di_ok | |
| trend_note = " [accel relaxed]" if adx_strong else "" | |
| macd_val = f"MACD:{row['macd']:.5f} Sig:{row['macd_sig']:.5f}" | |
| return (ok, | |
| f"MACD:{'✅' if macd_ok else '❌'}{trend_note} ({macd_val}) " | |
| f"ADX:{row['adx']:.1f}{'✅' if adx_ok else '❌'} " | |
| f"DI:{'✅' if di_ok else '❌'}") | |