Spaces:
Running
Running
| from __future__ import annotations | |
| from dataclasses import dataclass, asdict | |
| from typing import List, Dict, Any | |
| import numpy as np | |
| import pandas as pd | |
| from data.oanda import pip, digs | |
| class Zone: | |
| zone_type: str # support/resistance/demand/supply | |
| low: float | |
| high: float | |
| source: str # swing/base/impulse | |
| timeframe: str | |
| strength: float | |
| touches: int = 1 | |
| def to_dict(self) -> Dict[str, Any]: | |
| d = asdict(self) | |
| d["mid"] = round((self.low + self.high) / 2, 6) | |
| return d | |
| def _local_extrema(series: pd.Series, window: int = 3, mode: str = "max") -> List[int]: | |
| idxs: List[int] = [] | |
| if len(series) < (2 * window + 1): | |
| return idxs | |
| vals = series.reset_index(drop=True) | |
| for i in range(window, len(vals) - window): | |
| seg = vals.iloc[i - window:i + window + 1] | |
| if mode == "max" and vals.iloc[i] >= seg.max(): | |
| idxs.append(i) | |
| elif mode == "min" and vals.iloc[i] <= seg.min(): | |
| idxs.append(i) | |
| return idxs | |
| def _merge_zones(zones: List[Zone], instrument: str) -> List[Zone]: | |
| if not zones: | |
| return [] | |
| tol = 12 * pip(instrument) | |
| zones_sorted = sorted(zones, key=lambda z: (z.low + z.high) / 2) | |
| merged: List[Zone] = [] | |
| for z in zones_sorted: | |
| if not merged: | |
| merged.append(z) | |
| continue | |
| prev = merged[-1] | |
| prev_mid = (prev.low + prev.high) / 2 | |
| cur_mid = (z.low + z.high) / 2 | |
| if abs(cur_mid - prev_mid) <= tol and prev.zone_type == z.zone_type: | |
| total_strength = prev.strength + z.strength | |
| new_low = min(prev.low, z.low) | |
| new_high = max(prev.high, z.high) | |
| merged[-1] = Zone( | |
| zone_type=prev.zone_type, | |
| low=new_low, | |
| high=new_high, | |
| source=f"{prev.source}+{z.source}", | |
| timeframe=prev.timeframe if prev.strength >= z.strength else z.timeframe, | |
| strength=round(total_strength, 2), | |
| touches=prev.touches + z.touches, | |
| ) | |
| else: | |
| merged.append(z) | |
| return merged | |
| def build_swing_zones(df: pd.DataFrame, instrument: str, timeframe: str, window: int = 3) -> List[Zone]: | |
| if df.empty: | |
| return [] | |
| zsize = max(8 * pip(instrument), float(df["atr"].iloc[-1]) * 0.25 if "atr" in df.columns else 8 * pip(instrument)) | |
| highs = _local_extrema(df["high"], window, mode="max") | |
| lows = _local_extrema(df["low"], window, mode="min") | |
| zones: List[Zone] = [] | |
| for i in highs[-8:]: | |
| price = float(df["high"].iloc[i]) | |
| zones.append(Zone("resistance", price - zsize / 2, price + zsize / 2, "swing", timeframe, 2.0)) | |
| for i in lows[-8:]: | |
| price = float(df["low"].iloc[i]) | |
| zones.append(Zone("support", price - zsize / 2, price + zsize / 2, "swing", timeframe, 2.0)) | |
| return zones | |
| def build_base_zones(df: pd.DataFrame, instrument: str, timeframe: str, window: int = 5) -> List[Zone]: | |
| if len(df) < window + 2: | |
| return [] | |
| zones: List[Zone] = [] | |
| p = pip(instrument) | |
| for i in range(window, len(df) - 2): | |
| seg = df.iloc[i - window:i] | |
| seg_range = float(seg["high"].max() - seg["low"].min()) | |
| atr = float(seg["atr"].mean()) if "atr" in seg.columns else max(15 * p, seg_range) | |
| if seg_range > max(0.6 * atr, 12 * p): | |
| continue | |
| next_bar = df.iloc[i] | |
| breakout_up = next_bar["close"] > seg["high"].max() + 0.2 * atr | |
| breakout_down = next_bar["close"] < seg["low"].min() - 0.2 * atr | |
| if breakout_up: | |
| zones.append(Zone("demand", float(seg["low"].min()), float(seg["high"].max()), "base", timeframe, 2.5)) | |
| elif breakout_down: | |
| zones.append(Zone("supply", float(seg["low"].min()), float(seg["high"].max()), "base", timeframe, 2.5)) | |
| return zones[-8:] | |
| def build_impulse_zones(df: pd.DataFrame, instrument: str, timeframe: str) -> List[Zone]: | |
| if len(df) < 10: | |
| return [] | |
| zones: List[Zone] = [] | |
| p = pip(instrument) | |
| for i in range(2, len(df)): | |
| bar = df.iloc[i] | |
| prev = df.iloc[i - 1] | |
| atr = float(bar.get("atr", 0) or prev.get("atr", 0) or 20 * p) | |
| body = abs(float(bar["close"] - bar["open"])) | |
| if body < 1.2 * atr: | |
| continue | |
| if bar["close"] > bar["open"]: | |
| low = min(float(prev["low"]), float(bar["open"])) | |
| high = max(float(prev["high"]), float(bar["open"])) | |
| zones.append(Zone("demand", low, high, "impulse", timeframe, 3.0)) | |
| else: | |
| low = min(float(prev["low"]), float(bar["open"])) | |
| high = max(float(prev["high"]), float(bar["high"])) | |
| zones.append(Zone("supply", low, high, "impulse", timeframe, 3.0)) | |
| return zones[-8:] | |
| def build_zone_map(df_wtf: pd.DataFrame, df_htf: pd.DataFrame, df_mtf: pd.DataFrame, instrument: str) -> Dict[str, Any]: | |
| zones: List[Zone] = [] | |
| zones += build_swing_zones(df_wtf, instrument, "W1", window=2) | |
| zones += build_swing_zones(df_htf, instrument, "D1", window=3) | |
| zones += build_swing_zones(df_mtf, instrument, "H4", window=3) | |
| zones += build_base_zones(df_htf, instrument, "D1") | |
| zones += build_base_zones(df_mtf, instrument, "H4") | |
| zones += build_impulse_zones(df_htf, instrument, "D1") | |
| zones += build_impulse_zones(df_mtf, instrument, "H4") | |
| merged = _merge_zones(zones, instrument) | |
| merged.sort(key=lambda z: z.strength, reverse=True) | |
| return { | |
| "zones": [z.to_dict() for z in merged], | |
| "supports": [z.to_dict() for z in merged if z.zone_type in ("support", "demand")], | |
| "resistances": [z.to_dict() for z in merged if z.zone_type in ("resistance", "supply")], | |
| } | |
| def nearest_zone(price: float, zones: List[Dict[str, Any]], prefer_above: bool | None = None) -> Dict[str, Any] | None: | |
| if not zones: | |
| return None | |
| best = None | |
| best_dist = float("inf") | |
| for z in zones: | |
| mid = (z["low"] + z["high"]) / 2 | |
| if prefer_above is True and mid < price: | |
| continue | |
| if prefer_above is False and mid > price: | |
| continue | |
| dist = abs(mid - price) | |
| if dist < best_dist: | |
| best_dist = dist | |
| best = z | |
| return best | |
| def evaluate_location(price: float, direction: int, zone_map: Dict[str, Any], instrument: str) -> Dict[str, Any]: | |
| p = pip(instrument) | |
| supports = zone_map.get("supports", []) | |
| resistances = zone_map.get("resistances", []) | |
| support = nearest_zone(price, supports, prefer_above=False) or nearest_zone(price, supports, None) | |
| resistance = nearest_zone(price, resistances, prefer_above=True) or nearest_zone(price, resistances, None) | |
| def dist_pips(z: Dict[str, Any] | None) -> float | None: | |
| if not z: | |
| return None | |
| mid = (z["low"] + z["high"]) / 2 | |
| return round(abs(mid - price) / p, 1) | |
| support_dist = dist_pips(support) | |
| resistance_dist = dist_pips(resistance) | |
| in_support = bool(support and support["low"] <= price <= support["high"]) | |
| in_resistance = bool(resistance and resistance["low"] <= price <= resistance["high"]) | |
| near_support = bool(support_dist is not None and support_dist <= 20) | |
| near_resistance = bool(resistance_dist is not None and resistance_dist <= 20) | |
| status = "safe" | |
| if direction == 1 and (near_resistance or in_resistance): | |
| status = "near_resistance" | |
| elif direction == -1 and (near_support or in_support): | |
| status = "near_support" | |
| elif in_support: | |
| status = "in_support" | |
| elif in_resistance: | |
| status = "in_resistance" | |
| return { | |
| "status": status, | |
| "support": support, | |
| "resistance": resistance, | |
| "distance_to_support_pips": support_dist, | |
| "distance_to_resistance_pips": resistance_dist, | |
| "in_support": in_support, | |
| "in_resistance": in_resistance, | |
| "near_support": near_support, | |
| "near_resistance": near_resistance, | |
| } | |
| def zone_room_to_target(price: float, direction: int, zone_map: Dict[str, Any], instrument: str) -> Dict[str, Any]: | |
| p = pip(instrument) | |
| if direction == 1: | |
| target_zone = nearest_zone(price, zone_map.get("resistances", []), prefer_above=True) | |
| else: | |
| target_zone = nearest_zone(price, zone_map.get("supports", []), prefer_above=False) | |
| if not target_zone: | |
| return {"room_pips": None, "target_zone": None} | |
| mid = (target_zone["low"] + target_zone["high"]) / 2 | |
| room = abs(mid - price) / p | |
| return {"room_pips": round(room, 1), "target_zone": target_zone} | |