import logging import os from datetime import datetime from typing import Any, Dict, List, Optional from fastapi import HTTPException from app.v1.services.zerodha.client import ZerodhaClient from app.v1.services.gpt_engine import prepare_market_data_prompt, call_chatgpt_analysis from app.v1.services.tegpt.config import TEGPT_VERBOSE_LOGS from app.v1.services.tegpt.zerodha_services import fetch_market_data from app.v1.utils.snapshot_sanitize import compact_analysis_for_persistence logger = logging.getLogger(__name__) def analyze_symbol_service( db, zerodha_client: ZerodhaClient, symbol: str, timeframes: List[str], question: str, context: str, user_id: str, include_market_data: bool = False, ) -> Dict[str, Any]: """Analyze single symbol. Logging is intentionally minimal at INFO. Enable verbose tracing via `TEGPT_VERBOSE_LOGS=true`. """ try: symbol = (symbol or "").strip().upper() # ---- Cache: reuse recent stored analysis instead of calling GPT repeatedly ---- cache_minutes = int( os.getenv( f"GPT_ANALYSIS_CACHE_MINUTES_{str(context or '').strip().upper()}", os.getenv( "GPT_ANALYSIS_CACHE_MINUTES_DEFAULT", os.getenv("GPT_ANALYSIS_CACHE_MINUTES", "0"), ), ) ) if cache_minutes > 0: try: key_timeframes = sorted([str(tf) for tf in (timeframes or [])]) cached = db["analyses"].find_one( { "symbol": symbol, "user_id": user_id, "context": context, "question": question, "timeframes": key_timeframes, }, sort=[("timestamp", -1), ("_id", -1)], ) if cached and isinstance(cached.get("timestamp"), datetime): age_s = (datetime.utcnow() - cached["timestamp"]).total_seconds() if age_s >= 0 and age_s <= float(cache_minutes) * 60.0: cached_analysis = cached.get("analysis") or {} if isinstance(cached_analysis, dict) and cached_analysis: cached_analysis = dict(cached_analysis) cached_analysis["analysis_id"] = str(cached.get("_id")) cached_analysis["timestamp"] = cached["timestamp"].isoformat() cached_analysis["symbol"] = symbol cached_analysis["cache"] = {"hit": True, "age_seconds": round(age_s, 1)} if include_market_data: md = cached.get("market_data") if isinstance(md, dict) and md: cached_analysis["market_data"] = md logger.info("♻️ USING CACHED GPT ANALYSIS for %s (age=%.1fs)", symbol, age_s) return cached_analysis except Exception: logger.exception("Cache lookup failed for %s", symbol) logger.info("🔍 STARTING ANALYSIS for %s", symbol) timeframes = sorted([str(tf) for tf in (timeframes or [])]) market_data = fetch_market_data(zerodha_client, symbol, timeframes, db=db) # Never persist raw candles/quote snapshots. def _compact_market_data(md: Dict[str, Any]) -> Dict[str, Any]: if not isinstance(md, dict): return {} out: Dict[str, Any] = {} for k in ( "instrument_token", "stock_id", "indicators", "strategies", "pivots", "fib", "error", ): if k in md and md.get(k) is not None: out[k] = md.get(k) return out market_data_compact = _compact_market_data(market_data) for tf, candles in market_data.get("candles", {}).items(): if candles: logger.info( " 📊 %s: %d candles, Latest: O:%s, C:%s", str(tf).upper(), len(candles), (candles[-1] or {}).get("open"), (candles[-1] or {}).get("close"), ) if market_data.get("error"): raise HTTPException(status_code=400, detail=f"Market data error: {market_data['error']}") logger.info("🤖 CHATGPT ANALYSIS STARTING for %s", symbol) def _as_utc_dt(v): if v is None: return None if isinstance(v, datetime): return v if isinstance(v, str): try: return datetime.fromisoformat(v.replace("Z", "+00:00")).replace(tzinfo=None) except Exception: return None return None def _get_latest_market_intelligence() -> Dict[str, Any]: """Best-effort broad market context for the GPT prompt.""" try: doc = db["market_intelligence_summary"].find_one( {"type": "latest"}, sort=[("updated_at", -1), ("captured_at", -1), ("_id", -1)], ) if not doc: return {} payload = doc.get("payload") or {} ts = _as_utc_dt(doc.get("captured_at") or doc.get("updated_at") or doc.get("created_at")) def _norm_key(v: Any) -> str: s = str(v or "").strip().upper() out_chars = [] for ch in s: out_chars.append(ch if ch.isalnum() else " ") return " ".join("".join(out_chars).split()) def _best_sector_match(sector_value: str, sectors_debug: Dict[str, Any]) -> Optional[str]: if not sector_value or not isinstance(sectors_debug, dict) or not sectors_debug: return None target = _norm_key(sector_value) if not target: return None target_tokens = set(target.split()) best = None best_score = 0 for k in sectors_debug.keys(): kk = _norm_key(k) if not kk: continue kk_tokens = set(kk.split()) overlap = len(target_tokens & kk_tokens) if any(t in kk for t in target_tokens): overlap += 1 if overlap > best_score: best_score = overlap best = k return best if best_score >= 1 else None out: Dict[str, Any] = {} if ts: out["as_of"] = ts.isoformat() + "Z" if isinstance(payload, dict): if payload.get("market_bias") is not None: out["market_bias"] = payload.get("market_bias") if payload.get("overall_risk") is not None: out["risk_regime"] = payload.get("overall_risk") if payload.get("volatility_state") is not None: out["volatility_regime"] = payload.get("volatility_state") if payload.get("sector_strength") is not None: out["sector_strength"] = payload.get("sector_strength") try: stock = db["stocks"].find_one( {"symbol": symbol}, {"_id": 0, "symbol": 1, "sector": 1, "industry": 1, "name": 1}, ) except Exception: stock = None if isinstance(stock, dict): sector_val = (stock.get("sector") or "").strip() industry_val = (stock.get("industry") or "").strip() if sector_val: out["symbol_sector"] = sector_val if industry_val: out["symbol_industry"] = industry_val sectors_debug = ((payload.get("diagnostics") or {}).get("sectors") or {}) if isinstance(payload, dict) else {} match_key = _best_sector_match(sector_val or industry_val, sectors_debug) if isinstance(sectors_debug, dict) else None if match_key and isinstance(sectors_debug.get(match_key), dict): diag = sectors_debug.get(match_key) or {} vol = diag.get("volatility") if isinstance(diag.get("volatility"), dict) else {} out["symbol_sector_trend"] = { "index": match_key, "class": diag.get("class"), "ret": diag.get("ret"), "volatility_state": vol.get("state"), } return out except Exception: return {} market_intelligence = _get_latest_market_intelligence() # IMPORTANT: never send raw candles/quotes to GPT. Use compact computed features only. prompt = prepare_market_data_prompt(symbol, market_data_compact, question, context, market_intelligence=market_intelligence) logger.info("📝 PROMPT PREPARED for %s - Length: %d chars", symbol, len(prompt)) def _fallback_analysis_from_features(md: Dict[str, Any]) -> Dict[str, Any]: """Deterministic, non-GPT fallback using computed indicators/strategy flags.""" def _pick_tf(md_: Dict[str, Any]) -> str: for tf_ in ("5minute", "15minute", "30minute", "day"): if (md_.get("indicators") or {}).get(tf_): return tf_ keys = list((md_.get("indicators") or {}).keys()) return keys[0] if keys else "5minute" tf = _pick_tf(md) ind = (md.get("indicators") or {}).get(tf, {}) or {} st = (md.get("strategies") or {}) per_tf = (st.get("per_timeframe") or {}) if isinstance(st, dict) else {} flags = per_tf.get(tf, {}) if isinstance(per_tf, dict) else {} rsi = ind.get("rsi") macd_hist = ind.get("macd_hist") close = ind.get("close") vwap = ind.get("vwap") trend = (flags.get("trend") or {}) # StrategyFeatureCalculator emits direction in {uptrend, downtrend, sideways, unknown}. # Older code expected {bullish, bearish, sideways}. Normalize for stable decisions. raw_trend_dir = str(trend.get("direction") or "sideways").strip().lower() if raw_trend_dir in {"uptrend", "bullish", "up"}: trend_dir = "bullish" elif raw_trend_dir in {"downtrend", "bearish", "down"}: trend_dir = "bearish" elif raw_trend_dir in {"unknown", "na", "n/a", ""}: trend_dir = "sideways" else: trend_dir = raw_trend_dir trend_strength = (trend.get("strength") or 0) rb = (flags.get("range_breakout") or {}) rb_status = (rb.get("status") or "none") orb = (flags.get("opening_range") or {}) orb_status = (orb.get("status") or "none") mb = (flags.get("momentum_burst") or {}) mom_status = (mb.get("status") or "none") score = 50.0 rationale = [] if trend_dir == "bullish": score += 10 rationale.append(f"{tf} trend bullish") elif trend_dir == "bearish": score += 10 rationale.append(f"{tf} trend bearish") try: if float(trend_strength) >= 2: score += 5 rationale.append("trend strength elevated") except Exception: pass if rb_status in {"breakout_up", "breakout_down"}: score += 10 rationale.append(f"range breakout: {rb_status}") if orb_status in {"orb_breakout_up", "orb_breakout_down"}: score += 8 rationale.append(f"opening-range breakout: {orb_status}") # StrategyFeatureCalculator uses status in {momentum_burst, normal, unknown} if mom_status in {"momentum_burst", "strong", "strong_up", "strong_down"}: score += 8 rationale.append(f"momentum burst: {mom_status}") try: if rsi is not None: r = float(rsi) if r >= 60: score += 5 rationale.append("RSI strong") elif r <= 40: score += 5 rationale.append("RSI weak") except Exception: pass try: if macd_hist is not None: mh = float(macd_hist) if mh > 0: score += 4 rationale.append("MACD momentum positive") elif mh < 0: score += 4 rationale.append("MACD momentum negative") except Exception: pass try: if close is not None and vwap is not None: c = float(close) vw = float(vwap) if c > vw: score += 4 rationale.append("price above VWAP") elif c < vw: score += 4 rationale.append("price below VWAP") except Exception: pass score = max(0.0, min(100.0, score)) decision = "HOLD" if context == "long_buy": if trend_dir == "bullish" and score >= 60: decision = "BUY" elif trend_dir == "bearish" and score >= 70: decision = "SELL" elif context == "short_sell": if trend_dir == "bearish" and score >= 60: decision = "SELL" elif trend_dir == "bullish" and score >= 70: decision = "BUY" else: # If we have an objective breakout trigger, allow a decision even if trend is 'sideways'. has_trigger = rb_status in {"breakout_up", "breakout_down"} or orb_status in {"orb_breakout_up", "orb_breakout_down"} or mom_status in {"momentum_burst"} if (trend_dir == "bullish" or (has_trigger and rb_status == "breakout_up")) and score >= 65: decision = "BUY" elif (trend_dir == "bearish" or (has_trigger and rb_status == "breakout_down")) and score >= 65: decision = "SELL" confidence = "LOW" if score >= 75: confidence = "HIGH" elif score >= 60: confidence = "MEDIUM" decision_probability = float(score) if not rationale: rationale = ["Computed fallback from indicators/strategy flags"] return { "decision": decision, "confidence": confidence, "score": int(round(score)), "decision_probability": decision_probability, "risk_profile": "conservative" if score < 70 else "aggressive", "early_breakout": bool( rb_status in {"breakout_up", "breakout_down"} or orb_status in {"orb_breakout_up", "orb_breakout_down"} or mom_status in {"momentum_burst"} ), "early_breakout_comment": "fallback computed (GPT unavailable)", "rationale": rationale[:7], "technical_indicators": { "trend": "bullish" if trend_dir == "bullish" else "bearish" if trend_dir == "bearish" else "sideways", "momentum": "strong" if score >= 70 else "neutral" if score >= 55 else "weak", "volume": "normal", "support": None, "resistance": None, "accumulation_distribution": "neutral", "pattern_summary": str(((flags.get("pattern") or {}).get("pattern") or ""))[:120], }, "risk_reward_ratio": 1.5, "time_horizon": "intraday", "notes": "GPT unavailable; deterministic fallback used", } if TEGPT_VERBOSE_LOGS: logger.info("🚀 SENDING TO CHATGPT for %s...", symbol) else: logger.debug("[teGPT] Sending to GPT | %s", symbol) analysis = call_chatgpt_analysis(prompt) used_gpt_fallback = False # --- Hybrid decision safety-net --- # GPT is intentionally conservative and often returns HOLD when evidence is mixed. # In practice, we also have deterministic strategy/indicator features already computed. # If those features indicate a strong, directionally aligned setup, we can override # HOLD -> BUY/SELL. This is especially useful during testing to avoid 'all HOLD' # outputs even when the market data is healthy. try: enable_rule_override = str(os.getenv("TEGPT_RULE_OVERRIDE_ENABLED", "1")).strip() == "1" # Default slightly lower so we get actionable signals when deterministic features are strong. override_min_score = float(os.getenv("TEGPT_RULE_OVERRIDE_MIN_SCORE", "65")) override_require_trigger = str(os.getenv("TEGPT_RULE_OVERRIDE_REQUIRE_TRIGGER", "1")).strip() == "1" if enable_rule_override and isinstance(analysis, dict): d0 = str(analysis.get("decision") or "").upper() if d0 == "HOLD": rule = _fallback_analysis_from_features(market_data) rd = str(rule.get("decision") or "").upper() rs = rule.get("score") try: rs_f = float(rs) if rs is not None else 0.0 except Exception: rs_f = 0.0 # Require objective trigger(s) unless explicitly disabled. ok_trigger = True if override_require_trigger: ok_trigger = bool(rule.get("early_breakout")) if not ok_trigger: # Also accept explicit strategy flags as triggers. st = (market_data.get("strategies") or {}) if isinstance(market_data, dict) else {} per_tf = (st.get("per_timeframe") or {}) if isinstance(st, dict) else {} tf = "5minute" feats = per_tf.get(tf) if isinstance(per_tf, dict) else None if isinstance(feats, dict): rb = feats.get("range_breakout") or {} orb = feats.get("opening_range") or {} mb = feats.get("momentum_burst") or {} ok_trigger = ( (rb.get("status") in {"breakout_up", "breakout_down"}) or (orb.get("status") in {"orb_breakout_up", "orb_breakout_down"}) or (mb.get("status") == "momentum_burst") ) if rd in {"BUY", "SELL"} and rs_f >= override_min_score and ok_trigger: analysis["decision"] = rd # Keep both keys populated for downstream compatibility. analysis["confidence"] = "HIGH" if rs_f >= 85 else "MEDIUM" analysis["conviction"] = "HIGH" if rs_f >= 85 else "MEDIUM" analysis["score"] = int(round(rs_f)) analysis["decision_probability"] = float(min(100.0, max(0.0, rs_f))) # Preserve GPT rationale, but explain the override. analysis.setdefault("rationale", []) if isinstance(analysis.get("rationale"), list): analysis["rationale"] = ( [ f"Decision overridden from HOLD using rule-based signals (score={int(round(rs_f))}).", ] + analysis["rationale"] )[:7] analysis["decision_source"] = "RULE_OVERRIDE" analysis["rule_signals"] = { "decision": rd, "score": int(round(rs_f)), "early_breakout": bool(rule.get("early_breakout")), } except Exception: logger.debug("[teGPT] Rule override evaluation failed", exc_info=True) if isinstance(analysis, dict) and analysis.get("error_type") == "insufficient_quota": logger.warning("⚠️ GPT quota exceeded for %s; using fallback analysis", symbol) analysis = _fallback_analysis_from_features(market_data) analysis["gpt_error_type"] = "insufficient_quota" analysis["decision_source"] = "FALLBACK_QUOTA" used_gpt_fallback = True try: decision = (analysis.get("decision") if isinstance(analysis, dict) else None) or "N/A" conf = (analysis.get("confidence") if isinstance(analysis, dict) else None) or "N/A" prob = (analysis.get("decision_probability") if isinstance(analysis, dict) else None) except Exception: decision, conf, prob = "N/A", "N/A", None log_tag = "GPT FALLBACK" if used_gpt_fallback else "GPT OK" logger.info( "[teGPT] %s | %s | decision=%s | confidence=%s | prob=%s", log_tag, symbol, str(decision).upper(), str(conf).upper(), prob if prob is not None else "—", ) if TEGPT_VERBOSE_LOGS: logger.info("🎯 CHATGPT RESPONSE for %s:", symbol) logger.info(" 📊 Decision: %s", analysis.get("decision", "N/A")) logger.info(" 🎯 Confidence: %s", analysis.get("confidence", "N/A")) logger.info(" 🎯 Decision %%: %s", analysis.get("decision_probability", "N/A")) logger.info(" 💰 Price Target: %s", analysis.get("price_target", "N/A")) logger.info(" 🛡️ Stop Loss: %s", analysis.get("stop_loss", "N/A")) logger.info(" 🛡️ SL Zone: %s", analysis.get("stop_loss_zone", {})) logger.info(" 📈 Entry Price: %s", analysis.get("entry_price", "N/A")) logger.info(" 🔍 Rationale: %s", analysis.get("rationale", [])) if analysis.get("decision") not in ["BUY", "SELL", "HOLD"]: logger.warning("⚠️ Invalid decision '%s' for %s, defaulting to HOLD", analysis.get("decision"), symbol) analysis["decision"] = "HOLD" def _infer_current_price(md: Dict[str, Any]) -> Optional[float]: try: q = (md.get("quote") or {}) last_price = q.get("last_price") if last_price is not None: lp = float(last_price) if lp > 0: return lp except Exception: pass candles_by_tf = md.get("candles") or {} for tf in ("5minute", "15minute", "30minute", "day"): candles = candles_by_tf.get(tf) or [] if not candles: continue last = candles[-1] or {} close = last.get("close") try: if close is not None: c = float(close) if c > 0: return c except Exception: continue return None current_price = _infer_current_price(market_data) analysis["current_price"] = current_price def _safe_float(v: Any) -> Optional[float]: try: if v is None or isinstance(v, bool): return None f = float(v) if f != f or f in (float("inf"), float("-inf")): return None return f except Exception: return None def _env_float(name: str, default: float, *, min_value: float = 0.0, max_value: float = 1e9) -> float: raw = os.getenv(name) if raw is None or str(raw).strip() == "": return float(default) try: v = float(str(raw).strip()) except Exception: return float(default) if v != v or v in (float("inf"), float("-inf")): return float(default) return float(max(min_value, min(max_value, v))) def _normalize_entry_zone(a: Dict[str, Any]) -> Optional[Dict[str, float]]: z = a.get("entry_zone") if isinstance(z, dict): low = _safe_float(z.get("low")) high = _safe_float(z.get("high")) if low is not None and high is not None and low > 0 and high > 0: if high < low: low, high = high, low return {"low": float(low), "high": float(high)} entry = _safe_float(a.get("entry_price")) if entry is None: entry = current_price if entry is None or entry <= 0: return None buf = _env_float("AI_ENTRY_ZONE_BUFFER_PCT", 0.001, min_value=0.0, max_value=0.02) return {"low": float(entry) * (1.0 - buf), "high": float(entry) * (1.0 + buf)} def _ensure_sl_side(*, decision: str, sl: Optional[float], zone: Optional[Dict[str, float]]) -> Optional[float]: if sl is None or not zone: return sl zl = _safe_float(zone.get("low")) zh = _safe_float(zone.get("high")) if zl is None or zh is None or zl <= 0 or zh <= 0: return sl sl_buffer_pct = _env_float("ENTRY_SL_BUFFER_PCT", 0.001, min_value=0.0, max_value=0.02) d = (decision or "").upper() if d == "BUY": max_ok = float(zl) * (1.0 - float(sl_buffer_pct)) if not (float(sl) < float(zl)): return max_ok return min(float(sl), max_ok) if d == "SELL": min_ok = float(zh) * (1.0 + float(sl_buffer_pct)) if not (float(sl) > float(zh)): return min_ok return max(float(sl), min_ok) return sl def _normalize_targets(a: Dict[str, Any]) -> List[float]: arr = a.get("targets") if isinstance(arr, list): vals = [] for x in arr: f = _safe_float(x) if f is not None and f > 0: vals.append(float(f)) return vals[:2] return [] def _evaluate_entry_trigger(*, decision: str, zone: Dict[str, float], md: Dict[str, Any]) -> Dict[str, Any]: reasons: List[str] = [] state = "WAITING_FOR_ENTRY" d = (decision or "").upper() candles_by_tf = md.get("candles") or {} series_5m = candles_by_tf.get("5minute") if isinstance(candles_by_tf, dict) else None series_15m = candles_by_tf.get("15minute") if isinstance(candles_by_tf, dict) else None trigger_tf = None series = None if isinstance(series_5m, list) and series_5m: trigger_tf = "5minute" series = series_5m elif isinstance(series_15m, list) and series_15m: trigger_tf = "15minute" series = series_15m else: return {"reasons": reasons, "state": state} last = series[-1] if isinstance(series[-1], dict) else {} prev = series[-2] if len(series) >= 2 and isinstance(series[-2], dict) else None def _cnum(c, key): if not isinstance(c, dict): return None return _safe_float(c.get(key) or c.get(key[:1])) c_open = _cnum(last, "open") c_close = _cnum(last, "close") c_high = _cnum(last, "high") c_low = _cnum(last, "low") zone_low = float(zone.get("low") or 0.0) zone_high = float(zone.get("high") or 0.0) if zone_low <= 0 or zone_high <= 0: return {"reasons": reasons, "state": state} if d == "BUY": if c_close is not None and c_close >= zone_high: reasons.append("5M_CLOSE_ABOVE_ZONE" if trigger_tf == "5minute" else "15M_CLOSE_ABOVE_ZONE") elif c_open is not None and c_open > zone_high: reasons.append("5M_GAP_OPEN_ABOVE_ZONE" if trigger_tf == "5minute" else "15M_GAP_OPEN_ABOVE_ZONE") if prev is not None: prev_close = _cnum(prev, "close") if ( prev_close is not None and prev_close >= zone_high and c_low is not None and c_low <= zone_high and c_close is not None and c_close > zone_high ): reasons.append("BREAKOUT_RETEST_HOLD" if trigger_tf == "5minute" else "BREAKOUT_RETEST_HOLD_15M") elif d == "SELL": if c_close is not None and c_close <= zone_low: reasons.append("5M_CLOSE_BELOW_ZONE" if trigger_tf == "5minute" else "15M_CLOSE_BELOW_ZONE") elif c_open is not None and c_open < zone_low: reasons.append("5M_GAP_OPEN_BELOW_ZONE" if trigger_tf == "5minute" else "15M_GAP_OPEN_BELOW_ZONE") if prev is not None: prev_close = _cnum(prev, "close") if ( prev_close is not None and prev_close <= zone_low and c_high is not None and c_high >= zone_low and c_close is not None and c_close < zone_low ): reasons.append("BREAKDOWN_RETEST_HOLD" if trigger_tf == "5minute" else "BREAKDOWN_RETEST_HOLD_15M") if reasons: state = "ENTRY_ACTIVATED" return {"reasons": reasons, "state": state} decision_upper = str(analysis.get("decision") or "").upper() if decision_upper in {"BUY", "SELL"}: zone = _normalize_entry_zone(analysis) if zone: analysis["entry_zone"] = zone sl = _safe_float(analysis.get("stop_loss")) sl = _ensure_sl_side(decision=decision_upper, sl=sl, zone=zone) if sl is not None: analysis["exec_sl"] = float(sl) t = _normalize_targets(analysis) if t: analysis["exec_targets"] = t rr = _safe_float(analysis.get("risk_reward_ratio")) if rr is not None: analysis["exec_rr_ratio"] = float(rr) trig = _evaluate_entry_trigger(decision=decision_upper, zone=zone, md=market_data) analysis["entry_trigger_reason"] = trig.get("reasons") or [] analysis["signal_state"] = trig.get("state") or "WAITING_FOR_ENTRY" else: analysis.setdefault("signal_state", "WAITING_FOR_ENTRY") logger.info("✅ CHATGPT ANALYSIS FORMATTED for %s - Decision: %s", symbol, analysis["decision"]) features = { "indicators": market_data.get("indicators", {}), "strategies": market_data.get("strategies", {}), "pivots": market_data.get("pivots", {}), "fib": market_data.get("fib", {}), } analysis["features"] = features # ---- Deterministic taxonomy: trend + setup ---- # Goal: stable, UI-friendly fields driven by computed flags (not GPT prose). try: decision_upper = str(analysis.get("decision") or "HOLD").strip().upper() strategies = features.get("strategies") if isinstance(features, dict) else {} per_tf = (strategies.get("per_timeframe") or {}) if isinstance(strategies, dict) else {} def _pick_tf_for_taxonomy() -> Optional[str]: for tf in ("15minute", "5minute", "30minute", "day", "week", "month"): if isinstance(per_tf.get(tf), dict) and per_tf.get(tf): return tf for k, v in (per_tf or {}).items(): if isinstance(v, dict) and v: return str(k) return None tf_tax = _pick_tf_for_taxonomy() flags = per_tf.get(tf_tax) if tf_tax and isinstance(per_tf, dict) else {} flags = flags if isinstance(flags, dict) else {} rb_status = str(((flags.get("range_breakout") or {}) if isinstance(flags.get("range_breakout"), dict) else {}).get("status") or "") orb_status = str(((flags.get("opening_range") or {}) if isinstance(flags.get("opening_range"), dict) else {}).get("status") or "") mom_status = str(((flags.get("momentum_burst") or {}) if isinstance(flags.get("momentum_burst"), dict) else {}).get("status") or "") trend_dir = str(((flags.get("trend") or {}) if isinstance(flags.get("trend"), dict) else {}).get("direction") or "") if decision_upper == "BUY": analysis["trend"] = "BULLISH" if orb_status == "orb_breakout_up" or rb_status == "breakout_up": analysis["setup"] = "BREAKOUT" elif mom_status == "momentum_burst": analysis["setup"] = "RALLY" elif trend_dir == "uptrend": analysis["setup"] = "TREND" else: analysis["setup"] = "TREND" elif decision_upper == "SELL": analysis["trend"] = "BEARISH" if orb_status == "orb_breakout_down" or rb_status == "breakout_down": analysis["setup"] = "BREAKDOWN" elif mom_status == "momentum_burst": analysis["setup"] = "WEAKNESS" elif trend_dir == "downtrend": analysis["setup"] = "WEAKNESS" else: analysis["setup"] = "WEAKNESS" else: analysis["trend"] = "NO_TREND" analysis["setup"] = "NONE" # Backward-compatible compact label for chips if GPT didn't provide one. if not isinstance(analysis.get("trend_label"), str) or not str(analysis.get("trend_label") or "").strip(): if analysis.get("trend") == "BULLISH": analysis["trend_label"] = f"Bullish - {str(analysis.get('setup') or 'Trend').title()}" elif analysis.get("trend") == "BEARISH": analysis["trend_label"] = f"Bearish - {str(analysis.get('setup') or 'Weakness').title()}" else: analysis["trend_label"] = "" except Exception: # Never fail the analysis response because taxonomy parsing failed. analysis.setdefault("trend", "NO_TREND") analysis.setdefault("setup", "NONE") try: from app.v1.services.entry_engine import build_execution_plan plan = build_execution_plan(decision=str(analysis.get("decision") or "").upper(), market_data=market_data) if plan is not None: analysis["entry_engine_entry_zone"] = plan.entry_zone analysis["entry_engine_entry_trigger_reason"] = plan.entry_trigger_reason analysis["entry_engine_exec_sl"] = plan.sl analysis["entry_engine_exec_targets"] = plan.targets analysis["entry_engine_exec_rr_ratio"] = plan.rr_ratio analysis["entry_engine_signal_state"] = plan.state analysis["entry_engine_diagnostics"] = plan.diagnostics analysis.setdefault("entry_zone", plan.entry_zone) analysis.setdefault("entry_trigger_reason", plan.entry_trigger_reason) if analysis.get("exec_sl") is None and plan.sl is not None: analysis["exec_sl"] = plan.sl if analysis.get("exec_targets") is None and plan.targets: analysis["exec_targets"] = plan.targets if analysis.get("exec_rr_ratio") is None and plan.rr_ratio is not None: analysis["exec_rr_ratio"] = plan.rr_ratio analysis.setdefault("signal_state", plan.state) analysis["entry_trigger_fired"] = bool(analysis.get("entry_trigger_reason")) missed = (plan.diagnostics or {}).get("missed_trade_reasons") analysis["entry_missed_trade_reasons"] = missed if isinstance(missed, list) else [] logger.info( "[EntryEngine] %s %s state=%s zone=%s triggers=%s", symbol, str(analysis.get("decision") or "").upper(), plan.state, plan.entry_zone, plan.entry_trigger_reason, ) except Exception: logger.exception("[EntryEngine] Failed to compute execution plan for %s", symbol) instrument_token = market_data.get("instrument_token") if instrument_token is not None: analysis["instrument_token"] = instrument_token analysis_doc = { "symbol": symbol, "user_id": user_id, "analysis": compact_analysis_for_persistence(analysis), "market_data": market_data_compact, "features": features, "question": question, "context": context, "timeframes": timeframes, "timestamp": datetime.utcnow(), } result = db["analyses"].insert_one(analysis_doc) analysis["analysis_id"] = str(result.inserted_id) analysis["timestamp"] = analysis_doc["timestamp"].isoformat() analysis["symbol"] = symbol if include_market_data: # Return full runtime snapshot to the caller if requested, # but do not store it in the database. analysis["market_data"] = market_data logger.info("✅ ANALYSIS COMPLETE for %s - Testing Mode", symbol) return analysis except HTTPException: raise except Exception as e: logger.exception("Symbol analysis failed for %s", symbol) raise HTTPException(status_code=500, detail=str(e)) def bulk_analyze_service( db, zerodha_client: ZerodhaClient, symbols: List[str], analysis_type: str, timeframes: List[str], max_concurrent: int, user_id: str, include_market_data: bool = False, ) -> List[Dict[str, Any]]: """Bulk analyze multiple symbols with detailed progress logging.""" logger.info("🔄 BULK ANALYSIS STARTED - %d symbols, Type: %s", len(symbols), analysis_type) logger.info("📋 SYMBOLS TO ANALYZE: %s", symbols) if not symbols: logger.warning("⚠️ NO SYMBOLS PROVIDED for bulk analysis") return [] questions = { "short_sell": ( "Provide comprehensive intraday trading analysis for this TOP GAINER. " "Decide between BUY (intraday long), SELL (intraday short), or HOLD (no trade). " "Use HOLD only when the setup is clearly low quality (score < 30) or the evidence is very mixed. " "When the up-move is strong and healthy, prefer BUY; when there are multiple objective reversal/exhaustion " "signals (e.g. RSI>70 at resistance, bearish patterns, volume/price divergence), prefer SELL." ), "long_buy": ( "Provide comprehensive intraday trading analysis for this TOP LOSER / beaten-down stock. " "Decide between BUY (intraday long), SELL (intraday short), or HOLD (no trade). " "Use HOLD only when the setup is clearly low quality (score < 30) or there is no clear edge. " "When price is deeply oversold near support with improving momentum/accumulation, prefer BUY; " "reserve SELL for cases where the downtrend remains very strong with fresh breakdowns." ), "general": ( "Provide comprehensive intraday trading analysis with a clear decision: " "BUY (intraday long), SELL (intraday short), or HOLD (no trade). " "Score the setup 0-100 and avoid HOLD on high-quality setups (score >= 60) unless there is a very specific risk." ), } question = questions.get(analysis_type, questions["general"]) logger.info("❓ ANALYSIS QUESTION TYPE: %s", analysis_type) results: List[Dict[str, Any]] = [] successful_analyses = 0 failed_analyses = 0 batch_size = min(max_concurrent, 5) logger.info("📦 PROCESSING IN BATCHES - Batch size: %d", batch_size) for i in range(0, len(symbols), batch_size): batch_symbols = symbols[i : i + batch_size] batch_num = (i // batch_size) + 1 total_batches = (len(symbols) + batch_size - 1) // batch_size logger.info("📦 BATCH %d/%d: %s", batch_num, total_batches, batch_symbols) for symbol_idx, sym in enumerate(batch_symbols, 1): try: logger.info("🎯 ANALYZING %s (%d/%d in batch)", sym, symbol_idx, len(batch_symbols)) analysis = analyze_symbol_service( db=db, zerodha_client=zerodha_client, symbol=sym, timeframes=timeframes, question=question, context="general", user_id=user_id, include_market_data=bool(include_market_data), ) results.append(analysis) successful_analyses += 1 logger.info("✅ COMPLETED %s - Decision: %s", sym, analysis.get("decision", "N/A")) except Exception as e: failed_analyses += 1 logger.error("❌ FAILED %s: %s", sym, str(e)) results.append( { "symbol": sym, "decision": "HOLD", "confidence": "LOW", "rationale": [f"Analysis failed: {str(e)}"], "error": str(e), "timestamp": datetime.utcnow().isoformat(), } ) if i + batch_size < len(symbols): import time time.sleep(1) logger.info("🏁 BULK ANALYSIS COMPLETE") logger.info(" ✅ Successful: %d", successful_analyses) logger.info(" ❌ Failed: %d", failed_analyses) logger.info(" 📊 Total Results: %d", len(results)) decisions: Dict[str, int] = {} for result in results: decision = result.get("decision", "UNKNOWN") decisions[decision] = decisions.get(decision, 0) + 1 logger.info("🎯 DECISION BREAKDOWN: %s", decisions) return results