Spaces:
Running
Running
File size: 7,162 Bytes
aa79b15 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """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}
|