""" ========================================= CV6 Autonomous Engine — Market Scanner Scans ALL supported NSE symbols. User NEVER selects stocks manually. Returns ranked candidate list. ========================================= """ import random import time from typing import Dict, List # Full NSE symbol universe — same master list used across the app ALL_NSE_SYMBOLS = [ # Indices "NIFTY 50", "BANKNIFTY", "FINNIFTY", # Nifty 50 constituents "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", # Midcap additions "PIDILITIND", "HAVELLS", "VOLTAS", "TORNTPHARM", "LUPIN", "BIOCON", "MUTHOOTFIN", "CHOLAFIN", "IDFCFIRSTB", "FEDERALBNK", "BANDHANBNK", "RBLBANK", "AUBANK", "NAUKRI", "ZOMATO", "PAYTM", "NYKAA", "DELHIVERY", "POLICYBZR", "LTIMindtree", "MPHASIS", "COFORGE", "PERSISTENT", "INFOEDGE", "IRCTC", "INDIAMART", "DEEPAKNTR", "ATGL", "GODREJCP", "DABUR", "EMAMILTD", "COLPAL", "MARICO", "TATAPOWER", "ADANIGREEN", "ADANITRANS", "CANBK", "PNB", "BANKBARODA", "UNIONBANK", ] _SEED_PRICES: Dict[str, float] = { "NIFTY 50": 24732, "BANKNIFTY": 52800, "FINNIFTY": 23100, "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, } class ScanResult: """Result for one symbol from the scanner.""" __slots__ = ( "symbol", "ltp", "open", "high", "low", "change_pct", "volume", "momentum_score", "volume_score", "breakout_score", "opportunity_score", "signal", "strategy", "source", "sector", # AI phase annotations (set by autonomous_engine._phase_analyze) "_ai_signal", "_ai_confidence", "_ai_reason", ) def __init__(self, symbol: str, price_data: dict): self.symbol = symbol self.ltp = price_data.get("ltp", 0) self.open = price_data.get("open", self.ltp) self.high = price_data.get("high", self.ltp) self.low = price_data.get("low", self.ltp) self.change_pct = price_data.get("change_pct", 0) self.volume = price_data.get("volume", 0) self.source = price_data.get("source", "simulated") self.momentum_score = 0.0 self.volume_score = 0.0 self.breakout_score = 0.0 self.opportunity_score = 0.0 self.signal = "NEUTRAL" self.strategy = "" self.sector = "Other" # AI phase (filled by autonomous_engine._phase_analyze) self._ai_signal = "" self._ai_confidence = 0.0 self._ai_reason = "" def to_dict(self) -> dict: return { "symbol": self.symbol, "ltp": round(self.ltp, 2), "change_pct": round(self.change_pct, 2), "volume": self.volume, "momentum": round(self.momentum_score, 1), "volume_score":round(self.volume_score, 1), "breakout": round(self.breakout_score, 1), "score": round(self.opportunity_score, 1), "signal": self.signal, "strategy": self.strategy, "source": self.source, } class MarketScanner: """ Scans all NSE symbols. Calls live price endpoint per symbol (batched). Applies technical scoring filters. Returns ranked candidate list. """ def __init__(self): self.last_scan_time = 0.0 self.last_results: List[ScanResult] = [] def _fetch_price(self, symbol: str) -> dict: """Fetch price using the same logic as /market/live endpoint.""" try: # Import the live price function directly (same process) from app.api.market_api import get_live_price return get_live_price(symbol=symbol) except Exception: # Fallback seed simulation base = _SEED_PRICES.get(symbol, 1000) rng = random.Random(int(time.time() // 60) + hash(symbol)) chg = (rng.random() - 0.48) * 0.01 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.002, 2), "low": round(min(op, ltp) * 0.998, 2), "change_pct": round(chg * 100, 2), "volume": int(rng.uniform(50000, 5_000_000)), "source": "simulated", } def _score(self, r: ScanResult) -> None: """Score 0-100. Higher = better opportunity.""" chg = r.change_pct rng = random.Random(int(time.time() // 30) + hash(r.symbol)) # Momentum: ±3% range → 0-50 points momentum = min(50.0, max(0.0, (abs(chg) / 3.0) * 50.0)) r.momentum_score = momentum # Volume activity: simulated relative volume 0-30 points rel_vol = rng.uniform(0.5, 2.5) r.volume_score = min(30.0, rel_vol * 15.0) # Breakout: price near high of day = 0-20 points if r.high > r.low: pos = (r.ltp - r.low) / (r.high - r.low) r.breakout_score = pos * 20.0 else: r.breakout_score = 10.0 total = r.momentum_score + r.volume_score + r.breakout_score r.opportunity_score = min(100.0, total) # Signal direction if chg > 0.5: r.signal = "BUY" elif chg < -0.5: r.signal = "SELL" else: r.signal = "NEUTRAL" # Best fitting strategy if abs(chg) > 1.5: r.strategy = "SUPERTREND" elif rel_vol > 1.8: r.strategy = "VWAP" elif abs(chg) > 0.8: r.strategy = "EMA_CROSSOVER" else: r.strategy = "RSI_REVERSAL" def scan(self, strategies_enabled: list, min_score: float = 60.0) -> List[ScanResult]: """ Full market scan. Returns list of ScanResult sorted by opportunity_score desc. """ results: List[ScanResult] = [] # Scan all symbols (batch-friendly — reuses in-process price function) for symbol in ALL_NSE_SYMBOLS: try: price = self._fetch_price(symbol) r = ScanResult(symbol, price) self._score(r) # Only include tradeable signals if r.signal != "NEUTRAL" and r.strategy in strategies_enabled: results.append(r) except Exception: continue # Sort by opportunity score descending results.sort(key=lambda x: x.opportunity_score, reverse=True) self.last_scan_time = time.time() self.last_results = results return results def top_candidates(self, n: int = 10, min_score: float = 60.0) -> List[ScanResult]: return [r for r in self.last_results if r.opportunity_score >= min_score][:n]