llm_client.py

"""
llm_client.py — Shared LLM provider chain for all modules.

Single source of truth for OpenRouter → Ollama → Groq → Cerebras → HuggingFace provider
routing, dynamic reordering based on availability, cooldown tracking, and retry logic.
Imported by:
  - ai_forecast.py  (via `from llm_client import make_chat_call`)
  - news_sentiment.py
  - self_learning.py

Public API
----------
  make_chat_call(messages, max_tokens, temperature, fast_fail_on_rate_limit,
                 task_offset, max_retries) -> (content, provider, model)

  Raises RuntimeError if all providers are unavailable after max_retries passes.

Dynamic provider ordering
-------------------------
  Providers are sorted at each call by current availability:
    1. Available (not daily-exhausted, cooldown expired) — fewest recent failures first
    2. Rate-limited (cooldown active) — soonest recovery first
    3. Daily exhausted — lowest priority

  When ALL cloud providers are daily-exhausted → Ollama is promoted to Phase 1
  (before any cloud provider). This avoids wasting time on guaranteed 429s.

  Daily status resets automatically at midnight IST.
"""

from __future__ import annotations

import logging
import os
import threading
import time
from datetime import datetime, timedelta, timezone
from typing import Dict, List

import re

import requests

# Load .env before reading API keys
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    pass

logger = logging.getLogger(__name__)

# ── Unified provider status (replaces individual _*_DISABLED_UNTIL globals) ──
# avail_at: Unix timestamp — provider is usable again after this point (0 = now)
# daily_exhausted: True when daily quota is fully consumed (402/daily-quota 429)
# fail_streak: consecutive failure count — used for secondary sort
_PROVIDER_STATUS: dict = {
    "openrouter":  {"avail_at": 0.0, "daily_exhausted": False, "fail_streak": 0},
    "groq":        {"avail_at": 0.0, "daily_exhausted": False, "fail_streak": 0},
    "cerebras":    {"avail_at": 0.0, "daily_exhausted": False, "fail_streak": 0},
    "huggingface": {"avail_at": 0.0, "daily_exhausted": False, "fail_streak": 0},
    # Extra free tiers — large independent daily capacity (Gemini ~1,500 req/day, SambaNova
    # persistent free 70B, NVIDIA NIM free 70B/Nemotron). Appended LAST so the happy path is
    # unchanged, but the dynamic availability sort auto-promotes them to the front the moment the
    # first four degrade — which is exactly when the exhaustion→single-Ollama funnel used to bite.
    # No-op without keys.
    "gemini":      {"avail_at": 0.0, "daily_exhausted": False, "fail_streak": 0},
    "sambanova":   {"avail_at": 0.0, "daily_exhausted": False, "fail_streak": 0},
    "nvidia":      {"avail_at": 0.0, "daily_exhausted": False, "fail_streak": 0},
}
_PROVIDER_DAILY_RESET: str = ""   # "YYYY-MM-DD" IST string; reset daily flags on date change

_LLM_COOLDOWN_SECS: int = 600
_OPENROUTER_COOLDOWN_SECS: int = 60
_ALL_PROVIDERS_RETRY_WAIT_SECS: int = 2

_LLM_LOCK = threading.Lock()
_LLM_DISABLED_UNTIL: float = 0.0  # global kill-switch (all providers failed simultaneously)

# Limits concurrent cloud LLM calls. Ollama is handled separately via _OLLAMA_SEMAPHORE.
# 16 concurrent: enough for all watchlist stocks × 4 TFs without artificial throttling.
_LLM_SEMAPHORE = threading.Semaphore(16)

# Limit Ollama to 1 concurrent call: the HF Space serializes internally anyway,
# so multiple concurrent requests just queue inside Ollama and each waits N×40s.
_OLLAMA_SEMAPHORE = threading.Semaphore(1)

# Cache Ollama health check result to avoid hammering /api/tags on every call.
_OLLAMA_HEALTH_LAST_CHECK: float = 0.0
_OLLAMA_HEALTH_RESULT: bool = False
_OLLAMA_HEALTH_TTL: int = 60
# BUGFIX: a NEGATIVE (unhealthy) probe result must expire fast. A cold-starting HF Space
# returns False for ~20-40s and then becomes healthy; caching that False for the full 60s TTL
# locked every subsequent call out of Ollama (never re-probed, never warmed) — this is what
# SKIPped 47/72 predictions in the last batch run. Positive results keep the 60s TTL (which is
# all we need to avoid a /api/tags storm on a Space that is already up).
_OLLAMA_HEALTH_NEG_TTL: int = 10

# After a chat timeout or inference failure, back off for this many seconds before retrying.
# Prevents cascading 45s timeouts when the HF Space is unresponsive.
_OLLAMA_INFER_BACKOFF_UNTIL: float = 0.0
_OLLAMA_INFER_BACKOFF_SECS: int = 120  # 2 min — shorter than 300s so Ollama is re-tried mid-batch
# A warmup failure just means the model is still loading (cold start); it recovers in seconds,
# so back off only briefly. The old 300s backoff after a single cold-start warmup miss took
# Ollama offline for half of a ~10-min batch run, cascading into AI-unavailable SKIPs.
_OLLAMA_WARMUP_BACKOFF_SECS: int = 45
_OLLAMA_CHAT_TIMEOUT: int = 70          # 70s: warmup confirms model is loaded, so 70s is enough for 512 tokens

# Canonical cloud provider order (original preference before runtime reordering)
_CLOUD_PROVIDERS = ["openrouter", "groq", "cerebras", "huggingface", "gemini", "sambanova", "nvidia"]
_PROVIDER_ORDER = _CLOUD_PROVIDERS


# ── Provider status helpers ───────────────────────────────────────────────────

def _maybe_daily_reset() -> None:
    """Reset daily_exhausted flags at midnight IST. No-op if already reset today."""
    global _PROVIDER_DAILY_RESET
    today = datetime.now(timezone(timedelta(hours=5, minutes=30))).strftime("%Y-%m-%d")
    with _LLM_LOCK:
        if _PROVIDER_DAILY_RESET != today:
            _PROVIDER_DAILY_RESET = today
            for s in _PROVIDER_STATUS.values():
                s["daily_exhausted"] = False
                # Keep fail_streak — it decays naturally on success, not on reset
            logger.info("LLM: Daily provider status reset (IST midnight)")


def _get_cloud_order() -> list[str]:
    """
    Return cloud providers sorted by current availability.
    Scores: (0=available, 1=rate-limited, 2=daily-exhausted) then secondary sub-sort.
    Available providers with fewer failures come first; daily-exhausted come last.
    """
    _maybe_daily_reset()
    now = time.time()

    def _score(name: str):
        s = _PROVIDER_STATUS[name]
        if s["daily_exhausted"]:
            return (2, s["fail_streak"])
        if s["avail_at"] > now:
            return (1, s["avail_at"] - now)   # sooner recovery = higher priority
        return (0, s["fail_streak"])           # available; fewest failures first

    return sorted(_CLOUD_PROVIDERS, key=_score)


def _all_cloud_daily_exhausted() -> bool:
    """True when every cloud provider has consumed its daily quota."""
    _maybe_daily_reset()
    return all(_PROVIDER_STATUS[n]["daily_exhausted"] for n in _CLOUD_PROVIDERS)


def unavailability_is_recoverable() -> bool:
    """After make_chat_call raised, classify the outage as TEMPORARY vs HARD.

    TEMPORARY (returns True): at least one keyed cloud provider is only cooling down /
    burst-rate-limited (NOT daily-exhausted) and will recover once its short per-minute
    window lapses, OR Ollama is configured and not in a hard backoff. In this case the
    caller should mark the failure as a retryable 'timeout' — the frontend keeps refetching
    and the forecast fills in a moment later when a provider resets.

    HARD (returns False): every keyed cloud provider is daily-exhausted (real quota gone
    until midnight IST) AND Ollama is unconfigured / in backoff — nothing will work soon,
    so 'ai_unavailable' is the honest reason.

    Note: a provider with no API key can never recover on its own, so it doesn't count as
    a recoverable source (this is checked via provider_key_status()).
    """
    _maybe_daily_reset()
    keys = provider_key_status()
    # A keyed cloud provider that hit only a transient (per-minute) limit will reset soon.
    for name in _CLOUD_PROVIDERS:
        if keys.get(name) and not _PROVIDER_STATUS[name]["daily_exhausted"]:
            return True
    # Ollama has no rate limits — recoverable if configured and not in a hard backoff window.
    if keys.get("ollama") and time.time() >= _OLLAMA_INFER_BACKOFF_UNTIL:
        return True
    return False


def _mark_ok(name: str) -> None:
    with _LLM_LOCK:
        s = _PROVIDER_STATUS[name]
        s["avail_at"] = 0.0
        s["fail_streak"] = max(0, s["fail_streak"] - 1)


def _mark_rate_limited(name: str, cooldown: float = 60.0) -> None:
    with _LLM_LOCK:
        s = _PROVIDER_STATUS[name]
        s["avail_at"] = max(s["avail_at"], time.time() + cooldown)
        s["fail_streak"] += 1


def _mark_daily_exhausted(name: str) -> None:
    with _LLM_LOCK:
        s = _PROVIDER_STATUS[name]
        s["daily_exhausted"] = True
        s["fail_streak"] += 1
    logger.warning("LLM: %s marked daily-exhausted — will skip until midnight IST reset", name)


def _is_daily_quota_429(body: str) -> bool:
    """Classify a 429 response body as a daily quota exhaustion (bench until reset) rather than a
    transient per-minute RPM limit (short cooldown). Gemini/Google free-tier daily caps say
    'exceeded your current quota, please check your plan and billing' and reference *PerDay quota
    metrics; per-minute limits reference *PerMinute and a short retryDelay. Returns False when
    ambiguous so a recoverable RPM 429 is never sidelined for the whole day. Without this, a
    daily-dead provider that sits first in cloud_order gets retried (and 429s) on every call of a
    150-stock scan, deepening the funnel to slow Ollama."""
    if not body:
        return False
    b = body.lower()
    if "per minute" in b or "perminute" in b or "requests per minute" in b:
        return False
    return (
        "per day" in b or "perday" in b or "requests per day" in b
        or "check your plan and billing" in b
        or "exceeded your current quota" in b
        or "insufficient_quota" in b
        or "depleted" in b
    )


def _is_provider_available(name: str, fast_fail: bool = False) -> bool:
    """True if this provider should be tried right now."""
    _maybe_daily_reset()
    s = _PROVIDER_STATUS[name]
    if s["daily_exhausted"]:
        return False
    if fast_fail:
        return True  # fast_fail ignores temporary cooldowns
    return s["avail_at"] <= time.time()


def reset_ollama_state() -> None:
    """Clear Ollama's transient backoff + health cache so the next call gets a fresh probe.

    Called between deferred-retry rounds (backtest anti-skip loop, and the production
    background-fill pass) so a stock deferred because Ollama was mid-backoff gets a genuine
    fresh Ollama attempt in the next round — instead of being skipped again for the whole
    remaining backoff window. Does NOT touch cloud `daily_exhausted` flags (those are real
    quota state and reset only at midnight IST)."""
    global _OLLAMA_INFER_BACKOFF_UNTIL, _OLLAMA_HEALTH_RESULT, _OLLAMA_HEALTH_LAST_CHECK
    with _LLM_LOCK:
        _OLLAMA_INFER_BACKOFF_UNTIL = 0.0
        _OLLAMA_HEALTH_RESULT = False
        _OLLAMA_HEALTH_LAST_CHECK = 0.0  # force a re-probe (neg-TTL already expired)


# ── Diagnostic: provider key presence + isolated live probe ──────────────────
# Powers /api/provider-status?probe=1 so the ACTUAL provider state on a deployed HF
# Space (env comes from Secrets, not local .env) is observable. A provider can look
# healthy locally yet be unconfigured or rate-limited on the Space — this is the #1
# cause of "AI unavailable" that can't be reproduced by testing on the terminal.
# (env_key, chat_completions_url, model_env_var, default_model) — all OpenAI-compatible.
_PROBE_CONFIG: dict = {
    "openrouter":  ("OPENROUTER_API_KEY", "https://openrouter.ai/api/v1/chat/completions", "OPENROUTER_BEST_FREE_MODEL", "openai/gpt-oss-120b:free"),
    "groq":        ("GROQ_API_KEY", "https://api.groq.com/openai/v1/chat/completions", "GROQ_MODEL", "llama-3.3-70b-versatile"),
    "cerebras":    ("CEREBRAS_API_KEY", "https://api.cerebras.ai/v1/chat/completions", "CEREBRAS_MODEL", "llama-3.3-70b"),
    "huggingface": ("HF_TOKEN", "https://router.huggingface.co/novita/v3/openai/chat/completions", "HF_INFERENCE_MODEL", "meta-llama/Llama-3.1-8B-Instruct"),
    "gemini":      ("GEMINI_API_KEY", "https://generativelanguage.googleapis.com/v1beta/openai/chat/completions", "GEMINI_MODEL", "gemini-flash-latest"),
    "sambanova":   ("SAMBANOVA_API_KEY", "https://api.sambanova.ai/v1/chat/completions", "SAMBANOVA_MODEL", "Meta-Llama-3.3-70B-Instruct"),
    "nvidia":      ("NVIDIA_API_KEY", "https://integrate.api.nvidia.com/v1/chat/completions", "NVIDIA_MODEL", "nvidia/llama-3.3-nemotron-super-49b-v1"),
}


def provider_key_status() -> dict:
    """Which provider API keys are present in the current environment.

    Reads env vars directly, so on a deployed HF Space it reveals which Secrets are
    actually configured there — the difference that makes AI work locally (.env) but
    fail on the Space when a key was never synced via export_env_secrets.py."""
    st = {name: bool(os.environ.get(cfg[0], "").strip()) for name, cfg in _PROBE_CONFIG.items()}
    st["ollama"] = bool(os.environ.get("OLLAMA_ENDPOINT", "").strip())
    return st


def probe_provider(name: str, timeout: int = 12) -> dict:
    """Make a single isolated 1-token call to one provider and report the result.

    Diagnostic only — does NOT mutate the shared _PROVIDER_STATUS routing flags, so
    probing never perturbs live prediction routing. Returns
    {configured, ok, status, model, latency_ms, error}."""
    if name == "ollama":
        ep = os.environ.get("OLLAMA_ENDPOINT", "").strip()
        if not ep:
            return {"configured": False, "ok": False, "error": "OLLAMA_ENDPOINT not set"}
        try:
            from ollama_client import check_ollama_health, get_ollama_model
            t0 = time.time()
            ok = check_ollama_health(ep, timeout=timeout)
            return {"configured": True, "ok": bool(ok),
                    "model": get_ollama_model(ep) if ok else None,
                    "latency_ms": round((time.time() - t0) * 1000)}
        except Exception as e:
            return {"configured": True, "ok": False, "error": str(e)[:200]}

    cfg = _PROBE_CONFIG.get(name)
    if not cfg:
        return {"configured": False, "ok": False, "error": f"unknown provider {name}"}
    env_key, url, model_env, default_model = cfg
    api_key = os.environ.get(env_key, "").strip()
    if not api_key:
        return {"configured": False, "ok": False, "error": f"{env_key} not set"}
    model = (os.environ.get(model_env) or default_model).strip()
    t0 = time.time()
    try:
        resp = requests.post(
            url,
            headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
            json={"model": model, "messages": [{"role": "user", "content": "Reply with OK"}],
                  "temperature": 0.0, "max_tokens": 5},
            timeout=timeout,
        )
        latency = round((time.time() - t0) * 1000)
        ok = resp.status_code == 200
        out = {"configured": True, "ok": ok, "status": resp.status_code,
               "model": model, "latency_ms": latency}
        if not ok:
            out["error"] = resp.text[:200]
        return out
    except Exception as e:
        return {"configured": True, "ok": False, "model": model,
                "latency_ms": round((time.time() - t0) * 1000), "error": str(e)[:200]}


def make_chat_call(
    messages: List[Dict],
    max_tokens: int,
    temperature: float = 0.3,
    fast_fail_on_rate_limit: bool = False,
    task_offset: int = 0,
    max_retries: int = 0,
    call_timeout: int = 20,
    preferred_provider: str | None = None,
) -> tuple[str, str, str]:
    """
    Call an LLM with the given messages.
    Provider order is dynamic: best available cloud provider(s) → Ollama → remaining cloud.
    When all cloud providers are daily-exhausted, Ollama is promoted to Phase 1.

    task_offset rotates the starting provider among the ones currently available, so concurrent
    calls for different stocks/tasks spread across providers instead of all piling onto the same
    "best" one. preferred_provider, if given and available, is forced to the front instead
    (used to pin a specific provider when a caller wants to avoid the automatic rotation).

    Retries up to max_retries passes (default: AI_MAX_RETRIES env var, fallback 3).
    fast_fail_on_rate_limit=True: single pass, no Ollama, no sleep (backtest / low-latency).

    Returns (content, provider, model).
    Raises RuntimeError when all retries are exhausted.
    """
    _call_start = time.time()

    with _LLM_LOCK:
        if time.time() < _LLM_DISABLED_UNTIL:
            raise RuntimeError("LLM providers on cooldown — rate limit hit recently")

    # ── Retry helper: transient network errors only (not 429) ─────────────────
    def _retry_post(fn, retries: int = 2, backoff: float = 1.5):
        # fast_fail: fail immediately on first timeout — don't burn 40s retrying a slow provider
        if fast_fail_on_rate_limit:
            retries = 0
        last_exc = None
        for attempt in range(retries + 1):
            try:
                return fn()
            except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as exc:
                last_exc = exc
                if attempt < retries:
                    time.sleep(backoff * (attempt + 1))
        raise last_exc  # type: ignore[misc]

    # ── Provider 1: OpenRouter ────────────────────────────────────────────────
    def _try_openrouter():
        if not _is_provider_available("openrouter", fast_fail_on_rate_limit):
            return None
        api_key = os.environ.get("OPENROUTER_API_KEY", "").strip()
        if not api_key:
            logger.debug("OpenRouter skipped — OPENROUTER_API_KEY not set")
            return None
        model = (os.environ.get("OPENROUTER_BEST_FREE_MODEL") or "openai/gpt-oss-120b:free").strip()
        fallback_chain_raw = os.environ.get("OPENROUTER_FREE_MODELS", "")
        fallback_models = [m.strip() for m in fallback_chain_raw.split(",") if m.strip() and m.strip() != model]
        models_to_try = [model] + fallback_models[:9]

        for try_model in models_to_try:
            try:
                resp = _retry_post(lambda m=try_model: requests.post(
                    "https://openrouter.ai/api/v1/chat/completions",
                    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                    json={"model": m, "messages": messages, "temperature": temperature, "max_tokens": max_tokens},
                    timeout=12,
                ))
                if resp.status_code == 429:
                    _body = resp.text
                    if "free-models-per-day" in _body or "free_models_per_day" in _body:
                        _mark_daily_exhausted("openrouter")
                        return None
                    logger.warning("OpenRouter rate-limited on %s (429) — trying next model", try_model)
                    continue
                if resp.status_code == 401:
                    logger.error("OpenRouter auth failed (401) — check OPENROUTER_API_KEY")
                    return None
                if resp.status_code in (400, 404, 422):
                    logger.debug("OpenRouter model %s unavailable (%s) — trying next", try_model, resp.status_code)
                    continue
                if resp.status_code == 200:
                    content = (((resp.json().get("choices") or [{}])[0].get("message") or {}).get("content") or "").strip()
                    content = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip()
                    if content and not content.lstrip().startswith("<"):
                        _mark_ok("openrouter")
                        return content, "openrouter", try_model
                    if content:
                        logger.debug("OpenRouter %s — content starts with '<', skipping", try_model)
                    else:
                        logger.debug("OpenRouter %s returned empty content", try_model)
                else:
                    logger.warning("OpenRouter %s status %s — body: %s", try_model, resp.status_code, resp.text[:200])
            except Exception as exc:
                logger.debug("OpenRouter call failed for %s: %s", try_model, exc)

        cooldown = 5 if fast_fail_on_rate_limit else _OPENROUTER_COOLDOWN_SECS
        _mark_rate_limited("openrouter", cooldown)
        logger.warning("OpenRouter exhausted (no model produced content) — cooldown %.0fs", cooldown)
        return None

    # ── Provider 2: Groq ──────────────────────────────────────────────────────
    def _try_groq():
        if not _is_provider_available("groq", fast_fail_on_rate_limit):
            return None
        api_key = os.environ.get("GROQ_API_KEY", "").strip()
        if not api_key:
            logger.debug("Groq skipped — GROQ_API_KEY not set")
            return None
        primary = "llama-3.3-70b-versatile"
        fallback_chain_raw = os.environ.get("GROQ_FREE_MODELS", "")
        fallbacks = [m.strip() for m in fallback_chain_raw.split(",") if m.strip() and m.strip() != primary]
        models_to_try = [primary] + fallbacks[:3]

        for try_model in models_to_try:
            try:
                resp = _retry_post(lambda m=try_model: requests.post(
                    "https://api.groq.com/openai/v1/chat/completions",
                    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                    json={"model": m, "messages": messages, "temperature": temperature, "max_tokens": max_tokens},
                    timeout=15,
                ))
                if resp.status_code == 429:
                    logger.warning("Groq rate-limited on %s (429) — trying next model", try_model)
                    if not fast_fail_on_rate_limit:
                        time.sleep(1)
                    continue
                if resp.status_code == 401:
                    logger.error("Groq auth failed (401) — check GROQ_API_KEY")
                    return None
                if resp.status_code in (404, 422):
                    logger.debug("Groq model %s unavailable (%s) — trying next", try_model, resp.status_code)
                    continue
                if resp.status_code == 200:
                    content = (((resp.json().get("choices") or [{}])[0].get("message") or {}).get("content") or "").strip()
                    content = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip()
                    if content and not content.lstrip().startswith("<"):
                        _mark_ok("groq")
                        return content, "groq", try_model
                    if content:
                        logger.debug("Groq %s — content starts with '<', skipping", try_model)
                    else:
                        logger.debug("Groq %s returned empty content", try_model)
                else:
                    logger.warning("Groq %s status %s — body: %s", try_model, resp.status_code, resp.text[:200])
            except Exception as exc:
                logger.debug("Groq call failed for %s: %s", try_model, exc)

        _mark_rate_limited("groq", 5 if fast_fail_on_rate_limit else 60)
        return None

    # ── Provider 3: Cerebras ──────────────────────────────────────────────────
    def _try_cerebras():
        if not _is_provider_available("cerebras", fast_fail_on_rate_limit):
            return None
        api_key = os.environ.get("CEREBRAS_API_KEY", "").strip()
        if not api_key:
            logger.debug("Cerebras skipped — CEREBRAS_API_KEY not set")
            return None
        primary = os.environ.get("CEREBRAS_MODEL", "llama-3.3-70b").strip()
        fallback_chain_raw = os.environ.get("CEREBRAS_FALLBACK_MODELS", "llama-3.1-8b")
        fallbacks = [m.strip() for m in fallback_chain_raw.split(",") if m.strip() and m.strip() != primary]
        models_to_try = [primary] + fallbacks[:2]

        for try_model in models_to_try:
            try:
                resp = _retry_post(lambda m=try_model: requests.post(
                    "https://api.cerebras.ai/v1/chat/completions",
                    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                    json={"model": m, "messages": messages, "temperature": temperature, "max_tokens": max_tokens},
                    timeout=20,
                ))
                if resp.status_code == 429:
                    logger.warning("Cerebras rate-limited on %s (429) — trying next model", try_model)
                    continue
                if resp.status_code == 401:
                    logger.error("Cerebras auth failed (401) — check CEREBRAS_API_KEY")
                    return None
                if resp.status_code in (400, 404, 422):
                    logger.debug("Cerebras model %s unavailable (%s) — trying next", try_model, resp.status_code)
                    continue
                if resp.status_code == 200:
                    content = (((resp.json().get("choices") or [{}])[0].get("message") or {}).get("content") or "").strip()
                    content = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip()
                    if content and not content.lstrip().startswith("<"):
                        _mark_ok("cerebras")
                        return content, "cerebras", try_model
                    if content:
                        logger.debug("Cerebras %s — content starts with '<', skipping", try_model)
                    else:
                        logger.debug("Cerebras %s returned empty content", try_model)
                else:
                    logger.warning("Cerebras %s status %s — body: %s", try_model, resp.status_code, resp.text[:200])
            except Exception as exc:
                logger.debug("Cerebras call failed for %s: %s", try_model, exc)

        _mark_rate_limited("cerebras", 5 if fast_fail_on_rate_limit else 60)
        return None

    # ── Provider 4: HuggingFace ───────────────────────────────────────────────
    def _try_huggingface():
        if not _is_provider_available("huggingface", fast_fail_on_rate_limit):
            return None
        api_key = os.environ.get("HF_TOKEN", "").strip()
        if not api_key:
            logger.debug("HF Inference API skipped — HF_TOKEN not set")
            return None
        primary = (os.environ.get("HF_INFERENCE_MODEL") or "meta-llama/Llama-3.1-8B-Instruct").strip()
        fallback_chain_raw = os.environ.get("HF_INFERENCE_FALLBACK_MODELS", "")
        fallbacks = [m.strip() for m in fallback_chain_raw.split(",") if m.strip() and m.strip() != primary]
        models_to_try = [primary] + fallbacks[:3]

        _HF_ENDPOINTS = [
            "https://router.huggingface.co/novita/v3/openai/chat/completions",
            "https://api-inference.huggingface.co/v1/chat/completions",
        ]
        for try_model in models_to_try:
            try:
                endpoint = _HF_ENDPOINTS[0]
                try:
                    resp = _retry_post(lambda ep=endpoint, m=try_model: requests.post(
                        ep,
                        headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                        json={"model": m, "messages": messages, "temperature": temperature, "max_tokens": max_tokens},
                        timeout=30,
                    ))
                except Exception:
                    endpoint = _HF_ENDPOINTS[1]
                    resp = _retry_post(lambda ep=endpoint, m=try_model: requests.post(
                        ep,
                        headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                        json={"model": m, "messages": messages, "temperature": temperature, "max_tokens": max_tokens},
                        timeout=30,
                    ))
                if resp.status_code == 429:
                    logger.warning("HF Inference rate-limited on %s (429) — trying next model", try_model)
                    if not fast_fail_on_rate_limit:
                        time.sleep(1)
                    continue
                if resp.status_code == 401:
                    logger.error("HF Inference auth failed (401) — check HF_TOKEN")
                    return None
                if resp.status_code == 402:
                    _mark_daily_exhausted("huggingface")
                    return None
                if resp.status_code in (400, 404, 422, 503):
                    logger.debug("HF Inference model %s unavailable (%s) — trying next", try_model, resp.status_code)
                    continue
                if resp.status_code == 200:
                    content = (((resp.json().get("choices") or [{}])[0].get("message") or {}).get("content") or "").strip()
                    content = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip()
                    if content and not content.lstrip().startswith("<"):
                        _mark_ok("huggingface")
                        return content, "huggingface", try_model
                    if content:
                        logger.debug("HF Inference %s — content starts with '<', skipping", try_model)
                    else:
                        logger.debug("HF Inference %s returned empty content", try_model)
                else:
                    logger.warning("HF Inference %s status %s — body: %s", try_model, resp.status_code, resp.text[:200])
            except Exception as exc:
                logger.debug("HF Inference call failed for %s: %s", try_model, exc)

        _mark_rate_limited("huggingface", 5 if fast_fail_on_rate_limit else 30)
        return None

    # ── Extra free tiers: Gemini + SambaNova (OpenAI-compatible) ──────────────
    # Ported from research/providers_ext.py per PRODUCTION_DELTA.md. Large independent daily
    # capacity so `_all_cloud_daily_exhausted()` rarely becomes true → the exhaustion→Ollama
    # funnel stops firing. Both no-op (return None) when their key is unset.
    def _try_openai_compatible(name: str, base_url: str, api_key: str, models_to_try: list[str],
                               daily_on_429: bool):
        """Shared driver for OpenAI-compatible /chat/completions providers.

        Tries EVERY model in models_to_try before giving up on the provider — a 429/503/404 on
        one model (e.g. gemini-flash-latest under high demand) falls through to a sibling model
        (e.g. gemini-flash-lite-latest, which has separate capacity) instead of abandoning the
        whole provider. Only after ALL models fail is the provider marked down. If every failure
        was a quota 429 and daily_on_429 is set, the provider is marked daily-exhausted; otherwise
        a short cooldown is applied so a transient per-minute 429 doesn't sideline it for the day.
        """
        saw_daily_429 = False
        for try_model in models_to_try:
            try:
                resp = _retry_post(lambda m=try_model: requests.post(
                    f"{base_url}/chat/completions",
                    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                    json={"model": m, "messages": messages, "temperature": temperature, "max_tokens": max_tokens},
                    timeout=20,
                ))
                if resp.status_code == 429:
                    # Try the next model first — sibling models often have independent quota. Only
                    # after all models 429 do we mark the provider down (daily vs short cooldown).
                    # A 429 whose body signals a DAILY quota/billing exhaustion is benched till the
                    # reset even when daily_on_429 is False (per-minute default), so a daily-dead
                    # provider stops being retried first on every call of a batch scan.
                    logger.warning("%s rate-limited on %s (429) — trying next model", name, try_model)
                    if daily_on_429 or _is_daily_quota_429(resp.text or ""):
                        saw_daily_429 = True
                    if not fast_fail_on_rate_limit:
                        time.sleep(1)
                    continue
                if resp.status_code == 401:
                    logger.error("%s auth failed (401) — check API key", name)
                    return None
                if resp.status_code in (400, 404, 410, 422, 503):
                    # 410 = model retired ("no longer available"); 404 = unknown model;
                    # 503 = transient high demand. All: roll over to the next sibling model.
                    logger.debug("%s model %s unavailable (%s) — trying next", name, try_model, resp.status_code)
                    continue
                if resp.status_code == 200:
                    content = (((resp.json().get("choices") or [{}])[0].get("message") or {}).get("content") or "").strip()
                    content = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip()
                    if content and not content.lstrip().startswith("<"):
                        _mark_ok(name)
                        return content, name, try_model
                    logger.debug("%s %s returned empty/invalid content", name, try_model)
                else:
                    logger.warning("%s %s status %s — body: %s", name, try_model, resp.status_code, resp.text[:200])
            except Exception as exc:
                logger.debug("%s call failed for %s: %s", name, try_model, exc)
        # All models failed. If we only ever saw quota 429s, mark daily-exhausted so the provider
        # drops to the back until midnight IST; otherwise a short cooldown keeps it in rotation.
        if saw_daily_429:
            _mark_daily_exhausted(name)
        else:
            _mark_rate_limited(name, 5 if fast_fail_on_rate_limit else 60)
        return None

    def _try_gemini():
        if not _is_provider_available("gemini", fast_fail_on_rate_limit):
            return None
        api_key = os.environ.get("GEMINI_API_KEY", "").strip()
        if not api_key:
            logger.debug("Gemini skipped — GEMINI_API_KEY not set")
            return None
        # NOTE: the older `gemini-2.5-flash` / `gemini-2.5-flash-lite` names now return HTTP 404
        # ("no longer available to new users") on the OpenAI-compat endpoint — they silently killed
        # this whole provider. Use the alias names that stay valid across model generations, with a
        # multi-model fallback so a 503 (high demand) or 429 (per-model quota) on the primary falls
        # through to a sibling with independent capacity. daily_on_429=False: Gemini free-tier 429s
        # are usually per-minute RPM limits that recover in seconds, so a short cooldown keeps it in
        # rotation instead of benching it until midnight.
        primary = (os.environ.get("GEMINI_MODEL") or "gemini-flash-latest").strip()
        fallback_raw = os.environ.get(
            "GEMINI_FALLBACK_MODELS", "gemini-flash-lite-latest,gemini-2.0-flash"
        )
        fallbacks = [m.strip() for m in fallback_raw.split(",") if m.strip() and m.strip() != primary]
        models_to_try = [primary] + fallbacks[:3]
        return _try_openai_compatible(
            "gemini", "https://generativelanguage.googleapis.com/v1beta/openai",
            api_key, models_to_try, daily_on_429=False,
        )

    def _try_sambanova():
        if not _is_provider_available("sambanova", fast_fail_on_rate_limit):
            return None
        api_key = os.environ.get("SAMBANOVA_API_KEY", "").strip()
        if not api_key:
            logger.debug("SambaNova skipped — SAMBANOVA_API_KEY not set")
            return None
        primary = (os.environ.get("SAMBANOVA_MODEL") or "Meta-Llama-3.3-70B-Instruct").strip()
        fallback_raw = os.environ.get("SAMBANOVA_FALLBACK_MODELS", "Meta-Llama-3.1-8B-Instruct")
        fallbacks = [m.strip() for m in fallback_raw.split(",") if m.strip() and m.strip() != primary]
        models_to_try = [primary] + fallbacks[:2]
        return _try_openai_compatible(
            "sambanova", "https://api.sambanova.ai/v1",
            api_key, models_to_try, daily_on_429=False,
        )

    def _try_nvidia():
        # NVIDIA NIM (build.nvidia.com) — free API key, OpenAI-compatible, large independent
        # daily capacity across many models (Llama-3.3-70B, Nemotron, DeepSeek, Qwen). Zero cost.
        # daily_on_429=False: NIM free-tier 429s are per-minute RPM limits that recover in seconds,
        # so a short cooldown keeps it in rotation instead of benching it until midnight.
        if not _is_provider_available("nvidia", fast_fail_on_rate_limit):
            return None
        api_key = os.environ.get("NVIDIA_API_KEY", "").strip()
        if not api_key:
            logger.debug("NVIDIA NIM skipped — NVIDIA_API_KEY not set")
            return None
        primary = (os.environ.get("NVIDIA_MODEL") or "nvidia/llama-3.3-nemotron-super-49b-v1").strip()
        fallback_raw = os.environ.get(
            "NVIDIA_FALLBACK_MODELS",
            "meta/llama-3.1-8b-instruct",
        )
        fallbacks = [m.strip() for m in fallback_raw.split(",") if m.strip() and m.strip() != primary]
        models_to_try = [primary] + fallbacks[:3]
        return _try_openai_compatible(
            "nvidia", "https://integrate.api.nvidia.com/v1",
            api_key, models_to_try, daily_on_429=False,
        )

    # ── Provider name → function map ─────────────────────────────────────────
    _PROVIDER_FNS = {
        "openrouter":  _try_openrouter,
        "groq":        _try_groq,
        "cerebras":    _try_cerebras,
        "huggingface": _try_huggingface,
        "gemini":      _try_gemini,
        "sambanova":   _try_sambanova,
        "nvidia":      _try_nvidia,
    }

    # ── Ollama (own server, no rate limits — handled outside cloud semaphore) ─
    def _try_ollama_fn():
        try:
            from ollama_client import ollama_chat, get_ollama_model, check_ollama_health
            _ep = os.environ.get("OLLAMA_ENDPOINT", "").strip()
            if not _ep:
                return None
            global _OLLAMA_HEALTH_LAST_CHECK, _OLLAMA_HEALTH_RESULT, _OLLAMA_INFER_BACKOFF_UNTIL
            _now_h = time.time()
            # Skip entirely if a recent inference timed out — don't burn another 45s
            with _LLM_LOCK:
                if _now_h < _OLLAMA_INFER_BACKOFF_UNTIL:
                    logger.debug("Ollama skipped — inference backoff active for %.0fs", _OLLAMA_INFER_BACKOFF_UNTIL - _now_h)
                    return None
                # A cached False expires fast (NEG_TTL) so a waking Space is re-probed; a cached
                # True is trusted for the full TTL to avoid probing a healthy Space repeatedly.
                _ttl = _OLLAMA_HEALTH_TTL if _OLLAMA_HEALTH_RESULT else _OLLAMA_HEALTH_NEG_TTL
                _cv = (_now_h - _OLLAMA_HEALTH_LAST_CHECK) < _ttl
                _ch = _OLLAMA_HEALTH_RESULT if _cv else None
            _fresh_check = _ch is None
            if _ch is None:
                _ch = check_ollama_health(_ep, timeout=35)  # tolerate HF Space cold-start (was 8s, then 20s; 2026-07-31: measured real cold-start ~25.8s, 20s was failing this check almost every time)
                with _LLM_LOCK:
                    _OLLAMA_HEALTH_LAST_CHECK = time.time()
                    _OLLAMA_HEALTH_RESULT = _ch
            if not _ch:
                return None
            _m = get_ollama_model(_ep)
            # Fresh health check means Space just woke from sleep — warmup before real call
            # so the model is loaded and the real inference doesn't hang.
            if _fresh_check:
                from ollama_client import warmup_ollama
                _warm = warmup_ollama(_ep, model=_m, timeout=30)
                if not _warm:
                    with _LLM_LOCK:
                        _OLLAMA_INFER_BACKOFF_UNTIL = time.time() + _OLLAMA_WARMUP_BACKOFF_SECS
                        _OLLAMA_HEALTH_RESULT = False
                    logger.warning("Ollama warmup failed — model still loading, backing off %ds", _OLLAMA_WARMUP_BACKOFF_SECS)
                    return None
                logger.info("Ollama warmup succeeded — model warm, proceeding with inference")
            with _OLLAMA_SEMAPHORE:
                _r = ollama_chat(messages, endpoint=_ep, model=_m, timeout=_OLLAMA_CHAT_TIMEOUT)
            if _r:
                logger.info("LLM: Ollama succeeded")
                return _r, "ollama", _m
            # Inference returned None (timeout or empty) — set backoff so we don't retry immediately
            with _LLM_LOCK:
                _OLLAMA_INFER_BACKOFF_UNTIL = time.time() + _OLLAMA_INFER_BACKOFF_SECS
                _OLLAMA_HEALTH_RESULT = False  # also invalidate health so fast path re-checks later
            logger.warning("Ollama inference failed — backing off for %ds", _OLLAMA_INFER_BACKOFF_SECS)
        except Exception as _e:
            logger.debug("Ollama call failed: %s", _e)
        return None

    # ── Dynamic dispatch ──────────────────────────────────────────────────────
    # Order is re-evaluated on every call:
    #   - Available cloud providers (not exhausted, cooldown expired) come first
    #   - When ALL cloud providers are daily-exhausted, Ollama is promoted to Phase 1
    #   - Otherwise Ollama is the LAST resort — tried only after every cloud provider
    #     (each exhausting its own model list) has failed
    #   - fast_fail runs Ollama once OUTSIDE this pass (see below), so a retry storm doesn't
    #     re-probe the slow Space every pass
    #
    # task_offset rotation (2026-07-17 fix): previously `best = cloud_order[0]` picked the SAME
    # provider for every concurrent call regardless of task_offset — so a whole batch (e.g. 54
    # backtest predictions) all tried the identical "best" provider first, exhausted its rate
    # limit together, then ALL cascaded to the next provider together, repeating the storm down
    # the chain. task_offset (assigned round-robin per stock/task in ai_forecast.py) now rotates
    # the starting pick among the providers that are CURRENTLY AVAILABLE — spreading concurrent
    # calls across providers instead of piling them onto one, while still respecting the
    # cooldown/exhaustion sort (an unavailable provider is never promoted ahead of an available one).
    def _one_pass():
        cloud_order = _get_cloud_order()
        available = [p for p in cloud_order if _is_provider_available(p, fast_fail_on_rate_limit)]
        if preferred_provider and preferred_provider in cloud_order and _is_provider_available(preferred_provider, fast_fail_on_rate_limit):
            ordered = [preferred_provider] + [p for p in cloud_order if p != preferred_provider]
        elif len(available) > 1:
            start = available[task_offset % len(available)]
            ordered = [start] + [p for p in cloud_order if p != start]
        else:
            ordered = cloud_order
        best = ordered[0]
        rest = ordered[1:]
        all_cloud_exhausted = _all_cloud_daily_exhausted()

        # Phase 1: best available cloud OR Ollama if all cloud are daily-exhausted
        if all_cloud_exhausted:
            logger.info("LLM: All cloud providers daily-exhausted — trying Ollama first")
            r = _try_ollama_fn()
            if r is not None:
                return r
        else:
            _LLM_SEMAPHORE.acquire()
            try:
                r = _PROVIDER_FNS[best]()
                if r is not None:
                    return r
            finally:
                _LLM_SEMAPHORE.release()

        # Phase 2: Remaining cloud providers in dynamic order.
        # Try EVERY remaining fast cloud provider (each exhausting its own model list) BEFORE
        # falling to slow local Ollama — the providers have independent quotas, so a rate-limited
        # "best" provider says nothing about the others. (Previously Ollama was tried here, in the
        # middle, which funnelled to a ~34s local call after only the first provider failed while
        # 5 healthy fast providers went untried.)
        if rest:
            _LLM_SEMAPHORE.acquire()
            try:
                for name in rest:
                    r = _PROVIDER_FNS[name]()
                    if r is not None:
                        return r
            finally:
                _LLM_SEMAPHORE.release()

        # Phase 3: Ollama last resort — only when ALL cloud is genuinely down.
        # Skipped under fast_fail (that path runs Ollama once as a last-resort OUTSIDE _one_pass,
        # so a retry storm doesn't re-probe the slow Space every pass). Skipped when all cloud was
        # daily-exhausted because Phase 1 already tried Ollama first.
        if not fast_fail_on_rate_limit and not all_cloud_exhausted:
            r = _try_ollama_fn()
            if r is not None:
                return r

        return None

    result = _one_pass()
    if result is not None:
        return result

    if fast_fail_on_rate_limit:
        # Cloud providers all failed. Ollama has no rate limits — always try it last.
        # Use the shared health cache so a 150-stock scan doesn't burn _OLLAMA_CHAT_TIMEOUT per stock.
        global _OLLAMA_HEALTH_LAST_CHECK, _OLLAMA_HEALTH_RESULT, _OLLAMA_INFER_BACKOFF_UNTIL
        _ep_check = os.environ.get("OLLAMA_ENDPOINT", "").strip()
        if _ep_check:
            from ollama_client import ollama_chat, get_ollama_model, check_ollama_health
            with _LLM_LOCK:
                _now_lr = time.time()
                # Skip if a recent inference timed out
                if _now_lr < _OLLAMA_INFER_BACKOFF_UNTIL:
                    logger.debug("Ollama last-resort skipped — backoff active for %.0fs", _OLLAMA_INFER_BACKOFF_UNTIL - _now_lr)
                else:
                    # Negative results expire fast (NEG_TTL) so a fast-path 20s probe that missed a
                    # cold-starting Space does NOT block this longer cold-start probe.
                    _ttl = _OLLAMA_HEALTH_TTL if _OLLAMA_HEALTH_RESULT else _OLLAMA_HEALTH_NEG_TTL
                    _cached_valid = (_now_lr - _OLLAMA_HEALTH_LAST_CHECK) < _ttl
                    _ch = _OLLAMA_HEALTH_RESULT if _cached_valid else None
                    _now_lr = None  # signal: proceed
            if _now_lr is None:  # not in backoff
                _fresh_lr = _ch is None
                if _ch is None:
                    # No recent result — do the longer probe (HF Space may be cold-starting).
                    # 2026-07-31: raised 25s->35s — a direct curl measurement of the live Space
                    # showed a real cold-start response time of ~25.8s, i.e. the old 25s timeout
                    # was failing this health check almost every time by a hair, right before the
                    # Space finished waking up — so Ollama was silently never reached as a fallback
                    # (every rate-limited batch fell through to "all providers unavailable" instead
                    # of the no-rate-limit Ollama fallback this path exists for).
                    _ch = check_ollama_health(_ep_check, timeout=35)
                    with _LLM_LOCK:
                        _OLLAMA_HEALTH_LAST_CHECK = time.time()
                        _OLLAMA_HEALTH_RESULT = _ch
                if _ch:
                    _m = get_ollama_model(_ep_check)
                    # Warmup on fresh health check — prevents 45s hang on cold-start model loading
                    if _fresh_lr:
                        from ollama_client import warmup_ollama
                        if not warmup_ollama(_ep_check, model=_m, timeout=30):
                            with _LLM_LOCK:
                                _OLLAMA_INFER_BACKOFF_UNTIL = time.time() + _OLLAMA_WARMUP_BACKOFF_SECS
                                _OLLAMA_HEALTH_RESULT = False
                            logger.warning("Ollama last-resort warmup failed — backing off %ds", _OLLAMA_WARMUP_BACKOFF_SECS)
                            raise RuntimeError("All LLM providers unavailable — Ollama cold-start backoff")
                        logger.info("Ollama last-resort warmup succeeded")
                    with _OLLAMA_SEMAPHORE:
                        _r = ollama_chat(messages, endpoint=_ep_check, model=_m, timeout=_OLLAMA_CHAT_TIMEOUT)
                    if _r:
                        logger.info("LLM: Ollama (last-resort fast_fail path) succeeded")
                        return _r, "ollama", _m
                    # Inference failed — backoff so next call doesn't wait again
                    with _LLM_LOCK:
                        _OLLAMA_INFER_BACKOFF_UNTIL = time.time() + _OLLAMA_INFER_BACKOFF_SECS
                        _OLLAMA_HEALTH_RESULT = False
                    logger.warning("Ollama last-resort inference failed — backing off %ds", _OLLAMA_INFER_BACKOFF_SECS)
                else:
                    logger.debug("LLM: Ollama last-resort skipped (cached unhealthy)")
        raise RuntimeError("All LLM providers unavailable — all rate-limited or unconfigured")

    # ── Retry loop ────────────────────────────────────────────────────────────
    if max_retries <= 0:
        max_retries = int(os.getenv("AI_MAX_RETRIES", "3"))

    for _retry_attempt in range(max_retries - 1):
        _wait = float(_ALL_PROVIDERS_RETRY_WAIT_SECS)
        logger.warning(
            "All LLM providers unavailable (attempt %d/%d) — waiting %.1fs before retry",
            _retry_attempt + 2, max_retries, _wait,
        )
        time.sleep(_wait)
        result = _one_pass()
        if result is not None:
            return result

    raise RuntimeError(
        f"All LLM providers unavailable after {max_retries} attempts — all rate-limited or unconfigured"
    )


if __name__ == "__main__":
    print("Testing LLM provider chain...")
    order = _get_cloud_order()
    print(f"Current provider order: {order}")
    for name in _CLOUD_PROVIDERS:
        s = _PROVIDER_STATUS[name]
        print(f"  {name}: daily_exhausted={s['daily_exhausted']}, avail_at={s['avail_at']:.0f}, fail_streak={s['fail_streak']}")
    try:
        content, provider, model = make_chat_call(
            [{"role": "user", "content": "Say 'OK' and nothing else."}],
            max_tokens=5,
            temperature=0.0,
            fast_fail_on_rate_limit=True,
            max_retries=1,
        )
        print(f"Provider: {provider}  Model: {model}")
        print(f"Response: {content}")
    except RuntimeError as e:
        print(f"No provider available: {e}")