#!/usr/bin/env python3 """ predictor_core.py — Unified Indian equity prediction engine (v2). Integrates: • S1/S2/S3/MFS/NIRA/PED/SUPER signal generators (from trial_run.py) • ML feature scoring (from ml_combiner.py) • MacroContext gates (from macro_context.py) • Earnings calendar blackout (yfinance + trial_run.build_earnings_blackout) • AI news sentiment (news_sentiment.py → Claude Haiku) • Hard VIX / Nifty EMA200 enforcement (no longer advisory only) Public API: predict_stock_v2(ticker, start_date, end_date) → dict rank_stocks_v2(start_date, end_date, universe, capital) → dict timeframe_to_dates(tf) → (start_date, end_date) """ from __future__ import annotations import sys, os, warnings, math, time, logging, threading warnings.filterwarnings("ignore") sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from concurrent.futures import ThreadPoolExecutor, as_completed def _sf(v, d: int = 2, fallback=None): """Round float safely; return fallback if value is NaN or Inf.""" try: r = round(float(v), d) return r if math.isfinite(r) else fallback except (TypeError, ValueError): return fallback def _fn(v, default: float = 0.0) -> float: """Return v if finite, otherwise default (used to keep NaN out of sums).""" try: f = float(v) return f if math.isfinite(f) else default except (TypeError, ValueError): return default import numpy as np import pandas as pd import yfinance as yf from datetime import datetime, timedelta, date as _date_cls from typing import Optional # ── MULTI-SOURCE DATA FETCHER ──────────────────────────────────────────────── from data_sources import fetch_ohlcv, fetch_market_data, get_cached_ohlcv, fetch_live_price, update_cached_ohlcv # ── REQUEST CONTEXT (for tracking data merges + validation) ─────────────────── from request_context import RequestContext # ── STRATEGY SIGNAL GENERATORS (from trial_run.py) ─────────────────────────── from trial_run import ( rsi, atr, obv, adx_s, macd_h, gen_s1, gen_s2, gen_s3, gen_mfs, gen_nira, gen_ped, gen_supertrend, gen_s4, gen_s5, gen_s6, gen_s4v2, gen_s5v2, gen_s6v2, gen_s7, gen_s8, gen_s9, gen_s10, gen_s11, gen_s_capflow, gen_s_confluence_trio, gen_s_seasonal, gen_s12, gen_s13, gen_s14, gen_s15, gen_s16, gen_s17, gen_s18, gen_s19, gen_s20, ) # ── ML FEATURE FUNCTIONS (from ml_combiner.py) ─────────────────────────────── from ml_combiner import bollinger_position, ema_stack_score, shadow_flag # ── MACRO CONTEXT (from macro_context.py) ──────────────────────────────────── try: from macro_context import MacroContext _HAS_MACRO = True except ImportError: _HAS_MACRO = False # ── AI NEWS SENTIMENT ───────────────────────────────────────────────────────── from news_sentiment import fetch_and_analyze # ── AI DIRECTIONAL FORECAST ─────────────────────────────────────────────────── from ai_forecast import get_ai_forecast # ── SECTOR PULSE ───────────────────────────────────────────────────────────── try: from sector_pulse import get_sector_pulse, get_sector_for_ticker _HAS_SECTOR_PULSE = True except ImportError: _HAS_SECTOR_PULSE = False # ── FUNDAMENTALS ───────────────────────────────────────────────────────────── try: from fundamentals import get_fundamentals _HAS_FUNDAMENTALS = True except ImportError: _HAS_FUNDAMENTALS = False # ── TIMEFRAME HELPER ───────────────────────────────────────────────────────── TIMEFRAME_DAYS: dict[str, int] = {"INTRADAY": 0, "1D": 1, "3D": 3, "5D": 5, "1W": 7} def timeframe_to_dates(tf: str) -> tuple[str, str]: """Convert 'INTRADAY'/'1D'/'3D'/'5D' to (start_date, end_date) strings. On weekdays start_date = today (unchanged). On weekends/holidays start_date advances to the next trading day so predictions target "buy on Monday" rather than the past weekend. INTRADAY (n == 0) is a same-day horizon: start == end == the trading day, with no weekend buffer. Callers detect INTRADAY downstream via start == end. """ from datetime import date from market_calendar import next_trading_day n = TIMEFRAME_DAYS.get(tf.upper(), 5) today = date.today() start = next_trading_day(today) # no-op on trading days; Mon on Sat/Sun if n == 0: # Same-day intraday window — target is the trading day itself. return start.strftime("%Y-%m-%d"), start.strftime("%Y-%m-%d") end = start + timedelta(days=n + 2) # weekend buffer anchored from start return start.strftime("%Y-%m-%d"), end.strftime("%Y-%m-%d") # ── NSE UNIVERSE (dynamic — fetched from Yahoo Finance screener) ───────────── from universe import get_universe TICKER_NAMES: dict[str, str] = {} # populated on first predict/rank call DEFAULT_UNIVERSE: list[str] = [] # populated on first predict/rank call def _init_universe() -> None: """Load universe from cache (or fetch from YF) on first use.""" global TICKER_NAMES, DEFAULT_UNIVERSE if not TICKER_NAMES: TICKER_NAMES = get_universe() DEFAULT_UNIVERSE = list(TICKER_NAMES.keys()) # ── BACKTEST-DERIVED EXPECTED RETURNS (from trading_strategies_india.md v2) ── # (win_rate, avg_win%, avg_loss%) per strategy at ~21-day horizon _STRATEGY_STATS_DEFAULT = { # ══════════════════════════════════════════════════════════════════════════ # ALL WIN RATES BELOW ARE NSE-VERIFIED (276 stocks, 2019-2024 backtest) # NOT from US research claims. Mode A = all signals. Mode B = VIX<18 filter. # Format: (win_rate, avg_win_pct, avg_loss_pct) # ══════════════════════════════════════════════════════════════════════════ # ── Original strategies ────────────────────────────────────────────────── # S1: Shadow recovery C5 fires rarely — only 2 signals on 276 stocks; using 30-stock estimate "S1": (0.63, 5.8, 3.4), # C5 shadow recovery — ~2-3/year, 63% NSE estimate # S2-MFS-NIRA: all WEAK at every horizon — no genuine edge beyond market drift "S2": (0.58, 12.5, 4.2), # Momentum Breakout — 58.3% at 1M (best, N=4054) "S3": (0.58, 8.0, 2.0), # EMA Trend — 57.5% at 1M (best, N=29358) "MFS": (0.60, 10.0, 4.0), # Multi-Factor — 59.8% at 1M (best, N=54932) "NIRA": (0.57, 9.0, 3.5), # Index Reconstitution — 57.2% at 1M "PED": (0.56, 4.5, 2.5), # Post-Earnings Drift — 55.5% at 1M "SUPER": (0.52, 7.0, 3.5), # Supertrend crossover — no backtest data (est.) # ── S4: Connors RSI(2) — NSE-verified (NOT US 77% claim) ───────────────── # v1 now uses RSI(2)<2 (doubleyourreturns.in NSE 2004-2017: 68-71% with <2 vs 56% with <5) # v2: 66.2% at 1M Mode A (N=544); expected ~70% with RSI(2)<2 threshold "S4": (0.63, 3.5, 2.8), # RSI(2)<2 v1 — upgraded from 56% to 63% (NSE-calibrated) "S4V2": (0.70, 3.2, 2.5), # RSI(2)<3 v2 — 70% Mode A (NSE doubleyourreturns.in) # ── S5: DMA Pullback — NSE-verified (NOT US 80% claim) ─────────────────── # v1: 56.5% at 1M Mode A (N=42574). v2: 58.5% at 2W Mode A (N=8932) "S5": (0.57, 4.2, 3.1), # DMA Pullback v1 — 57% NSE-verified (was wrongly 80%) "S5V2": (0.59, 4.0, 2.8), # DMA Pullback v2 — 58.5% (tighter RSI5<30 filter) # ── S6: Momentum RSI Dip — BEST VERIFIED STRATEGY ───────────────────────── # Mode A 2W: 65.3% (MEDIUM confidence, N=498) # Mode B 3D: 70.9% (HIGH confidence, N=141, excess +11.3%) ← BREAKTHROUGH # Mode B 1M: 63.8% (WEAK) # Using Mode B 3D result as the operating win rate (VIX filter is applied at predict time) "S6": (0.65, 5.5, 3.2), # MomDip v1 — 65% Mode A, 70.9% Mode B 3D (HIGH!) # ── S6v2: Enhanced MomDip — NEW BEST STRATEGY ──────────────────────────── # Mode B 1M: 76.5% (LOW confidence, N=34) — FIRST result above 75%! # Mode C 1D: 73.3% (MEDIUM confidence, N=15) # Mode C 1M: 93.3% (MEDIUM confidence, N=15) — spectacular but small N # Using conservative Mode B estimate; will update with live paper trades "S6V2": (0.69, 5.2, 3.0), # MomDip v2 — 69% Mode A-B blend; 76.5% Mode B 1M # ── S7: Multi-day Capitulation — NEW ───────────────────────────────────── # Mode A 1W: 59.4% (MEDIUM confidence, N=212, excess +3.8%) # Mode C 3D: 70.8% (MEDIUM confidence, N=24) "S7": (0.61, 4.5, 2.8), # 3-bar capitulation — 61% Mode A; 70.8% Mode C 3D # ── S8: RSI Triple Confluence — STATISTICALLY CONFIRMED NEW STRATEGY ───── # Mode A 3D: 59.2% (HIGH confidence, N=282, excess +5.7%) ← VERIFIED EDGE # Mode A 1D: 60.1%, Mode A 1M: 63.3% (MEDIUM) # Mode C 1M: 73.0% (MEDIUM confidence, N=37) "S8": (0.60, 4.8, 2.9), # RSI 3-period confluence — HIGH at 3D (N=282) # ── S9: MACD-ADX Momentum — WEAK, include for completeness ─────────────── # Mode A 1M: 58.0% (WEAK). Mode B/C show no improvement. "S9": (0.58, 6.0, 3.5), # MACD-ADX crossover — 58% 1M, no real edge # ── S10: 20-Day Low in Uptrend — MODERATE SIGNAL ───────────────────────── # Mode A 2W: 61.4% (WEAK, N=2465), 1W: 60.6%, 3D: 59.8% # Mode B 3D: 57.4% (LOW, N=674) "S10": (0.61, 5.0, 3.0), # 20D Low Uptrend — 61% Mode A 2W; moderate signal # ── S11: Confluence Gate — PROMISING, SMALL N ──────────────────────────── # Mode A 3D: 61.2% (WEAK, N=98). Mode B: ~62%. Mode C: 70.6% (N=17) # Small sample limits confidence; keep as high-quality rare signal "S11": (0.62, 5.5, 3.0), # S8+S6v2 confluence — 62% Mode A, 70.6% Mode C # ── v4 Research-derived strategies (Jun 2026 — NSE-verified Jun 2026) ─────── # S_CAPFLOW: Capitulation 3-red + volume spike + Nifty bull. NSE backtest 378 stocks 2019-24. # Mode A 3D: 58.7% (N=1457, WEAK). Mode B 1M: 52.3%. Much weaker than estimated. "S_CAPFLOW": (0.587, 3.2, 2.4), # Capitulation vol-spike — 58.7% NSE-verified 3D # S_CTRIO: Triple RSI confluence (RSI2<5 + RSI5<30 + RSI14<35) + SMA200 + ADX>20. # Mode A 3D: 71.0% HIGH (N=124, excess +16.9%). Mode B 3D: 70.0% HIGH (N=70). BEST SIGNAL. "S_CTRIO": (0.710, 5.8, 2.5), # Triple RSI confluence — 71.0% NSE-verified 3D HIGH # S_SEASONAL: Santa Claus rally Dec20-Jan5. NSE 20-year study: 80-85%. # Mode A 2W: 70.8%. Mode B 1W: HIGH confidence (excess +10.2%). Mode B 1M: HIGH. "S_SEASONAL": (0.708, 5.2, 2.8), # Santa rally — 70.8% NSE-verified 2W # ── v5 New strategies (Jun 2026 NSE-verified on 378 stocks 2019-24) ───────── # S12: Post-Budget seasonal. Documented 80% for Nifty index; individual stocks: 50.7%. # The index-level edge does NOT translate to stock-level. Signals too many stocks (N=2238). "S12": (0.507, 2.0, 2.5), # Post-Budget seasonal — 50.7% NSE-verified (weak on stocks) # S13: Oct-Nov seasonal. Documented 90% for Nifty; individual stocks: 57.1%. Same issue. "S13": (0.571, 3.5, 2.8), # Oct-Nov seasonal — 57.1% NSE-verified # S14: EMA20 touch in uptrend. Mode A 1M: 56.3%. Not as strong as designed. "S14": (0.563, 3.0, 2.8), # EMA20 dip — 56.3% NSE-verified # S15: NR7 Inside Bar. Mode A 1M: 55.7%. Marginal edge. "S15": (0.557, 2.8, 3.0), # NR7+IB — 55.7% NSE-verified # S16: StochRSI Oversold Recovery. Mode A 1W: 62.3% MEDIUM. Better than expected. "S16": (0.623, 3.8, 2.8), # StochRSI recovery — 62.3% NSE-verified 1W MEDIUM # ── v6 Famous Analyst Strategies (NSE-verified Jun 2026, 378-stock universe) ── # S17: TTM Squeeze + NR7. Best: Mode A 1M=59.6% WEAK; Mode C 1M=55.8% HIGH (N=95, excess+5.3%). # US strategies don't port well to NSE individual stocks at short horizons. "S17": (0.596, 4.2, 2.8), # TTM Squeeze — 59.6% Mode A 1M (WEAK on stocks; best under full macro) # S18: RSI Bullish Divergence. Best: Mode A 1D=54.3% LOW (N=254); Mode C 1D=61.5% LOW (N=13). "S18": (0.543, 0.8, 2.5), # RSI divergence — 54.3% Mode A 1D LOW (N=254) # S19: VCP — Minervini Volatility Contraction Pattern. Best: Mode A 1M=60.5% WEAK (N=806); Mode C 1W=62.5% LOW (N=48). "S19": (0.605, 3.7, 3.0), # VCP breakout — 60.5% Mode A 1M WEAK (best NSE result) # S20: Gap-Up + Volume Surge. Weak on NSE: Mode A 1M=58.7% WEAK; negative excess vs Nifty all horizons. "S20": (0.587, 4.0, 2.8), # Gap-up + vol surge — 58.7% Mode A 1M WEAK (negative Nifty excess) } # Which strategies are valid predictors for each timeframe. # Filters active_strategies before computing expected returns to avoid # averaging a 5D strategy's stats into a 1D prediction. STRATEGY_TIMEFRAME_MAP: dict[str, list[str]] = { # INTRADAY reuses only the fastest-reacting, oversold-bounce / short-momentum # signals — swing-oriented strategies don't resolve within a single session. "INTRADAY": ["S1", "S4", "S4V2", "S8", "S16", "S_CTRIO"], "1D": ["S1", "S4", "S4V2", "S7", "S8", "S11", "PED", "S_CAPFLOW", "S_CTRIO", "S15", "S16", "S20"], "3D": ["S1", "S4", "S4V2", "S5", "S5V2", "S6", "S6V2", "S7", "S8", "S9", "S11", "SUPER", "PED", "S_CAPFLOW", "S_CTRIO", "S14", "S15", "S16", "S17", "S18", "S20"], "5D": ["S2", "S5", "S5V2", "S6", "S6V2", "S9", "S10", "S11", "MFS", "NIRA", "SUPER", "S_CAPFLOW", "S_CTRIO", "S_SEASONAL", "S12", "S13", "S14", "S15", "S16", "S17", "S18", "S19"], "1M": ["S6V2", "S_SEASONAL", "S19"], "1W": ["S6", "S6V2", "S9", "S10", "S11", "MFS", "SUPER", "S_CAPFLOW", "S_SEASONAL", "S14", "S15", "S16", "S17", "S19"], } # Estimated next-open slippage buffer (vs prior close) used to propose a realistic # entry limit. This is also the default "no-chase" threshold at execution time. ENTRY_BUFFER_BY_TIMEFRAME: dict[str, float] = { "INTRADAY": 0.001, # 0.10% — tightest, entry is the live same-session price "1D": 0.002, # 0.20% "3D": 0.003, # 0.30% "5D": 0.005, # 0.50% "1W": 0.007, # 0.70% — wider buffer for longer hold } # INTRADAY minimum favorable move: a same-session trade must be able to clear ~1% to be worth # the round-trip cost. Verified achievable on historical data (3.05M NSE day-rows, 3,706 series): # a favorable intraday move >= 1% occurred on 89.2% of days (daily range >= 1% on 93.5%, median # range 3.42%). So 1% filters only the genuinely flat setups. Env-overridable. INTRADAY_MIN_MOVE_PCT: float = float(os.getenv("INTRADAY_MIN_MOVE_PCT", "1.0")) # Minimum gap kept between the near and far intraday bounds after both are floored to >=1%, so the # range never collapses to a single point (e.g. [1.00%, 1.10%]). Env INTRADAY_MIN_SPREAD_PCT. INTRADAY_MIN_SPREAD_PCT: float = float(os.getenv("INTRADAY_MIN_SPREAD_PCT", "0.1")) # ── TIGHT DIRECTIONAL BAND for INTRADAY + 1D (user request 2026-07-30) ──────────────────────── # A NARROW band centered on a volatility-scaled expected move so the mean/target is a single clear # number (e.g. 1.00%–1.25%), instead of a wide [1.0, 2.0] intraday band or the flat ±1% 1D band. # center = clamp(mult × ATR%, min_center, cap); band = [center − half, center + half]; near floored # to `floor` (keeps the whole move >= 1%). Mirrors ai_forecast._TIGHT_BAND (same env knobs) so the # AI path and the strategy/no-AI path produce identical bands. 3D/5D are unaffected. _TIGHT_BAND: dict = { "INTRADAY": { "mult": float(os.getenv("INTRADAY_BAND_ATR_MULT", "0.33")), "min_center": float(os.getenv("INTRADAY_BAND_MIN_CENTER", "1.10")), "cap": float(os.getenv("INTRADAY_BAND_CAP", "2.0")), "half": float(os.getenv("INTRADAY_BAND_HALF_PCT", "0.125")), "floor": float(os.getenv("INTRADAY_MIN_MOVE_PCT", "1.0")), }, "1D": { "mult": float(os.getenv("ONE_D_BAND_ATR_MULT", "0.45")), "min_center": float(os.getenv("ONE_D_BAND_MIN_CENTER", "1.10")), "cap": float(os.getenv("ONE_D_BAND_CAP", "2.5")), "half": float(os.getenv("ONE_D_BAND_HALF_PCT", "0.125")), "floor": float(os.getenv("ONE_D_BAND_FLOOR", "1.0")), }, } def _tight_band_mag(tf_label: str, atr_pct: float): """Tight directional band as (near_mag, far_mag) magnitudes (near < far), or None for 3D/5D.""" cfg = _TIGHT_BAND.get(tf_label) if not cfg: return None center = min(cfg["cap"], max(cfg["min_center"], cfg["mult"] * (atr_pct or 0.0))) h = cfg["half"] lo, hi = center - h, center + h if lo < cfg["floor"]: lo, hi = cfg["floor"], cfg["floor"] + 2 * h return round(lo, 2), round(hi, 2) # AI range mode (see ai_forecast._AI_RANGE_MODE). "containment" (default) → do NOT tighten the # INTRADAY/1D band; keep the wide AI/strategy prediction interval so the move lands inside the shown # range ~85% of the time. "target" → apply the tight directional band below. Env AI_RANGE_MODE. _AI_RANGE_MODE = os.getenv("AI_RANGE_MODE", "containment").strip().lower() # ── 1D RANGE-ONLY POLICY ───────────────────────────────────────────────────── # Next-day (1D) DIRECTION has a proven ~74% accuracy ceiling: no point-in-time feature (trend, # momentum, ml_score, intraday close-strength) reliably separates hits from misses, and strong # closes actually mean-revert. A directional 1D target is therefore structurally unreachable # ~1-in-4 times (worse in a falling market) — the "did not reach 1D target" problem. So 1D is an # honest RANGE-ONLY call rather than a coin-flip directional bet. # # The band is FLAT by policy (±1%), NOT ATR-scaled. An ATR-scaled band (the old # max(3.5%, 1.30×ATR%)) ballooned to ±5%+ on volatile stocks — so wide the user can't bet on it # ("somewhere between ₹10.7k and ₹11.9k" = no information) and so wide it always "hit", inflating # the hit-rate without predicting anything. A flat ±1% band is a real, actionable "stays within 1%" # claim and keeps 1D consistent with the NEUTRAL band everywhere else in the codebase # (ai_forecast._NEUT_RANGE, database._SNAP_NEUT, research.range_model._NEUT_FLAT, # ml_predictor._neutral_range). Env-overridable; set FORCE_1D_RANGE_ONLY=0 to restore directional # 1D behaviour, or ONE_D_RANGE_HALF_PCT to widen/narrow the flat band. FORCE_1D_RANGE_ONLY: bool = os.getenv("FORCE_1D_RANGE_ONLY", "0") != "0" ONE_D_RANGE_HALF_PCT: float = float(os.getenv("ONE_D_RANGE_HALF_PCT", "1.0")) # flat falsifiable half-width def _load_live_strategy_stats() -> dict: """ Override hardcoded stats with live paper trading results when a strategy has ≥ 30 closed trades — gives adapting win rate / avg return estimates. """ stats = dict(_STRATEGY_STATS_DEFAULT) try: import database as db rows = db.get_signal_accuracy() for row in rows: sig = row.get("signal", "").upper() if sig not in stats: continue total = row.get("total", 0) if total < 30: continue win_rate = (row.get("win_rate") or 0) / 100.0 avg_pnl = row.get("avg_pnl_pct") or 0 if win_rate <= 0: continue # Estimate avg_win / avg_loss from win_rate and avg_pnl: # avg_pnl = wr * avg_win - lr * avg_loss # Use current avg_win/avg_loss ratio to split old_wr, old_win, old_loss = stats[sig] old_rr = old_win / old_loss if old_loss else 2.0 lr = 1 - win_rate # avg_pnl = wr * avg_win - lr * (avg_win / rr) # avg_pnl = avg_win * (wr - lr / rr) → solve for avg_win denom = win_rate - (lr / old_rr) if abs(denom) > 1e-6: new_win = max(0.5, min(avg_pnl / denom, 20.0)) # cap at 20× so corrupt rows can't inflate EV new_loss = max(0.1, new_win / old_rr) stats[sig] = (round(win_rate, 3), round(new_win, 2), round(new_loss, 2)) except Exception: pass return stats _STRATEGY_STATS = _load_live_strategy_stats() # ── MARKET DATA CACHE (5-min TTL — shared across all parallel stock downloads) ─ _DATA_CACHE: dict = {} _DATA_CACHE_TTL = 300 # seconds # ── OHLCV CACHE (5-min TTL) ─────────────────────────────────────────────────── # When the same ticker is predicted across multiple timeframes (1D/3D/5D) in the # watchlist flow, this prevents 3 separate Yahoo Finance downloads per ticker. _OHLCV_CACHE: dict = {} _OHLCV_CACHE_TTL = 300 # seconds # Per-key fetch locks: prevent multiple parallel TF threads from racing to # fetch the same (ticker, period) from the network simultaneously. _OHLCV_FETCH_LOCKS: dict = {} _OHLCV_FETCH_LOCKS_LOCK = threading.Lock() # ── RANK RESULT CACHE (10-min TTL) ──────────────────────────────────────────── # rank_stocks_v2 scans 150 stocks — cache the full result so repeated UI hits # (e.g. dashboard reload) don't re-run the entire scan within the same session. _RANK_CACHE: dict = {} _RANK_CACHE_TTL = 600 # seconds # ── PREDICTION RESULT CACHE (5-min TTL) ─────────────────────────────────────── # Watchlist runs 3 TFs × N tickers in one parallel pool — without this cache a # reload within 5 min re-runs every full prediction pipeline call. _PRED_CACHE: dict = {} _PRED_CACHE_TTL = 300 # seconds def clear_runtime_caches() -> dict: """Clear in-memory caches used by ranking/prediction flows. Used by API-level force-refresh endpoints to guarantee a fresh recompute. Returns basic counts for observability/debugging. """ cleared = { "market_cache": len(_DATA_CACHE), "ohlcv_cache": len(_OHLCV_CACHE), "rank_cache": len(_RANK_CACHE), "pred_cache": len(_PRED_CACHE), } _DATA_CACHE.clear() _OHLCV_CACHE.clear() _RANK_CACHE.clear() _PRED_CACHE.clear() return cleared # ── LOG THROTTLE (avoid per-ticker warning spam) ──────────────────────────── _LAST_NIFTY_WARN_TS = 0.0 _NIFTY_WARN_TTL = 300 # seconds _NIFTY_WARN_LOCK = threading.Lock() def _warn_nifty_unavailable_once() -> None: global _LAST_NIFTY_WARN_TS now = time.time() with _NIFTY_WARN_LOCK: if now - _LAST_NIFTY_WARN_TS >= _NIFTY_WARN_TTL: logging.warning("Nifty market data unavailable — Nifty-dependent signals skipped (throttled)") _LAST_NIFTY_WARN_TS = now def _get_market_cache(period: str = "1y"): """Return (nifty_c, vix_c) Series, cached for 5 minutes so parallel stock downloads don't each re-fetch a full year of Nifty + VIX data. Uses fetch_market_data() which tries NSE unofficial → Twelve Data → Yahoo. Only caches successful (non-empty) Nifty results so transient failures allow a retry on the next call instead of poisoning the cache.""" key = f"mkt_{period}" entry = _DATA_CACHE.get(key) if entry and time.time() - entry["ts"] < _DATA_CACHE_TTL: return entry["nifty_c"], entry["vix_c"] days = {"1y": 365, "2y": 730}.get(period, 365) nifty_c, vix_c = fetch_market_data(period_days=days) if nifty_c is None or len(nifty_c) == 0: # Don't cache a failed result — allow immediate retry on next call logging.warning("Market data fetch returned empty Nifty — not caching, will retry") return pd.Series(dtype=float), vix_c _DATA_CACHE[key] = {"ts": time.time(), "nifty_c": nifty_c, "vix_c": vix_c} return nifty_c, vix_c # ── DATA LOADER ─────────────────────────────────────────────────────────────── def _load_ticker_data(ticker: str, period: str = "1y"): """ Download OHLCV for ticker using the multi-source fallback chain. Nifty + VIX come from the 5-min cache (fetched once across all parallel calls). OHLCV is also cached for 5 minutes so multiple TF predictions for the same ticker (watchlist 1D/3D/5D flow) share a single Yahoo Finance download. Returns (sc, sh, sl, sv, nifty_c, vix_c) suitable for signal generators. """ nifty_c, vix_c = _get_market_cache(period) key = f"{ticker}_{period}" # Fast path: no lock needed when already cached. entry = _OHLCV_CACHE.get(key) if entry and time.time() - entry["ts"] < _OHLCV_CACHE_TTL: sc, sh, sl, sv = entry["ohlcv"] return sc, sh, sl, sv, nifty_c, vix_c # Slow path: serialize concurrent fetches for the same key so parallel TF # threads don't all hit the network at once for the same ticker. with _OHLCV_FETCH_LOCKS_LOCK: if key not in _OHLCV_FETCH_LOCKS: _OHLCV_FETCH_LOCKS[key] = threading.Lock() fetch_lock = _OHLCV_FETCH_LOCKS[key] with fetch_lock: entry = _OHLCV_CACHE.get(key) if entry and time.time() - entry["ts"] < _OHLCV_CACHE_TTL: sc, sh, sl, sv = entry["ohlcv"] return sc, sh, sl, sv, nifty_c, vix_c sc, sh, sl, sv = fetch_ohlcv(ticker, period=period) _OHLCV_CACHE[key] = {"ts": time.time(), "ohlcv": (sc, sh, sl, sv)} return sc, sh, sl, sv, nifty_c, vix_c # ── MARKET GATES (HARD ENFORCEMENT) ───────────────────────────────────────── def _get_vix() -> tuple[float, str]: _, vix_c = _get_market_cache() if vix_c is None or len(vix_c) == 0: return 18.0, "MODERATE (18.0) — VIX unavailable, using fallback" v = float(vix_c.iloc[-1]) import math if v <= 0 or math.isnan(v): logging.warning("VIX value invalid (%s) — using neutral 18.0 to prevent Mode B gate misfiring", v) return 18.0, "MODERATE (18.0) — VIX invalid, using fallback" if v < 15: label = f"LOW ({v:.1f}) — Full size" elif v < 20: label = f"MODERATE ({v:.1f}) — Use 67% position size" elif v < 25: label = f"HIGH ({v:.1f}) — Use 50% position size" else: label = f"EXTREME ({v:.1f}) — NO NEW TRADES" return v, label def _get_vix_declining() -> bool: """Returns True if VIX 5-day EMA slope is negative (declining trend). When combined with VIX<18 this is the Mode B condition that pushes S6 from 65% → 70.9% accuracy (HIGH confidence, N=141 in 276-stock backtest). """ _, vix_c = _get_market_cache() if vix_c is None or len(vix_c) < 6: return False slope = float(vix_c.ewm(span=5).mean().diff().iloc[-1]) return slope < 0 def _get_nifty_gate() -> tuple[bool, str]: """Returns (is_above_ema200, description). Fetches 2 years of Nifty data so EMA200 has ~304 warm bars (vs ~52 with 1y), making the current EMA200 level accurate. Requires 3 of the last 5 closes below the warm EMA200 to trigger CAUTION — majority-of-week rule filters noise.""" nifty_c, _ = _get_market_cache("2y") # 2y → EMA200 is fully warmed up if nifty_c is None or len(nifty_c) < 3: return True, "Nifty data unavailable — assuming OK" ema200_s = nifty_c.ewm(span=200).mean() ema200 = float(ema200_s.iloc[-1]) price = float(nifty_c.iloc[-1]) window = min(5, len(nifty_c)) below_days = int((nifty_c.iloc[-window:] < ema200_s.iloc[-window:]).sum()) above = below_days < 3 # 3+ of 5 days below triggers defensive mode label = ( f"OK — Nifty ₹{price:,.0f} above EMA200 ₹{ema200:,.0f}" if above else f"CAUTION — Nifty ₹{price:,.0f} BELOW EMA200 ₹{ema200:,.0f} ({below_days}/5 days, defensive mode)" ) return above, label _MACRO_GATE_CACHE: dict = {"data": None, "ts": 0.0} _MACRO_GATE_TTL = 300 # 5 minutes — macro conditions don't change intraday def _get_macro_gate() -> tuple[bool, str]: """Returns (global_risk_on, description). Results cached for 5 minutes.""" if not _HAS_MACRO: return True, "Macro data unavailable — assuming Risk-ON" now = time.time() if _MACRO_GATE_CACHE["data"] is not None and (now - _MACRO_GATE_CACHE["ts"]) < _MACRO_GATE_TTL: return _MACRO_GATE_CACHE["data"] end = datetime.now().strftime("%Y-%m-%d") start = (datetime.now() - timedelta(days=90)).strftime("%Y-%m-%d") mc = MacroContext() mc.load(start, end) # _features is shifted by 1 day so today's date is never in the index. # Use build_mask with ffill to get the most recent available T-1 row. _today_ts = pd.Timestamp.now().normalize() _mask = mc.build_mask(pd.DatetimeIndex([_today_ts])) ok = bool(_mask.iloc[0]) if not _mask.empty else True # Re-fetch the full feature row for the description. feat = mc.get(_today_ts - pd.Timedelta(days=1)) if not feat: # Fallback: scan backwards up to 5 days for a valid row. for _d in range(1, 6): feat = mc.get(_today_ts - pd.Timedelta(days=_d)) if feat: break parts = [] if not feat.get("sp500_trend", True): parts.append("S&P500 weak") if not feat.get("usdinr_stable", True): parts.append("USD/INR volatile") if feat.get("crude_spike", False): parts.append("crude spike") desc = "Risk-ON" if ok else f"Risk-OFF ({', '.join(parts) or 'macro headwinds'})" result = (ok, desc) _MACRO_GATE_CACHE["data"] = result _MACRO_GATE_CACHE["ts"] = time.time() return result # ── EARNINGS BLACKOUT ───────────────────────────────────────────────────────── def get_earnings_status(ticker: str, window: int = 5) -> dict: """Check if ticker has earnings within `window` calendar days.""" try: t = yf.Ticker(ticker) ed = t.earnings_dates if ed is None or ed.empty: return {"next_date": None, "days_away": None, "in_blackout": False, "warning": None} today = datetime.now().date() # Sort descending so the nearest upcoming date is encountered first. # yfinance returns earnings_dates in undocumented order — sorting avoids # returning a distant date when a near one is also in the window. for edate in sorted(ed.index, key=lambda d: abs((d.date() - today).days) if hasattr(d, "date") else abs((pd.Timestamp(d).date() - today).days)): edate_d = edate.date() if hasattr(edate, "date") else pd.Timestamp(edate).date() days_away = (edate_d - today).days if -window <= days_away <= 30: in_blackout = abs(days_away) <= window warning = ( f"Earnings {'in' if days_away >= 0 else 'were'} " f"{abs(days_away)} day{'s' if abs(days_away) != 1 else ''} " f"{'away' if days_away >= 0 else 'ago'} — 5-day blackout applies" if in_blackout else None ) return { "next_date": str(edate_d), "days_away": days_away, "in_blackout": in_blackout, "warning": warning, } except Exception: pass return {"next_date": None, "days_away": None, "in_blackout": False, "warning": None} # ── STRATEGY SIGNAL RUNNER ──────────────────────────────────────────────────── def run_strategy_signals( ticker: str, sc: pd.DataFrame, sh: pd.DataFrame, sl: pd.DataFrame, sv: pd.DataFrame, nifty_c: pd.Series, lookback_days: int = 5, vix_c=None, ) -> dict: """ Run S1–S6/MFS/NIRA/PED/SUPER on recent data. A signal is 'active' if it fired within the last `lookback_days` trading days. """ recent = set(sc.index[-lookback_days:]) def _fired(sigs): return any(d in recent and t == ticker for d, t in sigs) results = {} signal_errors: list[str] = [] def _run(name, fn): try: results[name] = _fired(fn()) except Exception as e: results[name] = False logging.warning("Signal %s failed for %s: %s", name, ticker, e) signal_errors.append(f"{name}: {e}") _run("S1", lambda: gen_s1(sc, sh, sl, sv, nifty_c)) _run("S2", lambda: gen_s2(sc, sh, sl, sv, nifty_c)) _run("S3", lambda: gen_s3(sc, sh, sl, sv)) _run("MFS", lambda: gen_mfs(sc, sh, sl, sv, nifty_c)) _run("NIRA", lambda: gen_nira(sc, sh, sl, sv, nifty_c)) _run("PED", lambda: gen_ped(sc, sh, sl, sv)) _run("SUPER", lambda: gen_supertrend(sc, sh, sl)) _run("S4", lambda: gen_s4(sc, sh, sl, sv, vix_c)) _run("S5", lambda: gen_s5(sc, sh, sl, sv, vix_c)) _run("S6", lambda: gen_s6(sc, sh, sl, sv, nifty_c, vix_c)) # ── New high-accuracy strategies (v2 tightened + S7-S11) ───────────────── _run("S4V2", lambda: gen_s4v2(sc, sh, sl, sv, nifty_c, vix_c)) _run("S5V2", lambda: gen_s5v2(sc, sh, sl, sv, nifty_c, vix_c)) _run("S6V2", lambda: gen_s6v2(sc, sh, sl, sv, nifty_c, vix_c)) _run("S7", lambda: gen_s7(sc, sh, sl, sv, nifty_c, vix_c)) _run("S8", lambda: gen_s8(sc, sh, sl, sv, nifty_c, vix_c)) _run("S9", lambda: gen_s9(sc, sh, sl, sv, nifty_c, vix_c)) _run("S10", lambda: gen_s10(sc, sh, sl, sv, nifty_c, vix_c)) _run("S11", lambda: gen_s11(sc, sh, sl, sv, nifty_c, vix_c)) # ── v4 Research-derived strategies (Jun 2026) ───────────────────────────── _run("S_CAPFLOW", lambda: gen_s_capflow(sc, sh, sl, sv, nifty_c, vix_c)) _run("S_CTRIO", lambda: gen_s_confluence_trio(sc, sh, sl, sv, nifty_c, vix_c)) _run("S_SEASONAL", lambda: gen_s_seasonal(sc, sh, sl, sv, nifty_c, vix_c)) # ── v5 New high-accuracy strategies (documented NSE >75% + technical) ───── _run("S12", lambda: gen_s12(sc, sh, sl, sv, nifty_c, vix_c)) _run("S13", lambda: gen_s13(sc, sh, sl, sv, nifty_c, vix_c)) _run("S14", lambda: gen_s14(sc, sh, sl, sv, nifty_c, vix_c)) _run("S15", lambda: gen_s15(sc, sh, sl, sv, nifty_c, vix_c)) _run("S16", lambda: gen_s16(sc, sh, sl, sv, nifty_c, vix_c)) _run("S17", lambda: gen_s17(sc, sh, sl, sv, nifty_c, vix_c)) _run("S18", lambda: gen_s18(sc, sh, sl, sv, nifty_c, vix_c)) _run("S19", lambda: gen_s19(sc, sh, sl, sv, nifty_c, vix_c)) _run("S20", lambda: gen_s20(sc, sh, sl, sv, nifty_c, vix_c)) active = [k for k, v in results.items() if v] return {"signals": results, "active": active, "count": len(active), "signal_errors": signal_errors} # ── ML FEATURE SCORE ────────────────────────────────────────────────────────── def get_ml_feature_score( ticker: str, sc: pd.DataFrame, sh: pd.DataFrame, sl: pd.DataFrame, sv: pd.DataFrame, nifty_c: pd.Series, vix_c: Optional[pd.Series], ) -> dict: """ Compute the 10 ml_combiner features for the current bar. Returns a 0–100 composite score and individual feature values. Higher score = historically more bullish feature state. """ c = sc[ticker].dropna() h = sh[ticker].reindex(c.index).ffill() l = sl[ticker].reindex(c.index).ffill() v = sv[ticker].reindex(c.index).ffill() if len(c) < 30: return {"score": 50, "probability": 0.5, "features": {}} # ── Feature 1: RSI (inverted: low RSI = bullish for mean reversion) ──── rsi_val = float(rsi(c).iloc[-1]) # Rescale: RSI 20=bullish(+1), RSI 80=bearish(-1) rsi_feat = (50 - rsi_val) / 50 # +1 at RSI=0, -1 at RSI=100 # ── Feature 2: Bollinger position (0=lower band, 1=upper band) ───────── bb_pos_val = float(bollinger_position(c).iloc[-1]) if len(c) >= 20 else 0.5 bb_feat = 1 - bb_pos_val # low bb_pos = bullish (near lower band) # ── Feature 3: EMA stack (0–1, higher = more bullish) ────────────────── ema_stack_val = float(ema_stack_score(c).iloc[-1]) if len(c) >= 20 else 0.5 # ── Feature 4: Volume ratio ───────────────────────────────────────────── v20 = float(v.rolling(20).mean().iloc[-1]) vol_ratio = float(v.iloc[-1]) / v20 if v20 > 0 else 1.0 vol_feat = min(vol_ratio / 2.0, 1.0) # 2x volume = max score # ── Feature 5: OBV z-score ───────────────────────────────────────────── ob = obv(c, v) ob_mu = ob.rolling(20).mean().iloc[-1] ob_std = ob.rolling(20).std().iloc[-1] obv_z = float((ob.iloc[-1] - ob_mu) / ob_std) if ob_std and ob_std > 0 else 0.0 obv_feat = max(-1.0, min(1.0, obv_z / 2.0)) # normalize to -1..+1 # ── Feature 6: Relative strength vs Nifty (3M) ───────────────────────── ni = nifty_c.reindex(c.index).ffill() rs3m = float((c.iloc[-1] / c.iloc[-63] - 1) - (ni.iloc[-1] / ni.iloc[-63] - 1)) if len(c) >= 63 else 0.0 rs_feat = max(-1.0, min(1.0, rs3m * 5)) # ±20% RS = ±1 # ── Feature 7: MACD histogram sign ───────────────────────────────────── mh = macd_h(c) macd_val = float(mh.iloc[-1]) if len(c) >= 27 else 0.0 macd_feat = 1.0 if macd_val > 0 else 0.0 # ── Feature 8: ADX (trend strength) ──────────────────────────────────── adx_val = float(adx_s(h, l, c).iloc[-1]) if len(c) >= 15 else 20.0 adx_feat = min(adx_val / 50.0, 1.0) # ── Feature 9: Shadow recovery (intraday bounce) ──────────────────────── shadow_val = float(shadow_flag(c, l).iloc[-1]) if len(c) >= 20 else 0.0 # ── Feature 10: VIX level (lower = better environment) ───────────────── vix_feat = 0.5 if vix_c is not None and not vix_c.empty: vix_val = float(vix_c.dropna().iloc[-1]) vix_feat = max(0.0, 1.0 - vix_val / 30.0) # VIX=0→1.0, VIX=30→0.0 # ── Feature 11: Supertrend direction (price above ST line = bullish) ──── def _st_dir_last(cs, hs, ls, period=10, mult=3.0): if len(cs) < period + 1: return 1 atr_v = atr(hs, ls, cs, period) # atr(high, low, close, n) hl2 = (hs + ls) / 2 up_raw = (hl2 + mult * atr_v).values dn_raw = (hl2 - mult * atr_v).values cv = cs.values n = len(cv) upper, lower = up_raw.copy(), dn_raw.copy() dirn = np.ones(n, dtype=int) for i in range(1, n): upper[i] = min(up_raw[i], upper[i-1]) if cv[i-1] <= upper[i-1] else up_raw[i] lower[i] = max(dn_raw[i], lower[i-1]) if cv[i-1] >= lower[i-1] else dn_raw[i] if cv[i] > upper[i-1]: dirn[i] = 1 elif cv[i] < lower[i-1]: dirn[i] = -1 else: dirn[i] = dirn[i-1] return int(dirn[-1]) st_feat = 1.0 if _st_dir_last(c, h, l) > 0 else 0.0 # ── Weighted composite (weights reflect LR-like importance from backtest) ─ weights = { "ema_stack": 0.22, # trend alignment — most predictive in S2/S3 "rs_3m": 0.17, # relative strength — key S2/MFS filter "obv": 0.13, # volume accumulation "macd": 0.12, # momentum confirmation "vol_ratio": 0.10, # conviction signal "adx": 0.08, # trend strength "supertrend": 0.06, # trend direction (Supertrend 10,3) "rsi": 0.05, # oversold (S1 specific) "bb_pos": 0.03, # mean reversion zone "vix": 0.02, # macro environment "shadow": 0.02, # S1 intraday bounce } feat_vals = { "ema_stack": _fn(ema_stack_val, 0.5), "rs_3m": _fn((rs_feat + 1) / 2, 0.5), "obv": _fn((obv_feat + 1) / 2, 0.5), "macd": _fn(macd_feat, 0.5), "vol_ratio": _fn(vol_feat, 0.5), "adx": _fn(adx_feat, 0.5), "supertrend": _fn(st_feat, 0.5), "rsi": _fn((rsi_feat + 1) / 2, 0.5), "bb_pos": _fn(bb_feat, 0.5), "vix": _fn(vix_feat, 0.5), "shadow": _fn(shadow_val, 0.0), } raw_composite = sum(weights[k] * feat_vals[k] for k in weights) score = int(round(raw_composite * 100)) # Convert to probability (logistic-style, steepness=8) prob = 1 / (1 + math.exp(-8 * (raw_composite - 0.5))) # ── Two-tier gate (mirrors random_ai_test two-tier logic) ──────────────── # Hard prerequisites: vol_ratio, macd, RSI must all be clearly bullish/bearish # before allowing HIGH probability. Any failure caps at 0.72 (MEDIUM ceiling). # This prevents the weighted composite from reaching HIGH on EMA alignment # alone when volume is thin or momentum has already turned. vol_feat_raw = feat_vals.get("vol_ratio", 0.5) # 0..1 (2× avg = 1.0) macd_feat_raw = feat_vals.get("macd", 0.5) # 1.0=positive, 0.0=negative rsi_feat_raw = feat_vals.get("rsi", 0.5) # higher = more bullish (oversold) adx_feat_raw = feat_vals.get("adx", 0.5) bull_hard_ok = ( vol_feat_raw >= 0.45 and # vol ≥ 0.9× avg (normalised: 0.9/2=0.45) macd_feat_raw >= 0.9 and # MACD histogram positive adx_feat_raw >= 0.4 # ADX > 20 (trending market) # Note: Nifty 5D momentum checked externally via nifty_ok in _calc_confidence ) bear_hard_ok = ( vol_feat_raw >= 0.6 and macd_feat_raw <= 0.1 and # MACD firmly negative adx_feat_raw >= 0.4 ) if raw_composite > 0.5 and not bull_hard_ok: prob = min(prob, 0.72) elif raw_composite < 0.5 and not bear_hard_ok: prob = max(prob, 0.28) return { "score": score, "probability": _sf(prob, 3, 0.5), "upgraded": bool(prob > 0.65) if math.isfinite(prob) else False, "features": { "rsi": _sf(rsi_val, 1), "bb_pos": _sf(bb_pos_val, 2), "ema_stack": _sf(ema_stack_val, 2), "vol_ratio": _sf(vol_ratio, 2), "obv_z": _sf(obv_z, 2), "rs_3m_pct": _sf(rs3m * 100, 1), "macd_pos": bool(macd_val > 0) if math.isfinite(macd_val) else None, "adx": _sf(adx_val, 1), "shadow": bool(shadow_val), "supertrend": bool(st_feat), }, } # ── DIRECTIONAL PRICE FORECAST (ML-based, independent of signal layer) ──────── def _directional_forecast( ml_probability: float, tf_label: str, nifty_ok: bool, vix_level: float, ) -> dict: """ Predicts price direction (BULLISH / BEARISH / NEUTRAL) from the ML feature probability. This is a price forecast — separate from the trade recommendation (direction field, which requires a buy signal to fire). Returns a dict with predicted_direction, predicted_return_lo, predicted_return_hi. """ tf_scale = {"INTRADAY": 0.20, "1D": 0.35, "3D": 0.65, "5D": 1.0, "1W": 1.3}.get(tf_label, 1.0) regime_mult = 0.7 if not nifty_ok else 1.0 vix_mult = 0.8 if vix_level > 20 else 1.0 p_bull = ml_probability p_bear = 1.0 - ml_probability if p_bull >= 0.60: strength = (p_bull - 0.60) / 0.40 # 0→1 as probability goes 0.60→1.00 lo = round((0.5 + strength * 2.0) * tf_scale * regime_mult * vix_mult, 2) hi = round((1.5 + strength * 5.0) * tf_scale * regime_mult * vix_mult, 2) return {"predicted_direction": "BULLISH", "predicted_return_lo": lo, "predicted_return_hi": hi} elif p_bear >= 0.60: strength = (p_bear - 0.60) / 0.40 # Bearish returns are negative; regime dampener not applied (bear markets can be fast) hi = round(-(0.5 + strength * 2.0) * tf_scale, 2) lo = round(-(1.5 + strength * 5.0) * tf_scale, 2) return {"predicted_direction": "BEARISH", "predicted_return_lo": lo, "predicted_return_hi": hi} else: lo = round(-0.5 * tf_scale, 2) hi = round(0.5 * tf_scale, 2) return {"predicted_direction": "NEUTRAL", "predicted_return_lo": lo, "predicted_return_hi": hi} # ── EXPECTED RETURN CALCULATOR ──────────────────────────────────────────────── def _calc_expected_return( active_strategies: list[str], ml_upgraded: bool, n_trading_days: int, nifty_ok: bool, macro_ok: bool, vix_level: float, ) -> tuple[float, float, str, dict]: """ Returns (lo%, hi%, direction, backtest_stats) based on actual backtest stats. backtest_stats contains raw (undampened) historical performance for UI display. """ scale = 1.0 if not active_strategies: return 0.0, 0.0, "NO TRADE", {} # Compute weighted EV from each active strategy's backtest stats evs, win_rates, avg_wins, avg_losses = [], [], [], [] for s in active_strategies: if s in _STRATEGY_STATS: wr, avg_win, avg_loss = _STRATEGY_STATS[s] ev = wr * avg_win - (1 - wr) * avg_loss evs.append(ev) win_rates.append(wr) avg_wins.append(avg_win) avg_losses.append(avg_loss) if not evs: return 0.0, round(3.0 * scale, 2), "SLIGHTLY BULLISH", {} avg_ev = sum(evs) / len(evs) avg_wr = sum(win_rates) / len(win_rates) raw_avg_win = sum(avg_wins) / len(avg_wins) raw_avg_loss = sum(avg_losses) / len(avg_losses) # Signal count multiplier (more convergent signals = wider expected range) n_sig = len(active_strategies) signal_mult = 1.0 + 0.15 * (n_sig - 1) # +15% per additional signal # ML upgrade adds 10% ml_mult = 1.10 if ml_upgraded else 1.0 # Nifty below EMA200 = reduce by 40% nifty_mult = 0.60 if not nifty_ok else 1.0 # Macro risk-off = reduce by 20% macro_mult = 0.80 if not macro_ok else 1.0 # VIX adjustment vix_mult = 1.0 if vix_level < 20 else 0.70 adjusted_ev = avg_ev * signal_mult * ml_mult * nifty_mult * macro_mult * vix_mult * scale lo = round(min(adjusted_ev * 0.4, adjusted_ev * 1.8), 2) # conservative bound hi = round(max(adjusted_ev * 0.4, adjusted_ev * 1.8), 2) # optimistic bound if avg_wr >= 0.60 and n_sig >= 2: direction = "BULLISH" elif avg_wr >= 0.50 or n_sig >= 1: direction = "SLIGHTLY BULLISH" else: direction = "NEUTRAL" # Downgrade if regime is adverse if not nifty_ok or vix_level > 20: direction = "SLIGHTLY BULLISH" if direction == "BULLISH" else "NEUTRAL" # Build dampener description for UI dampeners = [] if not nifty_ok: dampeners.append("Nifty= 20: dampeners.append("VIX≥20 −30%") total_dampener = round((1 - nifty_mult * macro_mult * vix_mult) * 100) backtest_stats = { "win_rate_pct": round(avg_wr * 100, 1), "avg_win_pct": round(raw_avg_win, 2), "avg_loss_pct": round(raw_avg_loss, 2), "dampener_pct": total_dampener, "dampeners": dampeners, "signals_used": [s for s in active_strategies if s in _STRATEGY_STATS], } return lo, hi, direction, backtest_stats def _calc_confidence( n_signals: int, ml_upgraded: bool, news_label: str, nifty_ok: bool, vix_level: float, active_strategies: list | None = None, vix_declining: bool = False, ml_probability: float = 0.5, nifty_trending: bool = False, qm_score: float = 0.0, stage2_breadth: float = 0.5, fii_regime: str = "NEUTRAL", pcr: float | None = None, macro_ok: bool = False, ) -> tuple[str, dict]: """ Confidence scoring with VIX-gated Mode B and Mode S (strict) bonuses. Returns (confidence_label, breakdown_dict). Mode B (VIX<18+declining): backtest-verified boost. S6 Mode B 3D = 70.9% HIGH (N=141, excess +11.3%) S8 Mode A 3D = HIGH (N=282, excess +5.7%) S6v2 Mode B 1M = 76.5% — first above-75% result Mode S (strict): VIX<15 + Nifty trending + ML>0.62 + 2+ signals. Adds +4 bonus (vs Mode B's +3) to push reliable HIGH above 75% threshold. S17 Quality-Momentum: normalized 12M return/vol > 1.5 adds +2 bonus. BacktestIndia.com 18.5yr study: 78% annual win rate. Weinstein Stage 2 breadth gate: <30% stocks advancing = bear market penalty. FII flow and PCR contrarian adjustments for institutional regime. """ active_strategies = active_strategies or [] breakdown: dict = {} base = 0 base += n_signals * 2 base += 3 if ml_upgraded else 0 base += 2 if news_label == "BULLISH" else (-1 if news_label == "BEARISH" else 0) base -= 3 if not nifty_ok else 0 base -= 2 if vix_level > 20 else 0 breakdown["base"] = base score = base # S_CTRIO=71% HIGH, S6=68.3% Mode B HIGH, S8=59.3% HIGH, S_SEASONAL=70.8% NSE-verified # S17/S18/S19/S20 verified WEAK-LOW on NSE stocks (none reached Mode B ≥65%) high_acc_signals = {"S6", "S6V2", "S8", "S11", "S7", "S_CTRIO", "S_SEASONAL", "S16"} has_high_acc = bool(set(active_strategies) & high_acc_signals) # ── Mode B bonus (VIX<18 + declining) ────────────────────────────────────── vix_mode_b = vix_level < 18 and vix_declining mode_b_bonus = 3 if (vix_mode_b and has_high_acc) else 0 score += mode_b_bonus breakdown["mode_b"] = mode_b_bonus # ── Mode S bonus (strict: VIX<15 + Nifty trending up + ML>0.62 + 2+ signals) ── vix_mode_s = vix_level < 15 and nifty_ok and nifty_trending mode_s_bonus = 4 if (vix_mode_s and n_signals >= 2 and ml_probability > 0.62 and has_high_acc) else 0 score += mode_s_bonus breakdown["mode_s"] = mode_s_bonus # ── S_CTRIO extra (71.0% HIGH, excess +16.9% vs Nifty — system's best verified signal) ── ctrio_bonus = 3 if ("S_CTRIO" in active_strategies and (vix_mode_b or vix_mode_s)) else 0 score += ctrio_bonus breakdown["ctrio_bonus"] = ctrio_bonus # ── Quality-Momentum bonus (BacktestIndia 18.5yr: 78% annual win rate) ── qm_bonus = 2 if (qm_score > 1.5 and nifty_ok) else 0 score += qm_bonus breakdown["qm_bonus"] = qm_bonus # ── Weinstein Stage 2 breadth gate ───────────────────────────────────────── if stage2_breadth < 0.30: breadth_adj = -3 # bear market: <30% of stocks in Stage 2 (advancing) elif stage2_breadth > 0.60: breadth_adj = 1 # strong bull else: breadth_adj = 0 score += breadth_adj breakdown["breadth_adj"] = breadth_adj # ── FII flow regime bonus/penalty ────────────────────────────────────────── fii_adj = {"FII_STRONG_BUY": 2, "FII_BUY": 1, "NEUTRAL": 0, "FII_SELLING_DII_ABSORBING": 0, "RISK_OFF": -2}.get(fii_regime, 0) score += fii_adj breakdown["fii_adj"] = fii_adj # ── PCR contrarian adjustment ─────────────────────────────────────────────── if pcr is not None: pcr_adj = 1 if pcr > 1.25 else (-1 if pcr < 0.80 else 0) else: pcr_adj = 0 score += pcr_adj breakdown["pcr_adj"] = pcr_adj # ── Mode C bonus (full macro alignment: VIX<18 declining + macro_ok + HIGH signal) ── # Mode C results: S4v2=78.5%, S8=73.3%, S11=76.2%, S6v2=87.5%, SCT=75.0% mode_c_bonus = 2 if (macro_ok and vix_mode_b and has_high_acc) else 0 score += mode_c_bonus breakdown["mode_c"] = mode_c_bonus breakdown["total"] = score if score >= 7 and nifty_ok: label = "HIGH" elif score >= 3: label = "MEDIUM" else: label = "LOW" return label, breakdown # ── MAIN PREDICTION API ─────────────────────────────────────────────────────── def predict_stock_v2(ticker: str, start_date: str, end_date: str, _market_ctx: dict | None = None, _run_ai_forecast: bool = False, _ai_fast_mode: bool = False, _ai_fast_fail_on_rate_limit: bool = False, _skip_fresh_fetch: bool = False, _skip_news: bool = False) -> dict: """ Full prediction for one NSE ticker over [start_date, end_date]. Pass _market_ctx (from rank_stocks_v2) to skip redundant market gate fetches. Pass _run_ai_forecast=True (from watchlist) to call Claude with strategy context. Pass _skip_fresh_fetch=True to use cached OHLCV + live price (avoids network timeouts). """ _init_universe() company = TICKER_NAMES.get(ticker, ticker.replace(".NS", "").replace(".BO", "")) # ── Prediction cache (5-min TTL) — skip full pipeline on rapid reloads ─── _pred_key = ( f"{ticker}|{start_date}|{end_date}|{int(_run_ai_forecast)}" f"|{int(_ai_fast_mode)}|{int(_ai_fast_fail_on_rate_limit)}" ) _pred_hit = _PRED_CACHE.get(_pred_key) if _pred_hit and time.time() - _pred_hit["ts"] < _PRED_CACHE_TTL: return _pred_hit["result"] # ── Request context (tracks data merges, validation, timing) ─────────────── ctx = RequestContext(ticker) cache_age_days = None def _normalize_news_payload(news_obj: object) -> dict: """Guarantee a stable news payload shape for downstream fields.""" if not isinstance(news_obj, dict): news_obj = {} label = str(news_obj.get("label") or "NEUTRAL").upper() if label not in ("BULLISH", "BEARISH", "NEUTRAL"): label = "NEUTRAL" try: score = int(news_obj.get("score", 0)) except Exception: score = 0 summary = str(news_obj.get("summary") or "No recent news found.") key_headline = str(news_obj.get("key_headline") or "") headlines = news_obj.get("headlines") if not isinstance(headlines, list): headlines = [] headlines_dated = news_obj.get("headlines_dated") if not isinstance(headlines_dated, list): headlines_dated = [] latest_date = str(news_obj.get("latest_date") or "") source = str(news_obj.get("source") or "none") return { "label": label, "score": score, "summary": summary, "key_headline": key_headline, "headlines": headlines, "headlines_dated": headlines_dated, "latest_date": latest_date, "source": source, } # ── Trading days in range ──────────────────────────────────────────────── try: s = datetime.strptime(start_date, "%Y-%m-%d") e = datetime.strptime(end_date, "%Y-%m-%d") n_trading = max(1, int((e - s).days * 5 / 7)) except Exception: n_trading = 10 # ── Market gates (HARD enforcement) ───────────────────────────────────── if _market_ctx: vix_level = _market_ctx["vix_level"] vix_label = _market_ctx["vix_label"] nifty_ok = _market_ctx["nifty_ok"] nifty_label = _market_ctx["nifty_label"] macro_ok = _market_ctx["macro_ok"] macro_label = _market_ctx["macro_label"] else: vix_level, vix_label = _get_vix() nifty_ok, nifty_label = _get_nifty_gate() macro_ok, macro_label = _get_macro_gate() # Hard block: VIX > 25 if vix_level > 25: return { "ticker": ticker, "company": company, "start_date": start_date, "end_date": end_date, "direction": "NO TRADE", "confidence": "BLOCKED", "no_trade_reason": "vix_block", "reason": f"India VIX {vix_level:.1f} > 25 — risk rules prohibit new positions", "expected_return_range": "N/A", "ret_lo": 0, "ret_hi": 0, "midpoint": 0, "signal_count": 0, "vix": {"level": vix_level, "label": vix_label}, "nifty_gate": nifty_label, "macro": macro_label, "signals": {}, "active_strategies": [], "ml": {}, "news": {}, "earnings": {}, "price": None, } # ── Download data ───────────────────────────────────────────────────────── try: sc, sh, sl, sv, nifty_c, vix_c = _load_ticker_data(ticker, period="2y") if ticker not in sc.columns or sc[ticker].dropna().empty: return {"ticker": ticker, "error": "No price data available"} bars = sc[ticker].dropna() if len(bars) < 200: return {"ticker": ticker, "error": f"Insufficient history: {len(bars)} bars (need 200+)"} if nifty_c is None or len(nifty_c) == 0: # Degrade gracefully: Nifty-dependent signals will return False, # but price-only signals (S3, PED, SUPER, etc.) still run. _warn_nifty_unavailable_once() nifty_c = pd.Series(dtype=float) price = float(bars.iloc[-1]) # Production predictions: start_date == today. Backtest passes historical dates. # start_date is a "YYYY-MM-DD" string — parse to a date before comparing. if isinstance(start_date, str): _start_d = datetime.strptime(start_date, "%Y-%m-%d").date() elif isinstance(start_date, datetime): _start_d = start_date.date() else: _start_d = start_date _is_today_prediction = (_start_d >= _date_cls.today()) except Exception as ex: return {"ticker": ticker, "error": str(ex)} # ── Run strategy signals ────────────────────────────────────────────────── sig_result = run_strategy_signals(ticker, sc, sh, sl, sv, nifty_c, vix_c=vix_c) # ── ML feature score ────────────────────────────────────────────────────── ml = get_ml_feature_score(ticker, sc, sh, sl, sv, nifty_c, vix_c) # ── Parallel I/O: news + FII/PCR + intraday (concurrent network calls) ───── def _fetch_news_io(): if _skip_news: return {"label": "NEUTRAL", "score": 0, "source": "skip", "summary": "", "headlines": [], "key_headline": ""} return fetch_and_analyze(ticker, company) def _fetch_fii_io(): try: from fii_flow import get_fii_dii_flow, get_nifty_pcr _fii = get_fii_dii_flow() _regime = _fii.get("regime", "NEUTRAL") _data = {"fii_net": _fii.get("fii_net"), "dii_net": _fii.get("dii_net"), "regime": _regime} _pcr = get_nifty_pcr() return _regime, _data, (float(_pcr) if _pcr is not None else None) except Exception: return "NEUTRAL", {"fii_net": None, "dii_net": None, "regime": "NEUTRAL"}, None def _fetch_intraday_io(): try: from intraday_live import get_live_intraday_context return get_live_intraday_context(ticker) except Exception as _id_err: logging.debug("intraday_live failed for %s: %s", ticker, _id_err) return {"data_available": False, "reason": "import or download failed"} def _fetch_sector_io(): if not _HAS_SECTOR_PULSE: return None try: return get_sector_pulse() except Exception as _sp_err: logging.debug("sector_pulse failed: %s", _sp_err) return None # Fundamentals are expensive and yfinance-backed. Restrict by default to # AI/watchlist flows so universe ranking does not hammer the provider. _fetch_fundamentals_enabled = _run_ai_forecast or (os.getenv("ENABLE_FUNDAMENTALS_ON_RANK", "0") == "1") def _fetch_fundamentals_io(): if not _fetch_fundamentals_enabled: return None if not _HAS_FUNDAMENTALS: return None try: return get_fundamentals(ticker) except Exception as _fu_err: logging.debug("fundamentals failed for %s: %s", ticker, _fu_err) return None def _fetch_earnings_io(): return get_earnings_status(ticker) def _fetch_live_price_io(): # Only meaningful for live/today predictions; backtest always returns None. if not _is_today_prediction: return None try: lp = fetch_live_price(ticker, allow_delayed=True) return float(lp) if lp and lp > 0 else None except Exception: return None _io_workers = 5 + (1 if _fetch_fundamentals_enabled else 0) + 1 # +1 for live price _IO_TIMEOUT = 10 # single wall-clock cap for the whole set (not per-future) # NOTE: do NOT use `with ThreadPoolExecutor(...) as _pool:` here — the context # manager's __exit__ calls shutdown(wait=True), which blocks until every # already-running task finishes (news_sentiment's LLM call can run far longer # than _IO_TIMEOUT on provider failover/lock contention). That silently defeated # the cap and wasted the extra wait, since _safe_result() below had already # fallen back to the default by then — surfacing as "news failing" on the first # watchlist check for a ticker (the straggler's result populates news_sentiment's # own cache in the background, so the next check reads it correctly). Explicit # executor + shutdown(wait=False) makes the timeout a true hard cap. _pool = ThreadPoolExecutor(max_workers=_io_workers) try: _f_news = _pool.submit(_fetch_news_io) _f_fii = _pool.submit(_fetch_fii_io) _f_intraday = _pool.submit(_fetch_intraday_io) _f_sector = _pool.submit(_fetch_sector_io) _f_earnings = _pool.submit(_fetch_earnings_io) _f_fundamentals = _pool.submit(_fetch_fundamentals_io) if _fetch_fundamentals_enabled else None _f_live = _pool.submit(_fetch_live_price_io) # Single wall-clock wait instead of 6 sequential result(timeout=10) calls. # Previously worst-case was 6 × 10s = 60s; now it's _IO_TIMEOUT seconds total. _all_futs = [f for f in [_f_news, _f_fii, _f_intraday, _f_sector, _f_earnings, _f_fundamentals, _f_live] if f] import concurrent.futures as _cf _cf.wait(_all_futs, timeout=_IO_TIMEOUT) def _safe_result(fut, default): if fut is None: return default try: return fut.result(timeout=0) except Exception: return default news = _safe_result(_f_news, {"label": "NEUTRAL", "score": 0, "summary": "No recent news found.", "source": "none"}) fii_regime, fii_data, pcr_value = _safe_result(_f_fii, ("NEUTRAL", {}, None)) intraday_dict = _safe_result(_f_intraday, {}) sector_pulse_data = _safe_result(_f_sector, None) earnings = _safe_result(_f_earnings, {"in_blackout": False, "days_to_earnings": None}) fund_data = _safe_result(_f_fundamentals, None) _live_price = _safe_result(_f_live, None) finally: _pool.shutdown(wait=False) # For production predictions, use live price as the AI's price anchor so the model # sees the current market level (not yesterday's close). Signals and stop-loss # calculations still use the OHLCV price — only the AI context is updated. _ai_current_price = _live_price if _live_price else price news = _normalize_news_payload(news) # Sector position — is this ticker's sector leading or lagging? _ticker_sector = get_sector_for_ticker(ticker) if _HAS_SECTOR_PULSE else None _sector_leading = False _sector_lagging = False if sector_pulse_data and _ticker_sector: _sector_leading = _ticker_sector in sector_pulse_data.get("leading_sectors", []) _sector_lagging = _ticker_sector in sector_pulse_data.get("lagging_sectors", []) # ── Compute prediction ──────────────────────────────────────────────────── # Filter to strategies that are appropriate for this timeframe to avoid # diluting 1D predictions with 5D strategy stats and vice versa. # n_trading=2 is still a 1D pick — timeframe_to_dates adds a 2-day weekend buffer. # A same-day window (start == end) is the INTRADAY horizon — timeframe_to_dates # emits start == end only for INTRADAY, so this uniquely identifies it. if start_date == end_date: _tf_label = "INTRADAY" else: _tf_label = "1D" if n_trading <= 2 else ("3D" if n_trading <= 4 else ("5D" if n_trading <= 6 else "1W")) _relevant = STRATEGY_TIMEFRAME_MAP.get(_tf_label, sig_result["active"]) _active_for_ev = [s for s in sig_result["active"] if s in _relevant] ret_lo, ret_hi, direction, backtest_stats = _calc_expected_return( _active_for_ev, ml["upgraded"], n_trading, nifty_ok, macro_ok, vix_level, ) # Directional price forecast — independent of whether a buy signal fired price_forecast = _directional_forecast(ml["probability"], _tf_label, nifty_ok, vix_level) # ── Pre-compute context needed by AI forecast prompt ───────────────────── # VIX declining check — required for Mode B confidence bonus vix_declining = _get_vix_declining() # Nifty 5D trending check (for Mode S strict gate) try: nifty_trending = bool(nifty_c.iloc[-1] > nifty_c.iloc[-5]) except Exception: nifty_trending = False # S17 Quality-Momentum score (risk-adjusted 12M momentum, BacktestIndia 18.5yr 78% WR) try: _c = sc[ticker].dropna() _ret_12m = float(_c.iloc[-1] / _c.iloc[-252] - 1) if len(_c) >= 252 else 0.0 _vol_12m = float(_c.pct_change().tail(252).std() * (252 ** 0.5)) qm_score = _ret_12m / _vol_12m if _vol_12m > 0 else 0.0 except Exception: qm_score = 0.0 # Stage 2 breadth: would require the full universe sc DataFrame. # predict_stock_v2 only has the single-ticker sc, so we default to neutral (0.5) # and skip the breadth adjustment. rank_stocks_v2 could pre-compute this separately. stage2_breadth = 0.5 if stage2_breadth < 0.30: breadth_regime = "BEAR" elif stage2_breadth > 0.60: breadth_regime = "BULL" else: breadth_regime = "NEUTRAL" # Mode C: full macro alignment — VIX<18 declining + macro_ok (all global macro favorable) _vix_mode_b_pre = vix_level < 18 and vix_declining mode_c_active = bool(macro_ok and _vix_mode_b_pre) mode_c_label = ( "FULL MACRO ALIGNMENT (Mode C) — 73-87% accuracy on HIGH signals" if mode_c_active else "" ) # AI directional forecast — runs for all watchlist predictions when _run_ai_forecast=True ai_forecast: dict | None = None _af_err_msg: str | None = None _ai_soft_fail = False # True when the AI outage is transient (provider will reset soon / Ollama up) if _run_ai_forecast: try: # Build indicator snapshot in ₹ for Claude context _c = sc[ticker].dropna() _h = sh[ticker].dropna() _lo = sl[ticker].dropna() # INTRADAY: the daily OHLCV series ends at yesterday's close, so RSI / Bollinger / # momentum computed off it would describe YESTERDAY — not today's intraday move. # Append the live price as today's provisional bar so the indicators the AI reasons # about reflect the current session (e.g. a stock that has already dropped intraday # shows a lower RSI / BB position). INTRADAY-only and only when the live price differs # from the last close; backtest (no live price) and 1D/3D are unaffected. if _tf_label == "INTRADAY" and _live_price and len(_c) >= 1: try: _last_close = float(_c.iloc[-1]) if _last_close > 0 and abs(_live_price - _last_close) / _last_close > 0.001: _today_ts = pd.Timestamp.now().normalize() if len(_c.index) == 0 or _c.index[-1] != _today_ts: # Use today's intraday high/low when available, else the live price. _t_hi = _live_price _t_lo = _live_price if isinstance(intraday_dict, dict) and intraday_dict.get("data_available"): _oh = intraday_dict.get("orb_high") _ol = intraday_dict.get("orb_low") if _oh: _t_hi = max(_live_price, float(_oh)) if _ol: _t_lo = min(_live_price, float(_ol)) _c = pd.concat([_c, pd.Series([_live_price], index=[_today_ts])]) _h = pd.concat([_h, pd.Series([_t_hi], index=[_today_ts])]) if len(_h) else _h _lo = pd.concat([_lo, pd.Series([_t_lo], index=[_today_ts])]) if len(_lo) else _lo except Exception as _iv_err: logging.debug("intraday live-bar append failed for %s: %s", ticker, _iv_err) _indicators = {} _ohlcv_df = None try: # Keys must match what ai_forecast._build_context_block() reads _indicators["close"] = round(float(_c.iloc[-1]), 2) if len(_c) >= 1 else None _indicators["rsi14"] = round(float(rsi(_c, 14).iloc[-1]), 1) if len(_c) >= 14 else None _indicators["rsi5"] = round(float(rsi(_c, 5).iloc[-1]), 1) if len(_c) >= 5 else None _indicators["rsi2"] = round(float(rsi(_c, 2).iloc[-1]), 1) if len(_c) >= 2 else None # Bollinger Bands (20, 2σ) _bb_sma = _c.rolling(20).mean() _bb_std = _c.rolling(20).std() _indicators["bb_lower"] = round(float((_bb_sma - 2 * _bb_std).iloc[-1]), 2) _indicators["bb_mid"] = round(float(_bb_sma.iloc[-1]), 2) _indicators["bb_upper"] = round(float((_bb_sma + 2 * _bb_std).iloc[-1]), 2) _indicators["ema20"] = round(float(_c.ewm(span=20).mean().iloc[-1]), 2) _indicators["ema50"] = round(float(_c.ewm(span=50).mean().iloc[-1]), 2) if len(_c) >= 50 else None _indicators["ema200"] = round(float(_c.ewm(span=200).mean().iloc[-1]), 2) if len(_c) >= 200 else None # atr signature: atr(h, l, c, n=14) _atr_s = atr(_h, _lo, _c, 14) _indicators["atr14"] = round(float(_atr_s.iloc[-1]), 2) # ADX (trend strength) if len(_c) >= 15: _indicators["adx14"] = round(float(adx_s(_h, _lo, _c).iloc[-1]), 1) # Volume ratio _vol20 = sv[ticker].rolling(20).mean().iloc[-1] if ticker in sv.columns else None _vol_now = sv[ticker].iloc[-1] if ticker in sv.columns else None if _vol20 and _vol_now: _indicators["vol_ratio"] = round(float(_vol_now) / (float(_vol20) + 1e-9), 2) # MACD signal (histogram = MACD line - signal line) _macd_line = _c.ewm(span=12).mean() - _c.ewm(span=26).mean() _indicators["macd_signal"] = round(float((_macd_line - _macd_line.ewm(span=9).mean()).iloc[-1]), 4) # OBV trend if ticker in sv.columns and len(_c) >= 10: _obv_series = obv(_c, sv[ticker].dropna()) _obv_slope = _obv_series.diff(5).iloc[-1] _indicators["obv_trend"] = "rising" if _obv_slope > 0 else ("falling" if _obv_slope < 0 else "flat") # Short-term momentum — critical direction signals used by synthesis prompt _price_now = float(_c.iloc[-1]) if len(_c) >= 10: _indicators["return_10d"] = round((_price_now / float(_c.iloc[-10]) - 1) * 100, 1) if len(_c) >= 20: _indicators["return_20d"] = round((_price_now / float(_c.iloc[-20]) - 1) * 100, 1) if len(_c) >= 63: _indicators["return_90d"] = round((_price_now / float(_c.iloc[-63]) - 1) * 100, 1) if len(_c) >= 252: _hi52 = float(_c.iloc[-252:].max()) _indicators["Dist_from_52W_High_%"] = round((_price_now / _hi52 - 1) * 100, 1) # Bollinger Band position: 0% = lower band, 100% = upper band if len(_c) >= 20: _bb_u = float((_bb_sma + 2 * _bb_std).iloc[-1]) _bb_l = float((_bb_sma - 2 * _bb_std).iloc[-1]) if _bb_u > _bb_l: _indicators["bb_pct"] = round((_price_now - _bb_l) / (_bb_u - _bb_l) * 100, 1) # Consecutive up/down days if len(_c) >= 6: _diffs = _c.iloc[-6:].diff().dropna() _up = _dn = 0 for _d in reversed(_diffs.values): if _d > 0 and _dn == 0: _up += 1 elif _d < 0 and _up == 0: _dn += 1 else: break if _up >= 2: _indicators["consec_days"] = f"+{_up} consecutive up" elif _dn >= 2: _indicators["consec_days"] = f"-{_dn} consecutive down" except Exception as _ind_err: logging.debug("indicator build for ai_forecast failed: %s", _ind_err) # Inject live intraday context so AI knows today's price vs yesterday's close. # Only added when there's a meaningful gap (>0.1%) and we have a live price. if _live_price and abs(_live_price - price) / price > 0.001: _indicators["live_price"] = round(_live_price, 2) _indicators["prev_close"] = round(price, 2) _indicators["intraday_change_pct"] = round((_live_price - price) / price * 100, 2) # Build OHLCV DataFrame for Claude try: _ohlcv_df = pd.DataFrame({ "High": sh[ticker].dropna(), "Low": sl[ticker].dropna(), "Close": sc[ticker].dropna(), "Volume": sv[ticker].dropna() if ticker in sv.columns else pd.Series(dtype=float), }).dropna().tail(252) except Exception: _ohlcv_df = None ai_forecast = get_ai_forecast( ticker, company, _tf_label, ml, nifty_ok, macro_ok, vix_level, news, current_price=_ai_current_price, indicators=_indicators, ohlcv_df=_ohlcv_df, mode_c_active=mode_c_active, vix_declining=vix_declining, market_breadth={ "stage2_pct": round(stage2_breadth * 100, 1), "regime": breadth_regime, }, fii_pcr={ "fii_net": fii_data.get("fii_net"), "dii_net": fii_data.get("dii_net"), "fii_regime": fii_regime, "pcr": round(pcr_value, 2) if pcr_value is not None else None, }, fundamentals=fund_data, _fast_mode=_ai_fast_mode, _fast_fail_on_rate_limit=_ai_fast_fail_on_rate_limit, ) except Exception as _af_err: _af_err_msg = str(_af_err) logging.warning("ai_forecast skipped for %s: %s", ticker, _af_err) # Classify the outage: a provider that is only per-minute rate-limited (or Ollama) # will recover shortly, so mark it retryable rather than a hard failure. try: from llm_client import unavailability_is_recoverable _ai_soft_fail = unavailability_is_recoverable() except Exception: _ai_soft_fail = False # AI forecast supersedes the ML directional forecast when it ran successfully. # Skip ai_unavailable results — they are not real predictions. _ai_real = ai_forecast and ai_forecast.get("source") != "ai_unavailable" if _ai_real and ai_forecast.get("direction") in ("BULLISH", "BEARISH", "NEUTRAL"): price_forecast["predicted_direction"] = ai_forecast["direction"] if ai_forecast.get("predicted_return_lo") is not None: price_forecast["predicted_return_lo"] = ai_forecast["predicted_return_lo"] if ai_forecast.get("predicted_return_hi") is not None: price_forecast["predicted_return_hi"] = ai_forecast["predicted_return_hi"] # AI-led mode: for timeframe prediction paths, use AI direction/returns as # the primary trade call instead of strategy-gated NO TRADE logic. # Do NOT override when AI was unavailable — fall back to no_trade instead. # Compute consecutive down days for post-selloff gate (used below regardless of AI) _consec_down = 0 try: _close_tail = sc[ticker].dropna().diff().iloc[-5:] for _d in reversed(_close_tail.values): if _d < 0: _consec_down += 1 else: break except Exception: pass if _run_ai_forecast and _ai_real and ai_forecast.get("direction") in ("BULLISH", "BEARISH", "NEUTRAL"): direction = ai_forecast["direction"] if ai_forecast.get("predicted_return_lo") is not None: ret_lo = round(float(ai_forecast["predicted_return_lo"]), 2) if ai_forecast.get("predicted_return_hi") is not None: ret_hi = round(float(ai_forecast["predicted_return_hi"]), 2) # Bear-market gate: Nifty below EMA200 → structural headwind. # Downgrade BULLISH → SLIGHTLY BULLISH (not NEUTRAL): top5 accepts SLIGHTLY BULLISH, # and the confidence penalty (-3 pts when nifty_ok=False) already filters weak setups. # Forcing NEUTRAL blocked all picks even when Nifty is only marginally below EMA200. if not nifty_ok and direction == "BULLISH": direction = "SLIGHTLY BULLISH" # Symmetric bear-market BEARISH gate (3D only): in a confirmed bear market, individual stocks # are just as likely to bounce as continue falling — BEARISH calls are 50/50 on NSE. # Data: 22/24 3D BEARISH misses on 2026-06-24 had nifty_ok=False (5/5 days below EMA200). # Force NEUTRAL for 3D BEARISH when market structure is bearish to avoid coin-flip BEARISH. if not nifty_ok and _tf_label == "3D" and direction == "BEARISH": direction = "NEUTRAL" # Post-selloff BEARISH gate: after 3+ consecutive down days, NSE stocks strongly mean-revert. # Data: 29/29 3D BEARISH misses had window_low > entry (stocks bounced after multi-day selloff). # Force NEUTRAL — this is valid for all TFs because a falling stock past 3 days tends to bounce. if _consec_down >= 3 and direction == "BEARISH": direction = "NEUTRAL" # Earnings blackout dampens expected return if earnings["in_blackout"]: ret_lo = round(ret_lo * 0.5, 2) ret_hi = round(ret_hi * 0.5, 2) direction = "NEUTRAL" # override during blackout midpoint = round((ret_lo + ret_hi) / 2, 2) # target_price_lo/hi and expected_target_price are computed later, after # entry_price (open-buffer estimate) is finalised, so R:R stays consistent. confidence, confidence_breakdown = _calc_confidence( sig_result["count"], ml["upgraded"], news["label"], nifty_ok, vix_level, active_strategies=sig_result["active"], vix_declining=vix_declining, ml_probability=ml.get("probability", 0.5), nifty_trending=nifty_trending, qm_score=qm_score, stage2_breadth=stage2_breadth, fii_regime=fii_regime, pcr=pcr_value, macro_ok=macro_ok, ) # ── Loophole-based confidence downgrades ────────────────────────────────── # Apply loophole penalties to catch and mitigate risky predictions loophole_penalty = 0 loophole_flags = [] # 1. NO_STRATEGY_SIGNALS: cap at MEDIUM if signals < 2 if sig_result["count"] < 2 and confidence in ("HIGH", "MEDIUM"): confidence = "MEDIUM" if sig_result["count"] > 0 else "LOW" loophole_flags.append("NO_STRATEGY_SIGNALS") loophole_penalty += 2 # 2. RSI_OVERBOUGHT: penalize HIGH if RSI > 65 if confidence == "HIGH" and direction == "BULLISH": rsi_val = ml.get("features", {}).get("rsi") if rsi_val and rsi_val > 65: confidence = "MEDIUM" loophole_flags.append("RSI_OVERBOUGHT") loophole_penalty += 1 confidence_breakdown["rsi_overbought"] = -1 # 3. BELOW_EMA50: downgrade BULLISH if price < EMA50 and RSI not oversold if direction == "BULLISH": rsi_val = ml.get("features", {}).get("rsi", 50) _ema50_raw = sc[ticker].dropna() _ema50_val = float(_ema50_raw.ewm(span=50).mean().iloc[-1]) if len(_ema50_raw) >= 50 else None if price and _ema50_val and price < _ema50_val and rsi_val >= 30: confidence = "LOW" loophole_flags.append("BELOW_EMA50") loophole_penalty += 2 # 4. BEARISH_NEWS vs BULLISH_CALL: downgrade if news conflicts strongly if direction == "BULLISH" and news.get("score", 0) <= -8 and confidence in ("HIGH", "MEDIUM"): confidence = "LOW" if confidence == "MEDIUM" else "MEDIUM" loophole_flags.append("BEARISH_NEWS_CONFLICT") loophole_penalty += 1 confidence_breakdown["news_conflict"] = -1 # 5. UNCERTAIN_ML: penalize if probability near 0.5 (50–50 chance) ml_prob = ml.get("probability", 0.5) if 0.48 <= ml_prob <= 0.52 and confidence in ("HIGH", "MEDIUM"): confidence = "MEDIUM" if confidence == "HIGH" else "LOW" loophole_flags.append("UNCERTAIN_ML") loophole_penalty += 1 confidence_breakdown["uncertain_ml"] = -1 confidence_breakdown["loophole_penalty"] = loophole_penalty if loophole_flags: confidence_breakdown["loopholes_found"] = loophole_flags if _run_ai_forecast: if _ai_real and ai_forecast.get("confidence") in ("LOW", "MEDIUM", "HIGH"): confidence = ai_forecast["confidence"] else: confidence = None # AI unavailable — no_trade_reason drives display, no fake value # ── EMA key levels & ATR14 stop loss ───────────────────────────────────── c = sc[ticker].dropna() h = sh[ticker].dropna() lo = sl[ticker].dropna() key_levels = { "ema20": round(float(c.ewm(span=20).mean().iloc[-1]), 2), "ema50": round(float(c.ewm(span=50).mean().iloc[-1]), 2) if len(c) >= 50 else None, "ema200": round(float(c.ewm(span=200).mean().iloc[-1]), 2) if len(c) >= 200 else None, } # ATR14-based stop loss, scaled to the holding period. # Full 1.5× is calibrated for 5D swing trades; shorter holds use a tighter multiplier # so the min R:R target aligns with the shorter predicted return range. try: atr14_series = atr(h, lo, c, 14) atr14_val = float(atr14_series.iloc[-1]) # ATR multiplier and R:R target keyed to _tf_label (not raw n_trading) # so the 1D weekend-buffer inflation (n_trading=2) doesn't bleed into 3D sizing. atr_mult = {"INTRADAY": 0.4, "1D": 0.7, "3D": 1.1, "5D": 1.5, "1W": 1.8}.get(_tf_label, 1.5) sl_risk = atr_mult * atr14_val # Final SL/target/R:R are computed later after entry_price is finalized # and with direction-aware logic (LONG vs SHORT). sl_price = None sl_pct = None sl_target = None actual_rr = None except Exception: atr14_val = None sl_risk = None sl_price = None sl_pct = None sl_target = None actual_rr = None no_trade_reason = ( "no_signal" if sig_result["count"] == 0 else "wrong_timeframe" if not _active_for_ev else None ) if _run_ai_forecast and _ai_real: # In AI-led timeframe prediction mode, expose directional call directly # from AI and do not suppress with strategy horizon gates. no_trade_reason = None elif _run_ai_forecast and not _ai_real: # AI was unavailable this pass. The ML forecast already renders instantly beside the # AI slot (ML-vs-AI), so the card is never empty — therefore we NEVER emit a terminal # 'ai_unavailable'. The AI cell always stays in the retryable 'timeout' state so the # frontend keeps refetching and the forecast fills in the moment a provider (or Ollama) # frees up. No ML value is substituted into the AI slot; the two stay independent. # (_ai_soft_fail is still computed above for logging/diagnostics but no longer gates # the reason — even a hard outage is treated as retryable, since providers reset.) no_trade_reason = "timeout" # NEUTRAL calls have no directional edge — midpoint = 0 → target = entry = current price, # making the risk block misleading. Suppress as no-trade regardless of AI availability. if direction == "NEUTRAL" and no_trade_reason is None: no_trade_reason = "neutral_signal" # ── Tight directional band for INTRADAY + 1D (user request 2026-07-30) ─── # A NARROW band centered on a volatility-scaled expected move so the mean/target is a single # clear number (e.g. 1.00%–1.25%), instead of a wide [1.0, 2.0] intraday band or a flat ±1% 1D # band. Reshapes ret_lo/ret_hi for BOTH the AI path (already tight from ai_forecast — idempotent) # and the strategy/no-AI path (whose _price_forecast bands are wide, e.g. 0.3–0.8%). 3D/5D keep # their existing wider bands. Only applies to actionable directional calls. if (no_trade_reason is None and _tf_label in _TIGHT_BAND and _AI_RANGE_MODE != "containment" and direction in ("BULLISH", "SLIGHTLY BULLISH", "BEARISH", "SELL")): _atr_pct = ((_indicators.get("atr14") or 0.0) / price * 100.0) if price else 0.0 _tb = _tight_band_mag(_tf_label, _atr_pct) if _tb: _lo_mag, _hi_mag = _tb if direction in ("BULLISH", "SLIGHTLY BULLISH"): ret_lo, ret_hi = _lo_mag, _hi_mag else: ret_lo, ret_hi = -_hi_mag, -_lo_mag midpoint = round((ret_lo + ret_hi) / 2, 2) # ── 1D range-only policy (proven direction-ceiling fix) ────────────────── # See FORCE_1D_RANGE_ONLY / ONE_D_RANGE_HALF_PCT at module top. 1D next-day DIRECTION caps # at ~74% accuracy, so a directional 1D target is unreachable ~1-in-4 times. Convert every # 1D call into an honest RANGE-ONLY (NEUTRAL) call with a FLAT ±1% falsifiable band — a real # "stays within 1%" claim the user can act on, not a ±5% ATR band that's unbettable. Applied # regardless of AI availability. Blocking states (VIX gate, AI timeout, data error) are left # untouched; only a directional call or the plain neutral_signal suppression is replaced. range_bound = False if (FORCE_1D_RANGE_ONLY and _tf_label == "1D" and no_trade_reason in (None, "neutral_signal")): direction = "NEUTRAL" _half_1d = round(ONE_D_RANGE_HALF_PCT, 2) ret_lo, ret_hi = -_half_1d, _half_1d midpoint = 0.0 range_bound = True no_trade_reason = None # range-only IS the informative call, not a suppressed no-trade # ── Realistic open-price entry estimate ───────────────────────────────── # price = yesterday's close. Next-day entry happens at market open, which # often gaps. Use a TF-aware buffer so the plan is executable and can also # be used as the no-chase threshold at execution time. _entry_buffer = ENTRY_BUFFER_BY_TIMEFRAME.get(_tf_label, 0.003) # If reversal risk is high (low ML score, low confidence), increase buffer # so entry price is lower = more SL margin before hitting max loss. _ml_score = ml.get("score", 50) if _ml_score < 40 or confidence == "LOW": _entry_buffer = _entry_buffer * 1.5 # 50% more conservative elif _ml_score < 50: _entry_buffer = _entry_buffer * 1.2 # 20% more conservative _is_actionable_buy = direction in ("BULLISH", "SLIGHTLY BULLISH") and no_trade_reason is None _is_actionable_sell = direction in ("SELL", "BEARISH") and no_trade_reason is None # When a live price is available (market open / intraday view), 1D/3D now anchor to it just # like INTRADAY: the realistic entry is the CURRENT live price and targets/SL are measured # from NOW, not from yesterday's close + an open-gap buffer. This keeps a BULLISH target ABOVE # the live price on a gap-up day (previously the close-anchored target could render BELOW the # shown live price) and matches the AI-forecast targets, which already anchor to the live price. # Backtest has no live price (_live_price is None) → falls back to the prior close, so # historical behaviour is unchanged. _has_live = _live_price is not None and _live_price > 0 _use_live = _has_live and _tf_label in ("INTRADAY", "1D", "3D") # NOTE: a "gapped past target" flag used to be computed here by comparing the live price to a # target derived from the PRIOR CLOSE. It was removed because 1D/3D targets are now re-anchored # to the live price (see _target_anchor below), so the displayed target always sits ahead of the # live price — making a close-anchored "already passed" warning contradict the visible target. gapped_past_target = False if _use_live: entry_price = round(_live_price, 2) entry_basis = "live" elif _is_actionable_buy: entry_price = round(price * (1 + _entry_buffer), 2) entry_basis = "est_open_conservative" if _ml_score < 50 else "est_open" elif _is_actionable_sell: entry_price = round(price * (1 - _entry_buffer), 2) entry_basis = "est_open_conservative" if _ml_score < 50 else "est_open" else: entry_price = round(price, 2) entry_basis = "last_close" # Recompute targets from the anchor so price-range, SL and R:R are all internally consistent # with the achievable entry. Anchor = live price when available (INTRADAY/1D/3D), else the # prior close (backtest / market closed). _target_anchor = _live_price if _use_live else price target_price_lo = round(_target_anchor * (1 + ret_lo / 100), 2) target_price_hi = round(_target_anchor * (1 + ret_hi / 100), 2) expected_target_price = round(_target_anchor * (1 + midpoint / 100), 2) # For a 1D range-only call, keep the embedded AI-forecast sub-line consistent: no directional # arrow, no BUY/SKIP chip, no "Target ₹" — just the NEUTRAL band. Otherwise the # secondary AI line (and ML-vs-AI agreement) would still show the LLM's discarded directional # bet, contradicting the range-only main call. if isinstance(ai_forecast, dict) and ai_forecast.get("source") != "ai_unavailable": ai_forecast = dict(ai_forecast) if range_bound: ai_forecast["direction"] = "NEUTRAL" ai_forecast["should_buy"] = None ai_forecast["range_bound"] = True # ALWAYS keep the AI sub-object's return band + ₹ targets in lockstep with the headline # ret_lo/ret_hi/midpoint (same anchor, same band). The frontend rescales the AI range from # af.predicted_return_lo/hi but the headline "Target" from midpoint — if those drift apart # (e.g. the tight-band reshape updated ret_lo/ret_hi but the AI kept its own slightly # different band), the Target renders OUTSIDE the range ("Target ₹1,611" under a # ₹1,621–₹1,627 range). Syncing here makes that impossible by construction. ai_forecast["predicted_return_lo"] = ret_lo ai_forecast["predicted_return_hi"] = ret_hi ai_forecast["target_price_lo"] = target_price_lo ai_forecast["target_price_hi"] = target_price_hi ai_forecast["expected_target_price"] = expected_target_price # Recalculate stop-loss from entry_price so risk % and R:R reflect actual cost. # SL ALWAYS below entry (max loss level). Direction affects target/gain only. if atr14_val and entry_price and sl_risk: sl_price = round(entry_price - sl_risk, 2) sl_pct = round(sl_risk / entry_price * 100, 1) # Target and R:R are direction-aware is_short_bias = direction in ("SELL", "BEARISH") if is_short_bias: # SHORT: gain when price goes DOWN from entry to target_lo sl_target = target_price_lo if ret_lo is not None and ret_lo < 0 else None _target_gain = (entry_price - target_price_lo) if target_price_lo else 0 else: # LONG: gain when price goes UP from entry to target_hi sl_target = target_price_hi if ret_hi is not None and ret_hi > 0 else None _target_gain = (target_price_hi - entry_price) if target_price_hi else 0 actual_rr = round(_target_gain / sl_risk, 1) if sl_risk > 0 and _target_gain > 0 else None # If there's no trade setup (no signal), suppress the directional forecast. # Don't show BULLISH/BEARISH when there's no actionable setup. if no_trade_reason: price_forecast = { "predicted_direction": None, "predicted_return_lo": None, "predicted_return_hi": None, } ai_forecast = None # Suppress AI forecast direction only when no trade setup # ── Phase 6: Price targets (Camarilla / ATR / PDH) ────────────────────────── try: from price_targets import get_price_targets _strategy_bias = "BULLISH" if direction in ("BULLISH", "SLIGHTLY BULLISH") else ( "BEARISH" if direction in ("SELL", "BEARISH") else "NEUTRAL") price_targets_dict = get_price_targets(ticker, sc, sh, sl, _strategy_bias, confidence) except Exception as _pt_err: logging.debug("price_targets failed for %s: %s", ticker, _pt_err) price_targets_dict = None # ── Net-of-cost reality check (price prediction ≠ profitable trading) ────── # A predicted move that can't clear NSE round-trip fees is not a tradeable edge # (the "tomorrow ≈ today" trap: tiny ±0.07% bands always "hit" but net a loss). try: from costs import cost_pct_for_timeframe, net_return_pct, clears_costs as _clears _cost_pct = cost_pct_for_timeframe(_tf_label) _best_case_move = max(abs(ret_lo), abs(ret_hi)) # most favourable bound _net_expected = net_return_pct(midpoint, _tf_label) _clears_costs = bool(_clears(_best_case_move, _tf_label)) and direction not in ("NEUTRAL", "NO TRADE") except Exception: _cost_pct, _net_expected, _clears_costs = None, None, None _pred_result = { "ticker": ticker, "company": company, "start_date": start_date, "end_date": end_date, "trading_days": n_trading, "price": round(price, 2), "round_trip_cost_pct": _cost_pct, "net_expected_return_pct": _net_expected, "clears_costs": _clears_costs, "timeframe": _tf_label, "direction": direction, "confidence": confidence, "no_trade_reason": no_trade_reason, "range_bound": range_bound, "ai_disclaimer": ( f"AI unavailable — direction and targets are from strategy/ML math only. Error: {_af_err_msg}" if (_run_ai_forecast and _af_err_msg) else None ), "expected_return_range": f"{ret_lo:+.2f}% to {ret_hi:+.2f}%", "ret_lo": ret_lo, "ret_hi": ret_hi, "midpoint": midpoint, "target_price_lo": target_price_lo, "target_price_hi": target_price_hi, "expected_target_price": expected_target_price, "expected_entry_price": entry_price, "entry_basis": entry_basis, "entry_buffer_pct": round(_entry_buffer * 100, 2), "max_chase_pct": round(_entry_buffer * 100, 2), "gapped_past_target": gapped_past_target, "predicted_direction": price_forecast["predicted_direction"], "predicted_return_lo": price_forecast["predicted_return_lo"], "predicted_return_hi": price_forecast["predicted_return_hi"], "active_strategies": [] if _run_ai_forecast else sig_result["active"], "signal_count": sig_result["count"], "signals": {} if _run_ai_forecast else sig_result["signals"], "signal_errors": [] if _run_ai_forecast else sig_result.get("signal_errors", []), "ml": { "score": ml["score"], "probability": ml["probability"], "upgraded": ml["upgraded"], "features": ml["features"], }, "news": { "label": news["label"], "score": news["score"], "summary": news["summary"], "key_headline": news.get("key_headline", ""), "headlines": news.get("headlines", []), "headlines_dated": news.get("headlines_dated", []), "latest_date": news.get("latest_date", ""), "source": news.get("source", "keywords"), }, "earnings": earnings, "key_levels": key_levels, "risk": { "atr14": round(atr14_val, 2) if atr14_val else None, "stop_loss": sl_price, "stop_loss_pct": sl_pct, "min_target": sl_target, "actual_rr": actual_rr, }, "trade_plan": { "expected_entry_price": entry_price, "entry_basis": entry_basis, "entry_buffer_pct": round(_entry_buffer * 100, 2), "max_chase_pct": round(_entry_buffer * 100, 2), "gapped_past_target": gapped_past_target, "expected_target_price": expected_target_price, "target_price_lo": target_price_lo, "target_price_hi": target_price_hi, "expected_return_range": f"{ret_lo:+.2f}% to {ret_hi:+.2f}%", "stop_loss": sl_price, "stop_loss_pct": sl_pct, "risk_reward": actual_rr, "holding_timeframe": _tf_label, }, "market": { "vix_level": round(vix_level, 1), "vix_label": vix_label, "nifty_ok": nifty_ok, "nifty_label": nifty_label, "macro_ok": macro_ok, "macro_label": macro_label, }, "confidence_breakdown": confidence_breakdown, "market_breadth": { "stage2_pct": round(stage2_breadth * 100, 1), "regime": breadth_regime, }, "fii_pcr": { "fii_net": fii_data.get("fii_net"), "dii_net": fii_data.get("dii_net"), "fii_regime": fii_regime, "pcr": round(pcr_value, 2) if pcr_value is not None else None, }, "ai_forecast": ai_forecast, "backtest_stats": backtest_stats, "mode_c_active": mode_c_active, "mode_c_label": mode_c_label, "price_targets": price_targets_dict, "intraday": intraday_dict, "sector": { "name": _ticker_sector, "leading": _sector_leading, "lagging": _sector_lagging, "rotation": sector_pulse_data.get("rotation_signal") if sector_pulse_data else None, }, "fundamentals": { "score": fund_data.get("fundamental_score") if fund_data else None, "pe_relative": fund_data.get("pe_relative") if fund_data else None, "debt_level": fund_data.get("debt_level") if fund_data else None, "revenue_trend": fund_data.get("revenue_trend") if fund_data else None, "roe_pct": fund_data.get("roe_pct") if fund_data else None, "summary": fund_data.get("summary") if fund_data else None, }, "metadata": { "request_id": ctx.request_id, "elapsed_ms": int(ctx.elapsed() * 1000), "cache_age_days": cache_age_days, "data_source": "cached+live" if _skip_fresh_fetch else "fresh_fetch", }, } # ── Save prediction snapshot for audit trail ──────────────────────────────── # Save when AI ran successfully (source="watchlist") or when AI was unavailable # but signals exist (source="watchlist_heuristic") — so the validation tab always # shows something for stocks that were actually evaluated. if _run_ai_forecast and direction not in ("NO TRADE", "HOLD"): _snap_source = "watchlist" if _ai_real else "watchlist_heuristic" try: from database import save_prediction_snapshot import json save_prediction_snapshot( ticker=ticker, timeframe=_tf_label, direction=direction, confidence=confidence, target_price_lo=target_price_lo, target_price_hi=target_price_hi, predicted_return_lo=ret_lo, predicted_return_hi=ret_hi, current_price=price, snapshot_source=_snap_source, snapshot_data=json.dumps({ "signals_active": sig_result.get("active", []), "ml_score": ml.get("score"), "news_score": news.get("score"), "vix_level": vix_level, "nifty_status": nifty_label, "ai_unavailable": not _ai_real, }), ) except Exception as _snap_err: logging.debug("Failed to save prediction snapshot for %s: %s", ticker, _snap_err) _PRED_CACHE[_pred_key] = {"ts": time.time(), "result": _pred_result} return _pred_result # ── UNIVERSE RANKING ────────────────────────────────────────────────────────── def rank_stocks_v2( start_date: str, end_date: str, universe: Optional[list[str]] = None, capital: Optional[float] = None, ) -> dict: """Rank all stocks in universe for the date range. Returns sorted results dict.""" _init_universe() tickers = list(dict.fromkeys(universe or DEFAULT_UNIVERSE)) # Return cached result if still fresh (avoids re-scanning 150 stocks on repeated UI hits) _cache_key = (start_date, end_date, tuple(tickers), capital) _cached = _RANK_CACHE.get(_cache_key) if _cached and time.time() - _cached["ts"] < _RANK_CACHE_TTL: return _cached["result"] vix_level, vix_label = _get_vix() nifty_ok, nifty_label = _get_nifty_gate() macro_ok, macro_label = _get_macro_gate() market_ctx = { "vix_level": vix_level, "vix_label": vix_label, "nifty_ok": nifty_ok, "nifty_label": nifty_label, "macro_ok": macro_ok, "macro_label": macro_label, } results, errors = [], [] def _predict_one(t): return predict_stock_v2(t, start_date, end_date, _market_ctx=market_ctx) # Keep concurrency moderate to avoid memory pressure / kill -9 (137) # when multiple expensive scans overlap (dashboard refresh + top5 calls). max_workers = min(len(tickers), 8) with ThreadPoolExecutor(max_workers=max_workers) as pool: futures = {pool.submit(_predict_one, t): t for t in tickers} for fut in as_completed(futures): try: pred = fut.result() except Exception as exc: errors.append(f"{futures[fut]}: {exc}") continue if "error" in pred: errors.append(f"{pred['ticker']}: {pred['error']}") else: results.append(pred) # Sort: NO TRADE last, then by confidence tier, then by midpoint desc conf_order = {"HIGH": 0, "MEDIUM": 1, "LOW": 2, "BLOCKED": 9} dir_order = {"BULLISH": 0, "SLIGHTLY BULLISH": 1, "NEUTRAL": 2, "SLIGHTLY BEARISH": 3, "BEARISH": 4, "NO TRADE": 9} results.sort(key=lambda x: ( conf_order.get(x["confidence"], 5), dir_order.get(x["direction"], 5), -x["midpoint"], )) for i, r in enumerate(results): r["rank"] = i + 1 # Capital allocation for top bullish picks buys = [r for r in results if r["direction"] in ("BULLISH", "SLIGHTLY BULLISH") and r["confidence"] != "BLOCKED"] if capital and capital > 0 and buys: top = buys[:6] per = round(capital / len(top), 0) for p in top: p["suggested_allocation"] = per shares = int(per / p["price"]) if p.get("price") and p["price"] > 0 else 0 p["suggested_shares"] = shares p["allocation_value"] = round(shares * p["price"], 2) out = { "start_date": start_date, "end_date": end_date, "capital": capital, "total_scanned": len(tickers), "total_scored": len(results), "errors": errors, "ranked": results, "market": { "vix_level": round(vix_level, 1), "vix_label": vix_label, "nifty_ok": nifty_ok, "nifty_label": nifty_label, "macro_ok": macro_ok, "macro_label": macro_label, }, } _RANK_CACHE[_cache_key] = {"ts": time.time(), "result": out} return out # ── SMOKE TEST ──────────────────────────────────────────────────────────────── if __name__ == "__main__": print("Running smoke test on RELIANCE.NS...") p = predict_stock_v2("RELIANCE.NS", "2026-06-15", "2026-06-30") print(f"Direction : {p['direction']}") print(f"Confidence : {p['confidence']}") print(f"Expected : {p['expected_return_range']}") print(f"Signals : {p['active_strategies']} ({p['signal_count']} active)") print(f"ML Score : {p['ml']['score']}/100 prob={p['ml']['probability']}") print(f"News : {p['news']['label']} ({p['news']['source']})") print(f"Earnings : {p['earnings']}") print(f"Market VIX : {p['market']['vix_label']}") print(f"Nifty Gate : {p['market']['nifty_label']}") print("SMOKE TEST PASSED")