""" ========================================= CV6 AI Context Builder — Phase 4 Enhancement ========================================= Prepares a COMPLETE market intelligence package BEFORE the AI is called, so the AI only has to ANALYSE — not calculate indicators again. 9 Context Sections: 1. Market Personality 2. Sector Rotation 3. Liquidity 4. Institutional Activity (graceful skip if unavailable) 5. Relative Strength 6. Multi-Timeframe Analysis 7. Market Breadth 8. Portfolio Impact 9. AI Confidence Inputs AI Output contract: SIGNAL: BUY | SELL | WAIT | REJECT CONFIDENCE: 0-100 REASON: RISK_LEVEL: LOW | MEDIUM | HIGH Does NOT replace the scanner, strategy, or risk modules. Only enriches the information package sent to the AI. ========================================= """ from __future__ import annotations import math import time from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple from loguru import logger # Real NSE data — fail gracefully if unavailable try: from app.market.nse_data import ( get_fii_dii, get_india_vix, get_market_breadth, get_nifty_change ) _NSE_AVAILABLE = True except ImportError: _NSE_AVAILABLE = False # Real news sentiment — fail gracefully try: from app.market.news_feed import get_news_summary _NEWS_AVAILABLE = True except ImportError: _NEWS_AVAILABLE = False # Volume analysis — fail gracefully try: from app.market.volume_analyzer import analyse_volume, VolumeAnalysis, fmt_volume _VOLUME_AVAILABLE = True except ImportError: _VOLUME_AVAILABLE = False # Market structure — fail gracefully try: from app.market.market_structure import ( analyse_structure, get_nifty_structure, MarketStructure, fmt_structure ) _STRUCTURE_AVAILABLE = True except ImportError: _STRUCTURE_AVAILABLE = False # SmartCandidate is already scored by scanner + ranker — read-only here try: from app.autonomous.smart_scanner import SmartCandidate except ImportError: SmartCandidate = Any # type: ignore # ─── 9-Section Context Dataclasses ──────────────────────────────────────────── @dataclass class MarketPersonalityCtx: personality: str = "UNKNOWN" # TRENDING | SIDEWAYS | VOLATILE | RANGE_BOUND | MOMENTUM nifty_change: float = 0.0 breadth_pct: float = 0.0 # advance / (advance+decline) vix: float = 0.0 description: str = "" @dataclass class SectorRotationCtx: stock_sector: str = "Other" sector_rank: int = 0 # 1 = strongest, higher = weaker sector_momentum: float = 0.0 # 0-100 strong_sectors: List[str] = field(default_factory=list) weak_sectors: List[str] = field(default_factory=list) sector_signal: str = "NEUTRAL" # STRONG | NEUTRAL | WEAK @dataclass class LiquidityCtx: avg_volume: int = 0 relative_volume: float = 1.0 # current / 20-day avg delivery_pct: float = 0.0 bid_ask_spread: float = 0.0 # % liquidity_score: float = 0.0 # 0-100 liquidity_grade: str = "N/A" # A / B / C / D @dataclass class InstitutionalCtx: available: bool = False fii_activity: str = "N/A" # BUY | SELL | NEUTRAL | N/A dii_activity: str = "N/A" block_deals: int = 0 bulk_deals: int = 0 delivery_trend: str = "N/A" # INCREASING | DECREASING | STABLE @dataclass class RelativeStrengthCtx: vs_nifty: float = 0.0 # stock_change - nifty_change vs_sector: float = 0.0 # stock_change - sector_avg_change rs_score: float = 0.0 # 0-100 rs_grade: str = "N/A" # VERY_STRONG | STRONG | NEUTRAL | WEAK | VERY_WEAK @dataclass class TimeframeSignal: timeframe: str = "" signal: str = "NEUTRAL" strength: float = 0.0 # 0-100 @dataclass class MultiTimeframeCtx: signals: List[TimeframeSignal] = field(default_factory=list) agreement_pct: float = 0.0 # % timeframes agreeing with daily signal dominant_signal: str = "NEUTRAL" # Most common signal across TFs @dataclass class MarketBreadthCtx: advance: int = 0 decline: int = 0 new_high: int = 0 new_low: int = 0 breadth_score: float = 0.0 # 0-100 breadth_signal: str = "NEUTRAL" # BULLISH | NEUTRAL | BEARISH @dataclass class PortfolioImpactCtx: current_open_positions: int = 0 sector_exposure_pct: float = 0.0 # % of portfolio in same sector correlation_risk: str = "LOW" # LOW | MEDIUM | HIGH capital_usage_pct: float = 0.0 risk_exposure_pct: float = 0.0 @dataclass class NewsSentimentCtx: available: bool = False total_articles: int = 0 bullish_count: int = 0 bearish_count: int = 0 sentiment: str = "N/A" # BULLISH | BEARISH | NEUTRAL | N/A sentiment_score: float = 0.0 # -1.0 to +1.0 headlines: List[str] = field(default_factory=list) key_events: List[str] = field(default_factory=list) summary: str = "" @dataclass class OICtx: available: bool = False symbol: str = "NIFTY" pcr: float = 0.0 total_call_oi: int = 0 total_put_oi: int = 0 max_call_oi_strike: int = 0 max_put_oi_strike: int = 0 oi_signal: str = "N/A" # BULLISH | BEARISH | NEUTRAL | N/A sentiment: str = "" @dataclass class VolumeCtx: available: bool = False symbol: str = "" volume_ratio: float = 0.0 volume_signal: str = "N/A" # CLIMAX | SPIKE | HIGH | NORMAL | LOW | N/A price_volume: str = "N/A" # CONFIRMED_UP | CONFIRMED_DOWN | DIVERGENCE | UNCONFIRMED | N/A volume_trend: str = "N/A" # RISING | FALLING | FLAT | N/A interpretation: str = "" @dataclass class MarketStructureCtx: available: bool = False symbol: str = "" trend: str = "N/A" # UPTREND | DOWNTREND | SIDEWAYS | REVERSAL_UP | REVERSAL_DOWN trend_score: int = 50 above_200ema: bool = False ema_200: float = 0.0 current_price: float = 0.0 trade_bias: str = "WAIT" # BUY | SELL | NEUTRAL | WAIT nearest_support: float = 0.0 nearest_resistance: float = 0.0 bos: bool = False bos_direction: str = "" structure_pattern: str = "" interpretation: str = "" @dataclass class AIConfidenceInputs: technical_score: float = 0.0 # From SmartScanner Stage 2 quality_score: float = 0.0 # From SmartScanner Stage 3 risk_score: float = 0.0 # From SmartScanner Stage 4 rank_score: float = 0.0 # From SmartRanker sector_score: float = 0.0 # From SmartRanker sector factor market_score: float = 0.0 # From SmartRanker market factor liquidity_score: float = 0.0 # From SmartRanker liquidity factor mtf_score: float = 0.0 # From SmartRanker MTF factor rs_score: float = 0.0 # From SmartRanker RS factor overall_confidence: float = 0.0 # Weighted composite (0-100) # ─── Master context package ─────────────────────────────────────────────────── @dataclass class AIContextPackage: """ Complete market intelligence package passed to AI. AI uses this to ONLY analyse — all indicators are pre-computed. """ symbol: str = "" signal_from_scanner: str = "" # BUY | SELL (from tech filter) strategy: str = "" ltp: float = 0.0 change_pct: float = 0.0 data_quality: str = "unknown" # "real" | "simulated" (TIER-1 1B) market: MarketPersonalityCtx = field(default_factory=MarketPersonalityCtx) sector: SectorRotationCtx = field(default_factory=SectorRotationCtx) liquidity: LiquidityCtx = field(default_factory=LiquidityCtx) institutional: InstitutionalCtx = field(default_factory=InstitutionalCtx) rs: RelativeStrengthCtx = field(default_factory=RelativeStrengthCtx) mtf: MultiTimeframeCtx = field(default_factory=MultiTimeframeCtx) breadth: MarketBreadthCtx = field(default_factory=MarketBreadthCtx) portfolio: PortfolioImpactCtx = field(default_factory=PortfolioImpactCtx) confidence: AIConfidenceInputs = field(default_factory=AIConfidenceInputs) # NEW: Real data sections news: NewsSentimentCtx = field(default_factory=NewsSentimentCtx) oi: OICtx = field(default_factory=OICtx) volume: VolumeCtx = field(default_factory=VolumeCtx) structure: MarketStructureCtx = field(default_factory=MarketStructureCtx) def to_dict(self) -> Dict[str, Any]: from dataclasses import asdict return asdict(self) def to_prompt(self) -> str: """ Convert context package to a rich, structured AI prompt. AI should NOT re-calculate indicators — only analyse. """ m = self.market sc = self.sector lq = self.liquidity ins = self.institutional rs = self.rs tf = self.mtf br = self.breadth po = self.portfolio ci = self.confidence mtf_lines = "\n".join( f" {ts.timeframe:8s} → {ts.signal:7s} (strength {ts.strength:.0f})" for ts in tf.signals ) or " N/A" inst_section = "" if ins.available: inst_section = f""" ── 4. INSTITUTIONAL ACTIVITY ──────────────────────────────────────── FII Activity : {ins.fii_activity} DII Activity : {ins.dii_activity} Block Deals : {ins.block_deals} Bulk Deals : {ins.bulk_deals} Delivery Trend : {ins.delivery_trend}""" else: inst_section = "\n── 4. INSTITUTIONAL ACTIVITY ──── [Data not available — skip] ────" prompt = f"""=== CV6 AI TRADING ANALYSIS REQUEST === Symbol : {self.symbol} LTP : ₹{self.ltp:.2f} Change : {self.change_pct:+.2f}% Scanner Signal : {self.signal_from_scanner} Strategy : {self.strategy} ── 1. MARKET PERSONALITY ───────────────────────────────────────────── Personality : {m.personality} Nifty Change : {m.nifty_change:+.2f}% Breadth : {m.breadth_pct:.0f}% stocks advancing VIX : {m.vix:.1f} Description : {m.description} ── 2. SECTOR ROTATION ──────────────────────────────────────────────── Stock Sector : {sc.stock_sector} Sector Rank : #{sc.sector_rank} (1=strongest) Sector Momentum: {sc.sector_momentum:.0f}/100 Sector Signal : {sc.sector_signal} Strong Sectors : {', '.join(sc.strong_sectors[:3]) or 'N/A'} Weak Sectors : {', '.join(sc.weak_sectors[:3]) or 'N/A'} ── 3. LIQUIDITY ────────────────────────────────────────────────────── Avg Volume : {lq.avg_volume:,} Relative Volume: {lq.relative_volume:.1f}x (current vs 20-day avg) Delivery % : {lq.delivery_pct:.1f}% Bid-Ask Spread : {lq.bid_ask_spread:.3f}% Liquidity Score: {lq.liquidity_score:.0f}/100 ({lq.liquidity_grade}) {inst_section} ── 5. RELATIVE STRENGTH ────────────────────────────────────────────── vs Nifty 50 : {rs.vs_nifty:+.2f}% ({('outperforming' if rs.vs_nifty > 0 else 'underperforming')}) vs Sector Avg : {rs.vs_sector:+.2f}% RS Score : {rs.rs_score:.0f}/100 RS Grade : {rs.rs_grade} ── 6. MULTI-TIMEFRAME ANALYSIS ─────────────────────────────────────── Agreement : {tf.agreement_pct:.0f}% of timeframes agree with daily signal Dominant Signal: {tf.dominant_signal} Timeframe Detail: {mtf_lines} ── 7. MARKET BREADTH ───────────────────────────────────────────────── Advance / Decline: {br.advance} / {br.decline} New 52W High : {br.new_high} New 52W Low : {br.new_low} Breadth Score : {br.breadth_score:.0f}/100 Breadth Signal : {br.breadth_signal} ── 8. PORTFOLIO IMPACT ─────────────────────────────────────────────── Open Positions : {po.current_open_positions} Sector Exposure: {po.sector_exposure_pct:.0f}% of capital in same sector Correlation : {po.correlation_risk} Capital Used : {po.capital_usage_pct:.0f}% Risk Exposure : {po.risk_exposure_pct:.0f}% ── 9. AI CONFIDENCE INPUTS ─────────────────────────────────────────── Technical Score : {ci.technical_score:.0f}/100 Quality Score : {ci.quality_score:.0f}/100 Risk Score : {ci.risk_score:.0f}/100 RS Score : {ci.rs_score:.0f}/100 MTF Score : {ci.mtf_score:.0f}/100 Sector Score : {ci.sector_score:.0f}/100 Market Score : {ci.market_score:.0f}/100 Liquidity Score : {ci.liquidity_score:.0f}/100 Overall Confidence: {ci.overall_confidence:.0f}/100 ── 10. NEWS & SENTIMENT ────────────────────────────────────────────── {self._fmt_news()} ── 11. OPEN INTEREST (NIFTY OPTIONS) ──────────────────────────────── {self._fmt_oi()} ── 12. VOLUME ANALYSIS ────────────────────────────────────────────── {self._fmt_volume()} ── 13. MARKET STRUCTURE ───────────────────────────────────────────── {self._fmt_structure()} ═══════════════════════════════════════════════════════════════════════ INSTRUCTION: Data quality: {self.data_quality.upper()}. When quality is REAL, the technical indicators (RSI/EMA/VWAP/ATR/Supertrend/volume) were computed from real OHLCV candles. When SIMULATED, treat those technicals as UNRELIABLE and do not base a trade on them. Macro/news/OI/institutional fields may be partial if unavailable. You are the final reviewer. Analyse the COMPLETE picture including: • News sentiment + FII/DII institutional flow • OI (PCR, max pain resistance/support) • Volume signal (spike = institutional; divergence = fake move) • Market structure (200 EMA bias, trend, BoS, S/R levels) CRITICAL RULE: Only BUY when above_200ema=True; only SELL when above_200ema=False. If structure trade_bias=WAIT, output WAIT regardless of other signals. Output EXACTLY: SIGNAL: BUY|SELL|WAIT|REJECT CONFIDENCE: 0-100 REASON: RISK_LEVEL: LOW|MEDIUM|HIGH ═══════════════════════════════════════════════════════════════════════""" return prompt def _fmt_news(self) -> str: n = self.news if not n.available: return "News data not available" lines = [ f"Sentiment : {n.sentiment} (score {n.sentiment_score:+.2f})", f"Articles : {n.total_articles} ({n.bullish_count} bullish, {n.bearish_count} bearish)", f"Summary : {n.summary}", ] for i, h in enumerate(n.headlines[:3], 1): lines.append(f"Headline {i} : {h[:100]}") for e in n.key_events[:2]: lines.append(f"NSE Filing : {e[:100]}") return "\n".join(lines) def _fmt_oi(self) -> str: o = self.oi if not o.available: return "OI data not available" return ( f"PCR : {o.pcr:.3f} ({'>1.2 bullish, <0.8 bearish'})\n" f"Call OI : {o.total_call_oi:,} (max at {o.max_call_oi_strike} = resistance)\n" f"Put OI : {o.total_put_oi:,} (max at {o.max_put_oi_strike} = support)\n" f"OI Signal : {o.oi_signal}\n" f"Interpretation : {o.sentiment}" ) def _fmt_volume(self) -> str: v = self.volume if not v.available: return "Volume data not available" return ( f"Signal : {v.volume_signal} (ratio {v.volume_ratio:.2f}x 20-day avg)\n" f"Price-Volume : {v.price_volume}\n" f"Volume Trend : {v.volume_trend}\n" f"Interpretation : {v.interpretation}" ) def _fmt_structure(self) -> str: s = self.structure if not s.available: return "Market structure data not available" ema_str = f"₹{s.ema_200:.2f}" if s.ema_200 else "N/A" bias_flag = " ← CRITICAL FILTER" if s.trade_bias in ("BUY", "SELL") else "" bos_str = f"YES ({s.bos_direction})" if s.bos else "No" return ( f"Trend : {s.trend} (score {s.trend_score}/100)\n" f"200 EMA : {ema_str} | Price: ₹{s.current_price:.2f}\n" f"Above 200 EMA : {'YES — BUY bias' if s.above_200ema else 'NO — SELL bias only'}\n" f"Structure : {s.structure_pattern}\n" f"Break of Struct: {bos_str}\n" f"Nearest Support: ₹{s.nearest_support:.2f}\n" f"Nearest Resist : ₹{s.nearest_resistance:.2f}\n" f"Trade Bias : {s.trade_bias}{bias_flag}\n" f"Interpretation : {s.interpretation}" ) # ─── AIContextBuilder ───────────────────────────────────────────────────────── class AIContextBuilder: """ Builds AIContextPackage for a SmartCandidate. All 9 sections populated; data-unavailable sections fail gracefully. """ def build( self, candidate: Any, portfolio_state: Optional[Any] = None, ) -> AIContextPackage: """ Build complete context for one candidate. Args: candidate: SmartCandidate (with .rank_score attached by SmartRanker) portfolio_state: CapitalPoolManager or similar (optional — for section 8) Returns: AIContextPackage ready for .to_prompt() """ # TIER-1 (1B): reflect whether the technicals attached to this # candidate are real (computed from OHLCV) or simulated. _rt = getattr(candidate, "_real_technicals", None) _dq = "real" if (_rt is not None and getattr(_rt, "is_real", False)) else \ ("real" if getattr(candidate, "_real_data_ok", False) else "simulated") pkg = AIContextPackage( symbol = candidate.symbol, signal_from_scanner = candidate.signal, strategy = candidate.strategy, ltp = candidate.ltp, change_pct = candidate.change_pct, data_quality = _dq, ) rank = getattr(candidate, "rank_score", None) pkg.market = self._build_market(rank) pkg.sector = self._build_sector(candidate, rank) pkg.liquidity = self._build_liquidity(candidate, rank) pkg.institutional = self._build_institutional(candidate) pkg.rs = self._build_relative_strength(candidate, rank) pkg.mtf = self._build_multi_timeframe(candidate, rank) pkg.breadth = self._build_breadth() pkg.portfolio = self._build_portfolio(candidate, portfolio_state) pkg.confidence = self._build_confidence_inputs(candidate, rank, pkg) pkg.news = self._build_news(candidate) pkg.oi = self._build_oi() pkg.volume = self._build_volume(candidate) pkg.structure = self._build_structure(candidate) logger.debug( f"[AIContext] Built for {candidate.symbol} | " f"Market={pkg.market.personality} | " f"MTF={pkg.mtf.agreement_pct:.0f}% | " f"Overall={pkg.confidence.overall_confidence:.0f}" ) return pkg # ── Section 1: Market Personality ──────────────────────────────────────── def _build_market(self, rank: Any) -> MarketPersonalityCtx: ctx = MarketPersonalityCtx() # Pull from rank_score if already detected if rank and rank.market_personality: ctx.personality = rank.market_personality # ── Real Nifty change from NSE ──────────────────────────────────── if _NSE_AVAILABLE: try: chg = get_nifty_change() if chg is not None: ctx.nifty_change = chg except Exception as e: logger.debug(f"[AICtx] Nifty change unavailable: {e}") # No fallback to random — leave at 0.0 if unavailable # ── Real India VIX from NSE ─────────────────────────────────────── if _NSE_AVAILABLE: try: vix = get_india_vix() if vix is not None: ctx.vix = vix except Exception as e: logger.debug(f"[AICtx] VIX unavailable: {e}") # No fallback to random — leave at 0.0 if unavailable # ── Real market breadth from NSE ────────────────────────────────── if _NSE_AVAILABLE: try: bdata = get_market_breadth() if bdata: adv = int(bdata.get("advance", 0)) dec = int(bdata.get("decline", 0)) ctx.breadth_pct = round(adv / (adv + dec) * 100, 1) if (adv + dec) else 0.0 except Exception as e: logger.debug(f"[AICtx] Breadth unavailable: {e}") # No fallback to random — leave at 0.0 if unavailable # Auto-detect personality from real data if not set by ranker if ctx.personality == "UNKNOWN": if ctx.vix > 20: ctx.personality = "VOLATILE" elif abs(ctx.nifty_change) > 0.8: ctx.personality = "TRENDING" elif ctx.breadth_pct > 60: ctx.personality = "MOMENTUM" elif ctx.breadth_pct < 40: ctx.personality = "SIDEWAYS" # else stays UNKNOWN — honest desc_map = { "TRENDING": "Market is in a directional trend — momentum strategies preferred", "SIDEWAYS": "Market moving sideways — range-bound strategies preferred", "VOLATILE": f"High volatility (VIX {ctx.vix:.1f}) — reduce position size", "RANGE_BOUND": "Market range-bound — supports and resistances are key", "MOMENTUM": f"Strong breadth ({ctx.breadth_pct:.0f}% advancing) — breakout strategies preferred", "UNKNOWN": "Market personality unclear — use caution", } ctx.description = desc_map.get(ctx.personality, "Market personality unclear") return ctx # ── Section 2: Sector Rotation ─────────────────────────────────────────── def _build_sector(self, candidate: Any, rank: Any) -> SectorRotationCtx: ctx = SectorRotationCtx(stock_sector=candidate.sector) try: from app.autonomous.smart_ranker import _refresh_sector_strengths strengths = _refresh_sector_strengths() # Rank sectors ranked_sectors = sorted(strengths.items(), key=lambda x: x[1], reverse=True) sector_names = [s for s, _ in ranked_sectors] # Stock's sector rank ctx.sector_rank = sector_names.index(candidate.sector) + 1 if candidate.sector in sector_names else 99 ctx.sector_momentum = round(strengths.get(candidate.sector, 50.0), 1) ctx.strong_sectors = sector_names[:3] ctx.weak_sectors = sector_names[-3:] if ctx.sector_momentum >= 70: ctx.sector_signal = "STRONG" elif ctx.sector_momentum >= 40: ctx.sector_signal = "NEUTRAL" else: ctx.sector_signal = "WEAK" except Exception: ctx.sector_signal = "NEUTRAL" ctx.sector_momentum = 50.0 return ctx # ── Section 3: Liquidity ───────────────────────────────────────────────── def _build_liquidity(self, candidate: Any, rank: Any) -> LiquidityCtx: ctx = LiquidityCtx() # Pull from rank_score if available if rank: ctx.relative_volume = rank.relative_volume ctx.delivery_pct = rank.delivery_pct ctx.bid_ask_spread = rank.bid_ask_spread # Avg volume estimate (volume / relative_volume) rel_v = ctx.relative_volume if ctx.relative_volume > 0 else 1.0 ctx.avg_volume = int(candidate.volume / rel_v) # Liquidity score (0-100) rv_pts = min(40, ctx.relative_volume * 15) del_pts = min(30, ctx.delivery_pct * 0.5) ba_pts = min(30, max(0, (0.2 - ctx.bid_ask_spread) / 0.2 * 30)) ctx.liquidity_score = round(rv_pts + del_pts + ba_pts, 1) if ctx.liquidity_score >= 80: ctx.liquidity_grade = "A" elif ctx.liquidity_score >= 60: ctx.liquidity_grade = "B" elif ctx.liquidity_score >= 40: ctx.liquidity_grade = "C" else: ctx.liquidity_grade = "D" return ctx # ── Section 4: Institutional Activity ──────────────────────────────────── def _build_institutional(self, candidate: Any) -> InstitutionalCtx: ctx = InstitutionalCtx(available=False) # ── Real FII/DII data from NSE ──────────────────────────────────── if _NSE_AVAILABLE: try: fii_data = get_fii_dii() if fii_data: ctx.available = True ctx.fii_activity = fii_data.get("fii_activity", "NEUTRAL") ctx.dii_activity = fii_data.get("dii_activity", "NEUTRAL") # Store net values in delivery_trend for prompt context fii_net = fii_data.get("fii_net", 0) dii_net = fii_data.get("dii_net", 0) ctx.delivery_trend = ( f"FII ₹{fii_net:+.0f}Cr, DII ₹{dii_net:+.0f}Cr " f"(Date: {fii_data.get('date', 'today')})" ) logger.info( f"[AICtx] FII/DII loaded: FII={ctx.fii_activity} " f"₹{fii_net:+.0f}Cr, DII={ctx.dii_activity} ₹{dii_net:+.0f}Cr" ) except Exception as e: logger.debug(f"[AICtx] FII/DII unavailable: {e}") return ctx # ── Section 5: Relative Strength ───────────────────────────────────────── def _build_relative_strength(self, candidate: Any, rank: Any) -> RelativeStrengthCtx: ctx = RelativeStrengthCtx() if rank: ctx.vs_nifty = rank.rs_vs_nifty ctx.rs_score = rank.rs_score * 5 # convert 0-20 to 0-100 # vs Sector average try: from app.api.market_api import get_sector_performance sector_chg = float(get_sector_performance(candidate.sector).get("change_pct", 0)) ctx.vs_sector = round(candidate.change_pct - sector_chg, 2) except Exception: ctx.vs_sector = 0.0 # unknown — not random # Grade if ctx.rs_score >= 80: ctx.rs_grade = "VERY_STRONG" elif ctx.rs_score >= 60: ctx.rs_grade = "STRONG" elif ctx.rs_score >= 40: ctx.rs_grade = "NEUTRAL" elif ctx.rs_score >= 20: ctx.rs_grade = "WEAK" else: ctx.rs_grade = "VERY_WEAK" return ctx # ── Section 6: Multi-Timeframe ──────────────────────────────────────────── def _build_multi_timeframe(self, candidate: Any, rank: Any) -> MultiTimeframeCtx: ctx = MultiTimeframeCtx() timeframes = ["5m", "15m", "30m", "1h", "daily", "weekly"] tf_signals: List[TimeframeSignal] = [] try: from app.api.market_api import get_multi_timeframe raw = get_multi_timeframe(candidate.symbol) for tf in timeframes: d = raw.get(tf, {}) tf_signals.append(TimeframeSignal( timeframe = tf, signal = d.get("signal", "NEUTRAL"), strength = float(d.get("strength", 50)), )) except Exception: # No random fallback — leave signals empty; AI prompt shows N/A pass ctx.signals = tf_signals # Agreement % agree = sum(1 for ts in tf_signals if ts.signal == candidate.signal) ctx.agreement_pct = round(agree / len(tf_signals) * 100, 1) if tf_signals else 0.0 # Dominant signal from collections import Counter counts = Counter(ts.signal for ts in tf_signals) ctx.dominant_signal = counts.most_common(1)[0][0] if counts else "NEUTRAL" if rank: # Sync with ranker's pre-computed value ctx.agreement_pct = rank.mtf_agreement_pct return ctx # ── Section 7: Market Breadth ───────────────────────────────────────────── def _build_breadth(self) -> MarketBreadthCtx: ctx = MarketBreadthCtx() # ── Real breadth from NSE ───────────────────────────────────────── data_loaded = False if _NSE_AVAILABLE: try: bdata = get_market_breadth() if bdata: ctx.advance = int(bdata.get("advance", 0)) ctx.decline = int(bdata.get("decline", 0)) ctx.new_high = int(bdata.get("new_high", 0)) ctx.new_low = int(bdata.get("new_low", 0)) data_loaded = True logger.info( f"[AICtx] Breadth loaded: " f"Adv={ctx.advance} Dec={ctx.decline} " f"52H={ctx.new_high} 52L={ctx.new_low}" ) except Exception as e: logger.debug(f"[AICtx] Breadth unavailable: {e}") if not data_loaded: # Leave as zeros — AI prompt will show 0/0 which is honest ctx.breadth_signal = "N/A" return ctx total = ctx.advance + ctx.decline or 1 breadth_ratio = ctx.advance / total ctx.breadth_score = round( breadth_ratio * 80 + (ctx.new_high / (ctx.new_high + ctx.new_low + 1)) * 20, 1 ) if breadth_ratio >= 0.60: ctx.breadth_signal = "BULLISH" elif breadth_ratio <= 0.40: ctx.breadth_signal = "BEARISH" else: ctx.breadth_signal = "NEUTRAL" return ctx # ── Section 8: Portfolio Impact ─────────────────────────────────────────── def _build_portfolio(self, candidate: Any, portfolio_state: Any) -> PortfolioImpactCtx: ctx = PortfolioImpactCtx() try: if portfolio_state is None: return ctx # Try to get pool manager data if hasattr(portfolio_state, "all_positions"): all_pos = portfolio_state.all_positions() ctx.current_open_positions = len(all_pos) # Sector exposure same_sector = [p for p in all_pos if getattr(p, "sector", "") == candidate.sector] if hasattr(portfolio_state, "total_deployed"): total_dep = portfolio_state.total_deployed() or 1 sec_dep = sum(getattr(p, "value", 0) for p in same_sector) ctx.sector_exposure_pct = round(sec_dep / total_dep * 100, 1) # Capital usage if hasattr(portfolio_state, "pools"): total_cap = sum( getattr(p, "initial_capital", 0) for p in portfolio_state.pools.values() ) used_cap = sum( getattr(p, "deployed_capital", 0) for p in portfolio_state.pools.values() ) ctx.capital_usage_pct = round(used_cap / total_cap * 100, 1) if total_cap else 0 # Correlation risk (simple: if sector exposure high → HIGH) if ctx.sector_exposure_pct >= 30: ctx.correlation_risk = "HIGH" elif ctx.sector_exposure_pct >= 15: ctx.correlation_risk = "MEDIUM" else: ctx.correlation_risk = "LOW" ctx.risk_exposure_pct = round(ctx.capital_usage_pct * 0.05, 1) # approx 5% VaR except Exception: pass return ctx # ── Section 9: AI Confidence Inputs ────────────────────────────────────── def _build_confidence_inputs( self, candidate: Any, rank: Any, pkg: AIContextPackage, ) -> AIConfidenceInputs: ci = AIConfidenceInputs() # Scores from SmartScanner (Stages 2-4) ci.technical_score = getattr(candidate, "tech_score", 0.0) ci.quality_score = getattr(candidate, "quality_score", 0.0) ci.risk_score = getattr(candidate, "risk_score", 0.0) # Scores from SmartRanker if rank: ci.rank_score = rank.total ci.rs_score = rank.rs_score * 5 # scale 0-20 → 0-100 ci.mtf_score = rank.mtf_score * 5 ci.sector_score = rank.sector_score * 5 ci.market_score = rank.market_score * 5 ci.liquidity_score = pkg.liquidity.liquidity_score # Overall confidence: weighted average of all components weights = { "technical": 0.25, "quality": 0.15, "risk": 0.10, "rs": 0.15, "mtf": 0.15, "sector": 0.10, "market": 0.05, "liquidity": 0.05, } raw_scores = { "technical": ci.technical_score, "quality": ci.quality_score, "risk": ci.risk_score, "rs": ci.rs_score, "mtf": ci.mtf_score, "sector": ci.sector_score, "market": ci.market_score, "liquidity": ci.liquidity_score, } ci.overall_confidence = round( sum(raw_scores[k] * w for k, w in weights.items()), 1 ) return ci # ── Section 10: News Sentiment ─────────────────────────────────────────── def _build_news(self, candidate: Any) -> NewsSentimentCtx: ctx = NewsSentimentCtx(available=False) if not _NEWS_AVAILABLE: return ctx try: summary = get_news_summary( symbol=candidate.symbol, company_name=getattr(candidate, "company_name", ""), ) ctx.available = True ctx.total_articles = summary.total_articles ctx.bullish_count = summary.bullish_count ctx.bearish_count = summary.bearish_count ctx.sentiment = summary.sentiment ctx.sentiment_score = summary.sentiment_score ctx.headlines = summary.headlines[:5] ctx.key_events = summary.key_events[:3] ctx.summary = summary.summary_text logger.info( f"[AICtx] News: {candidate.symbol} → {summary.sentiment} " f"({summary.total_articles} articles, score {summary.sentiment_score:+.2f})" ) except Exception as e: logger.debug(f"[AICtx] News unavailable for {candidate.symbol}: {e}") return ctx # ── Section 11: Open Interest ──────────────────────────────────────────── def _build_oi(self, symbol: str = "NIFTY") -> OICtx: ctx = OICtx(available=False, symbol=symbol) if not _NSE_AVAILABLE: return ctx try: from app.market.nse_data import get_option_chain_oi data = get_option_chain_oi(symbol) if data: ctx.available = True ctx.pcr = data.get("pcr", 0.0) ctx.total_call_oi = data.get("total_call_oi", 0) ctx.total_put_oi = data.get("total_put_oi", 0) ctx.max_call_oi_strike = data.get("max_call_oi_strike", 0) ctx.max_put_oi_strike = data.get("max_put_oi_strike", 0) ctx.oi_signal = data.get("oi_signal", "N/A") ctx.sentiment = data.get("sentiment", "") logger.info( f"[AICtx] OI: PCR={ctx.pcr:.3f} " f"Signal={ctx.oi_signal} " f"MaxCall={ctx.max_call_oi_strike} MaxPut={ctx.max_put_oi_strike}" ) except Exception as e: logger.debug(f"[AICtx] OI unavailable: {e}") return ctx # ── Section 12: Volume Analysis ────────────────────────────────────────── def _build_volume(self, candidate: Any) -> VolumeCtx: ctx = VolumeCtx(available=False, symbol=getattr(candidate, "symbol", "")) if not _VOLUME_AVAILABLE: return ctx try: va = analyse_volume(candidate.symbol) if va: ctx.available = True ctx.volume_ratio = va.volume_ratio ctx.volume_signal = va.volume_signal ctx.price_volume = va.price_volume ctx.volume_trend = va.volume_trend ctx.interpretation = va.interpretation logger.info( f"[AICtx] Volume: {candidate.symbol} → {va.volume_signal} " f"({va.volume_ratio:.1f}x) | {va.price_volume}" ) except Exception as e: logger.debug(f"[AICtx] Volume unavailable for {candidate.symbol}: {e}") return ctx # ── Section 13: Market Structure ───────────────────────────────────────── def _build_structure(self, candidate: Any) -> MarketStructureCtx: ctx = MarketStructureCtx(available=False, symbol=getattr(candidate, "symbol", "")) if not _STRUCTURE_AVAILABLE: return ctx try: ms = analyse_structure(candidate.symbol) if ms: ctx.available = True ctx.trend = ms.trend ctx.trend_score = ms.trend_score ctx.above_200ema = ms.above_200ema ctx.ema_200 = ms.ema_200 ctx.current_price = ms.current_price ctx.trade_bias = ms.trade_bias ctx.nearest_support = ms.nearest_support ctx.nearest_resistance = ms.nearest_resistance ctx.bos = ms.bos ctx.bos_direction = ms.bos_direction ctx.structure_pattern = ms.structure_pattern ctx.interpretation = ms.interpretation logger.info( f"[AICtx] Structure: {candidate.symbol} → {ms.trend} " f"| Bias={ms.trade_bias} | Above200EMA={ms.above_200ema} " f"| Score={ms.trend_score}" ) except Exception as e: logger.debug(f"[AICtx] Structure unavailable for {candidate.symbol}: {e}") return ctx # ── Singleton ───────────────────────────────────────────────────────────────── ai_context_builder = AIContextBuilder()