"""
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".*?", "", 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".*?", "", 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".*?", "", 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".*?", "", 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".*?", "", 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}")