""" AkDevz AI Market Intelligence — Pre-Market AI Research Engine Powered by Groq's openai/gpt-oss-120b. Conducts multi-asset pre-market screening across NIFTY, BANK NIFTY, and key F&O leaders. Identifies: 1. Best Assets to Trade Today (Highest probability setups) 2. Avoid / Do Not Trade Today (Chop, traps, theta decay zones) 3. Beginner Copilot Blueprint & Hinglish Mentor Briefing """ import sys import json import logging from typing import Dict, Any, List from pathlib import Path from datetime import datetime from zoneinfo import ZoneInfo import numpy as np PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from python.data.market_feed import MarketFeed, SYMBOL_REGISTRY from python.analytics.technical_engine import TechnicalEngine from python.analytics.support_resistance import SupportResistanceEngine from python.analytics.volatility_engine import VolatilityEngine from python.analytics.volume_engine import VolumeEngine from python.intelligence.groq_analyzer import GroqAnalyzer from python.config.settings import settings logger = logging.getLogger(__name__) IST = ZoneInfo("Asia/Kolkata") class PreMarketResearcher: # Core universe for pre-market morning scan CORE_UNIVERSE = [ "NSE:NIFTY50-INDEX", "NSE:BANKNIFTY-INDEX", "NSE:RELIANCE-EQ", "NSE:HDFCBANK-EQ", "NSE:ICICIBANK-EQ", "NSE:INFY-EQ", "NSE:TCS-EQ", "NSE:SBIN-EQ" ] def __init__(self, groq_analyzer: GroqAnalyzer = None): self.groq = groq_analyzer or GroqAnalyzer() def scan_asset_metrics(self, symbol: str) -> Dict[str, Any]: """Calculates authentic quantitative metrics for a single asset.""" info = MarketFeed.resolve_symbol_info(symbol) candles = MarketFeed.get_real_candles(symbol, count=100) if not candles or len(candles) < 20: return {} closes = np.array([c[4] for c in candles], dtype=np.float64) highs = np.array([c[2] for c in candles], dtype=np.float64) lows = np.array([c[3] for c in candles], dtype=np.float64) volumes = np.array([c[5] for c in candles], dtype=np.float64) ltp = float(closes[-1]) prev_close = float(closes[-2]) if len(closes) > 1 else ltp change_pct = round(((ltp - prev_close) / prev_close) * 100.0, 2) if prev_close > 0 else 0.0 ema_20 = TechnicalEngine.calculate_ema(closes, 20) ema_50 = TechnicalEngine.calculate_ema(closes, 50) vwap = TechnicalEngine.calculate_vwap(highs, lows, closes, volumes) rsi = TechnicalEngine.calculate_rsi(closes, 14) adx, _, _ = TechnicalEngine.calculate_adx(highs, lows, closes, 14) atr = TechnicalEngine.calculate_atr(highs, lows, closes, 14) st, st_dir = TechnicalEngine.calculate_supertrend(highs, lows, closes, 10, 3.0) rvol = VolumeEngine.calculate_rvol(volumes) bb_mid, bb_up, bb_low, _ = VolatilityEngine.calculate_bollinger_bands(closes) kc_mid, kc_up, kc_low = VolatilityEngine.calculate_keltner_channels(highs, lows, closes) squeeze = VolatilityEngine.detect_volatility_squeeze(bb_up, bb_low, kc_up, kc_low) pivots = SupportResistanceEngine.calculate_daily_pivots( high=float(np.max(highs[-75:])), low=float(np.min(lows[-75:])), close=ltp ) return { "symbol": symbol, "display_name": info["display_name"], "segment": info["segment"], "ltp": ltp, "change_pct": change_pct, "vwap": round(float(vwap[-1]), 2) if not np.isnan(vwap[-1]) else ltp, "above_vwap": bool(ltp >= vwap[-1]) if not np.isnan(vwap[-1]) else True, "ema_20": round(float(ema_20[-1]), 2) if not np.isnan(ema_20[-1]) else ltp, "ema_50": round(float(ema_50[-1]), 2) if not np.isnan(ema_50[-1]) else ltp, "rsi": round(float(rsi[-1]), 1) if not np.isnan(rsi[-1]) else 50.0, "adx": round(float(adx[-1]), 1) if not np.isnan(adx[-1]) else 20.0, "atr": round(float(atr[-1]), 2) if not np.isnan(atr[-1]) else round(ltp * 0.005, 2), "supertrend_direction": "BULLISH" if int(st_dir[-1]) == 1 else "BEARISH", "rvol": round(rvol, 2), "squeeze_state": squeeze.get("state", "NORMAL"), "pivot_r1": pivots["r1"], "pivot_s1": pivots["s1"], "pivot_pp": pivots["pivot"], } def generate_premarket_report(self) -> Dict[str, Any]: """ Executes pre-market screening over the universe and prompts Groq GPT-OSS-120B. """ universe_stats = [] for sym in self.CORE_UNIVERSE: stats = self.scan_asset_metrics(sym) if stats: universe_stats.append(stats) # Build prompt for Groq GPT prompt = f""" You are the Head of Institutional Trading and AI Mentor at AkDevz AI Market Intelligence. Market is about to open (or conducting pre-market scan). Below are the real quantitative metrics for key Indian Benchmark Indices and F&O market leaders. REAL MARKET SCREENING DATA: {json.dumps(universe_stats, indent=2)} YOUR MISSION: Act as a seasoned, elite Derivatives Trader (20+ years on Dalal Street). Provide a crystal-clear, beginner-friendly Pre-Market Intelligence Briefing. STRICT INSTRUCTIONS: 1. Ground your analysis strictly on the quantitative numbers above (LTP, VWAP, RSI, ADX, Pivots, Squeeze). 2. Categorize the instruments into: - "best_assets_to_trade": Exactly 2 to 3 instruments with the cleanest setups, high momentum, clear support/resistance, or volatility breakout potential. - "avoid_assets_today": Exactly 2 to 3 instruments that are choppy, stuck in range, showing low ADX (<20), or high risk of theta decay for option buyers. 3. Provide an intuitive "hinglish_mentor_voice" explanation so even an absolute beginner knows what to do today. 4. Provide a single "golden_rule_today". 5. Return ONLY a valid JSON object matching the exact format below. JSON FORMAT: {{ "market_overview": {{ "overall_bias": "BULLISH or BEARISH or RANGEBOUND_CHOPPY", "tone_description": "2 sentence executive outlook for Indian markets today based on Nifty and BankNifty data.", "nifty_key_levels": "Resistance at X, Support at Y, Pivot at Z", "banknifty_key_levels": "Resistance at X, Support at Y, Pivot at Z" }}, "best_assets_to_trade": [ {{ "symbol": "NSE:NIFTY50-INDEX", "display_name": "NIFTY 50", "trade_bias": "CALL BUY or PUT BUY or PULLBACK BUY", "suggested_strike": "e.g. 23100 CE", "entry_trigger": "Exact condition, e.g. 5m close above 23080", "stop_loss": 23000.0, "target_1": 23150.0, "target_2": 23220.0, "why_best_today": "Clear reasoning citing RSI/ADX/VWAP alignment.", "beginner_tip": "Plain simple advice for beginners." }} ], "avoid_assets_today": [ {{ "symbol": "NSE:BANKNIFTY-INDEX", "display_name": "BANK NIFTY", "danger_type": "THETA_DECAY / CHOPPY_RANGE / FALSE_BREAKOUT", "why_avoid": "Why beginners will lose money here (e.g. low ADX, oscillating around VWAP).", "safe_condition_to_reconsider": "What must happen before considering a trade." }} ], "golden_rule_today": "One mandatory discipline rule every trader must follow today.", "hinglish_mentor_voice": "Friendly Hinglish audio briefing explaining today's gameplan in plain words: Aaj market me kya plan rakhna hai, kahan paisa banega aur kahan loss se bachna hai.", "timestamp": "{datetime.now(IST).strftime('%Y-%m-%d %H:%M:%S IST')}", "model_name": "groq/openai/gpt-oss-120b" }} """ try: if not self.groq.client: raise RuntimeError("Groq client not initialized") completion = self.groq.client.chat.completions.create( model=self.groq.model, messages=[ {"role": "system", "content": "You are an institutional Indian derivatives strategist and beginner trading copilot. Return valid JSON only."}, {"role": "user", "content": prompt} ], temperature=0.7, max_completion_tokens=2500, top_p=0.9, reasoning_effort="medium", stream=False ) resp_text = completion.choices[0].message.content or "{}" resp_text = resp_text.strip() if resp_text.startswith("```json"): resp_text = resp_text[7:] elif resp_text.startswith("```"): resp_text = resp_text[3:] if resp_text.endswith("```"): resp_text = resp_text[:-3] parsed = json.loads(resp_text.strip()) parsed["universe_stats"] = universe_stats return parsed except Exception as e: logger.error(f"Pre-market AI generation failed: {e}") # Reliable deterministic fallback return self._fallback_premarket(universe_stats) def _fallback_premarket(self, stats: List[Dict[str, Any]]) -> Dict[str, Any]: """Deterministic fallback if AI API is temporarily unavailable.""" nifty = next((s for s in stats if "NIFTY50" in s["symbol"]), stats[0] if stats else {}) best = [s for s in stats if s.get("adx", 0) > 20 and s.get("above_vwap")][:2] avoid = [s for s in stats if s.get("adx", 0) <= 20][:2] return { "market_overview": { "overall_bias": "BULLISH" if nifty.get("above_vwap") else "CAUTIOUS", "tone_description": f"Markets are consolidating near fair value. NIFTY is trading at {nifty.get('ltp', 23053)} with RSI at {nifty.get('rsi', 55)}.", "nifty_key_levels": f"Pivot: {nifty.get('pivot_pp', 23050)}, R1: {nifty.get('pivot_r1', 23120)}, S1: {nifty.get('pivot_s1', 22980)}", "banknifty_key_levels": "Key Pivot 54,600, Resistance 54,850, Support 54,350" }, "best_assets_to_trade": [ { "symbol": b.get("symbol", "NSE:NIFTY50-INDEX"), "display_name": b.get("display_name", "NIFTY 50"), "trade_bias": "PULLBACK BUY" if b.get("above_vwap") else "BREAKDOWN SELL", "suggested_strike": f"{int(round(b.get('ltp', 23050)/50)*50)} CE", "entry_trigger": f"Enter on 5m candle close above {b.get('ltp', 23050)}", "stop_loss": round(b.get("ltp", 23050) - (b.get("atr", 40) * 1.2), 2), "target_1": round(b.get("ltp", 23050) + (b.get("atr", 40) * 1.5), 2), "target_2": round(b.get("ltp", 23050) + (b.get("atr", 40) * 2.5), 2), "why_best_today": f"Strong trend with ADX {b.get('adx', 22)} and price holding above VWAP.", "beginner_tip": "Do not rush into trade at 9:15 AM; wait for first 15m candle confirmation." } for b in best ], "avoid_assets_today": [ { "symbol": a.get("symbol", "NSE:BANKNIFTY-INDEX"), "display_name": a.get("display_name", "BANK NIFTY"), "danger_type": "CHOPPY_RANGE", "why_avoid": f"ADX is compressed at {a.get('adx', 18)}, indicating no trend and high theta decay risk.", "safe_condition_to_reconsider": "Wait for breakout above day's first hour high with RVOL > 1.5." } for a in avoid ], "golden_rule_today": "Never average a losing option position. If price breaks stop-loss, exit immediately.", "hinglish_mentor_voice": "Aaj market me patient rehna sabse zaroori hai. Sirf unhi stocks me trade karo jo VWAP ke upar strong volume ke saath chal rahe hain. Sideways stocks se door raho taaki premium decay na ho.", "timestamp": datetime.now(IST).strftime("%Y-%m-%d %H:%M:%S IST"), "model_name": "quant-rules-fallback", "universe_stats": stats } def main(): researcher = PreMarketResearcher() report = researcher.generate_premarket_report() print(json.dumps(report, indent=2)) if __name__ == "__main__": main()