""" ========================================= CV6 Smart Ranker — Phase 3 Enhancement ========================================= Plugs in AFTER Stage 4 (Risk Filter) of SmartScanner. Takes top-N Stage-4 candidates and applies a richer multi-factor composite score to produce the FINAL ranked list that goes to AI. New scoring dimensions: Factor 1 Relative Strength vs Nifty (0-20 pts) Factor 2 Multi-Timeframe Agreement (0-20 pts) Factor 3 Liquidity Quality (0-20 pts) Factor 4 Sector Momentum (0-20 pts) Factor 5 Market Personality Alignment (0-20 pts) Final rank_score (0-100) is combined with existing total_score: composite = 0.40 × total_score + 0.60 × rank_score This is ADDITIVE — existing scanner + strategy + risk scores are preserved and combined, not replaced. Configurable: top_n → max candidates sent to AI (default 5, range 1–10) ========================================= """ from __future__ import annotations import math import random import time from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple from loguru import logger # Import the SmartCandidate from the existing scanner (never replaced) from app.autonomous.smart_scanner import SmartCandidate, SECTOR_MAP # ── Market personality constants ────────────────────────────────────────────── class MarketPersonality: TRENDING = "TRENDING" SIDEWAYS = "SIDEWAYS" VOLATILE = "VOLATILE" RANGE_BOUND = "RANGE_BOUND" MOMENTUM = "MOMENTUM" # ── Sector strength table (refreshed each rank cycle) ───────────────────────── _SECTOR_STRENGTHS: Dict[str, float] = {} _LAST_SECTOR_REFRESH: float = 0.0 _SECTOR_REFRESH_INTERVAL = 300.0 # 5 minutes def _refresh_sector_strengths() -> Dict[str, float]: """ Compute sector momentum scores (0–100). Uses live data when available; falls back to seeded simulation. All sectors get a score. Higher = stronger momentum. """ global _SECTOR_STRENGTHS, _LAST_SECTOR_REFRESH now = time.time() if now - _LAST_SECTOR_REFRESH < _SECTOR_REFRESH_INTERVAL and _SECTOR_STRENGTHS: return _SECTOR_STRENGTHS sectors = [ "Banking", "IT", "Pharma", "Auto", "FMCG", "Energy", "Metals", "Infra", "Power", "NBFC", "Consumer", "Conglomerate", "Ports", "Other", ] rng = random.Random(int(now // _SECTOR_REFRESH_INTERVAL)) strengths = {} for s in sectors: # Base + seeded noise so rankings are stable per 5-min window strengths[s] = round(rng.uniform(20.0, 90.0), 1) # Try to get real data (graceful skip on failure) try: from app.api.market_api import get_sector_data live = get_sector_data() if isinstance(live, dict): for s, v in live.items(): strengths[s] = float(v) except Exception: pass _SECTOR_STRENGTHS = strengths _LAST_SECTOR_REFRESH = now return strengths def _detect_market_personality() -> str: """ Detect current market personality from Nifty + breadth. Returns one of the MarketPersonality constants. Gracefully falls back to simulation. """ try: from app.api.market_api import get_market_overview data = get_market_overview() nifty_change = abs(float(data.get("nifty_change_pct", 0))) advance = int(data.get("advance", 0)) decline = int(data.get("decline", 0)) total = advance + decline or 1 breadth = advance / total if nifty_change > 1.5: return MarketPersonality.VOLATILE if nifty_change > 0.8 and breadth > 0.6: return MarketPersonality.TRENDING if nifty_change > 0.8: return MarketPersonality.MOMENTUM if breadth < 0.45: return MarketPersonality.RANGE_BOUND return MarketPersonality.SIDEWAYS except Exception: # Seeded simulation — stable per 15-min window rng = random.Random(int(time.time() // 900)) return rng.choice([ MarketPersonality.TRENDING, MarketPersonality.SIDEWAYS, MarketPersonality.VOLATILE, MarketPersonality.RANGE_BOUND, MarketPersonality.MOMENTUM, ]) # ── Rank score attached to each SmartCandidate ─────────────────────────────── @dataclass class RankScore: """Multi-factor rank score attached to a SmartCandidate.""" rs_score: float = 0.0 # Relative Strength vs Nifty (0-20) mtf_score: float = 0.0 # Multi-Timeframe agreement (0-20) liquidity_score: float = 0.0 # Liquidity quality (0-20) sector_score: float = 0.0 # Sector momentum (0-20) market_score: float = 0.0 # Market personality alignment (0-20) total: float = 0.0 # Sum (0-100) composite: float = 0.0 # 0.4×existing + 0.6×rank (0-100) # Extra info for AI context market_personality: str = "" mtf_agreement_pct: float = 0.0 # % timeframes agreeing with signal rs_vs_nifty: float = 0.0 # stock change minus Nifty change relative_volume: float = 1.0 # current vol / 20-day avg vol delivery_pct: float = 0.0 # delivery % (0 if unavailable) bid_ask_spread: float = 0.0 # spread % (0 if unavailable) def to_dict(self) -> dict: return { "rs_score": round(self.rs_score, 1), "mtf_score": round(self.mtf_score, 1), "liquidity_score": round(self.liquidity_score, 1), "sector_score": round(self.sector_score, 1), "market_score": round(self.market_score, 1), "total": round(self.total, 1), "composite": round(self.composite, 1), "market_personality": self.market_personality, "mtf_agreement_pct": round(self.mtf_agreement_pct, 1), "rs_vs_nifty": round(self.rs_vs_nifty, 2), "relative_volume": round(self.relative_volume, 2), "delivery_pct": round(self.delivery_pct, 1), "bid_ask_spread": round(self.bid_ask_spread, 3), } class SmartRanker: """ Phase 3 ranking layer. Called AFTER SmartScanner Stage 4 and BEFORE AI Stage 5. Usage: ranked = smart_ranker.rank(stage4_candidates, top_n=5) Returns the same SmartCandidate objects but: - .rank_score attribute added (RankScore dataclass) - Sorted by composite score (best first) - Trimmed to top_n """ DEFAULT_TOP_N: int = 5 # configurable, range 1-10 def rank( self, candidates: List[SmartCandidate], top_n: int = DEFAULT_TOP_N, ) -> List[SmartCandidate]: """ Rank candidates and return top N. Attaches .rank_score to each candidate. """ if not candidates: return [] top_n = max(1, min(10, top_n)) # clamp 1–10 # Refresh shared market state once per cycle sector_strengths = _refresh_sector_strengths() market_personality = _detect_market_personality() # Get Nifty change for RS calculation nifty_change = self._get_nifty_change() scored: List[Tuple[SmartCandidate, RankScore]] = [] for c in candidates: rs = self._score_relative_strength(c, nifty_change) mtf = self._score_multi_timeframe(c) liq = self._score_liquidity(c) sec = self._score_sector_momentum(c, sector_strengths) mkt = self._score_market_alignment(c, market_personality) total = rs.rs_score + mtf.mtf_score + liq.liquidity_score + sec.sector_score + mkt.market_score rank = RankScore( rs_score = rs.rs_score, mtf_score = mtf.mtf_score, liquidity_score = liq.liquidity_score, sector_score = sec.sector_score, market_score = mkt.market_score, total = round(total, 1), composite = round(0.40 * c.total_score + 0.60 * total, 1), market_personality = market_personality, mtf_agreement_pct = mtf.mtf_agreement_pct, rs_vs_nifty = rs.rs_vs_nifty, relative_volume = liq.relative_volume, delivery_pct = liq.delivery_pct, bid_ask_spread = liq.bid_ask_spread, ) # Attach rank_score to candidate (additive, does not modify existing fields) c.rank_score = rank # type: ignore[attr-defined] scored.append((c, rank)) # Sort by composite (best first) scored.sort(key=lambda t: t[1].composite, reverse=True) result = [c for c, _ in scored[:top_n]] if result: best = result[0] rs = best.rank_score # type: ignore[attr-defined] logger.info( f"[SmartRanker] Top {top_n}/{len(candidates)} | " f"Market={market_personality} | " f"Best={best.symbol} composite={rs.composite:.1f} " f"(rank={rs.total:.1f} + existing={best.total_score:.1f})" ) return result # ── Factor 1: Relative Strength vs Nifty ───────────────────────────────── def _score_relative_strength( self, c: SmartCandidate, nifty_change: float ) -> RankScore: """ RS = stock_change - nifty_change Score: RS > +2% → 20 pts (strong outperformer) RS > +1% → 16 pts RS > +0% → 12 pts (beats Nifty) RS > -1% → 8 pts (slight lag) RS ≤ -1% → 4 pts (underperformer) """ rs = c.change_pct - nifty_change if rs > 2.0: pts = 20.0 elif rs > 1.0: pts = 16.0 elif rs > 0.0: pts = 12.0 elif rs > -1.0: pts = 8.0 else: pts = 4.0 # Invert for SELL signals (stock falling harder = better SELL candidate) if c.signal == "SELL": pts = 20.0 - pts + 4.0 r = RankScore() r.rs_score = round(pts, 1) r.rs_vs_nifty = round(rs, 2) return r # ── Factor 2: Multi-Timeframe Agreement ─────────────────────────────────── def _score_multi_timeframe(self, c: SmartCandidate) -> RankScore: """ Simulate 6 timeframes: 5m, 15m, 30m, 1h, daily, weekly. Each timeframe votes BUY or SELL/NEUTRAL. Agreement % with the candidate's signal → score. When live multi-TF data is available it will be used. Otherwise: seeded simulation (stable per 5-min window per symbol). """ timeframes = ["5m", "15m", "30m", "1h", "daily", "weekly"] signals: Dict[str, str] = {} # Try live data first try: from app.api.market_api import get_multi_timeframe mtf_data = get_multi_timeframe(c.symbol) for tf in timeframes: signals[tf] = mtf_data.get(tf, {}).get("signal", "NEUTRAL") except Exception: # Seeded simulation — higher timeframes more likely to agree with # the daily signal; short TFs noisier rng = random.Random(int(time.time() // 300) + hash(c.symbol + "mtf")) tf_weights = {"5m": 0.55, "15m": 0.60, "30m": 0.65, "1h": 0.70, "daily": 0.80, "weekly": 0.75} for tf, w in tf_weights.items(): agree = rng.random() < w signals[tf] = c.signal if agree else ("SELL" if c.signal == "BUY" else "BUY") # Count agreements agree_count = sum(1 for v in signals.values() if v == c.signal) agreement_pct = round(agree_count / len(timeframes) * 100, 1) # Score mapping if agreement_pct >= 83: pts = 20.0 # 5/6 or 6/6 elif agreement_pct >= 66: pts = 16.0 # 4/6 elif agreement_pct >= 50: pts = 12.0 # 3/6 elif agreement_pct >= 33: pts = 7.0 # 2/6 else: pts = 3.0 # 1/6 or 0/6 r = RankScore() r.mtf_score = round(pts, 1) r.mtf_agreement_pct = agreement_pct return r # ── Factor 3: Liquidity Quality ─────────────────────────────────────────── def _score_liquidity(self, c: SmartCandidate) -> RankScore: """ Liquidity sub-factors: a. Relative Volume (current vol / 20-day avg vol) b. Delivery % (prefer > 40%) c. Bid-Ask Spread (prefer < 0.1%) """ r = RankScore() # a. Relative volume rel_vol = c.volume_ratio # already computed in Stage 2 if rel_vol >= 3.0: rv_score = 10.0 elif rel_vol >= 2.0: rv_score = 8.0 elif rel_vol >= 1.5: rv_score = 6.0 elif rel_vol >= 1.0: rv_score = 4.0 else: rv_score = 1.0 r.relative_volume = rel_vol # b. Delivery % (try live, else simulate) delivery = 0.0 try: from app.api.market_api import get_delivery_data delivery = float(get_delivery_data(c.symbol).get("delivery_pct", 0)) except Exception: rng = random.Random(int(time.time() // 300) + hash(c.symbol + "del")) delivery = round(rng.uniform(20, 75), 1) r.delivery_pct = delivery if delivery >= 60: del_score = 5.0 elif delivery >= 40: del_score = 3.5 elif delivery >= 25: del_score = 2.0 else: del_score = 0.5 # c. Bid-ask spread (try live, else estimate from price) spread = 0.0 try: from app.api.market_api import get_order_book book = get_order_book(c.symbol) best_bid = float(book.get("bid", c.ltp * 0.9995)) best_ask = float(book.get("ask", c.ltp * 1.0005)) spread = (best_ask - best_bid) / c.ltp * 100 if c.ltp else 0 except Exception: # Large-cap spreads ~0.02-0.05%, mid/small ~0.05-0.2% rng = random.Random(int(time.time() // 300) + hash(c.symbol + "ba")) spread = round(rng.uniform(0.01, 0.15), 3) r.bid_ask_spread = spread if spread < 0.03: ba_score = 5.0 elif spread < 0.07: ba_score = 3.5 elif spread < 0.12: ba_score = 2.0 else: ba_score = 0.5 r.liquidity_score = round(rv_score + del_score + ba_score, 1) return r # ── Factor 4: Sector Momentum ───────────────────────────────────────────── def _score_sector_momentum( self, c: SmartCandidate, sector_strengths: Dict[str, float] ) -> RankScore: """ Map sector strength (0-100) → points (0-20). Strong sector + BUY signal → high score. Weak sector + SELL signal → high score too (aligned). """ strength = sector_strengths.get(c.sector, sector_strengths.get("Other", 50.0)) if c.signal == "BUY": pts = round(strength * 0.20, 1) # 0-20 pts elif c.signal == "SELL": pts = round((100 - strength) * 0.20, 1) # weak sector → good SELL else: pts = 10.0 r = RankScore() r.sector_score = pts return r # ── Factor 5: Market Personality Alignment ──────────────────────────────── def _score_market_alignment( self, c: SmartCandidate, market_personality: str ) -> RankScore: """ Alignment scoring: TRENDING → prefer MOMENTUM / SUPERTREND / EMA_CROSSOVER signals VOLATILE → prefer RSI_REVERSAL (mean-reversion) SIDEWAYS → prefer VWAP / MACD range plays RANGE_BOUND→ prefer RSI_REVERSAL / VWAP MOMENTUM → prefer EMA_CROSSOVER / SUPERTREND """ strategy = c.strategy alignment_map = { MarketPersonality.TRENDING: { "SUPERTREND": 20, "EMA_CROSSOVER": 18, "MACD": 15, "VWAP": 12, "RSI_REVERSAL": 8, }, MarketPersonality.VOLATILE: { "RSI_REVERSAL": 20, "MACD": 14, "VWAP": 12, "SUPERTREND": 10, "EMA_CROSSOVER": 8, }, MarketPersonality.SIDEWAYS: { "VWAP": 20, "MACD": 18, "RSI_REVERSAL": 14, "EMA_CROSSOVER": 10, "SUPERTREND": 8, }, MarketPersonality.RANGE_BOUND: { "RSI_REVERSAL": 20, "VWAP": 17, "MACD": 13, "EMA_CROSSOVER": 8, "SUPERTREND": 6, }, MarketPersonality.MOMENTUM: { "EMA_CROSSOVER": 20, "SUPERTREND": 18, "MACD": 14, "VWAP": 10, "RSI_REVERSAL": 8, }, } pts = alignment_map.get(market_personality, {}).get(strategy, 10) r = RankScore() r.market_score = float(pts) return r # ── Nifty change helper ─────────────────────────────────────────────────── def _get_nifty_change(self) -> float: """Return Nifty 50 % change today. Graceful fallback.""" try: from app.api.market_api import get_live_price data = get_live_price(symbol="NIFTY") return float(data.get("change_pct", 0.0)) except Exception: rng = random.Random(int(time.time() // 900)) return round(rng.uniform(-1.5, 1.5), 2) # ── Singleton ───────────────────────────────────────────────────────────────── smart_ranker = SmartRanker()