"""analysis/patterns.py — chart pattern detection (H&S, double top/bottom, flag, triangle, engulfing).""" import numpy as np from analysis.structure import detect_swings def detect_head_and_shoulders(df, bias): df = detect_swings(df) recent = df.iloc[-60:] if bias == -1: peaks = recent["swing_high"].dropna() if len(peaks) < 3: return False, "H&S: insufficient peaks", None p = peaks.values[-3:]; left, head, right = p[0], p[1], p[2] if not (head > left and head > right and abs(left-right)/head < 0.30): return False, "H&S: peak shape not valid", None troughs = recent["swing_low"].dropna() if len(troughs) < 2: return False, "H&S: no neckline troughs", None neckline = np.mean(troughs.values[-2:]) broken = df.iloc[-1]["close"] < neckline note = f"H&S: L:{left:.5f} H:{head:.5f} R:{right:.5f} Neckline:{neckline:.5f} Break:{'✅' if broken else '⏳ forming'}" return broken, note, neckline else: troughs = recent["swing_low"].dropna() if len(troughs) < 3: return False, "Inv H&S: insufficient troughs", None t = troughs.values[-3:]; left, head, right = t[0], t[1], t[2] if not (head < left and head < right and abs(left-right)/abs(head) < 0.30): return False, "Inv H&S: trough shape not valid", None peaks = recent["swing_high"].dropna() if len(peaks) < 2: return False, "Inv H&S: no neckline peaks", None neckline = np.mean(peaks.values[-2:]) broken = df.iloc[-1]["close"] > neckline note = f"Inv H&S: L:{left:.5f} H:{head:.5f} R:{right:.5f} Neckline:{neckline:.5f} Break:{'✅' if broken else '⏳ forming'}" return broken, note, neckline def detect_double_top_bottom(df, bias): df = detect_swings(df); recent = df.iloc[-50:] if bias == -1: peaks = recent["swing_high"].dropna() if len(peaks) < 2: return False, "Dbl Top: insufficient peaks" p1, p2 = peaks.values[-2], peaks.values[-1] similar = abs(p1-p2) / max(p1,p2) < 0.003 if not similar: return False, f"Dbl Top: peaks too different ({p1:.5f} vs {p2:.5f})" troughs = recent["swing_low"].dropna() if troughs.empty: return False, "Dbl Top: no trough between peaks" valley = troughs.values[-1] confirmed = df.iloc[-1]["close"] < valley note = f"Dbl Top: {p1:.5f} & {p2:.5f} Valley:{valley:.5f} Break:{'✅' if confirmed else '⏳'}" return confirmed and (min(p1,p2)-valley)/min(p1,p2) > 0.01, note else: troughs = recent["swing_low"].dropna() if len(troughs) < 2: return False, "Dbl Bot: insufficient troughs" t1, t2 = troughs.values[-2], troughs.values[-1] similar = abs(t1-t2) / max(abs(t1),abs(t2)) < 0.003 if not similar: return False, f"Dbl Bot: troughs too different ({t1:.5f} vs {t2:.5f})" peaks = recent["swing_high"].dropna() if peaks.empty: return False, "Dbl Bot: no peak between troughs" peak = peaks.values[-1] confirmed = df.iloc[-1]["close"] > peak note = f"Dbl Bot: {t1:.5f} & {t2:.5f} Peak:{peak:.5f} Break:{'✅' if confirmed else '⏳'}" return confirmed and (peak-max(t1,t2))/peak > 0.01, note def detect_flag(df, bias): pole = df.iloc[-20:-8]; flag = df.iloc[-8:] if pole.empty or flag.empty: return False, "Flag: insufficient data" pole_move = pole["close"].iloc[-1] - pole["close"].iloc[0] pole_pct = abs(pole_move) / pole["close"].iloc[0] * 100 flag_range = flag["high"].max() - flag["low"].min() flag_retrace = flag_range / abs(pole_move) if abs(pole_move) > 0 else 1 if pole_pct < 0.5: return False, f"Flag: pole too small ({pole_pct:.2f}%)" if flag_retrace > 0.65: return False, f"Flag: too deep ({flag_retrace:.0%} retrace)" price_now = df.iloc[-1]["close"] ema_ok = df.iloc[-1]["ma_fast"] > df.iloc[-2]["ma_fast"] if bias==1 else df.iloc[-1]["ma_fast"] < df.iloc[-2]["ma_fast"] if bias == 1: breakout = price_now > flag["high"].max() ok = pole_move > 0 and breakout and ema_ok return ok, f"Bull Flag: pole +{pole_pct:.2f}% retrace:{flag_retrace:.0%} Breakout:{'✅' if breakout else '⏳'}" breakout = price_now < flag["low"].min() ok = pole_move < 0 and breakout and ema_ok return ok, f"Bear Flag: pole -{pole_pct:.2f}% retrace:{flag_retrace:.0%} Breakout:{'✅' if breakout else '⏳'}" def detect_triangle(df, bias): import numpy as np recent = df.iloc[-30:].copy() if len(recent) < 15: return False, "Triangle: insufficient data" x = np.arange(len(recent)) high_slope = np.polyfit(x, recent["high"].values, 1)[0] low_slope = np.polyfit(x, recent["low"].values, 1)[0] price_now = df.iloc[-1]["close"] if bias == 1: ok = low_slope > 0.00001 and abs(high_slope) < abs(low_slope)*0.3 and price_now > recent["high"].max()*0.999 return ok, f"Asc Triangle: low:{low_slope:.6f} high:{high_slope:.6f} Breakout:{'✅' if ok else '⏳'}" ok = high_slope < -0.00001 and abs(low_slope) < abs(high_slope)*0.3 and price_now < recent["low"].min()*1.001 return ok, f"Desc Triangle: high:{high_slope:.6f} low:{low_slope:.6f} Breakout:{'✅' if ok else '⏳'}" def detect_engulfing(df, bias): if len(df) < 3: return False, "Engulfing: insufficient data" curr = df.iloc[-1]; prev = df.iloc[-2] curr_body = abs(curr["close"]-curr["open"]); prev_body = abs(prev["close"]-prev["open"]) if prev_body == 0: return False, "Engulfing: zero body candle" ratio = curr_body / prev_body if bias == 1: ok = prev["close"]curr["open"] and curr["open"]<=prev["close"] and curr["close"]>=prev["open"] and ratio>=1.5 return ok, f"Bullish Engulf: ratio:{ratio:.1f}x {'✅' if ok else '❌'}" ok = prev["close"]>prev["open"] and curr["close"]=prev["close"] and curr["close"]<=prev["open"] and ratio>=1.5 return ok, f"Bearish Engulf: ratio:{ratio:.1f}x {'✅' if ok else '❌'}" def run_all_patterns(df_mtf, df_ltf, bias): results = {} ok, note, neckline = detect_head_and_shoulders(df_mtf, bias) results["hs"] = {"ok": ok, "note": note, "neckline": neckline, "tf": "4H"} ok, note = detect_double_top_bottom(df_mtf, bias) results["double"] = {"ok": ok, "note": note, "tf": "4H"} ok, note = detect_triangle(df_mtf, bias) results["triangle"] = {"ok": ok, "note": note, "tf": "4H"} ok, note = detect_flag(df_ltf, bias) results["flag"] = {"ok": ok, "note": note, "tf": "1H"} ok, note = detect_engulfing(df_ltf, bias) results["engulfing"]= {"ok": ok, "note": note, "tf": "1H"} pattern_score = sum(1 for v in results.values() if v["ok"]) best = None if results["hs"]["ok"] and results["hs"].get("neckline"): best = {"type": "H&S", "neckline": results["hs"]["neckline"]} elif results["double"]["ok"]: best = {"type": "Double Top/Bottom"} return {"patterns": results, "pattern_score": pattern_score, "any_pattern": pattern_score > 0, "best": best}