""" ========================================= CV6 AI Trading OS — Market Structure Engine ========================================= Detects: • Swing Highs / Swing Lows (pivot points) • HH / HL (Higher High / Higher Low) → Uptrend • LH / LL (Lower High / Lower Low) → Downtrend • Trend UPTREND | DOWNTREND | SIDEWAYS | REVERSAL • Key Support & Resistance levels • 200 EMA trend filter (price above/below) • Structure break (BoS — Break of Structure) • Trend strength score 0–100 Real data from yfinance. No random / fake values. ========================================= """ from __future__ import annotations import time from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple from loguru import logger # ── Cache ───────────────────────────────────────────────────────────────────── _cache: Dict[str, Any] = {} _cache_ts: Dict[str, float] = {} _CACHE_TTL = 300 # 5 minutes def _cached(key: str, fn): now = time.time() if key in _cache and (now - _cache_ts.get(key, 0)) < _CACHE_TTL: return _cache[key] result = fn() if result is not None: _cache[key] = result _cache_ts[key] = now return _cache.get(key) # ── Dataclasses ──────────────────────────────────────────────────────────────── @dataclass class SwingPoint: index: int price: float kind: str # "HIGH" | "LOW" date: str # ISO date string @dataclass class MarketStructure: symbol: str trend: str # UPTREND | DOWNTREND | SIDEWAYS | REVERSAL_UP | REVERSAL_DOWN trend_score: int # 0-100 (100 = strong uptrend, 0 = strong downtrend) above_200ema: bool # True = price above 200 EMA → only BUY bias ema_200: float ema_50: float ema_20: float current_price: float swing_highs: List[float] = field(default_factory=list) # last 3 swing_lows: List[float] = field(default_factory=list) # last 3 key_resistance: List[float] = field(default_factory=list) key_support: List[float] = field(default_factory=list) structure_pattern: str = "" # HH-HL | LH-LL | HH-LL | LH-HL bos: bool = False # Break of Structure detected bos_direction: str = "" # UP | DOWN nearest_support: float = 0.0 nearest_resistance:float = 0.0 distance_to_support_pct: float = 0.0 distance_to_resistance_pct: float = 0.0 trade_bias: str = "NEUTRAL" # BUY | SELL | NEUTRAL | WAIT interpretation: str = "" # ── EMA helper ───────────────────────────────────────────────────────────────── def _ema(prices: List[float], period: int) -> Optional[float]: if len(prices) < period: return None k = 2.0 / (period + 1) ema = sum(prices[:period]) / period for p in prices[period:]: ema = p * k + ema * (1 - k) return round(ema, 2) # ── Swing point detection ────────────────────────────────────────────────────── def _find_swings(highs: List[float], lows: List[float], lookback: int = 5) -> Tuple[List[int], List[int]]: """ Detect swing highs and lows using a simple pivot method. A pivot high at i: high[i] is the highest in [i-lookback:i+lookback+1] A pivot low at i: low[i] is the lowest in [i-lookback:i+lookback+1] """ n = len(highs) pivot_highs: List[int] = [] pivot_lows: List[int] = [] for i in range(lookback, n - lookback): window_h = highs[max(0, i-lookback): i+lookback+1] window_l = lows [max(0, i-lookback): i+lookback+1] if highs[i] == max(window_h): pivot_highs.append(i) if lows[i] == min(window_l): pivot_lows.append(i) return pivot_highs, pivot_lows # ── Core analysis ────────────────────────────────────────────────────────────── def analyse_structure(symbol: str, period: str = "1y") -> Optional[MarketStructure]: cache_key = f"struct_{symbol}_{period}" return _cached(cache_key, lambda: _build_structure(symbol, period)) def _build_structure(symbol: str, period: str) -> Optional[MarketStructure]: try: import yfinance as yf ticker = symbol.upper() if not ticker.endswith(".NS") and not ticker.endswith(".BO"): ticker = ticker + ".NS" df = yf.download(ticker, period=period, interval="1d", auto_adjust=True, progress=False) if df is None or len(df) < 50: logger.warning(f"[Structure] {symbol}: insufficient data ({len(df) if df is not None else 0} bars)") return None closes = [float(x) for x in df["Close"].dropna().tolist()] highs = [float(x) for x in df["High"].dropna().tolist()] lows = [float(x) for x in df["Low"].dropna().tolist()] dates = [str(d)[:10] for d in df.index.tolist()] n = len(closes) current_price = closes[-1] # ── EMAs ────────────────────────────────────────────────────────────── ema_200 = _ema(closes, 200) or 0.0 ema_50 = _ema(closes, 50) or 0.0 ema_20 = _ema(closes, 20) or 0.0 above_200ema = current_price > ema_200 if ema_200 > 0 else True # ── Swing detection ──────────────────────────────────────────────────── lookback = 5 ph_idx, pl_idx = _find_swings(highs, lows, lookback=lookback) # Get last 4 swing highs and lows recent_highs = [highs[i] for i in ph_idx[-4:]] recent_lows = [lows[i] for i in pl_idx[-4:]] # ── HH/HL/LH/LL pattern ─────────────────────────────────────────────── def _classify_highs(sh: List[float]) -> str: if len(sh) < 2: return "" return "HH" if sh[-1] > sh[-2] else "LH" def _classify_lows(sl: List[float]) -> str: if len(sl) < 2: return "" return "HL" if sl[-1] > sl[-2] else "LL" h_label = _classify_highs(recent_highs) l_label = _classify_lows(recent_lows) pattern = f"{h_label}-{l_label}" if h_label and l_label else "–" # ── Trend determination ─────────────────────────────────────────────── if pattern == "HH-HL": trend = "UPTREND" elif pattern == "LH-LL": trend = "DOWNTREND" elif pattern == "HH-LL": trend = "REVERSAL_DOWN" # momentum weakening elif pattern == "LH-HL": trend = "REVERSAL_UP" # bottoming out else: trend = "SIDEWAYS" # ── Trend score 0-100 ───────────────────────────────────────────────── score = 50 # neutral base if trend == "UPTREND": score = 75 elif trend == "DOWNTREND": score = 25 elif trend == "REVERSAL_UP": score = 60 elif trend == "REVERSAL_DOWN": score = 40 else: score = 50 # EMA alignment bonus/penalty if above_200ema: score = min(100, score + 10) else: score = max(0, score - 10) if ema_50 > 0 and ema_200 > 0: if ema_20 > ema_50 > ema_200: score = min(100, score + 10) elif ema_20 < ema_50 < ema_200: score = max(0, score - 10) # ── Key S/R levels ──────────────────────────────────────────────────── key_resistance = sorted( set([round(h, 1) for h in recent_highs if h > current_price]), reverse=False )[:3] key_support = sorted( set([round(l, 1) for l in recent_lows if l < current_price]), reverse=True )[:3] nearest_resistance = key_resistance[0] if key_resistance else round(current_price * 1.05, 1) nearest_support = key_support[0] if key_support else round(current_price * 0.95, 1) dist_res = round((nearest_resistance - current_price) / current_price * 100, 2) dist_sup = round((current_price - nearest_support) / current_price * 100, 2) # ── Break of Structure (BoS) ────────────────────────────────────────── bos = False bos_dir = "" if len(recent_highs) >= 2 and current_price > recent_highs[-2]: bos = True bos_dir = "UP" # broke above prior swing high elif len(recent_lows) >= 2 and current_price < recent_lows[-2]: bos = True bos_dir = "DOWN" # broke below prior swing low # ── Trade bias ──────────────────────────────────────────────────────── if trend in ("UPTREND",) and above_200ema: bias = "BUY" elif trend in ("DOWNTREND",) and not above_200ema: bias = "SELL" elif trend in ("REVERSAL_UP", "SIDEWAYS") and above_200ema: bias = "BUY" elif trend in ("REVERSAL_DOWN", "SIDEWAYS") and not above_200ema: bias = "SELL" elif dist_sup < 1.0: bias = "WAIT" # too close to support — wait for confirmation else: bias = "NEUTRAL" # ── Interpretation ──────────────────────────────────────────────────── parts = [] parts.append(f"{trend} ({pattern})") parts.append(f"Price {'above' if above_200ema else 'BELOW'} 200 EMA ₹{ema_200:.0f}") if bos: parts.append(f"⚠ Break of Structure {bos_dir}") parts.append(f"Support ₹{nearest_support:.0f} ({dist_sup:.1f}% away) | " f"Resistance ₹{nearest_resistance:.0f} ({dist_res:.1f}% away)") parts.append(f"Bias: {bias}") interpretation = " | ".join(parts) return MarketStructure( symbol = symbol.upper(), trend = trend, trend_score = score, above_200ema = above_200ema, ema_200 = ema_200, ema_50 = ema_50, ema_20 = ema_20, current_price = current_price, swing_highs = recent_highs, swing_lows = recent_lows, key_resistance = key_resistance, key_support = key_support, structure_pattern = pattern, bos = bos, bos_direction = bos_dir, nearest_support = nearest_support, nearest_resistance = nearest_resistance, distance_to_support_pct = dist_sup, distance_to_resistance_pct = dist_res, trade_bias = bias, interpretation = interpretation, ) except Exception as e: logger.error(f"[Structure] {symbol}: {e}") return None def fmt_structure(ms: MarketStructure) -> str: """One-line summary for AI prompt.""" ema_flag = "✅ ABOVE 200EMA" if ms.above_200ema else "❌ BELOW 200EMA" bos_str = f" | BoS {ms.bos_direction}" if ms.bos else "" return ( f"{ms.symbol}: {ms.trend} ({ms.structure_pattern}) | " f"Score {ms.trend_score}/100 | {ema_flag} | " f"Bias: {ms.trade_bias} | " f"S₹{ms.nearest_support:.0f} R₹{ms.nearest_resistance:.0f}" f"{bos_str}" ) # ── Market-wide structure (Nifty / BankNifty) ───────────────────────────────── def get_index_structure(index: str = "NIFTY") -> Optional[MarketStructure]: """ Get market structure for Nifty / BankNifty index. Used as macro filter — only trade in direction of index trend. """ symbol_map = { "NIFTY": "^NSEI", "BANKNIFTY": "^NSEBANK", "SENSEX": "^BSESN", } ticker = symbol_map.get(index.upper(), "^NSEI") cache_key = f"idx_struct_{index}" return _cached(cache_key, lambda: _build_structure(ticker.replace("^", "IDX_"), "1y")) def get_nifty_structure() -> Optional[MarketStructure]: """Convenience: Nifty 50 market structure.""" try: import yfinance as yf cache_key = "nifty_structure" def _fetch(): df = yf.download("^NSEI", period="1y", interval="1d", auto_adjust=True, progress=False) if df is None or len(df) < 50: return None closes = [float(x) for x in df["Close"].dropna().tolist()] highs = [float(x) for x in df["High"].dropna().tolist()] lows = [float(x) for x in df["Low"].dropna().tolist()] current = closes[-1] ema_200 = _ema(closes, 200) or 0.0 ema_50 = _ema(closes, 50) or 0.0 ema_20 = _ema(closes, 20) or 0.0 above = current > ema_200 if ema_200 > 0 else True ph_idx, pl_idx = _find_swings(highs, lows, lookback=5) sh = [highs[i] for i in ph_idx[-4:]] sl = [lows[i] for i in pl_idx[-4:]] h_lbl = "HH" if len(sh) >= 2 and sh[-1] > sh[-2] else ("LH" if len(sh) >= 2 else "") l_lbl = "HL" if len(sl) >= 2 and sl[-1] > sl[-2] else ("LL" if len(sl) >= 2 else "") pattern = f"{h_lbl}-{l_lbl}" if h_lbl and l_lbl else "–" if pattern == "HH-HL": trend = "UPTREND" elif pattern == "LH-LL": trend = "DOWNTREND" elif pattern == "HH-LL": trend = "REVERSAL_DOWN" elif pattern == "LH-HL": trend = "REVERSAL_UP" else: trend = "SIDEWAYS" bias = "BUY" if (trend in ("UPTREND", "REVERSAL_UP") and above) else \ "SELL" if (trend in ("DOWNTREND", "REVERSAL_DOWN") and not above) else "NEUTRAL" score = {"UPTREND": 80, "DOWNTREND": 20, "REVERSAL_UP": 60, "REVERSAL_DOWN": 40, "SIDEWAYS": 50}.get(trend, 50) if above: score = min(100, score + 10) else: score = max(0, score - 10) key_res = sorted([round(h, 0) for h in sh if h > current])[:2] key_sup = sorted([round(l, 0) for l in sl if l < current], reverse=True)[:2] nr = key_res[0] if key_res else round(current * 1.03, 0) ns = key_sup[0] if key_sup else round(current * 0.97, 0) return MarketStructure( symbol="NIFTY", trend=trend, trend_score=score, above_200ema=above, ema_200=ema_200, ema_50=ema_50, ema_20=ema_20, current_price=current, swing_highs=sh, swing_lows=sl, key_resistance=key_res, key_support=key_sup, structure_pattern=pattern, bos=False, bos_direction="", nearest_support=ns, nearest_resistance=nr, distance_to_support_pct=round((current - ns) / current * 100, 2), distance_to_resistance_pct=round((nr - current) / current * 100, 2), trade_bias=bias, interpretation=f"Nifty {trend} ({pattern}) | {'Above' if above else 'Below'} 200EMA ₹{ema_200:.0f} | Bias: {bias}", ) return _cached(cache_key, _fetch) except Exception as e: logger.error(f"[Structure] Nifty fetch: {e}") return None