diff --git a/backend/routers/forecast.py b/backend/routers/forecast.py index f8182e6..44830ad 100644 --- a/backend/routers/forecast.py +++ b/backend/routers/forecast.py @@ -21,9 +21,11 @@ def _load_prophet(name: str): raise FileNotFoundError(f"Model not found: {path}") return joblib.load(path) +@lru_cache(maxsize=1) def _load_daily(): return pd.read_csv(os.path.join(ROOT,"data","daily_kpis.csv"), parse_dates=["ds"]) +@lru_cache(maxsize=1) def _load_ext(): return pd.read_csv(os.path.join(ROOT,"data","external_regs.csv"), parse_dates=["ds"]) diff --git a/backend/routers/pricing.py b/backend/routers/pricing.py index 671ad5f..5ceff79 100644 --- a/backend/routers/pricing.py +++ b/backend/routers/pricing.py @@ -19,6 +19,14 @@ def _load_forecast(): b = joblib.load(os.path.join(ROOT,"models","prophet_occupancy.joblib")) return b["forecast"] +@lru_cache(maxsize=1) +def _load_daily(): + return pd.read_csv(os.path.join(ROOT,"data","daily_kpis.csv"), parse_dates=["ds"]) + +@lru_cache(maxsize=1) +def _load_ext(): + return pd.read_csv(os.path.join(ROOT,"data","external_regs.csv"), parse_dates=["ds"]) + @router.get("/recommendation") def get_pricing( current_adr: float = Query(120.0, description="Current ADR in EUR"), @@ -26,8 +34,8 @@ def get_pricing( ): """Dynamic pricing recommendation based on Prophet demand forecast.""" forecast_df = _load_forecast() - daily = pd.read_csv(os.path.join(ROOT,"data","daily_kpis.csv"), parse_dates=["ds"]) - ext = pd.read_csv(os.path.join(ROOT,"data","external_regs.csv"), parse_dates=["ds"]) + daily = _load_daily() + ext = _load_ext() rec = _engine.recommend(forecast_df, daily, ext, horizon_days=horizon_days, current_adr=current_adr) diff --git a/backend/routers/sentiment.py b/backend/routers/sentiment.py index 8d5b3cc..846c4e4 100644 --- a/backend/routers/sentiment.py +++ b/backend/routers/sentiment.py @@ -1,16 +1,24 @@ """backend/routers/sentiment.py — NLP endpoints with engine info""" import os, sys +from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutTimeout from fastapi import APIRouter from pydantic import BaseModel from typing import List ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, ROOT) -from src.sentiment_engine import analyse, analyse_batch, get_active_engine +from src.sentiment_engine import (analyse, analyse_batch, get_active_engine, + _textblob_analyse) from src.hf_sentiment_engine import MODELS as HF_MODELS router = APIRouter() +# Endpoint-level latency cap. Even with the new in-engine budgets, we never +# want this endpoint to spend more than this before returning *something* — +# Render's proxy starts emitting 502 around 30s of upstream silence. +_REQUEST_DEADLINE_S = 12 +_EXEC = ThreadPoolExecutor(max_workers=8) + class ReviewText(BaseModel): text: str @@ -60,36 +68,45 @@ def engine_info(): }, } +def _textblob_fallback(text: str, reason: str) -> dict: + r = _textblob_analyse(text) + r["engine"] = f"TextBlob (fallback: {reason})" + return r + @router.post("/analyse") def analyse_single(body: ReviewText): """ - Resilient single-review analysis. If the active engine throws (timeout, - cold start, parse error, read-only cache, etc.), we walk down the tier - fallback rather than returning a 500. + Single-review analysis with a hard deadline. If the active engine + (HF / Claude) doesn't return within the request budget, we return a + TextBlob result instead of letting the request hang until Render's + proxy returns 502 Bad Gateway. """ import logging, traceback log = logging.getLogger("sentiment") + fut = _EXEC.submit(analyse, body.text) try: - return analyse(body.text) + return fut.result(timeout=_REQUEST_DEADLINE_S) + except FutTimeout: + # The underlying call keeps running and will populate the in-memory + # cache; the user just doesn't wait for it on this request. + log.warning(f"sentiment analyse exceeded {_REQUEST_DEADLINE_S}s; falling back to TextBlob") + return _textblob_fallback(body.text, f"deadline {_REQUEST_DEADLINE_S}s exceeded") except Exception as e: log.warning(f"sentiment analyse failed: {e}\n{traceback.format_exc()}") - # Last-ditch TextBlob fallback so the dashboard never sees a 500. - try: - from src.sentiment_engine import _textblob_analyse - r = _textblob_analyse(body.text) - r["engine"] = f"TextBlob (fallback after error: {type(e).__name__})" - return r - except Exception: - return { - "label": "Neutral", "polarity": 0.0, "confidence": 0.0, - "sarcasm_flag": False, "aspects": {}, "themes": [], - "engine": f"error: {type(e).__name__}: {e}", - } + return _textblob_fallback(body.text, f"{type(e).__name__}") @router.post("/analyse-batch") def analyse_batch_endpoint(body: BatchReviews): + # Same deadline pattern. Scale budget mildly with batch size but cap it + # so the request can never block the proxy. + deadline = min(_REQUEST_DEADLINE_S + 2 * len(body.reviews), 45) + fut = _EXEC.submit(analyse_batch, body.reviews) try: - results = analyse_batch(body.reviews) + results = fut.result(timeout=deadline) + except FutTimeout: + # Per-item TextBlob so the client never sees a 502 on /analyse-batch. + results = [_textblob_fallback(t, f"batch deadline {deadline}s exceeded") + for t in body.reviews] except Exception as e: return {"count": 0, "results": [], "error": f"{type(e).__name__}: {e}"} return {"count": len(results), "results": results} diff --git a/backend/routers/xai.py b/backend/routers/xai.py index 393334b..f0490bc 100644 --- a/backend/routers/xai.py +++ b/backend/routers/xai.py @@ -15,6 +15,11 @@ def _load_explainer(): return CancellationExplainer(os.path.join(ROOT,"models","cancellation_model.joblib")) +@lru_cache(maxsize=1) +def _load_bookings() -> pd.DataFrame: + # Cached so /global-importance doesn't re-parse 180k rows per request. + return pd.read_csv(os.path.join(ROOT,"data","bookings.csv")) + class BookingForXAI(BaseModel): hotel: str = "Resort Hotel" lead_time: int = 120 @@ -52,7 +57,7 @@ def explain_booking(booking: BookingForXAI): def global_importance(n_samples: int = 300): """Top-20 global feature importances from SHAP.""" exp = _load_explainer() - bk = pd.read_csv(os.path.join(ROOT,"data","bookings.csv")) + bk = _load_bookings() try: result = exp.explain_global(bk[FEATURES], n_samples=n_samples) return { diff --git a/src/hf_sentiment_engine.py b/src/hf_sentiment_engine.py index bc1517d..c73625a 100644 --- a/src/hf_sentiment_engine.py +++ b/src/hf_sentiment_engine.py @@ -34,7 +34,9 @@ from __future__ import annotations -import os, json, re, time, hashlib, logging +import os, json, re, time, hashlib, logging, threading +from concurrent.futures import ThreadPoolExecutor, as_completed +from functools import lru_cache from typing import Optional from pathlib import Path import requests @@ -71,34 +73,39 @@ "LABEL_2": "Positive", "positive": "Positive", } -CACHE_PATH = Path("data/hf_sentiment_cache.json") +# Disk cache removed in favor of in-memory cache (see _MEM_CACHE below). # ───────────────────────────────────────────────────────────────────────────── # Cache helpers # ───────────────────────────────────────────────────────────────────────────── -def _load_cache() -> dict: - if CACHE_PATH.exists(): - try: - with open(CACHE_PATH) as f: - return json.load(f) - except Exception: - return {} - return {} - -def _save_cache(cache: dict): - """Best-effort cache save. Never raises — read-only mounts (e.g. Docker - `./data:/app/data:ro`) and missing dirs just disable caching for the - current request.""" - try: - CACHE_PATH.parent.mkdir(exist_ok=True) - with open(CACHE_PATH, "w") as f: - json.dump(cache, f, indent=2) - except (OSError, PermissionError) as e: - logger.debug(f"HF cache disabled (cannot write {CACHE_PATH}): {e}") +# ─── In-memory cache ───────────────────────────────────────────────────────── +# Was: read+rewrite the full data/hf_sentiment_cache.json on every request. +# That's O(N) I/O per call and burns Render's ephemeral disk. +# Now: bounded process-local dict, thread-safe. Disk persistence removed — +# on free-tier the container restarts often, so a persistent cache is low-value. +_MEM_CACHE: dict[str, dict] = {} +_MEM_CACHE_MAX = 2048 +_MEM_CACHE_LOCK = threading.Lock() + +def _cache_get(key: str) -> Optional[dict]: + with _MEM_CACHE_LOCK: + return _MEM_CACHE.get(key) + +def _cache_put(key: str, value: dict) -> None: + # Don't cache failures; we want the next request to retry. + if value.get("_hf_failed"): + return + with _MEM_CACHE_LOCK: + if len(_MEM_CACHE) >= _MEM_CACHE_MAX: + # Cheap eviction: drop one arbitrary entry. Avoids importing OrderedDict. + _MEM_CACHE.pop(next(iter(_MEM_CACHE))) + _MEM_CACHE[key] = value def _cache_key(text: str, suffix: str = "") -> str: - return hashlib.md5(f"{text.strip().lower()}{suffix}".encode()).hexdigest()[:20] + # Include engine identifier so a degraded result doesn't poison a later + # warm-engine lookup. + return hashlib.md5(f"{text.strip().lower()}|{suffix}".encode()).hexdigest()[:20] # ───────────────────────────────────────────────────────────────────────────── @@ -110,9 +117,12 @@ class HFInferenceClient: Handles: auth, retries, cold-start waits (model loading), rate limits. """ - TIMEOUT = 25 - RETRY_ATTEMPTS = 3 - COLD_START_WAIT= 20 # HF cold-starts large models; wait and retry + # Tight budgets so a slow HF doesn't burn Render's proxy timeout. + # Previous values (TIMEOUT=25, RETRIES=3, WAIT=20) could spend up to + # ~400s × 3 models per request and surface as 502 Bad Gateway. + TIMEOUT = 8 + RETRY_ATTEMPTS = 1 + COLD_START_WAIT = 3 def __init__(self, token: Optional[str] = None): self.token = token or os.environ.get("HF_API_TOKEN", "") @@ -262,50 +272,66 @@ def __init__(self, token: Optional[str] = None): self._engine = f"HuggingFace ({MODELS['sentiment'].split('/')[-1]})" def analyse(self, text: str, use_cache: bool = True) -> dict: - key = _cache_key(text, "hf_v2") - cache = _load_cache() if use_cache else {} - if key in cache: - return cache[key] + key = _cache_key(text, "hf_v3") + if use_cache: + hit = _cache_get(key) + if hit is not None: + return hit - text = str(text).strip() + text = str(text).strip() result = self._run_pipeline(text) if use_cache: - cache[key] = result - _save_cache(cache) + _cache_put(key, result) return result def _run_pipeline(self, text: str) -> dict: - # ── 1. Sentiment ────────────────────────────────────────────────── - raw_sent = self.client.query( - MODELS["sentiment"], - {"inputs": text, "options": {"wait_for_model": True}}, - ) + # Fan out the 3 HF model calls concurrently — they're independent. + # Was: 3 sequential calls × up to 25s each = 75s+ per review, easily + # triggering Render's 502 proxy timeout. Now: bounded by the slowest + # call (~TIMEOUT seconds) instead of their sum. + # + # wait_for_model is OFF so HF returns 503 fast on a cold model + # rather than blocking up to 20s; we retry once briefly via the + # client's RETRY_ATTEMPTS, and if the model is still cold we fail + # over to the next tier (Claude / TextBlob) in sentiment_engine.py. + payloads = { + "sentiment": (MODELS["sentiment"], + {"inputs": text, + "options": {"wait_for_model": False}}), + "irony": (MODELS["irony"], + {"inputs": text, + "options": {"wait_for_model": False}}), + "zero_shot": (MODELS["zero_shot"], + {"inputs": text, + "parameters": {"candidate_labels": ASPECT_LABELS, + "multi_label": True}, + "options": {"wait_for_model": False}}), + } + + raw: dict[str, object] = {} + with ThreadPoolExecutor(max_workers=3) as ex: + futures = {ex.submit(self.client.query, model_id, payload): name + for name, (model_id, payload) in payloads.items()} + for fut in as_completed(futures): + name = futures[fut] + try: + raw[name] = fut.result() + except Exception as e: + logger.warning(f"HF {name} call raised: {e}") + raw[name] = None + + raw_sent = raw.get("sentiment") + raw_irony = raw.get("irony") + raw_zs = raw.get("zero_shot") + + # Sentiment is the only mandatory call. If it failed, signal so the + # caller can fall through to Claude / TextBlob without poisoning cache. if raw_sent is None: - # Model unreachable — return structured failure so caller can fallback return {"_hf_failed": True} label, polarity, confidence = _parse_sentiment(raw_sent) - - # ── 2. Irony / sarcasm ──────────────────────────────────────────── - raw_irony = self.client.query( - MODELS["irony"], - {"inputs": text, "options": {"wait_for_model": True}}, - ) sarcasm_flag = _parse_irony(raw_irony) if raw_irony else False - - # ── 3. Zero-shot aspect scoring ─────────────────────────────────── - raw_zs = self.client.query( - MODELS["zero_shot"], - { - "inputs": text, - "parameters": { - "candidate_labels": ASPECT_LABELS, - "multi_label": True, - }, - "options": {"wait_for_model": True}, - }, - ) aspects = {"room": None, "service": None, "food": None, "value": None, "location": None} @@ -337,14 +363,24 @@ def _run_pipeline(self, text: str) -> dict: } def analyse_batch(self, texts: list[str], - delay: float = 0.5) -> list[dict]: - results = [] - for i, text in enumerate(texts): - r = self.analyse(text) - results.append(r) - if i < len(texts) - 1: - time.sleep(delay) - return results + max_workers: int = 4) -> list[dict]: + # Was: sequential with a 0.5s sleep between calls — 30 reviews = + # 15s of pure sleep before any model work, which exceeded Render's + # proxy timeout. Now: bounded concurrency with no artificial sleep. + # HF's API will 429 if we overshoot; HFInferenceClient retries on 429. + if not texts: + return [] + results: list[Optional[dict]] = [None] * len(texts) + with ThreadPoolExecutor(max_workers=max_workers) as ex: + futs = {ex.submit(self.analyse, t): i for i, t in enumerate(texts)} + for fut in as_completed(futs): + i = futs[fut] + try: + results[i] = fut.result() + except Exception as e: + logger.warning(f"HF batch item {i} failed: {e}") + results[i] = {"_hf_failed": True} + return results # type: ignore[return-value] # ───────────────────────────────────────────────────────────────────────────── diff --git a/src/sentiment_engine.py b/src/sentiment_engine.py index f3738a0..43b9059 100644 --- a/src/sentiment_engine.py +++ b/src/sentiment_engine.py @@ -5,7 +5,7 @@ Tier 3: TextBlob """ from __future__ import annotations -import os, json, re, time, hashlib, logging +import os, json, re, time, hashlib, logging, threading from typing import Optional from pathlib import Path import pandas as pd @@ -29,25 +29,27 @@ except ImportError: _ANTHROPIC_OK = False -CACHE_PATH = Path("data/sentiment_cache.json") - -def _load_cache(): - if CACHE_PATH.exists(): - try: - with open(CACHE_PATH) as f: return json.load(f) - except: pass - return {} - -def _save_cache(cache): - """Best-effort cache save. Never raises — read-only mounts (e.g. Docker - `./data:/app/data:ro`) and missing dirs just disable caching for the - current request.""" - try: - CACHE_PATH.parent.mkdir(exist_ok=True) - with open(CACHE_PATH, "w") as f: - json.dump(cache, f, indent=2) - except (OSError, PermissionError) as e: - logger.debug(f"sentiment cache disabled (cannot write {CACHE_PATH}): {e}") +# ─── In-memory cache ───────────────────────────────────────────────────────── +# Was: load+rewrite data/sentiment_cache.json on every request (O(N) per call, +# burns Render's ephemeral disk). Now: bounded in-process dict, thread-safe. +# We also drop the previous "result cached under one key regardless of which +# engine actually produced it" — a TextBlob fallback no longer poisons later +# warm-engine lookups, because failures are never cached. +_MEM_CACHE: dict[str, dict] = {} +_MEM_CACHE_MAX = 2048 +_MEM_CACHE_LOCK = threading.Lock() + +def _cache_get(key: str) -> Optional[dict]: + with _MEM_CACHE_LOCK: + return _MEM_CACHE.get(key) + +def _cache_put(key: str, value: dict) -> None: + if not value or value.get("engine", "").startswith("error:"): + return + with _MEM_CACHE_LOCK: + if len(_MEM_CACHE) >= _MEM_CACHE_MAX: + _MEM_CACHE.pop(next(iter(_MEM_CACHE))) + _MEM_CACHE[key] = value def _key(text): return hashlib.md5(text.strip().lower().encode()).hexdigest()[:20] @@ -95,14 +97,17 @@ def _get_claude(): def analyse(text: str, use_cache: bool = True) -> dict: key = _key(text) - cache = _load_cache() if use_cache else {} - if key in cache: return cache[key] + if use_cache: + hit = _cache_get(key) + if hit is not None: + return hit + result = None - # Tier 1: HuggingFace + # Tier 1: HuggingFace (its own cache is in-memory now too) hf = _get_hf() if hf: try: - r = hf.analyse(text, use_cache=False) + r = hf.analyse(text, use_cache=True) if not r.get("_hf_failed"): result = r except Exception as e: logger.warning(f"HF failed: {e}") # Tier 2: Claude @@ -111,20 +116,30 @@ def analyse(text: str, use_cache: bool = True) -> dict: if claude: try: result = _claude_analyse(text, claude) except Exception as e: logger.warning(f"Claude failed: {e}") - # Tier 3: TextBlob + # Tier 3: TextBlob (always succeeds) if result is None: result = _textblob_analyse(text) + if use_cache: - cache[key] = result - _save_cache(cache) + _cache_put(key, result) return result -def analyse_batch(texts, delay=0.3): - results = [] - for i, text in enumerate(texts): - r = analyse(text) - results.append(r) - if ("HuggingFace" in r.get("engine","") or "Claude" in r.get("engine","")) and i < len(texts)-1: - time.sleep(delay) +def analyse_batch(texts, max_workers: int = 4): + # Was: sequential with a 0.3s sleep between every call — 30 reviews + # spent ~9s asleep before any work, which alone exceeded Render's + # proxy timeout. Now: bounded concurrency, no artificial sleep. + from concurrent.futures import ThreadPoolExecutor, as_completed + if not texts: + return [] + results: list[Optional[dict]] = [None] * len(texts) + with ThreadPoolExecutor(max_workers=max_workers) as ex: + futs = {ex.submit(analyse, t): i for i, t in enumerate(texts)} + for fut in as_completed(futs): + i = futs[fut] + try: + results[i] = fut.result() + except Exception as e: + logger.warning(f"sentiment batch item {i} failed: {e}") + results[i] = _textblob_analyse(texts[i]) return results def enrich_dataframe(df: pd.DataFrame, text_col: str = "text") -> pd.DataFrame: