Spaces:
Running
Running
| 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, | |
| } | |