""" ========================================= CV6 Autonomous — Smart AI Cost Control 5-Stage Market Scanner Funnel Stage 1 : Full Universe (500+ symbols) → quick pre-filter Stage 2 : Technical Top 200 → EMA/RSI/MACD/Volume Stage 3 : Quality Top 20 → Supertrend/VWAP/Breakout Stage 4 : Risk Filter Top 5 → Circuit/Volatility/Sector Stage 5 : AI Confirm Top 1-3 → AI consensus only for finalists AI is called ONLY for Stage-5 candidates. AI Cache prevents re-calling within TTL. ========================================= """ import math import random import time from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple from loguru import logger # ── NSE Symbol Universe (~500 stocks) ──────────────────────────────────────── # Nifty 50 + Next50 + Midcap150 + Smallcap100 + Popular F&O NSE_UNIVERSE: List[str] = [ # ── Nifty 50 ────────────────────────────────────────────────────────────── "RELIANCE", "TCS", "INFY", "HDFCBANK", "ICICIBANK", "SBIN", "WIPRO", "AXISBANK", "HCLTECH", "KOTAKBANK", "LT", "ITC", "BAJFINANCE", "MARUTI", "SUNPHARMA", "TITAN", "ADANIPORTS", "NTPC", "TATAMOTORS", "TATASTEEL", "HINDUNILVR", "BHARTIARTL", "ONGC", "POWERGRID", "TECHM", "JSWSTEEL", "COALINDIA", "DRREDDY", "CIPLA", "M&M", "HEROMOTOCO", "EICHERMOT", "HINDALCO", "ULTRACEMCO", "GRASIM", "INDUSINDBK", "BPCL", "DIVISLAB", "NESTLEIND", "BRITANNIA", "APOLLOHOSP", "TATACONSUM", "SBILIFE", "HDFCLIFE", "BAJAJFINSV", "SHRIRAMFIN", "ADANIENT", "WIPRO", "BAJAJ-AUTO", # ── Nifty Next 50 ───────────────────────────────────────────────────────── "ABB", "ADANIGREEN", "ADANITRANS", "AMBUJACEM", "AUROPHARMA", "BANDHANBNK", "BANKBARODA", "BERGEPAINT", "BIOCON", "BOSCHLTD", "CHOLAFIN", "COLPAL", "DABUR", "DLF", "EMAMILTD", "FEDERALBNK", "GAIL", "GODREJCP", "GODREJPROP", "HAL", "HAVELLS", "IDFCFIRSTB", "IOC", "IRCTC", "LUPIN", "MARICO", "MUTHOOTFIN", "NAUKRI", "NMDC", "OFSS", "PAGEIND", "PIDILITIND", "PIIND", "PNB", "SAIL", "SRF", "TATAPOWER", "TORNTPHARM", "UBL", "UNIONBANK", "UPL", "VEDL", "VOLTAS", "ZYDUSLIFE", # ── Nifty Midcap 100 ────────────────────────────────────────────────────── "ASTRAL", "ATGL", "AUBANK", "APLAPOLLO", "BALKRISIND", "BAYERCROP", "BEL", "BHARATFORG", "BHEL", "CANBK", "CANFINHOME", "CASTROLIND", "CEATLTD", "CENTURYTEX", "CESC", "CGPOWER", "COFORGE", "CONCOR", "CROMPTON", "CUMMINSIND", "CYIENT", "DEEPAKNTR", "DELHIVERY", "DIXON", "ENGINERSIN", "EXIDEIND", "FINCABLES", "FLUOROCHEM", "FORTIS", "GLENMARK", "GNFC", "GODAWARI", "GRINDWELL", "GSPL", "HFCL", "HINDCOPPER", "HINDPETRO", "HONAUT", "IBREALEST", "IDEA", "IDFC", "IPCALAB", "ISEC", "JTEKTINDIA", "JUBLFOOD", "KALYANKJIL", "KANSAINER", "KPITTECH", "LAURUSLABS", "LTF", "LTIMindtree", "LTTS", "LUXIND", "M&MFIN", "MANAPPURAM", "MASFIN", "MCX", "METROPOLIS", "MINDTREE", "MOTHERSON", "MPHASIS", "MRF", "MTAR", "NATIONALUM", "NATCOPHARM", "NBCC", "NCC", "NIACL", "NOCIL", "NUVAMA", "OBEROIRLTY", "ONGC", "ORIENTELEC", "PERSISTENT", "PETRONET", "PFC", "PGHH", "PNB", "POLICYBZR", "PRAJIND", "PRESTIGE", "PRINCEPIPE", "RAMCOCEM", "RATNAMANI", "RAYMOND", "RBLBANK", "REDINGTON", "RELAXO", "RENUKA", "RITES", "ROUTE", "SAFARI", "SANOFI", "SCHAEFFLER", "SOLARINDS", "SONACOMS", "SPARC", "STARCEMENT", "STARHEALTH", "SUPREMEIND", "SUVENPHAR", "SWSOLAR", "TANLA", "TASTYBITE", "TATACHEM", "TATACOMM", "TATAELXSI", "TATAINVEST", "TBOTEK", "THERMAX", "THYROCARE", "TIMKEN", "TITAGARH", "TORNTPOWER", "TRENT", "TRIDENT", "TRIVENI", "TTKPRESTIG", "TVSHLTD", "TVSSRICHAK", "UCOBANK", "UJJIVANSFB", "UNOMINDA", "VGUARD", "VBL", "VINATIORGA", "VSTIND", "WABAG", "WELCORP", "WELSPUNLIV", "WHIRLPOOL", "WIPRO", "WOCKPHARMA", "XCHANGING", "YESBANK", "ZEEL", "ZENITHEXPO", "ZENTEC", "ZOMATO", "ZYDUSWELL", # ── Tech / New-Age ──────────────────────────────────────────────────────── "PAYTM", "NYKAA", "INFOEDGE", "INDIAMART", "CARTRADE", "EASEMYTRIP", "MAPMYINDIA", "IXIGO", "AWFIS", "DRONEACHARYA", # ── F&O Popular ────────────────────────────────────────────────────────── "NIFTY", "BANKNIFTY", "FINNIFTY", "MIDCPNIFTY", "SENSEX", "BATAINDIA", "BSOFT", "CLEAN", "CLDX", "DELTACORP", "GNFC", "GMRINFRA", "HUDCO", "IBULHSGFIN", "IDBI", "INDUSTOWER", "JKCEMENT", "JSWENERGY", "KAJARIACER", "KPITTECH", "L&TFH", "LICHSGFIN", "LINDEINDIA", "LICI", "LTTS", "MAXHEALTH", "MAZDOCK", "MFSL", "MRPL", "NETWORK18", "NLCINDIA", "NYKAA", "PAYTM", "PCBL", "PVRINOX", "RECLTD", "RPOWER", "SAIL", "SJVN", "SKFINDIA", "SUNTV", "SUZLON", "TANLA", "TATATECH", "TV18BRDCST", ] # De-duplicate preserving order _seen: set = set() NSE_UNIVERSE = [s for s in NSE_UNIVERSE if s not in _seen and not _seen.add(s)] # type: ignore # Seed prices for simulation fallback _SEED_PRICES: Dict[str, float] = { "RELIANCE": 2950, "TCS": 3718, "INFY": 1956, "HDFCBANK": 1687, "ICICIBANK": 1232, "SBIN": 812, "WIPRO": 543, "AXISBANK": 1145, "HCLTECH": 1642, "KOTAKBANK": 1832, "LT": 3582, "ITC": 463, "BAJFINANCE": 6840, "MARUTI": 10850, "SUNPHARMA": 1720, "TITAN": 3340, "ADANIPORTS": 1180, "NTPC": 342, "TATAMOTORS": 965, "TATASTEEL": 145, "HINDUNILVR": 2620, "BHARTIARTL": 1680, "ONGC": 263, "POWERGRID": 320, "TECHM": 1562, "JSWSTEEL": 890, "COALINDIA": 415, "DRREDDY": 5740, "CIPLA": 1480, "M&M": 2880, "HEROMOTOCO": 4920, "EICHERMOT": 4780, "HINDALCO": 672, "ULTRACEMCO": 10900, "GRASIM": 2680, "INDUSINDBK": 950, "BPCL": 310, "DIVISLAB": 4850, "NESTLEIND": 2160, "BRITANNIA": 4960, "APOLLOHOSP": 6850, "TATACONSUM": 880, "SBILIFE": 1680, "HDFCLIFE": 690, "BAJAJFINSV": 1640, "SHRIRAMFIN": 620, "ADANIENT": 2680, "NIFTY": 24732, "BANKNIFTY": 52800, "FINNIFTY": 23100, "ZOMATO": 218, "NAUKRI": 7200, "PAYTM": 650, "NYKAA": 165, "IRCTC": 905, "INDIAMART": 3200, "TRENT": 5800, "DIXON": 14200, "POLYCAB": 6500, "PERSISTENT": 5200, "COFORGE": 8400, "LTIMindtree": 5300, } # ── Sector mapping ──────────────────────────────────────────────────────────── SECTOR_MAP: Dict[str, str] = { "RELIANCE": "Energy", "ONGC": "Energy", "BPCL": "Energy", "IOC": "Energy", "GAIL": "Energy", "ADANIGREEN": "Energy", "TATAPOWER": "Energy", "HDFCBANK": "Banking", "ICICIBANK": "Banking", "SBIN": "Banking", "AXISBANK": "Banking", "KOTAKBANK": "Banking", "INDUSINDBK": "Banking", "BANDHANBNK": "Banking", "IDFCFIRSTB": "Banking", "PNB": "Banking", "BANKBARODA": "Banking", "YESBANK": "Banking", "FEDERALBNK": "Banking", "TCS": "IT", "INFY": "IT", "WIPRO": "IT", "HCLTECH": "IT", "TECHM": "IT", "LTIMindtree": "IT", "MPHASIS": "IT", "COFORGE": "IT", "PERSISTENT": "IT", "CYIENT": "IT", "LTTS": "IT", "SUNPHARMA": "Pharma", "CIPLA": "Pharma", "DRREDDY": "Pharma", "DIVISLAB": "Pharma", "LUPIN": "Pharma", "BIOCON": "Pharma", "LT": "Infra", "BHEL": "Infra", "NTPC": "Power", "POWERGRID": "Power", "MARUTI": "Auto", "TATAMOTORS": "Auto", "M&M": "Auto", "HEROMOTOCO": "Auto", "EICHERMOT": "Auto", "BAJAJ-AUTO": "Auto", "ITC": "FMCG", "HINDUNILVR": "FMCG", "NESTLEIND": "FMCG", "BRITANNIA": "FMCG", "DABUR": "FMCG", "MARICO": "FMCG", "TITAN": "Consumer", "TRENT": "Consumer", "BAJFINANCE": "NBFC", "BAJAJFINSV": "NBFC", "CHOLAFIN": "NBFC", "MUTHOOTFIN": "NBFC", "TATASTEEL": "Metals", "JSWSTEEL": "Metals", "HINDALCO": "Metals", "ADANIENT": "Conglomerate", "ADANIPORTS": "Ports", } @dataclass class SmartCandidate: """A symbol that passed one or more stages.""" symbol: str sector: str = "Other" ltp: float = 0.0 open: float = 0.0 high: float = 0.0 low: float = 0.0 change_pct: float = 0.0 volume: int = 0 # Stage scores (0-100 each) tech_score: float = 0.0 # Stage 2 quality_score: float = 0.0 # Stage 3 risk_score: float = 0.0 # Stage 4 (higher = safer) total_score: float = 0.0 signal: str = "NEUTRAL" strategy: str = "" # Stage 5 AI results ai_signal: str = "" ai_confidence: float = 0.0 ai_reason: str = "" ai_from_cache: bool = False # Technical indicators rsi: float = 50.0 ema_9: float = 0.0 ema_21: float = 0.0 vwap_signal: str = "NEUTRAL" volume_ratio: float = 1.0 # current vol / avg vol is_circuit: bool = False stage_reached: int = 0 def to_dict(self) -> dict: return { "symbol": self.symbol, "sector": self.sector, "ltp": round(self.ltp, 2), "change_pct": round(self.change_pct, 2), "volume": self.volume, "tech_score": round(self.tech_score, 1), "quality_score": round(self.quality_score, 1), "risk_score": round(self.risk_score, 1), "total_score": round(self.total_score, 1), "signal": self.signal, "strategy": self.strategy, "ai_signal": self.ai_signal, "ai_confidence": round(self.ai_confidence, 1), "ai_reason": self.ai_reason, "ai_from_cache": self.ai_from_cache, "rsi": round(self.rsi, 1), "volume_ratio": round(self.volume_ratio, 2), "stage_reached": self.stage_reached, } class SmartScanner: """ 5-Stage funnel. Called once per engine cycle. Stage 1 → Stage 2 → Stage 3 → Stage 4 → Stage 5 500+ 200 20 5 1-3 """ # Stage thresholds (configurable via config) STAGE2_TOP_N = 200 STAGE3_TOP_N = 20 STAGE4_TOP_N = 5 STAGE5_TOP_N = 3 def __init__(self): self.last_scan_time: float = 0.0 self.last_results: List[SmartCandidate] = [] self.stage_stats: Dict[str, int] = {} # ── Public API ──────────────────────────────────────────────────────────── def scan( self, strategies_enabled: List[str], min_score: float = 60.0, max_sector_pct: float = 30.0, circuit_filter: bool = True, volatility_filter: bool = True, ) -> List[SmartCandidate]: """ Run full 5-stage scan. Returns Stage-4 list with AI results attached. """ t0 = time.time() # ── Stage 1: Universe price fetch ──────────────────────────────────── universe = self._stage1_fetch() logger.debug(f"[SmartScan] Stage1: {len(universe)} symbols fetched") # ── Stage 2: Technical filter → Top 200 ────────────────────────────── stage2 = self._stage2_technical(universe, strategies_enabled)[:self.STAGE2_TOP_N] logger.debug(f"[SmartScan] Stage2 (technical): {len(stage2)} candidates") # ── Stage 3: Quality filter → Top 20 ───────────────────────────────── stage3 = self._stage3_quality(stage2)[:self.STAGE3_TOP_N] logger.debug(f"[SmartScan] Stage3 (quality): {len(stage3)} candidates") # ── Stage 4: Risk filter → Top 5 ───────────────────────────────────── stage4 = self._stage4_risk( stage3, max_sector_pct=max_sector_pct, circuit_filter=circuit_filter, volatility_filter=volatility_filter, )[:self.STAGE4_TOP_N] logger.debug(f"[SmartScan] Stage4 (risk): {len(stage4)} candidates") # Update stage stats self.stage_stats = { "stage1": len(universe), "stage2": len(stage2), "stage3": len(stage3), "stage4": len(stage4), } elapsed = round(time.time() - t0, 2) logger.info(f"[SmartScan] Complete in {elapsed}s: {len(universe)}→{len(stage2)}→{len(stage3)}→{len(stage4)}") # ── Phase 3 Enhancement: Smart Ranking ─────────────────────────────── # Re-rank Stage-4 candidates using multi-factor composite score. # top_n is configurable (default = STAGE5_TOP_N). # This does NOT replace any existing stage — it re-orders the final list. try: from app.autonomous.smart_ranker import smart_ranker stage4 = smart_ranker.rank(stage4, top_n=self.STAGE5_TOP_N) logger.info(f"[SmartScan] SmartRanker → top {len(stage4)} ranked candidates") except Exception as rank_err: logger.warning(f"[SmartScan] SmartRanker skipped (non-fatal): {rank_err}") self.last_scan_time = time.time() self.last_results = stage4 return stage4 def top_candidates(self, n: int = 5, min_score: float = 60.0) -> List[SmartCandidate]: return [c for c in self.last_results if c.total_score >= min_score][:n] # ── Stage implementations ───────────────────────────────────────────────── def _stage1_fetch(self) -> List[SmartCandidate]: """Fetch prices for all universe symbols. Fast — uses in-process cache/sim.""" results = [] for symbol in NSE_UNIVERSE: try: price = self._fetch_price(symbol) c = SmartCandidate( symbol=symbol, sector=SECTOR_MAP.get(symbol, "Other"), ltp=price.get("ltp", 0), open=price.get("open", 0), high=price.get("high", 0), low=price.get("low", 0), change_pct=price.get("change_pct", 0), volume=price.get("volume", 0), stage_reached=1, ) # Pre-filter: skip zero prices if c.ltp > 0: results.append(c) except Exception: continue return results def _stage2_technical( self, candidates: List[SmartCandidate], strategies: List[str] ) -> List[SmartCandidate]: """ Technical scoring: - RSI momentum (0-30 pts) - EMA crossover signal (0-25 pts) - Volume surge (0-25 pts) - Price change momentum (0-20 pts) """ scored = [] for c in candidates: rng = random.Random(int(time.time() // 60) + hash(c.symbol)) # RSI simulation (50±30 range, seeded per minute) rsi = min(95, max(10, 50 + rng.normalvariate(0, 20))) c.rsi = round(rsi, 1) # RSI score: extreme values score higher (oversold/overbought) rsi_score = 0.0 if rsi < 35: rsi_score = 25.0 + (35 - rsi) * 0.5 # oversold (BUY) elif rsi > 65: rsi_score = 25.0 + (rsi - 65) * 0.5 # overbought (SELL) elif 40 <= rsi <= 60: rsi_score = 5.0 # neutral else: rsi_score = 15.0 # EMA crossover (simulated) ema9 = c.ltp * (1 + rng.uniform(-0.005, 0.01)) ema21 = c.ltp * (1 + rng.uniform(-0.01, 0.005)) c.ema_9 = round(ema9, 2) c.ema_21 = round(ema21, 2) ema_score = 20.0 if ema9 > ema21 else 5.0 # Volume surge (0-25 pts) vol_ratio = rng.uniform(0.3, 3.0) c.volume_ratio = round(vol_ratio, 2) vol_score = min(25.0, vol_ratio * 12.0) # Price momentum (0-20 pts) chg = abs(c.change_pct) mom_score = min(20.0, chg * 8.0) c.tech_score = round(rsi_score + ema_score + vol_score + mom_score, 1) c.stage_reached = 2 # Only include if change is meaningful (±0.3%) or high volume if abs(c.change_pct) >= 0.3 or vol_ratio > 1.5: scored.append(c) # Assign signal for c in scored: if c.change_pct > 0.5 and c.rsi < 70: c.signal = "BUY" elif c.change_pct < -0.5 and c.rsi > 30: c.signal = "SELL" else: c.signal = "NEUTRAL" # Pick strategy based on profile if c.rsi < 35 or c.rsi > 65: c.strategy = "RSI_REVERSAL" elif c.ema_9 > c.ema_21 and c.volume_ratio > 1.5: c.strategy = "EMA_CROSSOVER" elif c.volume_ratio > 2.0: c.strategy = "VWAP" elif abs(c.change_pct) > 1.5: c.strategy = "SUPERTREND" else: c.strategy = "MACD" # Filter by strategy enabled + non-neutral scored = [c for c in scored if c.signal != "NEUTRAL" and c.strategy in strategies] scored.sort(key=lambda x: x.tech_score, reverse=True) return scored def _stage3_quality(self, candidates: List[SmartCandidate]) -> List[SmartCandidate]: """ Quality filter: - Supertrend confirmation (0-30 pts) - VWAP position (0-25 pts) - Breakout strength (0-25 pts) - Candle pattern (0-20 pts) """ for c in candidates: rng = random.Random(int(time.time() // 30) + hash(c.symbol + "q3")) # Supertrend (above/below) st_score = rng.uniform(15, 30) if c.signal == "BUY" else rng.uniform(15, 30) if c.signal == "SELL" else 5.0 # VWAP position (price vs VWAP) vwap = c.ltp * rng.uniform(0.99, 1.01) if c.signal == "BUY" and c.ltp > vwap: vwap_score = 25.0 c.vwap_signal = "ABOVE" elif c.signal == "SELL" and c.ltp < vwap: vwap_score = 25.0 c.vwap_signal = "BELOW" else: vwap_score = 8.0 c.vwap_signal = "NEUTRAL" # Breakout: price near day high (BUY) or low (SELL) if c.high > c.low: if c.signal == "BUY": pos = (c.ltp - c.low) / (c.high - c.low) breakout_score = pos * 25.0 else: pos = (c.high - c.ltp) / (c.high - c.low) breakout_score = pos * 25.0 else: breakout_score = 12.5 # Candle pattern (simulated) candle_score = rng.uniform(5, 20) c.quality_score = round(st_score + vwap_score + breakout_score + candle_score, 1) c.stage_reached = 3 candidates.sort(key=lambda x: x.quality_score, reverse=True) return candidates def _stage4_risk( self, candidates: List[SmartCandidate], max_sector_pct: float, circuit_filter: bool, volatility_filter: bool, ) -> List[SmartCandidate]: """ Risk filter: - Circuit check (skip if hit circuit) - Volatility filter (skip if ATR too high/low) - Sector concentration (skip if sector overweight) - Liquidity (skip illiquid stocks) """ sector_count: Dict[str, int] = {} passed = [] for c in candidates: rng = random.Random(int(time.time() // 60) + hash(c.symbol + "risk")) # Circuit filter if circuit_filter: hit_circuit = abs(c.change_pct) >= 10.0 c.is_circuit = hit_circuit if hit_circuit: continue # Volatility filter (ATR simulation: skip if > 5% or < 0.1%) if volatility_filter: simulated_atr_pct = abs(c.change_pct) + rng.uniform(0, 2.0) if simulated_atr_pct > 6.0 or simulated_atr_pct < 0.1: continue # Sector concentration (count per sector) sec = c.sector if sector_count.get(sec, 0) >= 2: # max 2 from same sector continue sector_count[sec] = sector_count.get(sec, 0) + 1 # Liquidity check (price × volume > ₹10 lakh) liquidity = c.ltp * c.volume if c.volume > 0 and liquidity < 1_000_000: continue # Risk score (inverse of risk — higher = safer) diversification = 25.0 if sector_count.get(sec, 1) == 1 else 15.0 liquidity_score = min(25.0, math.log10(max(liquidity, 1)) * 3.0) volatility_ok = 25.0 if simulated_atr_pct < 3.0 else 15.0 circuit_ok = 25.0 c.risk_score = round(diversification + liquidity_score + volatility_ok + circuit_ok, 1) c.total_score = round((c.tech_score + c.quality_score + c.risk_score) / 3, 1) c.stage_reached = 4 passed.append(c) passed.sort(key=lambda x: x.total_score, reverse=True) return passed # ── Price fetch helper ──────────────────────────────────────────────────── def _fetch_price(self, symbol: str) -> dict: """Fetch price — live API if available, else seeded simulation.""" try: from app.api.market_api import get_live_price return get_live_price(symbol=symbol) except Exception: base = _SEED_PRICES.get(symbol, 500 + hash(symbol) % 4000) rng = random.Random(int(time.time() // 60) + hash(symbol)) chg = (rng.random() - 0.48) * 0.025 # ±2.5% ltp = round(base * (1 + chg), 2) op = round(base * (1 - abs(chg) * 0.3), 2) return { "symbol": symbol, "ltp": ltp, "open": op, "high": round(max(op, ltp) * 1.003, 2), "low": round(min(op, ltp) * 0.997, 2), "change_pct": round(chg * 100, 2), "volume": int(rng.uniform(100_000, 10_000_000)), "source": "simulated", } # ── Singleton ───────────────────────────────────────────────────────────────── smart_scanner = SmartScanner()