Spaces:
Running
Running
| """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"]<prev["open"] and curr["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"]<curr["open"] and curr["open"]>=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} | |