Spaces:
Running
Running
File size: 11,771 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 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 | from __future__ import annotations
from typing import Dict, Any, List
def detect_regime_v2(df_htf, df_mtf) -> Dict[str, Any]:
row_d = df_htf.iloc[-1]
row_h4 = df_mtf.iloc[-1]
adx = float(row_h4.get("adx", 0))
ema_gap = abs(float(row_h4["ma_fast"] - row_h4["ma_slow"]))
atr = float(row_h4.get("atr", 0) or 0)
recent = df_mtf.iloc[-20:]
hh = recent["high"].max()
ll = recent["low"].min()
recent_range = float(hh - ll)
compression = atr > 0 and recent_range < atr * 2.8
stacked_bull = bool(row_d.get("ema_stack_bull", False)) or (row_d["ma_fast"] > row_d["ma_slow"])
stacked_bear = bool(row_d.get("ema_stack_bear", False)) or (row_d["ma_fast"] < row_d["ma_slow"])
if compression:
regime = "COMPRESSION"
elif adx >= 23 and ema_gap > 0 and (stacked_bull or stacked_bear):
regime = "TREND"
elif adx < 18:
regime = "RANGE"
else:
regime = "TRANSITION"
return {
"regime": regime,
"adx_h4": round(adx, 1),
"compression": compression,
"ema_gap": ema_gap,
"recent_range": recent_range,
}
def _structure_flags(df_mtf, bias: int) -> Dict[str, bool]:
highs = list(df_mtf["high"].tail(12))
lows = list(df_mtf["low"].tail(12))
hh = highs[-1] > max(highs[-5:-1]) if len(highs) >= 6 else False
ll = lows[-1] < min(lows[-5:-1]) if len(lows) >= 6 else False
hl = lows[-1] > min(lows[-5:-1]) if len(lows) >= 6 else False
lh = highs[-1] < max(highs[-5:-1]) if len(highs) >= 6 else False
return {"HH": hh, "HL": hl, "LL": ll, "LH": lh, "bull_ok": hh and hl, "bear_ok": ll and lh}
def detect_setups(df_wtf, df_htf, df_mtf, df_ltf, htf_bias: int, mtf: Dict[str, Any], ltf: Dict[str, Any],
location: Dict[str, Any], regime: Dict[str, Any]) -> List[Dict[str, Any]]:
row_h4 = df_mtf.iloc[-1]
row_h1 = df_ltf.iloc[-1]
struct = _structure_flags(df_mtf, htf_bias)
setups: List[Dict[str, Any]] = []
# Trend pullback
if htf_bias == 1 and regime["regime"] == "TREND":
if row_h4["price_zone"] in ("between_pb_up", "above_both") and ltf.get("triggered"):
setups.append({
"name": "Trend Pullback",
"code": "trend_pullback_buy",
"direction": 1,
"quality_boost": 10,
"why": "Higher timeframe trend is bullish and price is pulling back into value for continuation.",
"invalidation": "Invalid if H4 loses bullish structure or breaks down through support.",
})
if htf_bias == -1 and regime["regime"] == "TREND":
if row_h4["price_zone"] in ("between_pb_dn", "below_both") and ltf.get("triggered"):
setups.append({
"name": "Trend Pullback",
"code": "trend_pullback_sell",
"direction": -1,
"quality_boost": 10,
"why": "Higher timeframe trend is bearish and price is pulling back into value for continuation.",
"invalidation": "Invalid if H4 loses bearish structure or reclaims resistance.",
})
# Breakout retest
if htf_bias == 1 and location.get("resistance"):
rz = location["resistance"]
resistance_mid = (rz["low"] + rz["high"]) / 2
recent = df_ltf.iloc[-6:]
broke = recent["close"].max() > resistance_mid
holds = row_h1["close"] > resistance_mid and row_h1["low"] <= resistance_mid
if broke and holds:
setups.append({
"name": "Breakout Retest",
"code": "breakout_retest_buy",
"direction": 1,
"quality_boost": 8,
"why": "Price broke resistance and is retesting it as support.",
"invalidation": "Invalid if the retest fails and price closes back below the broken level.",
})
if htf_bias == -1 and location.get("support"):
sz = location["support"]
support_mid = (sz["low"] + sz["high"]) / 2
recent = df_ltf.iloc[-6:]
broke = recent["close"].min() < support_mid
holds = row_h1["close"] < support_mid and row_h1["high"] >= support_mid
if broke and holds:
setups.append({
"name": "Breakout Retest",
"code": "breakout_retest_sell",
"direction": -1,
"quality_boost": 8,
"why": "Price broke support and is retesting it as resistance.",
"invalidation": "Invalid if price reclaims the broken support.",
})
# Range reversal
if regime["regime"] in ("RANGE", "TRANSITION"):
if location.get("near_resistance") and row_h1["close"] < row_h1["open"]:
setups.append({
"name": "Range Reversal",
"code": "range_reversal_sell",
"direction": -1,
"quality_boost": 7,
"why": "Price is near range resistance and showing rejection.",
"invalidation": "Invalid if price breaks and holds above range resistance.",
})
if location.get("near_support") and row_h1["close"] > row_h1["open"]:
setups.append({
"name": "Range Reversal",
"code": "range_reversal_buy",
"direction": 1,
"quality_boost": 7,
"why": "Price is near range support and showing rejection.",
"invalidation": "Invalid if price breaks and holds below range support.",
})
# HTF zone reversal
if location.get("in_support") and row_h1.get("rsi", 50) > 45 and row_h1["close"] > row_h1["open"]:
setups.append({
"name": "HTF Zone Reversal",
"code": "htf_zone_reversal_buy",
"direction": 1,
"quality_boost": 6,
"why": "Price is reacting from a higher-timeframe support/demand zone.",
"invalidation": "Invalid if the zone fails decisively.",
})
if location.get("in_resistance") and row_h1.get("rsi", 50) < 55 and row_h1["close"] < row_h1["open"]:
setups.append({
"name": "HTF Zone Reversal",
"code": "htf_zone_reversal_sell",
"direction": -1,
"quality_boost": 6,
"why": "Price is reacting from a higher-timeframe resistance/supply zone.",
"invalidation": "Invalid if the zone is broken cleanly.",
})
# Liquidity sweep reversal
recent = df_ltf.iloc[-8:]
prev_high = recent["high"].iloc[:-1].max()
prev_low = recent["low"].iloc[:-1].min()
if row_h1["high"] > prev_high and row_h1["close"] < prev_high:
setups.append({
"name": "Liquidity Sweep Reversal",
"code": "liquidity_sweep_sell",
"direction": -1,
"quality_boost": 7,
"why": "Price swept recent highs and closed back below them.",
"invalidation": "Invalid if price reclaims the swept highs.",
})
if row_h1["low"] < prev_low and row_h1["close"] > prev_low:
setups.append({
"name": "Liquidity Sweep Reversal",
"code": "liquidity_sweep_buy",
"direction": 1,
"quality_boost": 7,
"why": "Price swept recent lows and closed back above them.",
"invalidation": "Invalid if price loses the swept lows again.",
})
return setups
def hard_blockers(htf: Dict[str, Any], mtf: Dict[str, Any], ltf: Dict[str, Any], regime: Dict[str, Any], location: Dict[str, Any], direction: int) -> List[str]:
blockers: List[str] = []
if regime["regime"] == "COMPRESSION":
blockers.append("Market is in compression/chop")
if direction == 1 and htf.get("bias") != 1:
blockers.append("D1 bullish bias is not aligned")
if direction == -1 and htf.get("bias") != -1:
blockers.append("D1 bearish bias is not aligned")
checks = mtf.get("checks", {})
ema_key = "1_ema20_50"
if not checks.get(ema_key, (False,))[0]:
blockers.append("4H EMA20/EMA50 alignment failed")
zone = mtf.get("zone", "mixed")
if zone == "mixed":
blockers.append("4H zone is mixed")
if direction == 1 and location.get("near_resistance"):
blockers.append("Buy is too close to resistance")
if direction == -1 and location.get("near_support"):
blockers.append("Sell is too close to support")
if direction == 1 and not ltf.get("triggered"):
blockers.append("1H entry trigger not confirmed")
if direction == -1 and not ltf.get("triggered"):
blockers.append("1H entry trigger not confirmed")
return blockers
def compute_grade(score: float, blockers: List[str], setup_quality: int, location: Dict[str, Any], regime: Dict[str, Any], room_pips: float | None) -> str:
if blockers:
return "NO TRADE"
if room_pips is not None and room_pips < 25:
return "A"
if score >= 85 and setup_quality >= 8 and regime["regime"] == "TREND" and location["status"] == "safe":
return "A++"
if score >= 72:
return "A+"
if score >= 58:
return "A"
return "NO TRADE"
def build_setup_explanation(selected: Dict[str, Any] | None, grade: str, blockers: List[str], location: Dict[str, Any], regime: Dict[str, Any]) -> str:
if blockers:
return (
f"No trade. Setup is blocked because: " + "; ".join(blockers[:3]) + ". "
f"Current regime is {regime['regime'].lower()} and location is {location['status'].replace('_', ' ')}."
)
if not selected:
return "No clear setup. Market does not cleanly match one of the defined playbooks yet."
return (
f"Setup: {selected['name']}. {selected['why']} "
f"Current regime is {regime['regime'].lower()}. "
f"Location status: {location['status'].replace('_', ' ')}. "
f"Grade: {grade}. {selected['invalidation']}"
)
def build_decision_package(df_wtf, df_htf, df_mtf, df_ltf, htf, mtf, ltf, location, room, zone_map) -> Dict[str, Any]:
regime = detect_regime_v2(df_htf, df_mtf)
setups = detect_setups(df_wtf, df_htf, df_mtf, df_ltf, htf.get("bias", 0), mtf, ltf, location, regime)
selected = setups[0] if setups else None
direction = selected["direction"] if selected else 0
weighted_score = 0.0
if htf.get("bias") != 0:
weighted_score += 20
weighted_score += min(30, (mtf.get("score", 0) / max(mtf.get("max", 1), 1)) * 30)
weighted_score += 20 if location.get("status") == "safe" else 5
weighted_score += selected.get("quality_boost", 0) if selected else 0
weighted_score += 10 if ltf.get("triggered") else 0
blockers = hard_blockers(htf, mtf, ltf, regime, location, direction) if direction != 0 else ["No clear setup detected"]
grade = compute_grade(weighted_score, blockers, selected.get("quality_boost", 0) if selected else 0,
location, regime, room.get("room_pips"))
decision = "NO TRADE"
if grade == "A++":
decision = "AUTO TRADE"
elif grade == "A+":
decision = "REVIEW"
elif grade == "A":
decision = "WATCH"
return {
"regime_v2": regime,
"setups_detected": setups,
"selected_setup": selected,
"direction": direction,
"blockers": blockers if blockers != ["No clear setup detected"] else blockers,
"weighted_score": round(weighted_score, 1),
"grade_v2": grade,
"decision": decision,
"explanation": build_setup_explanation(selected, grade, blockers if direction != 0 else ["No clear setup detected"], location, regime),
"room_to_target": room,
"location": location,
"zone_map": zone_map,
}
|