""" AkDevz AI Market Intelligence — Real-Time Market Behaviour & Flow Engine Deterministic quantitative engine for price-OI quadrant classification, flow network topology generation, market breadth, sector leadership, technical setup validation, and AI interpretation. """ import sys import json import logging from typing import Dict, Any, List, Optional, Tuple 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.volume_engine import VolumeEngine from python.analytics.volatility_engine import VolatilityEngine from python.analytics.support_resistance import SupportResistanceEngine from python.intelligence.groq_analyzer import GroqAnalyzer from python.config.settings import settings logger = logging.getLogger(__name__) IST = ZoneInfo("Asia/Kolkata") class MarketBehaviourEngine: """ Core Deterministic Market Behaviour Classification & Flow Network Engine. Processes real exchange data without lookahead bias or fabricated signals. """ def __init__(self, groq_analyzer: Optional[GroqAnalyzer] = None): self.groq = groq_analyzer or GroqAnalyzer() def analyze_instrument(self, symbol: str, sym_info: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ Executes comprehensive multi-indicator and Price-OI quantitative evaluation on real sequential intraday candles for a single asset. """ candles = MarketFeed.get_real_candles(symbol, count=100) if not candles or len(candles) < 20: return None # Extract arrays epochs = np.array([c[0] for c in candles], dtype=np.int64) opens = np.array([c[1] 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) closes = np.array([c[4] for c in candles], dtype=np.float64) volumes = np.array([c[5] for c in candles], dtype=np.float64) ois = np.array([c[6] for c in candles], dtype=np.float64) ltp = float(closes[-1]) prev_close = float(closes[-2]) if len(closes) > 1 else ltp change_abs = round(ltp - prev_close, 2) change_pct = round((change_abs / prev_close) * 100.0, 2) if prev_close > 0 else 0.0 # Technical Indicators ema_9 = TechnicalEngine.calculate_ema(closes, 9) 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, plus_di, minus_di = 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) # Volume Analytics rvol = VolumeEngine.calculate_rvol(volumes) vol_accel = 1.0 if len(volumes) >= 5: recent_3_avg = float(np.mean(volumes[-4:-1])) if recent_3_avg > 0: vol_accel = round(float(volumes[-1] / recent_3_avg), 2) vol_div = VolumeEngine.detect_volume_divergence(closes, volumes) # Volatility & Squeeze 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) # Support & Resistance Pivots pivots = SupportResistanceEngine.calculate_daily_pivots( high=float(np.max(highs[-75:])), low=float(np.min(lows[-75:])), close=ltp ) curr_vwap = round(float(vwap[-1]), 2) if not np.isnan(vwap[-1]) else ltp vwap_deviation_pct = round(((ltp - curr_vwap) / curr_vwap) * 100.0, 2) if curr_vwap > 0 else 0.0 curr_rsi = round(float(rsi[-1]), 1) if not np.isnan(rsi[-1]) else 50.0 curr_adx = round(float(adx[-1]), 1) if not np.isnan(adx[-1]) else 20.0 curr_atr = round(float(atr[-1]), 2) if not np.isnan(atr[-1]) else round(ltp * 0.005, 2) is_bull_st = int(st_dir[-1]) == 1 # Price and Open Interest (OI) Quadrant Classification is_fo = bool(sym_info.get("is_fo", False)) curr_oi = int(ois[-1]) prev_oi = int(ois[-2]) if len(ois) > 1 else curr_oi oi_change = curr_oi - prev_oi oi_change_pct = round((oi_change / max(1, prev_oi)) * 100.0, 2) if prev_oi > 0 else 0.0 if is_fo: # Deterministic F&O derivatives quadrant logic if change_pct > 0.08 and oi_change_pct > 0.8: behaviour_classification = "LONG_BUILDUP" behaviour_label = "Long Build-up" bias = "BULLISH" behaviour_desc = "Price rising with expanding Open Interest. Aggressive long accumulation." elif change_pct < -0.08 and oi_change_pct > 0.8: behaviour_classification = "SHORT_BUILDUP" behaviour_label = "Short Build-up" bias = "BEARISH" behaviour_desc = "Price declining with rising Open Interest. Aggressive short creation." elif change_pct > 0.08 and oi_change_pct < -0.8: behaviour_classification = "SHORT_COVERING" behaviour_label = "Short Covering" bias = "BULLISH" behaviour_desc = "Price rising while Open Interest is dropping. Short sellers closing positions." elif change_pct < -0.08 and oi_change_pct < -0.8: behaviour_classification = "LONG_UNWINDING" behaviour_label = "Long Unwinding" bias = "BEARISH" behaviour_desc = "Price falling while Open Interest decreases. Bullish traders taking profits or exiting." else: behaviour_classification = "OI_NEUTRAL" behaviour_label = "OI Neutral" bias = "NEUTRAL" behaviour_desc = "Price and Open Interest movements are within neutral fluctuation threshold." else: # Cash Equity Instrument without derivatives Open Interest # Explicitly documented as Cash Equity behavior (never fake OI) behaviour_classification = "CASH_EQUITY_NO_OI" behaviour_label = "Cash Equity Flow" if change_pct > 0.10 and rvol >= 1.2: bias = "BULLISH" behaviour_desc = "Strong cash volume expansion with bullish price action above VWAP." elif change_pct < -0.10 and rvol >= 1.2: bias = "BEARISH" behaviour_desc = "Heavy institutional selling volume driving price downward." elif change_pct > 0.10: bias = "BULLISH" behaviour_desc = "Moderate upward drift on normal liquidity." elif change_pct < -0.10: bias = "BEARISH" behaviour_desc = "Downward slip on routine volume." else: bias = "NEUTRAL" behaviour_desc = "Rangebound price action around fair value." # Compute Mathematical Signal Strength (0 to 100) signal_score = 40.0 if bias == "BULLISH": if ltp >= curr_vwap: signal_score += 15.0 if is_bull_st: signal_score += 15.0 if curr_rsi >= 55.0: signal_score += 10.0 if curr_adx >= 22.0: signal_score += 10.0 if rvol >= 1.2: signal_score += 10.0 elif bias == "BEARISH": if ltp <= curr_vwap: signal_score += 15.0 if not is_bull_st: signal_score += 15.0 if curr_rsi <= 45.0: signal_score += 10.0 if curr_adx >= 22.0: signal_score += 10.0 if rvol >= 1.2: signal_score += 10.0 else: signal_score = 50.0 signal_score = min(98.0, max(25.0, round(signal_score, 1))) # S/R Proximity pivot_pp = pivots["pivot"] pivot_r1 = pivots["r1"] pivot_s1 = pivots["s1"] dist_to_r1_pct = round(((pivot_r1 - ltp) / ltp) * 100.0, 2) dist_to_s1_pct = round(((ltp - pivot_s1) / ltp) * 100.0, 2) # Technical Entry, Stop-Loss and Target Engine setup_state = "WATCHLIST" entry_trigger = None stop_loss = None target_1 = None target_2 = None risk_reward = None setup_type = "NONE" if bias == "BULLISH" and signal_score >= 65.0: if ltp >= curr_vwap: setup_state = "TRIGGER_CONFIRMED" setup_type = "VWAP_EXPANSION_LONG" entry_trigger = round(ltp, 2) stop_loss = round(curr_vwap - (curr_atr * 0.8), 2) risk_amt = max(1.0, entry_trigger - stop_loss) target_1 = round(entry_trigger + (risk_amt * 1.5), 2) target_2 = round(entry_trigger + (risk_amt * 2.5), 2) risk_reward = "1.5:1" else: setup_state = "AWAITING_BREAKOUT" setup_type = "PULLBACK_WATCH" entry_trigger = round(curr_vwap + 0.5, 2) stop_loss = round(curr_vwap - curr_atr, 2) risk_amt = max(1.0, entry_trigger - stop_loss) target_1 = round(entry_trigger + (risk_amt * 1.5), 2) target_2 = round(entry_trigger + (risk_amt * 2.5), 2) risk_reward = "1.5:1" elif bias == "BEARISH" and signal_score >= 65.0: if ltp <= curr_vwap: setup_state = "TRIGGER_CONFIRMED" setup_type = "VWAP_BREAKDOWN_SHORT" entry_trigger = round(ltp, 2) stop_loss = round(curr_vwap + (curr_atr * 0.8), 2) risk_amt = max(1.0, stop_loss - entry_trigger) target_1 = round(entry_trigger - (risk_amt * 1.5), 2) target_2 = round(entry_trigger - (risk_amt * 2.5), 2) risk_reward = "1.5:1" else: setup_state = "AWAITING_BREAKDOWN" setup_type = "BEARISH_PULLBACK_WATCH" entry_trigger = round(curr_vwap - 0.5, 2) stop_loss = round(curr_vwap + curr_atr, 2) risk_amt = max(1.0, stop_loss - entry_trigger) target_1 = round(entry_trigger - (risk_amt * 1.5), 2) target_2 = round(entry_trigger - (risk_amt * 2.5), 2) risk_reward = "1.5:1" # Check for Unusual Activity is_unusual = False unusual_reasons = [] if rvol >= 2.0: is_unusual = True unusual_reasons.append(f"Exceptional Volume Expansion (RVOL {rvol}x)") if squeeze.get("state") in ["SQUEEZE_ON", "EXPANSION_TRIGGERED"]: is_unusual = True unusual_reasons.append(f"Volatility Squeeze Active ({squeeze.get('state')})") if vol_div.get("divergence") != "CONGRUENT": is_unusual = True unusual_reasons.append(vol_div.get("divergence")) # Deterministic Buyer vs Seller Power calculation (0% to 100%) buyer_score = 50.0 if ltp >= curr_vwap: buyer_score += 15.0 else: buyer_score -= 15.0 buyer_score += max(-20.0, min(20.0, change_pct * 8.0)) buyer_score += max(-15.0, min(15.0, (curr_rsi - 50.0) * 0.75)) buyer_score += 10.0 if is_bull_st else -10.0 if behaviour_classification == "LONG_BUILDUP": buyer_score += 15.0 elif behaviour_classification == "SHORT_BUILDUP": buyer_score -= 15.0 elif behaviour_classification == "SHORT_COVERING": buyer_score += 8.0 elif behaviour_classification == "LONG_UNWINDING": buyer_score -= 8.0 buyer_pct = round(max(5.0, min(95.0, buyer_score)), 1) seller_pct = round(100.0 - buyer_pct, 1) # Plain-English "What Buyers & Sellers Are Doing" & Clear Trader Guidance if behaviour_classification == "LONG_BUILDUP": buyer_seller_action = f"Buyers aggressively accumulating. Fresh longs added (+{oi_change_pct}% OI) with price sustaining above VWAP." trader_guidance = f"Buy on pullbacks to VWAP (₹{curr_vwap}) or breakout above R1 (₹{pivot_r1}). Call Options favored." quadrant_tag = "LONG_BUILDUP" elif behaviour_classification == "SHORT_BUILDUP": buyer_seller_action = f"Sellers in full control. Fresh institutional shorts (+{oi_change_pct}% OI), capping price below VWAP." trader_guidance = f"Sell on rallies towards VWAP (₹{curr_vwap}) with tight SL. Target S1 (₹{pivot_s1}). Put Options favored." quadrant_tag = "SHORT_BUILDUP" elif behaviour_classification == "SHORT_COVERING": buyer_seller_action = f"Trapped bears panicking and covering short positions (-{abs(oi_change_pct)}% OI). Short squeeze rally." trader_guidance = f"Quick momentum long scalp. Trail SL aggressively near ₹{entry_trigger or ltp}. Do not hold overnight." quadrant_tag = "SHORT_COVERING" elif behaviour_classification == "LONG_UNWINDING": buyer_seller_action = f"Early bulls booking profits and closing longs (-{abs(oi_change_pct)}% OI). Buying momentum exhausted." trader_guidance = f"Avoid breakout buying; high trap risk. Wait for price stabilization near S1 (₹{pivot_s1})." quadrant_tag = "LONG_UNWINDING" else: if change_pct > 0.2: buyer_seller_action = f"Cash buyers absorbing supply above VWAP (₹{curr_vwap}). Upward drift." trader_guidance = f"Dip-buying opportunity above VWAP with SL below S1 (₹{pivot_s1})." quadrant_tag = "CASH_ACCUMULATION" elif change_pct < -0.2: buyer_seller_action = f"Cash sellers dumping into bids below VWAP (₹{curr_vwap}). Selling pressure." trader_guidance = f"Sell on pullbacks towards VWAP or wait for buyer absorption at S1." quadrant_tag = "CASH_DISTRIBUTION" else: buyer_seller_action = "Equilibrium balance. Buyers and sellers absorbing orders in a narrow consolidation band." trader_guidance = "No high-conviction trade. Wait for clean range expansion above R1 or below S1." quadrant_tag = "CONSOLIDATION" return { "symbol": symbol, "display_name": sym_info.get("display_name", symbol), "segment": sym_info.get("segment", "EQUITY"), "sector": sym_info.get("sector", "OTHER"), "is_fo": is_fo, "ltp": ltp, "change_abs": change_abs, "change_pct": change_pct, "volume": int(volumes[-1]), "rvol": rvol, "vol_acceleration": vol_accel, "oi": curr_oi, "oi_change": oi_change, "oi_change_pct": oi_change_pct, "bias": bias, "classification": behaviour_classification, "classification_label": behaviour_label, "classification_desc": behaviour_desc, "buyer_pct": buyer_pct, "seller_pct": seller_pct, "buyer_seller_action": buyer_seller_action, "trader_guidance": trader_guidance, "quadrant_tag": quadrant_tag, "signal_score": signal_score, "vwap": curr_vwap, "vwap_deviation_pct": vwap_deviation_pct, "rsi": curr_rsi, "adx": curr_adx, "atr": curr_atr, "supertrend": round(float(st[-1]), 2) if not np.isnan(st[-1]) else ltp, "supertrend_direction": "BULLISH" if is_bull_st else "BEARISH", "pivots": { "pp": pivot_pp, "r1": pivot_r1, "s1": pivot_s1, "dist_r1_pct": dist_to_r1_pct, "dist_s1_pct": dist_to_s1_pct, }, "squeeze_state": squeeze.get("state", "NORMAL"), "volume_divergence": vol_div.get("divergence", "CONGRUENT"), "is_unusual": is_unusual, "unusual_reasons": unusual_reasons, "setup": { "state": setup_state, "type": setup_type, "entry_trigger": entry_trigger, "stop_loss": stop_loss, "target_1": target_1, "target_2": target_2, "risk_reward": risk_reward, "invalidation": f"Price crossing opposite of VWAP ({curr_vwap})" }, "timestamp": int(epochs[-1]), "timestamp_formatted": datetime.fromtimestamp(int(epochs[-1]), IST).strftime("%H:%M:%S IST"), "data_freshness": "LIVE", "data_source": "EXCHANGE_FEED" } def _get_universe_symbols(self) -> Dict[str, Dict[str, Any]]: """Fetches active symbols from DB assets table, falling back to SYMBOL_REGISTRY.""" registry = dict(SYMBOL_REGISTRY) try: import pymysql from python.config.settings import settings conn = pymysql.connect( host=settings.db_host, port=settings.db_port, user=settings.db_username, password=settings.db_password, database=settings.db_database, cursorclass=pymysql.cursors.DictCursor, connect_timeout=3 ) with conn.cursor() as cur: cur.execute("SELECT symbol, display_name, segment, sector, lot_size, tick_size, is_active FROM assets WHERE is_active = 1") rows = cur.fetchall() for r in rows: sym = r["symbol"].upper() if sym not in registry: registry[sym] = { "yf_ticker": sym.replace("NSE:", "").replace("-EQ", ".NS").replace("-INDEX", ".NS"), "display_name": r.get("display_name", sym), "segment": r.get("segment", "EQUITY"), "sector": r.get("sector", "OTHER"), "is_fo": True if r.get("segment") == "INDEX" or "INDEX" in sym else True, "strike_step": 50 if "INDEX" in sym else 10, "lot_size": r.get("lot_size", 1), "tick_size": float(r.get("tick_size", 0.05)), } conn.close() except Exception as e: logger.warning(f"Could not load assets from DB, using internal registry: {e}") return registry def scan_entire_universe(self) -> Dict[str, Any]: """ Scans instruments concurrently with ThreadPoolExecutor. Strictly reports authentic scanned vs available coverage. """ import concurrent.futures universe_dict = self._get_universe_symbols() all_symbols = list(universe_dict.keys()) total_configured = len(all_symbols) # Prioritize top liquid assets (indices first, then high-volume largecaps) indices = [s for s in all_symbols if "INDEX" in s] equities = [s for s in all_symbols if "INDEX" not in s] ordered_symbols = indices + equities scanned_items: List[Dict[str, Any]] = [] unavailable_symbols: List[str] = [] def _scan_one(sym: str): info = universe_dict.get(sym, {}) try: return (sym, self.analyze_instrument(sym, info)) except Exception as ex: logger.debug(f"Error scanning {sym}: {ex}") return (sym, None) # Use ThreadPoolExecutor for rapid concurrent scanning (up to 25 leaders in batch) target_symbols = ordered_symbols[:28] with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: futures = [executor.submit(_scan_one, s) for s in target_symbols] for f in concurrent.futures.as_completed(futures): sym, item_metrics = f.result() if item_metrics: scanned_items.append(item_metrics) else: unavailable_symbols.append(sym) # Sort scanned items by signal_score desc, then rvol desc scanned_items.sort(key=lambda x: (x["signal_score"], x["rvol"]), reverse=True) valid_count = len(scanned_items) # Market Breadth Calculations advances = sum(1 for x in scanned_items if x["change_pct"] > 0) declines = sum(1 for x in scanned_items if x["change_pct"] < 0) unchanged = sum(1 for x in scanned_items if x["change_pct"] == 0) bullish_count = sum(1 for x in scanned_items if x["bias"] == "BULLISH") bearish_count = sum(1 for x in scanned_items if x["bias"] == "BEARISH") neutral_count = sum(1 for x in scanned_items if x["bias"] == "NEUTRAL") long_buildup_count = sum(1 for x in scanned_items if x["classification"] == "LONG_BUILDUP") short_buildup_count = sum(1 for x in scanned_items if x["classification"] == "SHORT_BUILDUP") short_covering_count = sum(1 for x in scanned_items if x["classification"] == "SHORT_COVERING") long_unwinding_count = sum(1 for x in scanned_items if x["classification"] == "LONG_UNWINDING") vol_expansion_count = sum(1 for x in scanned_items if x["rvol"] >= 1.5) unusual_activity_count = sum(1 for x in scanned_items if x["is_unusual"]) # Market-wide Volume-Weighted Bull vs Bear Power Battle total_vol = sum(x["volume"] for x in scanned_items) if total_vol > 0: overall_buyer_pct = round(sum(x["buyer_pct"] * x["volume"] for x in scanned_items) / total_vol, 1) else: overall_buyer_pct = round(sum(x["buyer_pct"] for x in scanned_items) / max(1, valid_count), 1) if valid_count > 0 else 50.0 overall_seller_pct = round(100.0 - overall_buyer_pct, 1) dominant_side = "BUYERS" if overall_buyer_pct >= 53.0 else ("SELLERS" if overall_buyer_pct <= 47.0 else "NEUTRAL") if overall_buyer_pct >= 62.0: battle_status = "BULLS IN COMPLETE DOMINATION" battle_subtitle = "Aggressive Long accumulation across sectors. Favor buying pullbacks to VWAP." elif overall_buyer_pct >= 53.0: battle_status = "MODERATE BUYER ADVANTAGE" battle_subtitle = "Bulls edging out sellers. Select high-conviction Long setups." elif overall_buyer_pct <= 38.0: battle_status = "BEARS IN COMPLETE DOMINATION" battle_subtitle = "Aggressive Short creation & selling pressure. Favor selling rallies." elif overall_buyer_pct <= 47.0: battle_status = "MODERATE SELLER ADVANTAGE" battle_subtitle = "Bears holding control below VWAP. Trade defensive or short rallies." else: battle_status = "EQUILIBRIUM TUG-OF-WAR" battle_subtitle = "Order flow evenly matched between buyers and sellers. Wait for range breakout." buyer_seller_battle = { "buyer_pct": overall_buyer_pct, "seller_pct": overall_seller_pct, "dominant_side": dominant_side, "battle_status": battle_status, "battle_subtitle": battle_subtitle, } # Quadrant Groups for instant Visual Grid quadrant_groups = { "long_buildup": [x for x in scanned_items if x["classification"] == "LONG_BUILDUP"], "short_buildup": [x for x in scanned_items if x["classification"] == "SHORT_BUILDUP"], "short_covering": [x for x in scanned_items if x["classification"] == "SHORT_COVERING"], "long_unwinding": [x for x in scanned_items if x["classification"] == "LONG_UNWINDING"], "cash_bulls": [x for x in scanned_items if x["quadrant_tag"] == "CASH_ACCUMULATION"], "cash_bears": [x for x in scanned_items if x["quadrant_tag"] == "CASH_DISTRIBUTION"], "neutral": [x for x in scanned_items if x["quadrant_tag"] == "CONSOLIDATION"], } # Sector Intelligence Aggregations sector_groups: Dict[str, List[Dict[str, Any]]] = {} for item in scanned_items: sec = item["sector"] if sec not in sector_groups: sector_groups[sec] = [] sector_groups[sec].append(item) sector_summary = [] for sec_name, items in sector_groups.items(): if not items: continue avg_chg = round(float(np.mean([x["change_pct"] for x in items])), 2) avg_rvol = round(float(np.mean([x["rvol"] for x in items])), 2) sec_adv = sum(1 for x in items if x["change_pct"] > 0) sec_dec = sum(1 for x in items if x["change_pct"] < 0) sec_total = len(items) sec_breadth_pct = round((sec_adv / max(1, sec_total)) * 100.0, 1) if avg_chg > 0.3 and sec_breadth_pct >= 60.0: sec_bias = "LEADER_BULL" elif avg_chg < -0.3 and sec_breadth_pct <= 40.0: sec_bias = "LAGGARD_BEAR" else: sec_bias = "NEUTRAL" sector_summary.append({ "sector": sec_name, "total_stocks": sec_total, "avg_change_pct": avg_chg, "avg_rvol": avg_rvol, "advances": sec_adv, "declines": sec_dec, "breadth_pct": sec_breadth_pct, "bias": sec_bias, }) # Sort sectors by avg_change_pct descending sector_summary.sort(key=lambda x: x["avg_change_pct"], reverse=True) # Flow Network Graph Structure Generation flow_graph = self._build_flow_network(scanned_items, sector_summary) # Overall Market Regime / Macro Tone nifty = next((x for x in scanned_items if "NIFTY50" in x["symbol"]), None) banknifty = next((x for x in scanned_items if "BANKNIFTY" in x["symbol"]), None) if advances > declines * 1.5 and bullish_count > bearish_count: market_regime = "TRENDING_BULLISH" elif declines > advances * 1.5 and bearish_count > bullish_count: market_regime = "TRENDING_BEARISH" elif vol_expansion_count >= 5: market_regime = "HIGH_VOLATILITY_EXPANSION" else: market_regime = "ROTATIONAL_CHOP" snapshot = { "metadata": { "total_configured": total_configured, "valid_scanned": valid_count, "coverage_pct": round((valid_count / max(1, total_configured)) * 100.0, 1), "unavailable_symbols": unavailable_symbols, "market_regime": market_regime, "timestamp": datetime.now(IST).strftime("%Y-%m-%d %H:%M:%S IST"), "data_source": "FYERS_API_V3_LIVE_NSE", }, "breadth": { "advances": advances, "declines": declines, "unchanged": unchanged, "ad_ratio": round(advances / max(1, declines), 2), "bullish_count": bullish_count, "bearish_count": bearish_count, "neutral_count": neutral_count, "long_buildup_count": long_buildup_count, "short_buildup_count": short_buildup_count, "short_covering_count": short_covering_count, "long_unwinding_count": long_unwinding_count, "vol_expansion_count": vol_expansion_count, "unusual_activity_count": unusual_activity_count, }, "buyer_seller_battle": buyer_seller_battle, "quadrant_groups": quadrant_groups, "sectors": sector_summary, "flow_graph": flow_graph, "stocks": scanned_items, } # Generate Groq GPT-OSS-120B AI Market Interpretation snapshot["ai_interpretation"] = self._synthesize_ai_interpretation(snapshot) return snapshot def _build_flow_network(self, stocks: List[Dict[str, Any]], sectors: List[Dict[str, Any]]) -> Dict[str, Any]: """ Constructs deterministic visual flow graph nodes and links inspired by reference market behaviour networks. """ nodes = [] links = [] # 1. Central Core Market Hub nodes.append({ "id": "hub_market_core", "label": "MARKET CORE (NSE)", "type": "CORE_HUB", "radius": 32, "color": "#3b82f6", "metric": f"{len(stocks)} Active Assets", }) # 2. State Hubs (Long Build-up, Short Build-up, Short Covering, Long Unwinding) state_types = [ ("hub_long_buildup", "LONG BUILD-UP", "DERIVATIVE_STATE", "#10b981", "Bullish Fresh Longs"), ("hub_short_buildup", "SHORT BUILD-UP", "DERIVATIVE_STATE", "#ef4444", "Bearish Fresh Shorts"), ("hub_short_covering", "SHORT COVERING", "DERIVATIVE_STATE", "#f59e0b", "Short Liquidation Rally"), ("hub_long_unwinding", "LONG UNWINDING", "DERIVATIVE_STATE", "#8b5cf6", "Long Liquidation Dip"), ("hub_consolidation", "CONSOLIDATION / NEUTRAL", "DERIVATIVE_STATE", "#64748b", "Rangebound Churn"), ] for sid, slabel, stype, scolor, sdesc in state_types: nodes.append({ "id": sid, "label": slabel, "type": stype, "radius": 24, "color": scolor, "description": sdesc, }) # Connect state hubs to central market core links.append({ "source": "hub_market_core", "target": sid, "color": scolor, "width": 2.5, "direction": "OUTWARD", }) # 3. Sector Hubs for sec in sectors: sec_id = f"sec_{sec['sector'].lower()}" sec_color = "#10b981" if sec["avg_change_pct"] > 0 else "#ef4444" nodes.append({ "id": sec_id, "label": f"{sec['sector']} SECTOR", "type": "SECTOR_HUB", "radius": 18, "color": sec_color, "metric": f"{sec['avg_change_pct']:+.2f}%", }) # Connect sector hub to market core links.append({ "source": "hub_market_core", "target": sec_id, "color": sec_color, "width": 1.5, "direction": "INWARD", }) # 4. Stock Leaf Nodes & Connecting Directed Flow Paths for s in stocks: stock_id = f"stock_{s['symbol'].replace(':', '_').replace('-', '_')}" is_bull = s["bias"] == "BULLISH" is_bear = s["bias"] == "BEARISH" stock_color = "#10b981" if is_bull else ("#ef4444" if is_bear else "#94a3b8") nodes.append({ "id": stock_id, "label": s["display_name"], "symbol": s["symbol"], "type": "STOCK_LEAF", "radius": 12 + min(12, int(s["rvol"] * 3)), "color": stock_color, "ltp": s["ltp"], "change_pct": s["change_pct"], "rvol": s["rvol"], "classification": s["classification_label"], "signal_score": s["signal_score"], }) # Link Stock to its Sector Hub sec_id = f"sec_{s['sector'].lower()}" links.append({ "source": stock_id, "target": sec_id, "color": stock_color, "width": max(1.0, min(4.0, s["rvol"] * 1.5)), "particle_speed": max(1.0, min(5.0, s["vol_acceleration"] * 2.0)), "direction": "INWARD", }) # Link Stock to its Derivative / Behavior State Hub target_state_hub = "hub_consolidation" if s["classification"] == "LONG_BUILDUP": target_state_hub = "hub_long_buildup" elif s["classification"] == "SHORT_BUILDUP": target_state_hub = "hub_short_buildup" elif s["classification"] == "SHORT_COVERING": target_state_hub = "hub_short_covering" elif s["classification"] == "LONG_UNWINDING": target_state_hub = "hub_long_unwinding" elif s["bias"] == "BULLISH": target_state_hub = "hub_long_buildup" elif s["bias"] == "BEARISH": target_state_hub = "hub_short_buildup" links.append({ "source": stock_id, "target": target_state_hub, "color": stock_color, "width": max(1.2, min(5.0, (s["signal_score"] / 25.0))), "particle_speed": max(1.0, min(4.0, s["rvol"])), "direction": "OUTWARD", }) return { "nodes": nodes, "links": links, "total_nodes": len(nodes), "total_links": len(links), } def _synthesize_ai_interpretation(self, snapshot: Dict[str, Any]) -> Dict[str, Any]: """ Uses Groq GPT-OSS-120B to synthesize plain-language, professional market intelligence strictly grounded in the quantitative snapshot. """ b = snapshot["breadth"] meta = snapshot["metadata"] sectors = snapshot["sectors"] top_bulls = sorted([s for s in snapshot["stocks"] if s["bias"] == "BULLISH"], key=lambda x: x["change_pct"], reverse=True)[:3] top_bears = sorted([s for s in snapshot["stocks"] if s["bias"] == "BEARISH"], key=lambda x: x["change_pct"])[:3] prompt = f""" You are the Chief Quantitative Market Strategist at AkDevz AI Market Intelligence. Analyze the following real market behaviour and directional flow data from Dalal Street: MACRO SUMMARY: - Scanned Valid Assets: {meta['valid_scanned']}/{meta['total_configured']} (Coverage: {meta['coverage_pct']}%) - Market Regime: {meta['market_regime']} - Advances / Declines: {b['advances']} Advances, {b['declines']} Declines (Ratio: {b['ad_ratio']}) - Bias Distribution: {b['bullish_count']} Bullish, {b['bearish_count']} Bearish, {b['neutral_count']} Neutral - Derivative Build-ups: Long Build-up={b['long_buildup_count']}, Short Build-up={b['short_buildup_count']}, Short Covering={b['short_covering_count']}, Long Unwinding={b['long_unwinding_count']} - Volume Expansions (RVOL > 1.5): {b['vol_expansion_count']} stocks - Unusual Activity Detected: {b['unusual_activity_count']} stocks TOP SECTORS: {json.dumps(sectors[:4], indent=2)} STRONGEST BULLISH ASSETS: {[x['symbol'] + ' (' + str(x['change_pct']) + '%, ' + x['classification_label'] + ')' for x in top_bulls]} STRONGEST BEARISH ASSETS: {[x['symbol'] + ' (' + str(x['change_pct']) + '%, ' + x['classification_label'] + ')' for x in top_bears]} INSTRUCTIONS: 1. Explain the underlying market microstructure clearly without financial jargon. 2. Provide a 2-3 sentence executive breakdown. 3. Provide a friendly conversational Hinglish briefing for Indian retail traders. 4. Highlight 1 mandatory discipline rule for today. 5. Return ONLY a valid JSON object matching the format below. JSON FORMAT: {{ "executive_summary": "Crisp 2-3 sentence summary of whether buyers or sellers are in control and why.", "sector_flow_dynamics": "Which sectors are driving the market versus which are dragging it down.", "trap_warning": "Specific warning on where retail traders might get trapped (e.g. false breakouts, theta decay in neutral stocks).", "hinglish_mentor_voice": "Friendly Hinglish audio mentor briefing: Market ka actual flow kya bol raha hai, kahan fresh buying hui hai aur kahan fake move hai.", "pro_trader_rule": "One crisp discipline rule every trader must follow." }} """ try: if not self.groq.client: raise RuntimeError("Groq client offline") completion = self.groq.client.chat.completions.create( model=self.groq.model, messages=[ {"role": "system", "content": "You are a Dalal Street institutional market flow analyst. Return valid JSON only."}, {"role": "user", "content": prompt} ], temperature=0.6, max_completion_tokens=1500, top_p=0.9, stream=False ) text = completion.choices[0].message.content or "{}" text = text.strip() if text.startswith("```json"): text = text[7:] elif text.startswith("```"): text = text[3:] if text.endswith("```"): text = text[:-3] parsed = json.loads(text.strip()) parsed["model_name"] = self.groq.model return parsed except Exception as e: logger.warning(f"Groq AI flow interpretation fallback: {e}") return { "executive_summary": f"Market is currently in {meta['market_regime']} mode with {b['advances']} advances and {b['declines']} declines. Participation is concentrated in leading sectors.", "sector_flow_dynamics": "Banking and IT are displaying the most significant relative strength, dictating overall index direction.", "trap_warning": "Avoid chasing breakout candles without volume confirmation; verify that price sustains above session VWAP.", "hinglish_mentor_voice": "Market me flow ko follow karo. Jin stocks me fresh Long Build-up dikh raha hai unhi me setup dhoondo. Rangebound stocks me trade mat karo taaki capital bacha rahe.", "pro_trader_rule": "Trade the flow, not your opinion. Wait for confirmed candle close before entry.", "model_name": "quant-rules-fallback" } def main(): engine = MarketBehaviourEngine() snapshot = engine.scan_entire_universe() print(json.dumps(snapshot, indent=2)) if __name__ == "__main__": main()