top5_picker.py

#!/usr/bin/env python3
"""
top5_picker.py — Finds the top 5 NSE stocks to invest in for the week,
with predictions across three timeframes (1D, 3D, 5D) run concurrently.

Usage (programmatic):
    from top5_picker import get_top5_picks
    result = get_top5_picks()
"""

from __future__ import annotations
import sys, os, warnings
import datetime as _dt
warnings.filterwarnings("ignore")
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Optional, Callable

from predictor_core import predict_stock_v2, DEFAULT_UNIVERSE, timeframe_to_dates

TIMEFRAMES = ["INTRADAY", "1D"]  # 3D + 5D removed — INTRADAY/1D are the shown horizons


def _slim_ml_tf(m: Optional[dict]) -> Optional[dict]:
    """Extract the frontend-facing subset of an ml_predictor TF object (predict_all_tf['tfs'][tf])."""
    if not m:
        return None
    return {
        "direction":             m.get("direction"),
        "confidence":            m.get("confidence"),
        "confidence_prob":       m.get("confidence_prob"),
        "high_conviction":       m.get("high_conviction"),
        "predicted_return_lo":   m.get("predicted_return_lo"),
        "predicted_return_hi":   m.get("predicted_return_hi"),
        "target_price_lo":       m.get("target_price_lo"),
        "target_price_hi":       m.get("target_price_hi"),
        "expected_target_price": m.get("expected_target_price"),
        "current_price":         m.get("current_price"),
    }


def _ml_conviction_score(ml_res: Optional[dict]) -> float:
    """ML selection score for a stock — INTRADAY/1D (matches the AI ranking horizons).
    confidence_prob × |expected return| × direction multiplier."""
    if not ml_res or not ml_res.get("available"):
        return 0.0
    tfs = ml_res.get("tfs", {}) or {}
    best = 0.0
    for tf in ("INTRADAY", "1D"):
        d = tfs.get(tf) or {}
        prob = d.get("confidence_prob") or 0.5
        cur = d.get("current_price") or 0
        exp = 0.0
        if cur and d.get("expected_target_price"):
            exp = (d["expected_target_price"] / cur - 1) * 100
        elif d.get("predicted_return_hi") is not None:
            exp = d["predicted_return_hi"]
        direction = d.get("direction")
        dir_mult = 1.0 if direction == "BULLISH" else 0.2 if direction == "BEARISH" else 0.4
        s = prob * abs(exp) * dir_mult
        if s > best:
            best = s
    return best

# Universe for top picks — the WHOLE NSE market (all cap tiers), sourced via
# universe.get_universe() (NSE full equity list → Nifty-500 CSV → static fallback),
# so mid/small-caps are included, not just the top 500 by market cap. Never empty even
# when Yahoo blocks the Hugging Face Spaces datacenter IP.
def _build_top5_universe(n: int = 0) -> list[str]:
    try:
        from universe import get_universe
        tickers = [t for t in get_universe().keys() if t.endswith(".NS")]
        return tickers[:n] if n and n > 0 else tickers
    except Exception:
        return []

_TOP5_UNIVERSE: list[str] = []

def _get_top5_universe(force_refresh: bool = False) -> list[str]:
    global _TOP5_UNIVERSE
    if force_refresh:
        _TOP5_UNIVERSE = []
    if not _TOP5_UNIVERSE:
        _TOP5_UNIVERSE = _build_top5_universe()
    return _TOP5_UNIVERSE or DEFAULT_UNIVERSE


# ── SECTOR-DRIVEN CANDIDATE POOL (replaces ML scoring for Top Picks) ──────────────────────────
# Instead of ranking the whole market by an ML score, Top Picks now draws from the most VOLATILE
# ("violent") NSE sectors and lets AI pick the good stocks. The candidate pool = large-cap
# constituents of every tracked sector, ORDERED by their sector's realized volatility (most
# volatile first), so AI scans the hot sectors' stocks first. Env TOP5_SECTOR_MODE=0 restores the
# old whole-market ML-scored scan.
# DEFAULT OFF: Top Picks now scans the WHOLE NSE market ranked by stock ATR% volatility
# (see _score_scan / _score_tf) with sector diversity applied to the final picks, instead of
# restricting the candidate pool to a handful of large-cap sector constituents. Set
# TOP5_SECTOR_MODE=1 to restore the restricted sector-pool scan.
_TOP5_SECTOR_MODE = os.getenv("TOP5_SECTOR_MODE", "0") != "0"

# ── Candidate-pool ROTATION (so refreshes surface different volatile names) ─────────────────────
# Without rotation, sector mode scans the SAME ~88 large-caps every time and — because every
# constituent is scored and the best top-N always win — the picks are identical each refresh.
# Rotation fixes this: each compute takes a *rotating window* of every volatile sector's stocks
# (offset from `_rotation_offset()`), so you keep scanning the most 'violent' sectors but see FRESH
# names. The offset is the IST CALENDAR DAY (so a NEW DAY rotates on a plain browser reload, and
# reloads WITHIN a day stay stable) plus a manual bump that ticks up on each force-refresh (so the
# refresh button also rotates within a day). Env TOP5_ROTATE=0 restores identical-every-time.
_SECTOR_ROTATE = os.getenv("TOP5_ROTATE", "1") != "0"
# Max constituents drawn from each sector per refresh. Must be < a typical sector's size for the
# rotation to actually change the winners (avg tracked sector has ~8-9 large-caps in the map), so 6
# leaves room to rotate a few in/out each refresh. Env TOP5_SECTOR_PER overrides.
_SECTOR_PER_SECTOR = int(os.getenv("TOP5_SECTOR_PER", "6") or "6")
# Manual bump advanced once per FORCE-refresh (the /api/top5?refresh=1 button) so the button varies
# picks within a single day, on top of the day-based baseline.
_SECTOR_MANUAL_BUMP = 0


def _ist_day_index() -> int:
    """Whole days elapsed (IST) since a fixed epoch — a stable per-calendar-day integer."""
    ist = _dt.timezone(_dt.timedelta(hours=5, minutes=30))
    return (_dt.datetime.now(ist).date() - _dt.date(2020, 1, 1)).days


def _rotation_offset(force_refresh: bool = False) -> int:
    """Rotation offset = IST calendar day + manual bump.

    Day component → a new trading day rotates automatically on a plain reload (cache expires at IST
    midnight, so the next reload recomputes with a new day index); reloads within the same day reuse
    the same offset (stable picks). Manual bump → each force-refresh advances it so the refresh
    button surfaces fresh names within a day. Survives process restarts (day is calendar-derived).
    """
    global _SECTOR_MANUAL_BUMP
    if force_refresh:
        _SECTOR_MANUAL_BUMP += 1
    return _ist_day_index() + _SECTOR_MANUAL_BUMP

# Only surface affordable stocks: Top Picks excludes any stock trading ABOVE this price (₹). Keeps
# the picks actionable for smaller position sizes and filters out very high-priced scrips.
# Env TOP5_MAX_PRICE overrides; set 0 to disable the cap.
_TOP5_MAX_PRICE = float(os.getenv("TOP5_MAX_PRICE", "5000") or "5000")

# Excludes penny stocks (₹ below this floor). Ranking is ATR-as-%-of-price PRIMARY (see
# _score_scan/_score_tf below) — on penny stocks that % is inflated by tick-size granularity and
# thin, wide-spread order books rather than real tradeable momentum, so they kept rising to the
# top while being unreliable to hit any predicted intraday range. Env TOP5_MIN_PRICE overrides;
# set 0 to disable the floor.
_TOP5_MIN_PRICE = float(os.getenv("TOP5_MIN_PRICE", "100") or "100")

# Sector diversity: cap how many final picks may come from any one mapped NSE sector so a single
# hot sector can't fill the whole list. Stocks with no sector mapping (most mid/small-caps) are
# never capped. Env TOP5_MAX_PER_SECTOR overrides; 0 disables the cap.
_MAX_PER_SECTOR = int(os.getenv("TOP5_MAX_PER_SECTOR", "3") or "3")


def _diversify_by_sector(picks: list[dict], top_n: int, max_per_sector: int) -> list[dict]:
    """Spread the final picks across NSE sectors.

    Walks the already-ranked ``picks`` and caps classified stocks at ``max_per_sector`` per
    sector; stocks whose sector can't be resolved are never capped. If diversity leaves the list
    short of ``top_n``, the capped-out overflow is used to backfill (in rank order). Order is
    otherwise preserved, so the volatility ranking is honored.
    """
    if max_per_sector <= 0:
        return picks[:top_n]
    try:
        from sector_pulse import get_sector_for_ticker
    except Exception:
        return picks[:top_n]
    out: list[dict] = []
    counts: dict[str, int] = {}
    overflow: list[dict] = []
    for p in picks:
        sec = get_sector_for_ticker(p.get("ticker", ""))
        if sec is not None:
            if counts.get(sec, 0) >= max_per_sector:
                overflow.append(p)
                continue
            counts[sec] = counts.get(sec, 0) + 1
        out.append(p)
        if len(out) >= top_n:
            return out
    for p in overflow:
        if len(out) >= top_n:
            break
        out.append(p)
    return out[:top_n]


def _sector_ranked_universe(rotate: bool = False, per_sector: int = 0, offset: int = 0) -> tuple[list[str], dict[str, float]]:
    """Return (ordered_tickers, {ticker: sector_volatility_pct}) for the sector-driven scan.

    Tickers are the large-cap constituents of the tracked NSE sectors, ordered by their sector's
    realized volatility (descending). Sectors missing a volatility read (e.g. a broken index
    ticker) sort last. Returns ([], {}) on failure so the caller falls back to the market scan.

    When ``rotate`` is True, each sector's constituent list is rotated by ``offset`` (typically the
    IST calendar day + manual bump, see _rotation_offset) and — when ``per_sector`` > 0 — trimmed
    to a rotating window of that many stocks. Sector (volatility) ORDER is always preserved, so
    every refresh keeps scanning the most volatile sectors first while surfacing DIFFERENT stocks
    from them. With rotate=False it behaves exactly as before (deterministic full list).
    """
    try:
        from sector_pulse import get_sector_volatility, get_sector_constituents
        vols = get_sector_volatility()
        cons = get_sector_constituents()
    except Exception:
        return [], {}
    if not cons:
        return [], {}
    vol_by_sector = {r["name"]: r["volatility_pct"] for r in vols}
    _default_vol = min((r["volatility_pct"] for r in vols), default=0.0)
    # Sectors WITH a volatility read first (vol-desc), then any unread sectors (default vol).
    ordered_sectors = [r["name"] for r in vols] + [s for s in cons if s not in vol_by_sector]
    off = offset if rotate else 0
    ordered: list[str] = []
    vol_by_ticker: dict[str, float] = {}
    for sec in ordered_sectors:
        sv = vol_by_sector.get(sec, _default_vol)
        members = [tk for tk in cons.get(sec, []) if tk not in vol_by_ticker]
        if not members:
            continue
        if rotate and len(members) > 1:
            k = off % len(members)
            members = members[k:] + members[:k]      # rotate this sector's stocks by the offset
        if per_sector and per_sector > 0:
            members = members[:per_sector]            # keep only a rotating window per sector
        for tk in members:
            ordered.append(tk)
            vol_by_ticker[tk] = sv
    return ordered, vol_by_ticker
    return ordered, vol_by_ticker


# Round-robin cursor over the cold (uncached) tail of the universe so successive scans
# sweep different cold stocks and gradually warm the whole market.
_SCAN_COLD_OFFSET = 0


def _order_and_cap_scan(universe: list[str], cap: int) -> list[str]:
    """Order the scan universe cache-first, then cap it.

    Already-cached stocks (watchlist mid/small-caps + prior scans) are scanned first so
    they resolve instantly, giving fast and complete first results. The cold remainder is
    rotated by a persistent cursor so each run scans a different cold slice — over
    successive runs the whole market is swept and the OHLCV cache fully warms.
    """
    global _SCAN_COLD_OFFSET
    try:
        # Match the period the scan (predict_stock_v2 → _load_ticker_data) actually
        # caches: "2y". Checking "1y" here (the old value) meant the warm set was almost
        # always empty, so cache-first ordering silently no-op'd. Phase 1b's ML pass also
        # reads the same "2y" row, so warm stocks skip the OHLCV re-fetch entirely.
        from data_sources import cached_tickers
        warm = cached_tickers("2y")
    except Exception:
        warm = set()

    warm_list = [t for t in universe if t in warm]
    cold_list = [t for t in universe if t not in warm]
    # Whole-market scan (cap<=0): keep every stock, just ordered cache-first so warm names
    # resolve instantly and the cold remainder still gets scanned (and warmed) this run.
    if cap <= 0 or len(universe) <= cap:
        return warm_list + cold_list
    n_cold = len(cold_list)
    if n_cold:
        off = _SCAN_COLD_OFFSET % n_cold
        cold_list = cold_list[off:] + cold_list[:off]

    selected = (warm_list + cold_list)[:cap]

    # Advance the cursor by how many cold stocks we actually scanned this run.
    cold_scanned = max(0, len(selected) - len(warm_list))
    if n_cold:
        _SCAN_COLD_OFFSET = (_SCAN_COLD_OFFSET + cold_scanned) % n_cold
    return selected



def _run_predict_with_ctx(
    ticker: str,
    tf: str,
    market_ctx: Optional[dict],
    run_ai: bool = False,
    ai_fast_mode: bool = False,
    skip_news: bool = False,
) -> tuple[str, str, dict]:
    """Run one timeframe prediction using shared market context when available."""
    start, end = timeframe_to_dates(tf)
    pred = predict_stock_v2(
        ticker,
        start,
        end,
        _market_ctx=market_ctx,
        _run_ai_forecast=run_ai,
        _ai_fast_mode=ai_fast_mode,
        _ai_fast_fail_on_rate_limit=True,
        _skip_news=skip_news,
    )
    return ticker, tf, pred


def _get_specialist_recommendation(ticker: str) -> Optional[dict]:
    """
    Check if this stock is a specialist for a specific timeframe (Intraday vs 1D).
    Returns {best_tf, accuracy, reason} if specialist found, else None.
    """
    try:
        from database import get_connection
        conn = get_connection()
        cursor = conn.cursor()

        query = """
        SELECT
          timeframe,
          COUNT(*) as total,
          SUM(CASE WHEN validation_result = 'HIT' THEN 1 ELSE 0 END) as hits,
          ROUND(CAST(SUM(CASE WHEN validation_result = 'HIT' THEN 1 ELSE 0 END) AS REAL) / COUNT(*), 3) as win_rate
        FROM prediction_snapshots
        WHERE ticker = ? AND validation_status = 'VALIDATED'
          AND validation_result IN ('HIT', 'MISS')
          AND timeframe IN ('INTRADAY', '1D')
        GROUP BY timeframe
        HAVING COUNT(*) >= 10
        """

        cursor.execute(query, (ticker,))
        rows = cursor.fetchall()
        conn.close()

        if len(rows) < 2:
            return None

        data = {row[0]: {"win_rate": row[3], "hits": row[2], "total": row[1]} for row in rows}

        intraday_wr = data.get("INTRADAY", {}).get("win_rate", 0)
        one_d_wr = data.get("1D", {}).get("win_rate", 0)

        if abs(intraday_wr - one_d_wr) >= 0.05:
            if intraday_wr > one_d_wr:
                return {
                    "best_tf": "INTRADAY",
                    "accuracy": f"{int(intraday_wr*100)}%",
                    "reason": f"Specialist: INTRADAY {int(intraday_wr*100)}% vs 1D {int(one_d_wr*100)}%"
                }
            else:
                return {
                    "best_tf": "1D",
                    "accuracy": f"{int(one_d_wr*100)}%",
                    "reason": f"Specialist: 1D {int(one_d_wr*100)}% vs INTRADAY {int(intraday_wr*100)}%"
                }
        return None
    except Exception:
        return None


def get_top5_picks(
    universe: Optional[list[str]] = None,
    top_n: int = 20,
    _universe_size: int = 0,
    force_universe_refresh: bool = False,
    progress_cb: Optional[Callable[[dict], None]] = None,
) -> dict:
    """
    Returns top N stocks with 1D/3D/5D predictions.

    Returns dict:
    {
      "picks": [
        {
          ...full prediction dict from 5D run (anchor),
          "rank": int,
          "timeframes": {
            "1D": {"expected_return_range", "midpoint", "ret_lo", "ret_hi", "direction", "confidence"},
            "3D": {...},
            "5D": {...}
          }
        },
        ...
      ],
      "market": {...},
      "generated_at": "YYYY-MM-DD HH:MM",
      "errors": [...]
    }
    """
    from datetime import datetime
    import time
    import logging
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger("top5")

    start_time = time.time()

    def _publish(payload: dict) -> None:
        """Emit a progress snapshot to the caller (best-effort; never raises)."""
        if progress_cb is None:
            return
        try:
            progress_cb(payload)
        except Exception as _pe:
            logger.warning("[TOP5] progress_cb failed: %s", _pe)


    # Step 1: signal-only 1D scan for candidate selection (no LLM).
    # Draws from the WHOLE NSE market so mid/small-caps are eligible, but caps the scan
    # (_universe_size) for fast first results. _order_and_cap_scan orders cache-first —
    # already-warmed stocks (watchlist mid/small-caps + prior scans) scan instantly — and
    # rotates the cold remainder across runs so the whole market is swept over time.
    # Phase 1 makes NO LLM calls, so cost is time, not API quota; LLM quota is spent only
    # on the shortlisted candidates in Phase 2.
    effective_universe = universe or _get_top5_universe(force_refresh=force_universe_refresh) or DEFAULT_UNIVERSE
    # Sector-driven mode: candidate pool = large-cap constituents of the most VOLATILE sectors,
    # ordered by sector volatility. AI (Phase 2) does the picking — ML is out of selection. Only
    # applies when the caller didn't pass an explicit universe. Falls back to the market scan on
    # any failure (e.g. sector index downloads blocked).
    # Rotation (TOP5_ROTATE, default ON) takes a rotating per-sector window each refresh so the
    # picks aren't the identical volatile names every time — see _sector_ranked_universe.
    _sector_vol_by_ticker: dict[str, float] = {}
    _sector_mode_active = False
    if _TOP5_SECTOR_MODE and not universe:
        _rot_off = _rotation_offset(force_universe_refresh)
        _sec_uni, _sector_vol_by_ticker = _sector_ranked_universe(
            rotate=_SECTOR_ROTATE, per_sector=(_SECTOR_PER_SECTOR if _SECTOR_ROTATE else 0),
            offset=_rot_off)
        if _sec_uni:
            effective_universe = _sec_uni
            _sector_mode_active = True
            logger.info("[TOP5] SECTOR MODE: %d stocks from %d volatility-ranked sectors (rotate=%s, offset=%d)",
                        len(_sec_uni), len(set(_sector_vol_by_ticker.values())),
                        _SECTOR_ROTATE, _rot_off)
    if not _sector_mode_active:
        # Whole-market scan, ordered cache-first. _universe_size=0 (default) keeps the entire
        # NSE market; a positive value caps the scan for faster first results.
        effective_universe = _order_and_cap_scan(effective_universe, _universe_size)

    import concurrent.futures as _cf

    # ── Phase 1: signal-only 1D scan for candidate selection (no LLM) ─────────
    # Uses signals + ML score only — fast (2-5s/ticker cold, <1s cached), no LLM.
    # Scan on 1D (not 5D) so the short-term signal set (S1/S4/S8/S16/S_CTRIO…)
    # fires — top5 targets profitable INTRADAY/1D trades, so candidates are
    # ranked on short-horizon merit. LLM quota is preserved for Phase 2.
    # Worker count + deadline scale with universe size so a full-market scan can
    # cover the whole pool once the OHLCV cache is warm; on a cold cache it covers
    # as many as fit within the deadline and the cache warms over successive runs.
    phase1_start = time.time()
    scan_preds: dict[str, dict] = {}
    timeouts = 0
    _uni_n = max(1, len(effective_universe))
    scan_workers = min(16, _uni_n)
    # ~0.2s/ticker of wall-clock budget (warm cache, parallel), floored at 120s and
    # capped at 600s so a stuck run can never hang the background thread forever.
    scan_deadline = min(600, max(120, int(_uni_n * 0.2)))
    logger.info("[TOP5] PHASE 1 start: scanning %d stocks (workers=%d, deadline=%ds)",
                _uni_n, scan_workers, scan_deadline)
    _publish({
        "computing": True, "phase": "scanning",
        "scanned": 0, "scan_total": _uni_n, "picks": [], "generated_at": None,
        "message": f"Scanning the NSE market — 0/{_uni_n} stocks",
    })
    scan_ex = ThreadPoolExecutor(max_workers=scan_workers)
    scan_futs = {
        scan_ex.submit(_run_predict_with_ctx, ticker, "1D", None, False, False, True): ticker
        for ticker in effective_universe
    }
    scan_done_n = 0
    try:
        for f in _cf.as_completed(scan_futs, timeout=scan_deadline):
            ticker = scan_futs[f]
            try:
                _ticker, _tf, pred = f.result()
                scan_preds[_ticker] = pred
            except Exception as e:
                scan_preds[ticker] = {}
                if "timeout" in str(e).lower():
                    timeouts += 1
            scan_done_n += 1
            # Emit scan progress every 25 stocks so the UI + HF logs show a live counter.
            if scan_done_n % 25 == 0 or scan_done_n == _uni_n:
                logger.info("[TOP5] PHASE 1 progress: %d/%d scanned (%.0fs)",
                            scan_done_n, _uni_n, time.time() - phase1_start)
                _publish({
                    "computing": True, "phase": "scanning",
                    "scanned": scan_done_n, "scan_total": _uni_n,
                    "picks": [], "generated_at": None,
                    "message": f"Scanning the NSE market — {scan_done_n}/{_uni_n} stocks",
                })
    except _cf.TimeoutError:
        logger.info("[TOP5] PHASE 1 deadline (%ds) reached at %d/%d scanned",
                    scan_deadline, scan_done_n, _uni_n)
    # Cancel and account for any futures that never completed within the deadline.
    for f, ticker in scan_futs.items():
        if not f.done():
            f.cancel()
            if ticker not in scan_preds:
                scan_preds[ticker] = {}
                timeouts += 1
    scan_ex.shutdown(wait=False)

    phase1_elapsed = time.time() - phase1_start
    msg1 = f"[TOP5] PHASE 1 (signal scan, no LLM): {len(scan_preds)}/{len(effective_universe)} done, {timeouts} timeout, {phase1_elapsed:.1f}s"
    print(msg1)
    logger.info(msg1)

    # ── Composite short-term profit score ─────────────────────────────────────
    # Goal: rank stocks by expected short-horizon (INTRADAY/1D) profit. This is
    # the WITHIN-TIER tiebreak; confidence tier (below) is the primary sort.
    # Score components (all multiplicative on ret_hi so absolute return is preserved):
    #   conf_mult:    HIGH=1.0 / MEDIUM=0.80 / LOW=0.55
    #   ml_factor:    1 + (ml_probability - 0.5) × 0.30   → range [0.85, 1.15]
    #   rr_factor:    1 + 0.12 if actual_rr >= 2.0 else 0  (rewards good risk/reward)
    #   sector_factor: 1.12 if sector leading / 0.90 if sector lagging / 1.0 neutral
    # Backtest findings (research/backtest_top5.py + iter64/sweep on 1D):
    #   BEARISH direction accuracy = 38%, avg P&L = -0.42% → EXCLUDED.
    #   BULLISH HIGH-conf 1D = +0.86%/trade, 69% win; MEDIUM = +0.01% break-even.
    #   Ranking on 1D + confidence-tier vs old 5D-composite lifts realized 1D P&L
    #   +0.22% → +0.43%/trade and win rate 54% → 60% (iter64, top-5/date).
    # → only BULLISH, MEDIUM+ confidence, HIGH tier first, then ret_hi × ML × sector.
    _CONF_MULT = {"HIGH": 1.0, "MEDIUM": 0.80}
    _ACCEPTED_DIRECTIONS = {"BULLISH", "SLIGHTLY BULLISH"}

    def _score_tf(p: dict) -> float:
        """Composite ranking score for a qualified TF prediction. Higher is better.

        Volatility (ATR%) is the PRIMARY term — picks are ranked by how volatile the stock is.
        AI GATES the pick in _pick_best_tf (must be BULLISH/SLIGHTLY BULLISH, MEDIUM+ confidence,
        ret_hi>0); here confidence / ML / R:R / sector apply only a light multiplicative tilt so a
        stronger AI setup ranks above an equally-volatile weaker one. Falls back to AI ret_hi when
        ATR is missing so a pick is never scored zero purely for a missing ATR read.
        """
        price = float(p.get("price") or 0.0)
        atr14 = float((p.get("risk") or {}).get("atr14") or 0.0)
        atr_pct = (atr14 / price * 100.0) if price else 0.0
        base = atr_pct if atr_pct > 0 else float(p.get("ret_hi") or 0.0)

        conf = p.get("confidence", "LOW")
        conf_mult = _CONF_MULT.get(conf, 0.55)

        ml_prob = float((p.get("ml") or {}).get("probability") or 0.5)
        # Sector mode is AI-only: drop the ML feature-score multiplier. Otherwise apply the usual
        # small [0.85,1.15] tilt.
        ml_factor = 1.0 if _sector_mode_active else 1.0 + (ml_prob - 0.5) * 0.30

        risk_data = p.get("risk") or {}
        actual_rr = risk_data.get("actual_rr")
        rr_factor = 1.12 if (actual_rr is not None and actual_rr >= 2.0) else 1.0

        sector_data = p.get("sector") or {}
        if sector_data.get("leading"):
            sector_factor = 1.12
        elif sector_data.get("lagging"):
            sector_factor = 0.90
        else:
            sector_factor = 1.0

        score = base * conf_mult * ml_factor * rr_factor * sector_factor
        # SLIGHTLY BULLISH = downgraded from BULLISH (bear-market Nifty gate or weak signals).
        # Apply a 0.65× penalty so genuine BULLISH picks always rank higher for the same setup.
        if p.get("direction") == "SLIGHTLY BULLISH":
            score *= 0.65
        # When AI is unavailable, apply a heavy penalty — signal-strong stocks still
        # surface but rank below LLM-confirmed ones.
        if p.get("no_trade_reason") == "ai_unavailable":
            score *= 0.40
        return score

    # Phase 1 was signal-only (no AI), so direction = NEUTRAL for many stocks in
    # bear markets. Select candidates by ML probability + signal count instead of
    # direction — Phase 2 AI will assign the real direction.
    def _score_scan(p: dict) -> float:
        # Sector mode: rank purely by the stock's SECTOR volatility (most "violent" sectors
        # first), tie-broken by the stock's own ATR% — no ML score involved. AI picks from this
        # order in Phase 2.
        if _sector_mode_active:
            sv = float(_sector_vol_by_ticker.get(p.get("ticker"), 0.0))
            atr14 = (p.get("risk") or {}).get("atr14") or 0
            price = p.get("price") or 1
            atr_pct = (atr14 / price * 100) if price else 0
            return sv * 100.0 + atr_pct  # sector volatility dominates; stock ATR breaks ties
        ml_prob = float((p.get("ml") or {}).get("probability") or 0.5)
        sig_count = int((p.get("ml") or {}).get("signal_count") or
                        len(p.get("active_strategies") or []))
        sector_lead = 1.1 if (p.get("sector") or {}).get("leading") else 1.0
        # Volatility (ATR%) is the PRIMARY ranker — surface the most volatile movers across the
        # whole NSE market. ML probability + signal count + sector-leading apply only a light
        # multiplicative tilt so, among similarly-volatile names, one with supporting signals
        # ranks above one without.
        atr14 = (p.get("risk") or {}).get("atr14") or 0
        price = p.get("price") or 1
        atr_pct = atr14 / price * 100 if price else 0
        tilt = 1.0 + (ml_prob - 0.5) * 0.4 + min(sig_count, 5) * 0.04
        return atr_pct * tilt * sector_lead

    valid_scan = [p for p in scan_preds.values() if p and not p.get("error")]
    # Price band: keep stocks trading in (_TOP5_MIN_PRICE, _TOP5_MAX_PRICE] (₹) — excludes both
    # penny stocks (unreliable ATR%-driven noise) and unaffordably high-priced scrips.
    if _TOP5_MAX_PRICE and _TOP5_MAX_PRICE > 0:
        _pre_n = len(valid_scan)
        valid_scan = [p for p in valid_scan if 0 < (p.get("price") or 0) <= _TOP5_MAX_PRICE]
        if _pre_n != len(valid_scan):
            logger.info("[TOP5] PRICE CAP ≤₹%.0f: %d/%d candidates kept",
                        _TOP5_MAX_PRICE, len(valid_scan), _pre_n)
    if _TOP5_MIN_PRICE and _TOP5_MIN_PRICE > 0:
        _pre_n = len(valid_scan)
        valid_scan = [p for p in valid_scan if (p.get("price") or 0) >= _TOP5_MIN_PRICE]
        if _pre_n != len(valid_scan):
            logger.info("[TOP5] PENNY FLOOR ≥₹%.0f: %d/%d candidates kept",
                        _TOP5_MIN_PRICE, len(valid_scan), _pre_n)
    valid_scan.sort(key=_score_scan, reverse=True)

    # Volatility-ranked candidate pool for Phase 2. Phase 1 scanned the WHOLE market and ranked
    # every stock by ATR% (volatility); we hand only the top-K most volatile names to the AI so
    # LLM usage stays bounded while the pool is drawn from the entire NSE market. AI then confirms
    # direction on these (Phase 2), and sector diversity is applied to the final picks.
    _PHASE2_POOL = int(os.getenv("TOP5_PHASE2_POOL", "80") or "80")
    candidates = valid_scan[:_PHASE2_POOL] if (_PHASE2_POOL and _PHASE2_POOL > 0) else valid_scan

    # ── Phase 1b: ML conviction pass (standalone quantile model, no LLM) ──────
    # The ml_predictor is local + instant + rate-limit-free, so it can score the
    # market as a genuine SELECTOR (not just a display layer). We score the top
    # signal-ranked candidates, then BLEND: the candidate order fed to Phase 2 is
    # the interleaved union of the signal-top and the ML-top, so AI confirms BOTH
    # sets. ML predictions are stashed per ticker (closure) so _assemble can attach
    # them + compute an ML/AI agreement verdict. Bounded to _ML_SCAN_N + a deadline.
    # SKIPPED in sector mode — there ML is deliberately out of the selection path (AI-only).
    _ML_SCAN_N = 300
    ml_preds_by_ticker: dict[str, dict] = {}
    try:
        from ml_predictor.infer import get_ml_predictor
        _mlp = get_ml_predictor()
    except Exception:
        _mlp = None

    if (not _sector_mode_active and _mlp is not None
            and getattr(_mlp, "available", False) and candidates):
        ml_pool = candidates
        _publish({
            "computing": True, "phase": "scanning",
            "scanned": _uni_n, "scan_total": _uni_n, "picks": [], "generated_at": None,
            "message": f"ML scoring {len(ml_pool)} shortlisted candidates…",
        })

        def _ml_one(p: dict):
            tk = p.get("ticker")
            try:
                return tk, _mlp.predict_all_tf(tk, live_price=p.get("price"))
            except Exception:
                return tk, None

        ml_start = time.time()
        _ml_deadline = 120
        _ml_ex = ThreadPoolExecutor(max_workers=min(8, len(ml_pool)))
        _ml_futs = {_ml_ex.submit(_ml_one, p): p for p in ml_pool}
        try:
            for fut in _cf.as_completed(_ml_futs, timeout=_ml_deadline):
                tk, res = fut.result()
                if res and res.get("available"):
                    ml_preds_by_ticker[tk] = res
        except _cf.TimeoutError:
            logger.info("[TOP5] PHASE 1b ML deadline (%ds) reached", _ml_deadline)
        for fut in _ml_futs:
            if not fut.done():
                fut.cancel()
        _ml_ex.shutdown(wait=False)
        logger.info("[TOP5] PHASE 1b (ML conviction): scored %d/%d in %.1fs",
                    len(ml_preds_by_ticker), len(ml_pool), time.time() - ml_start)

    # `candidates` stays strictly volatility-ranked (set above): the user wants the most volatile
    # stocks surfaced, so Phase 2 AI confirms direction on the top-volatility pool in that order.
    # ML predictions (ml_preds_by_ticker) are still attached per stock in _assemble for the
    # ML/AI-agreement verdict, but they no longer reorder the pool.

    market_from_scan = next((p.get("market", {}) for p in scan_preds.values() if p and p.get("market")), {})
    shared_ctx = {
        "vix_level": market_from_scan.get("vix_level", 18.0),
        "vix_label": market_from_scan.get("vix_label", "UNKNOWN — assume moderate"),
        "nifty_ok": market_from_scan.get("nifty_ok", True),
        "nifty_label": market_from_scan.get("nifty_label", ""),
        "macro_ok": market_from_scan.get("macro_ok", True),
        "macro_label": market_from_scan.get("macro_label", ""),
    } if market_from_scan else None

    if not candidates:
        return {
            "picks": [],
            "market": market_from_scan,
            "no_picks_reason": "No scan results available — universe fetch or OHLCV data failed",
            "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
        }

    # ATR multipliers and R:R targets per timeframe (must match predictor_core.py)
    _ATR_MULT = {"INTRADAY": 0.4, "1D": 0.7, "3D": 1.1, "5D": 1.5}
    _RR_MULT  = {"INTRADAY": 1.2, "1D": 1.5, "3D": 1.7, "5D": 2.0}

    def _derive_risk(price, atr14, tf):
        """Derive SL/target from ATR when AI response risk fields are missing."""
        if not price or not atr14 or tf not in _ATR_MULT:
            return None, None, None
        sl_risk  = _ATR_MULT[tf] * atr14
        sl_price = round(price - sl_risk, 2)
        sl_pct   = round(sl_risk / price * 100, 1)
        sl_tgt   = round(price + sl_risk * _RR_MULT[tf], 2)
        return sl_price, sl_pct, sl_tgt

    # For shortlisted top picks, compute full-debate AI predictions for all TFs.
    # 5D is re-run with full debate (not reused from the fast-mode scan) so all
    # three timeframe tabs on the card have consistent, high-quality forecasts.
    ai_jobs: list[tuple[str, str]] = []
    for stock in candidates:
        ticker = stock["ticker"]
        for tf in TIMEFRAMES:
            ai_jobs.append((ticker, tf))

    def _empty_ai_tf(price_val: float, reason: str = "ai_unavailable") -> dict:
        """Represent an unavailable/pending AI forecast for a timeframe cell.

        reason="pending" is used for partial (streaming) snapshots while Phase 2 is
        still running — the frontend renders a spinner for those cells. The FINAL
        result uses reason="ai_unavailable" for any TF that never resolved.
        """
        return {
            "expected_return_range": None,
            "midpoint": 0,
            "ret_lo": None,
            "ret_hi": None,
            "target_price_lo": None,
            "target_price_hi": None,
            "expected_target_price": None,
            "expected_entry_price": price_val,
            "direction": "NO TRADE",
            "confidence": "LOW",
            "no_trade_reason": reason,
            "signal_count": 0,
            "predicted_direction": None,
            "predicted_return_lo": None,
            "predicted_return_hi": None,
            "ai_forecast": None,
            "risk": {},
        }

    # Rank across the shown timeframes — volatility-primary (see _pick_best_tf).
    _RANK_TFS = ["INTRADAY", "1D"]

    def _pick_best_tf(ticker: str) -> tuple[str, float]:
        """Return the best short-term TF (INTRADAY/1D) and its VOLATILITY-primary ranking score.

        AI GATES the pick: the TF must be a resolved BULLISH / SLIGHTLY BULLISH call with MEDIUM+
        confidence and a positive best-case return. Among qualified TFs, the one with the higher
        volatility-primary composite (_score_tf — ATR% × AI-confidence/ML/R:R/sector tilt) wins,
        and that score is used to rank stocks against each other, so the most volatile AI-approved
        movers surface first. Returns ("1D", 0.0) when the stock has no eligible short-term setup
        so it sinks to the bottom of the ranking.
        """
        best, best_score = None, 0.0
        for tf in _RANK_TFS:
            pred = ai_preds.get((ticker, tf), {})
            if not pred or pred.get("error"):
                continue
            if pred.get("no_trade_reason"):
                continue
            if pred.get("direction") not in _ACCEPTED_DIRECTIONS:
                continue
            if pred.get("confidence") not in _CONF_MULT:
                continue
            ret_hi = float(pred.get("ret_hi") or 0.0)
            if ret_hi <= 0:
                continue
            s = _score_tf(pred)
            if s > best_score:
                best, best_score = tf, s
        if best is None:
            return "1D", 0.0
        # Sector mode: tilt the score by the stock's SECTOR volatility so that among AI-approved
        # picks, ones in more "violent" sectors rank higher. +10% per 1% sector ATR, capped at +50%.
        if _sector_mode_active:
            sv = float(_sector_vol_by_ticker.get(ticker, 0.0))
            best_score *= 1.0 + min(sv, 5.0) * 0.10
        return best, best_score

    def _assemble(partial: bool = False) -> list[dict]:
        """Assemble ranked pick cards from candidates + whatever AI predictions exist.

        Reused for both the final result (partial=False) and the streaming progress
        snapshots emitted during Phase 2 (partial=True). During a partial pass, a TF
        whose AI job hasn't finished yet is marked no_trade_reason="pending" (renders
        as a spinner) and only stocks that already have a qualifying resolved setup
        are surfaced — so cards appear as soon as they're ready instead of all at once.
        """
        _missing_reason = "pending" if partial else "ai_unavailable"
        assembled: list[dict] = []
        for stock in candidates:
            ticker = stock["ticker"]
            price = stock.get("price") or 0
            anchor_atr14 = ((ai_preds.get((ticker, "1D"), {}).get("risk") or {}).get("atr14")
                            or (ai_preds.get((ticker, "INTRADAY"), {}).get("risk") or {}).get("atr14"))
            timeframe_data: dict[str, dict] = {}
            ml_res = ml_preds_by_ticker.get(ticker)
            ml_tfs = (ml_res.get("tfs", {}) if (ml_res and ml_res.get("available")) else {}) or {}

            for tf in TIMEFRAMES:
                base_stock = ai_preds.get((ticker, tf), {})
                if not base_stock or base_stock.get("error"):
                    base_stock = _empty_ai_tf(price, _missing_reason)

                tf_risk = base_stock.get("risk", {}) or {}
                sl = tf_risk.get("stop_loss")
                tgt = tf_risk.get("min_target")
                sl_pct = tf_risk.get("stop_loss_pct")
                actual_rr = tf_risk.get("actual_rr")
                if sl is None and anchor_atr14:
                    sl, sl_pct, tgt = _derive_risk(price, anchor_atr14, tf)
                    pred_ret_hi = base_stock.get("ret_hi")
                    if pred_ret_hi and pred_ret_hi > 0 and price > 0:
                        tgt = round(price * (1 + pred_ret_hi / 100), 2)

                timeframe_data[tf] = {
                    "expected_return_range": base_stock.get("expected_return_range"),
                    "midpoint":     base_stock.get("midpoint", 0),
                    "ret_lo":       base_stock.get("ret_lo"),
                    "ret_hi":       base_stock.get("ret_hi"),
                    "target_price_lo": base_stock.get("target_price_lo"),
                    "target_price_hi": base_stock.get("target_price_hi"),
                    "expected_target_price": base_stock.get("expected_target_price"),
                    "expected_entry_price": base_stock.get("expected_entry_price", price),
                    "gapped_past_target": base_stock.get("gapped_past_target", False),
                    "direction":    base_stock.get("direction", "NO TRADE"),
                    "confidence":   base_stock.get("confidence", "LOW"),
                    "no_trade_reason":    base_stock.get("no_trade_reason"),
                    "range_bound":        base_stock.get("range_bound", False),
                    "signal_count":       0,
                    "predicted_direction":    base_stock.get("predicted_direction"),
                    "predicted_return_lo":    base_stock.get("predicted_return_lo"),
                    "predicted_return_hi":    base_stock.get("predicted_return_hi"),
                    "ai_forecast":            base_stock.get("ai_forecast"),
                    "ml":                     _slim_ml_tf(ml_tfs.get(tf)),
                    "stop_loss":     sl,
                    "stop_loss_pct": sl_pct,
                    "min_target":    tgt,
                    "actual_rr":     actual_rr,
                }

            pick = dict(stock)
            best_tf, best_score = _pick_best_tf(ticker)
            pick["best_tf"] = best_tf
            ai_anchor = ai_preds.get((ticker, best_tf), {})
            if ai_anchor and not ai_anchor.get("error"):
                pick["direction"] = ai_anchor.get("direction", pick.get("direction"))
                pick["confidence"] = ai_anchor.get("confidence", pick.get("confidence"))
                pick["news"] = ai_anchor.get("news", pick.get("news", {}))
                pick["risk"] = ai_anchor.get("risk", pick.get("risk", {}))
            pick["signals"] = {}
            pick["signal_count"] = 0
            pick["timeframes"] = timeframe_data

            # ── ML/AI agreement verdict (AI confirms the ML selection, or not) ──
            # ml_selected = this stock was scored by the ML selector this run.
            # Verdict on the best timeframe: confirmed (same dir) → boost so it
            # surfaces; disagree (opposite dir) → kept but flagged; mixed → neutral.
            pick["ml_selected"] = ticker in ml_preds_by_ticker
            ml_best = _slim_ml_tf(ml_tfs.get(best_tf))
            ai_best_dir = (ai_preds.get((ticker, best_tf), {}) or {}).get("direction")
            pick["ml_ai_verdict"] = None
            if ml_best and ml_best.get("direction") and ai_best_dir:
                m = str(ml_best["direction"]).upper()
                a = str(ai_best_dir).upper()
                if m == a and m in ("BULLISH", "BEARISH"):
                    pick["ml_ai_verdict"] = "confirmed"
                    best_score *= 1.10  # reward ML+AI consensus so it ranks higher
                elif (m == "BULLISH" and a == "BEARISH") or (m == "BEARISH" and a == "BULLISH"):
                    pick["ml_ai_verdict"] = "disagree"
                elif m in ("BULLISH", "BEARISH") and a in ("BULLISH", "BEARISH", "NEUTRAL"):
                    pick["ml_ai_verdict"] = "mixed"
            pick["_score_best"] = best_score
            assembled.append(pick)

        assembled.sort(key=lambda x: x.get("_score_best", 0.0), reverse=True)
        qualifying = [p for p in assembled if p.get("_score_best", 0) > 0]
        # Partial snapshots only show already-qualifying cards (progressive reveal);
        # the final result falls back to the raw list if nothing qualifies. Sector diversity
        # spreads the final picks across NSE sectors so one hot sector can't fill the whole list.
        base_pool = qualifying if (qualifying or partial) else assembled
        result_picks = _diversify_by_sector(base_pool, top_n, _MAX_PER_SECTOR)

        for i, p in enumerate(result_picks):
            p["rank"] = i + 1
            p.pop("_score_best", None)
            if not partial:
                # Specialist lookup hits the DB — skip it on partial (streaming) passes.
                specialist = _get_specialist_recommendation(p.get("ticker", ""))
                if specialist:
                    p["specialist_recommendation"] = specialist
        return result_picks

    # ── Phase 2: AI fast-mode predictions for top candidates (150s cap) ────────

    # Uses fast-mode (1 LLM call per ticker shared across TFs) instead of full
    # debate (4 LLM calls) — 4× fewer LLM calls while giving the same AI direction.
    ai_preds: dict[tuple[str, str], dict] = {}  # all TFs filled by Step 2
    phase2_start = time.time()
    phase2_elapsed = 0
    phase2_timeouts = 0

    # Degraded-mode: when all cloud providers are daily-exhausted, Ollama is the
    # sole fallback (semaphore=1, ~90s/call). 150 stocks × 3 TFs = 450 sequential
    # Ollama calls ≈ 11 hours — impossible in any deadline. Instead, promote Phase 1
    # signal+ML results (which already have direction/confidence/ret_lo/ret_hi) to
    # the 1D slot so stocks surface with real directional calls. INTRADAY and 3D
    # show "AI unavailable". Picks are signal-quality, not debate-quality, but
    # infinitely better than zero picks.
    try:
        from llm_client import _all_cloud_daily_exhausted as _cloud_exhausted
        _p2_degraded = _cloud_exhausted()
    except Exception:
        _p2_degraded = False

    if _p2_degraded and ai_jobs:
        for stock in candidates:
            ticker = stock["ticker"]
            p1 = scan_preds.get(ticker, {})
            if p1 and not p1.get("error") and p1.get("direction") in _ACCEPTED_DIRECTIONS:
                ai_preds[(ticker, "1D")] = p1  # real signal+ML data
            # INTRADAY and 3D slots left empty → _empty_ai_tf() below
        ai_jobs = []
        msg2 = f"[TOP5] PHASE 2 DEGRADED (cloud daily-exhausted): promoted Phase 1 data for {len(ai_preds)} stocks"
        print(msg2)
        logger.info(msg2)

    elif ai_jobs:
        logger.info("[TOP5] PHASE 2 start: %d candidates × %d TFs = %d AI jobs",
                    len(candidates), len(TIMEFRAMES), len(ai_jobs))
        _publish({
            "computing": True, "phase": "predicting",
            "predicted": 0, "predict_total": len(ai_jobs), "candidates": len(candidates),
            "picks": [], "market": market_from_scan, "generated_at": None,
            "message": f"Running AI on {len(candidates)} candidates — 0/{len(ai_jobs)}",
        })
        phase2_ex = ThreadPoolExecutor(max_workers=min(len(candidates), 6) * len(TIMEFRAMES))
        p2_futs = {
            phase2_ex.submit(_run_predict_with_ctx, ticker, tf, shared_ctx, True, True): (ticker, tf)
            for ticker, tf in ai_jobs
        }
        p2_done_n = 0
        _last_emit = time.time()
        try:
            for f in _cf.as_completed(p2_futs, timeout=150):
                ticker, tf = p2_futs[f]
                try:
                    _ticker, _tf, pred = f.result()
                    ai_preds[(_ticker, _tf)] = pred
                except Exception as e:
                    ai_preds[(ticker, tf)] = {}
                    if "timeout" in str(e).lower():
                        phase2_timeouts += 1
                p2_done_n += 1
                # Stream partial ranked picks so ready cards render immediately —
                # emit every 6 completions or at least every 4s, whichever first.
                if progress_cb and (p2_done_n % 6 == 0 or time.time() - _last_emit > 4):
                    _last_emit = time.time()
                    _partial = _assemble(partial=True)
                    logger.info("[TOP5] PHASE 2 progress: %d/%d jobs done, %d cards ready (%.0fs)",
                                p2_done_n, len(ai_jobs), len(_partial), time.time() - phase2_start)
                    _publish({
                        "computing": True, "phase": "predicting",
                        "predicted": p2_done_n, "predict_total": len(ai_jobs),
                        "candidates": len(candidates), "picks": _partial,
                        "market": market_from_scan, "generated_at": None,
                        "message": f"Running AI on candidates — {p2_done_n}/{len(ai_jobs)} ({len(_partial)} ready)",
                    })
        except _cf.TimeoutError:
            logger.info("[TOP5] PHASE 2 deadline (150s) reached at %d/%d jobs", p2_done_n, len(ai_jobs))
        # Cancel and account for any jobs that never completed within the deadline.
        for f, key in p2_futs.items():
            if not f.done():
                f.cancel()
                if key not in ai_preds:
                    ai_preds[key] = {}
                    phase2_timeouts += 1
        phase2_ex.shutdown(wait=False)
        phase2_elapsed = time.time() - phase2_start
        msg2 = f"[TOP5] PHASE 2 (AI fast): {len(ai_preds)}/{len(ai_jobs)} done, {phase2_timeouts} timeout, {phase2_elapsed:.1f}s"
        print(msg2)
        logger.info(msg2)

    # Step 3: Assemble the final ranked picks (partial=False → any still-missing TF
    # is marked ai_unavailable, and specialist recommendations are attached).
    picks = _assemble(partial=False)

    total_elapsed = time.time() - start_time
    msg_final = f"[TOP5] TOTAL time: {total_elapsed:.1f}s | Phase1: {phase1_elapsed:.1f}s | Phase2: {phase2_elapsed:.1f}s | {len(picks)} picks generated"
    print(msg_final)
    logger.info(msg_final)

    return {
        "picks": picks,
        "market": market_from_scan,
        "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
        "_timing": {
            "phase1_sec": round(phase1_elapsed, 1),
            "phase2_sec": round(phase2_elapsed if ai_jobs else 0, 1),
            "total_sec": round(total_elapsed, 1),
            "phase1_timeouts": timeouts,
            "phase1_stocks": len(scan_preds),
            "phase2_jobs": len(ai_jobs),
        }
    }


def get_weekly_picks(
    universe: Optional[list[str]] = None,
    top_n: int = 20,
    _universe_size: int = 150,
    force_universe_refresh: bool = False,
) -> dict:
    """
    Top N NSE stocks for a 5-10 day (1W) hold. Mirrors get_top5_picks() but anchors
    on the 1W timeframe and applies stricter R:R requirements suitable for longer holds.

    Returns same structure as get_top5_picks() with timeframes: {"3D", "5D", "1W"}.
    """
    from datetime import datetime

    _WEEKLY_TIMEFRAMES = ["3D", "5D", "1W"]
    _W_ATR_MULT = {"3D": 1.1, "5D": 1.5, "1W": 1.8}
    _W_RR_MULT  = {"3D": 1.7, "5D": 2.0, "1W": 2.5}

    effective_universe = universe or _get_top5_universe(force_refresh=force_universe_refresh) or DEFAULT_UNIVERSE
    effective_universe = effective_universe[:_universe_size] if _universe_size > 0 else effective_universe

    # Step 1: 1W AI scan for candidate selection
    scan_preds: dict[str, dict] = {}
    with ThreadPoolExecutor(max_workers=min(20, max(1, len(effective_universe)))) as executor:
        futures = {
            executor.submit(_run_predict_with_ctx, ticker, "1W", None, True, True): ticker
            for ticker in effective_universe
        }
        for future in as_completed(futures):
            ticker = futures[future]
            try:
                _ticker, _tf, pred = future.result(timeout=90)
                scan_preds[_ticker] = pred
            except Exception:
                scan_preds[ticker] = {}

    # Scoring — same formula as _score_5d but anchored on 1W ret_hi
    _W_CONF_MULT = {"HIGH": 1.0, "MEDIUM": 0.80}   # exclude LOW (only 84% hit rate)
    _W_ACCEPTED_DIRECTIONS = {"BULLISH", "SLIGHTLY BULLISH"}
    # _W_MIN_RET_HI / _W_MIN_RR removed — calibrated ret_hi (0.18%) is far below any
    # meaningful magnitude threshold; scoring handles ranking instead.

    def _score_1w(p: dict) -> float:
        ret_hi_val = float(p.get("ret_hi") or 0.0)
        conf_mult = _W_CONF_MULT.get(p.get("confidence", "LOW"), 0.55)
        ml_prob = float((p.get("ml") or {}).get("probability") or 0.5)
        ml_factor = 1.0 + (ml_prob - 0.5) * 0.30
        actual_rr = (p.get("risk") or {}).get("actual_rr")
        rr_factor = 1.12 if (actual_rr is not None and actual_rr >= 2.0) else 1.0
        sector_data = p.get("sector") or {}
        sector_factor = 1.12 if sector_data.get("leading") else (0.90 if sector_data.get("lagging") else 1.0)
        atr14 = (p.get("risk") or {}).get("atr14") or 0
        price = p.get("price") or 1
        atr_pct = atr14 / price * 100 if price else 0
        vol_factor = 1.0 + min(atr_pct / 4.0, 0.5)
        score = ret_hi_val * conf_mult * ml_factor * rr_factor * sector_factor * vol_factor
        if p.get("direction") == "SLIGHTLY BULLISH":
            score *= 0.65
        if p.get("no_trade_reason") == "ai_unavailable":
            score *= 0.40
        return score

    bullish = []
    for p in scan_preds.values():
        if not p or p.get("direction") not in _W_ACCEPTED_DIRECTIONS:
            continue
        if p.get("confidence") not in _W_CONF_MULT:
            continue
        if float(p.get("ret_hi") or 0.0) <= 0:
            continue
        bullish.append(p)

    bullish.sort(key=_score_1w, reverse=True)
    candidates = bullish

    market_from_scan = next((p.get("market", {}) for p in scan_preds.values() if p and p.get("market")), {})
    shared_ctx = {
        "vix_level": market_from_scan.get("vix_level", 18.0),
        "vix_label": market_from_scan.get("vix_label", "UNKNOWN — assume moderate"),
        "nifty_ok": market_from_scan.get("nifty_ok", True),
        "nifty_label": market_from_scan.get("nifty_label", ""),
        "macro_ok": market_from_scan.get("macro_ok", True),
        "macro_label": market_from_scan.get("macro_label", ""),
    } if market_from_scan else None

    if not candidates:
        return {
            "picks": [],
            "market": market_from_scan,
            "no_picks_reason": "No bullish 1W setups available in current market conditions",
            "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
        }

    # Step 2: Re-predict 3D/5D for top candidates using shared market context
    def _derive_risk_weekly(price, atr14, tf):
        if not price or not atr14:
            return None, None, None
        sl_risk  = _W_ATR_MULT[tf] * atr14
        sl_price = round(price - sl_risk, 2)
        sl_pct   = round(sl_risk / price * 100, 1)
        sl_tgt   = round(price + sl_risk * _W_RR_MULT[tf], 2)
        return sl_price, sl_pct, sl_tgt

    ai_jobs = [(s["ticker"], tf) for s in candidates for tf in _WEEKLY_TIMEFRAMES if tf != "1W"]
    ai_preds: dict[tuple[str, str], dict] = {(s["ticker"], "1W"): s for s in candidates}
    if ai_jobs:
        with ThreadPoolExecutor(max_workers=min(8, len(ai_jobs))) as executor:
            futures = {
                executor.submit(_run_predict_with_ctx, ticker, tf, shared_ctx, True, False): (ticker, tf)
                for ticker, tf in ai_jobs
            }
            for future in as_completed(futures):
                ticker, tf = futures[future]
                try:
                    _ticker, _tf, pred = future.result(timeout=90)
                    ai_preds[(_ticker, _tf)] = pred
                except Exception:
                    ai_preds[(ticker, tf)] = {}

    def _empty_weekly_tf(price_val: float) -> dict:
        return {
            "expected_return_range": None, "midpoint": 0,
            "ret_lo": None, "ret_hi": None,
            "target_price_lo": None, "target_price_hi": None,
            "expected_target_price": None, "expected_entry_price": price_val,
            "direction": "NO TRADE", "confidence": "LOW",
            "no_trade_reason": "ai_unavailable", "signal_count": 0,
            "predicted_direction": None, "predicted_return_lo": None,
            "predicted_return_hi": None, "ai_forecast": None, "risk": {},
        }

    picks = []
    for stock in candidates:
        ticker  = stock["ticker"]
        price   = stock.get("price") or 0
        anchor_atr14 = (ai_preds.get((ticker, "1W"), {}).get("risk") or {}).get("atr14")
        timeframe_data: dict[str, dict] = {}

        for tf in _WEEKLY_TIMEFRAMES:
            base_stock = ai_preds.get((ticker, tf), {})
            if not base_stock or base_stock.get("error"):
                base_stock = _empty_weekly_tf(price)

            tf_risk = base_stock.get("risk", {}) or {}
            sl = tf_risk.get("stop_loss")
            tgt = tf_risk.get("min_target")
            sl_pct = tf_risk.get("stop_loss_pct")
            actual_rr = tf_risk.get("actual_rr")
            if sl is None and anchor_atr14:
                sl, sl_pct, tgt = _derive_risk_weekly(price, anchor_atr14, tf)
                pred_ret_hi = base_stock.get("ret_hi")
                if pred_ret_hi and pred_ret_hi > 0 and price > 0:
                    tgt = round(price * (1 + pred_ret_hi / 100), 2)

            timeframe_data[tf] = {
                "expected_return_range": base_stock.get("expected_return_range"),
                "midpoint":     base_stock.get("midpoint", 0),
                "ret_lo":       base_stock.get("ret_lo"),
                "ret_hi":       base_stock.get("ret_hi"),
                "target_price_lo": base_stock.get("target_price_lo"),
                "target_price_hi": base_stock.get("target_price_hi"),
                "expected_target_price": base_stock.get("expected_target_price"),
                "expected_entry_price": base_stock.get("expected_entry_price", price),
                "direction":    base_stock.get("direction", "NO TRADE"),
                "confidence":   base_stock.get("confidence", "LOW"),
                "no_trade_reason":    base_stock.get("no_trade_reason"),
                "range_bound":        base_stock.get("range_bound", False),
                "signal_count":       0,
                "predicted_direction":    base_stock.get("predicted_direction"),
                "predicted_return_lo":    base_stock.get("predicted_return_lo"),
                "predicted_return_hi":    base_stock.get("predicted_return_hi"),
                "ai_forecast":            base_stock.get("ai_forecast"),
                "stop_loss":     sl,
                "stop_loss_pct": sl_pct,
                "min_target":    tgt,
                "actual_rr":     actual_rr,
            }

        pick = dict(stock)
        ai_anchor = ai_preds.get((ticker, "1W"), {})
        if ai_anchor and not ai_anchor.get("error"):
            pick["direction"] = ai_anchor.get("direction", pick.get("direction"))
            pick["confidence"] = ai_anchor.get("confidence", pick.get("confidence"))
            pick["news"] = ai_anchor.get("news", pick.get("news", {}))
            pick["risk"] = ai_anchor.get("risk", pick.get("risk", {}))
        pick["signals"] = {}
        pick["signal_count"] = 0
        pick["timeframes"] = timeframe_data
        pick["_score_1w"] = _score_1w(ai_anchor if (ai_anchor and not ai_anchor.get("error")) else stock)
        picks.append(pick)

    picks.sort(key=lambda x: x.get("_score_1w", 0.0), reverse=True)
    picks = picks[:top_n]
    for i, p in enumerate(picks):
        p["rank"] = i + 1
        p.pop("_score_1w", None)

    return {
        "picks": picks,
        "market": market_from_scan,
        "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
    }