Initial GPU-accelerated FinStream API

Dockerfile

@@ -0,0 +1,20 @@
+FROM pytorch/pytorch:2.2.0-cuda12.1-cudnn8-runtime
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    TRANSFORMERS_CACHE=/cache
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    gcc \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 7860
+
+CMD ["sh", "-c", "uvicorn main:app --host 0.0.0.0 --port ${PORT:-7860}"]

README.md

@@ -1,10 +1,24 @@
 ---
-title: FinStream API
-emoji: 🌍
-colorFrom: blue
-colorTo: red
+title: FinStream Sentiment API
+emoji: 📈
+colorFrom: green
+colorTo: blue
 sdk: docker
-pinned: false
+app_port: 7860
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# FinStream Sentiment API
+
+GPU-accelerated FastAPI backend for FinStream financial sentiment analysis.
+
+## API Endpoints
+
+- `GET /health` - Service and model status
+- `POST /predict` - Single text sentiment analysis
+- `POST /analyze-csv` - Batch CSV analysis
+
+## Model
+
+- **Model**: hitenvk22/FinStream-Sentiment
+- **Architecture**: distilroberta-base
+- **Task**: 3-class financial sentiment (bullish, neutral, bearish)

app/api/routes.py

@@ -0,0 +1,177 @@
+from io import BytesIO
+import logging
+import os
+import re
+from uuid import uuid4
+
+import pandas as pd
+from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile, status
+from fastapi.responses import FileResponse
+from fastapi.concurrency import run_in_threadpool
+
+from ..schemas import (
+    BatchAnalysisResponse,
+    BatchAnalysisSummary,
+    HealthResponse,
+    PredictRequest,
+    PredictResponse,
+)
+
+
+router = APIRouter()
+logger = logging.getLogger("finstream.api")
+
+TEXT_COLUMN_HINTS = (
+    "text", "message", "sentence", "content", "news",
+    "headline", "comment", "description", "article", "body", "post",
+)
+
+REPORTS_DIR = "/tmp/reports"
+os.makedirs(REPORTS_DIR, exist_ok=True)
+
+
+def _get_model_manager(request: Request):
+    return request.app.state.model_manager
+
+
+def _normalize_column_name(column_name: str) -> str:
+    return re.sub(r"[^a-z0-9]+", "", column_name.lower())
+
+
+def _detect_text_column(frame: pd.DataFrame) -> str:
+    if frame.empty:
+        raise ValueError("CSV file is empty")
+    normalized_columns = {col: _normalize_column_name(str(col)) for col in frame.columns}
+    for column, normalized in normalized_columns.items():
+        if normalized in TEXT_COLUMN_HINTS or any(hint in normalized for hint in TEXT_COLUMN_HINTS):
+            return column
+    object_columns = frame.select_dtypes(include=["object", "string"]).columns.tolist()
+    if object_columns:
+        scored_columns = []
+        for column in object_columns:
+            series = frame[column].dropna().astype(str).str.strip()
+            if series.empty:
+                continue
+            average_length = series.str.len().mean()
+            non_empty_ratio = (series != "").mean()
+            scored_columns.append((float(average_length * non_empty_ratio), column))
+        if scored_columns:
+            scored_columns.sort(reverse=True)
+            return scored_columns[0][1]
+        return object_columns[0]
+    return frame.columns[0]
+
+
+@router.get("/health", response_model=HealthResponse)
+async def health_check(request: Request) -> HealthResponse:
+    mm = _get_model_manager(request)
+    return HealthResponse(
+        status="ok" if mm.is_ready else "degraded",
+        model_loaded=mm.is_ready,
+        device=mm.device,
+        model_name=mm.model_name,
+    )
+
+
+@router.post("/predict", response_model=PredictResponse)
+async def predict(payload: PredictRequest, request: Request) -> PredictResponse:
+    mm = _get_model_manager(request)
+    if not mm.is_ready:
+        raise HTTPException(status_code=503, detail="Model is not ready")
+    try:
+        result = await run_in_threadpool(mm.predict, payload.text)
+        return PredictResponse(**result)
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logger.exception("Prediction failed")
+        raise HTTPException(status_code=500, detail="Prediction failed") from exc
+
+
+@router.post("/analyze-csv", response_model=BatchAnalysisResponse)
+async def analyze_csv(
+    request: Request,
+    file: UploadFile = File(...),
+    report_id: str | None = Form(default=None),
+) -> BatchAnalysisResponse:
+    mm = _get_model_manager(request)
+    if not mm.is_ready:
+        raise HTTPException(status_code=503, detail="Model is not ready")
+    if not file.filename.lower().endswith(".csv"):
+        raise HTTPException(status_code=400, detail="Please upload a CSV file")
+
+    try:
+        raw_bytes = await file.read()
+        if not raw_bytes:
+            raise ValueError("Uploaded CSV file is empty")
+
+        frame = pd.read_csv(BytesIO(raw_bytes))
+        detected_text_column = _detect_text_column(frame)
+        working_frame = frame.copy()
+        working_frame[detected_text_column] = (
+            working_frame[detected_text_column].fillna("").astype(str).str.strip()
+        )
+        working_frame = working_frame[working_frame[detected_text_column] != ""]
+
+        if working_frame.empty:
+            raise ValueError("No non-empty text rows were found in the CSV")
+
+        texts = working_frame[detected_text_column].tolist()
+        predictions = mm.predict_batch(texts)
+
+        rows = []
+        for idx, (text, pred) in enumerate(zip(texts, predictions), start=1):
+            label = str(pred.get("label", "unknown")).lower()
+            if label == "positive":
+                label = "bullish"
+            elif label == "negative":
+                label = "bearish"
+            rows.append({
+                "row_number": idx,
+                "message": text,
+                "predicted_label": label,
+                "confidence": float(pred.get("confidence", 0.0)),
+            })
+
+        pred_frame = pd.DataFrame(rows)
+        counts = pred_frame["predicted_label"].value_counts().to_dict()
+        total = len(pred_frame)
+        bullish_c = counts.get("bullish", 0)
+        neutral_c = counts.get("neutral", 0)
+        bearish_c = counts.get("bearish", 0)
+        unknown_c = counts.get("unknown", 0)
+
+        rid = report_id.strip() if report_id and report_id.strip() else f"FSR-{uuid4().hex[:10].upper()}"
+        net_sent = round(((bullish_c - bearish_c) / total), 4) if total else 0.0
+        avg_conf = round(float(pred_frame["confidence"].mean()), 4)
+
+        net_label = "positive" if net_sent > 0.12 else ("negative" if net_sent < -0.12 else "mixed")
+
+        summary = BatchAnalysisSummary(
+            report_id=rid,
+            detected_text_column=detected_text_column,
+            total_rows=total,
+            analyzed_rows=total,
+            bullish_count=bullish_c,
+            neutral_count=neutral_c,
+            bearish_count=bearish_c,
+            unknown_count=unknown_c,
+            bullish_pct=round((bullish_c / total) * 100, 2) if total else 0.0,
+            neutral_pct=round((neutral_c / total) * 100, 2) if total else 0.0,
+            bearish_pct=round((bearish_c / total) * 100, 2) if total else 0.0,
+            unknown_pct=round((unknown_c / total) * 100, 2) if total else 0.0,
+            net_sentiment=net_sent,
+            net_sentiment_label=net_label,
+            average_confidence=avg_conf,
+            report_pdf_url=f"/reports/{rid}.pdf",
+        )
+
+        return BatchAnalysisResponse(
+            summary=summary,
+            predictions=pred_frame.reset_index(drop=True).to_dict(orient="records"),
+        )
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logger.exception("CSV analysis failed")
+        raise HTTPException(status_code=500, detail=f"CSV analysis failed: {exc}") from exc

app/schemas.py

@@ -0,0 +1,48 @@
+from pydantic import BaseModel
+
+
+class PredictRequest(BaseModel):
+    text: str
+
+
+class PredictResponse(BaseModel):
+    label: str
+    confidence: float
+
+
+class HealthResponse(BaseModel):
+    status: str
+    model_loaded: bool
+    device: str
+    model_name: str
+
+
+class BatchPredictionItem(BaseModel):
+    row_number: int
+    message: str
+    predicted_label: str
+    confidence: float
+
+
+class BatchAnalysisSummary(BaseModel):
+    report_id: str
+    detected_text_column: str
+    total_rows: int
+    analyzed_rows: int
+    bullish_count: int
+    neutral_count: int
+    bearish_count: int
+    unknown_count: int
+    bullish_pct: float
+    neutral_pct: float
+    bearish_pct: float
+    unknown_pct: float
+    net_sentiment: float
+    net_sentiment_label: str
+    average_confidence: float
+    report_pdf_url: str
+
+
+class BatchAnalysisResponse(BaseModel):
+    summary: BatchAnalysisSummary
+    predictions: list[dict]

app/services/model_service.py

@@ -0,0 +1,210 @@
+from __future__ import annotations
+
+import logging
+import re
+from threading import Lock
+
+import torch
+from anyio import to_thread
+from transformers import AutoModelForSequenceClassification, AutoTokenizer, pipeline
+
+
+logger = logging.getLogger("finstream.model")
+
+POSITIVE_WORDS = {
+    "beat", "beats", "bullish", "climb", "climbs", "climbed",
+    "gain", "gains", "gained", "growth", "higher",
+    "improve", "improves", "improved", "improvement", "improvements",
+    "outperform", "outperforms", "outperformed",
+    "profit", "profits", "profitable", "profitability",
+    "rally", "rallies", "rallied",
+    "rise", "rises", "rose", "risen",
+    "surge", "surges", "surged",
+    "strong", "stronger", "strongly",
+    "up", "uptick", "upside", "positive", "record",
+    "boost", "boosts", "boosted",
+    "upgrade", "upgrades", "upgraded",
+    "exceed", "exceeds", "exceeded",
+    "expand", "expands", "expanded", "expansion",
+    "accelerate", "accelerates", "accelerated",
+    "recover", "recovers", "recovered", "recovery",
+    "rebound", "rebounds", "rebounded",
+    "jump", "jumps", "jumped", "soar", "soars", "soared",
+    "dividend", "dividends", "buyback", "buybacks",
+    "upward", "uptrend", "bull", "upswing", "breakout",
+    "optimistic", "optimism", "momentum",
+    "upbeat", "win", "wins", "won", "success", "successful",
+}
+
+NEGATIVE_WORDS = {
+    "bearish", "decline", "declines", "declined",
+    "drop", "drops", "dropped",
+    "fall", "falls", "fell", "fallen",
+    "loss", "losses", "lost",
+    "miss", "misses", "missed",
+    "pressure", "pressures", "pressured",
+    "risk", "risks", "risky",
+    "selloff", "selloffs",
+    "slump", "slumps", "slumped",
+    "soft", "softer", "softness",
+    "weak", "weaker", "weakness", "weaknesses", "weaken", "weakens", "weakened",
+    "down", "downturn", "downturns", "downside", "downgrade",
+    "negative",
+    "cut", "cuts", "cutting",
+    "lower", "lowers", "lowered",
+    "reduce", "reduces", "reduced", "reduction",
+    "layoff", "layoffs", "bankrupt", "bankruptcy", "debt",
+    "default", "defaults",
+    "delay", "delays", "delayed",
+    "suspend", "suspends", "suspended", "suspension",
+    "worst", "worse", "worsen", "worsens", "worsened",
+    "volatile", "volatility",
+    "uncertainty", "uncertain",
+    "plunge", "plunges", "plunged",
+    "tumble", "tumbles", "tumbled",
+    "slide", "slides", "slid",
+    "crash", "crashes", "crashed",
+    "recession", "inflation", "inflationary",
+    "underperform", "underperforms", "underperformed",
+}
+
+
+def _normalize_label(raw_label: str) -> str:
+    normalized = raw_label.strip().lower()
+    if normalized in {"positive", "bullish", "label_1", "1", "pos"}:
+        return "bullish"
+    if normalized in {"negative", "bearish", "label_0", "0", "neg"}:
+        return "bearish"
+    if normalized in {"neutral", "label_2", "2"}:
+        return "neutral"
+    if "pos" in normalized:
+        return "bullish"
+    if "neg" in normalized:
+        return "bearish"
+    return normalized
+
+
+class SentimentModelManager:
+    def __init__(self, model_name: str) -> None:
+        self.model_name = model_name
+        self.device = "cpu"
+        self._device_index = -1
+        self._pipeline = None
+        self._load_error: str | None = None
+
+    @property
+    def is_ready(self) -> bool:
+        return self._pipeline is not None and self._load_error is None
+
+    @property
+    def load_error(self) -> str | None:
+        return self._load_error
+
+    async def load_async(self) -> None:
+        await to_thread.run_sync(self.load)
+
+    def load(self) -> None:
+        if self._pipeline is not None:
+            return
+
+        try:
+            self.device = "cuda" if torch.cuda.is_available() else "cpu"
+            self._device_index = 0 if torch.cuda.is_available() else -1
+            logger.info("Loading model %s on %s", self.model_name, self.device)
+
+            tokenizer = AutoTokenizer.from_pretrained(self.model_name)
+            model = AutoModelForSequenceClassification.from_pretrained(
+                self.model_name,
+                torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
+                low_cpu_mem_usage=True,
+            )
+            model.eval()
+
+            self._pipeline = pipeline(
+                task="sentiment-analysis",
+                model=model,
+                tokenizer=tokenizer,
+                device=self._device_index,
+                truncation=True,
+                framework="pt",
+            )
+            self._load_error = None
+            logger.info("Model loaded successfully on %s", self.device)
+        except Exception as exc:
+            self._load_error = str(exc)
+            logger.exception("Failed to load sentiment model")
+
+    @staticmethod
+    def _stem(token: str) -> str:
+        if len(token) <= 4:
+            return token
+        for suffix in ["ability", "abilities", "ification", "ifications",
+                        "ization", "izations", "isation", "isations",
+                        "ationally", "isation", "ization",
+                        "iveness", "fulness", "iousness",
+                        "ments", "ment", "ances", "ance",
+                        "eness", "ness", "ship",
+                        "able", "ably", "ible",
+                        "ally", "wise", "like",
+                        "ious", "eous", "uous",
+                        "sion", "tion", "sions", "tions",
+                        "ised", "ized", "ising", "izing",
+                        "ative", "itive", "tive",
+                        "less", "proof", "ward",
+                        "ing", "ings",
+                        "ed", "es", "er", "est", "ly"]:
+            if token.endswith(suffix) and len(token) - len(suffix) >= 3:
+                return token[:-len(suffix)]
+        return token
+
+    def _rule_based_predict(self, text: str) -> dict[str, float | str]:
+        tokens = re.findall(r"[a-zA-Z']+", text.lower())
+        if not tokens:
+            return {"label": "neutral", "confidence": 0.5}
+        stemmed_tokens = [self._stem(t) for t in tokens]
+        positive_hits = sum(
+            1 for i, t in enumerate(tokens)
+            if t in POSITIVE_WORDS or stemmed_tokens[i] in POSITIVE_WORDS
+        )
+        negative_hits = sum(
+            1 for i, t in enumerate(tokens)
+            if t in NEGATIVE_WORDS or stemmed_tokens[i] in NEGATIVE_WORDS
+        )
+        total_hits = positive_hits + negative_hits
+        score = positive_hits - negative_hits
+        if total_hits == 0:
+            return {"label": "neutral", "confidence": 0.5}
+        confidence = min(0.95, max(0.55, 0.55 + (abs(score) / total_hits) * 0.35))
+        if score > 0:
+            return {"label": "bullish", "confidence": round(confidence, 4)}
+        if score < 0:
+            return {"label": "bearish", "confidence": round(confidence, 4)}
+        return {"label": "neutral", "confidence": round(0.5 + (positive_hits / total_hits) * 0.1, 4)}
+
+    def predict(self, text: str) -> dict[str, float | str]:
+        if self._pipeline is None:
+            return self._rule_based_predict(text)
+        with torch.no_grad():
+            output = self._pipeline(text)
+        prediction = output[0] if isinstance(output, list) else output
+        return {
+            "label": _normalize_label(prediction.get("label", "unknown")),
+            "confidence": float(prediction.get("score", 0.0)),
+        }
+
+    def predict_batch(self, texts: list[str]) -> list[dict[str, float | str]]:
+        if self._pipeline is None:
+            return [self._rule_based_predict(text) for text in texts]
+        with torch.no_grad():
+            output = self._pipeline(texts)
+        if isinstance(output, dict):
+            output = [output]
+        results = []
+        for prediction in output:
+            if isinstance(prediction, list):
+                prediction = prediction[0]
+            results.append({
+                "label": _normalize_label(prediction.get("label", "unknown")),
+                "confidence": float(prediction.get("score", 0.0)),
+            })
+        return results

main.py

@@ -0,0 +1,85 @@
+from contextlib import asynccontextmanager
+import logging
+import os
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from starlette.requests import Request
+from fastapi.exceptions import RequestValidationError
+
+from app.api.routes import router as api_router
+from app.services.model_service import SentimentModelManager
+
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+)
+logger = logging.getLogger("finstream")
+
+MODEL_NAME = os.getenv("MODEL_NAME", "hitenvk22/FinStream-Sentiment")
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    mm = SentimentModelManager(model_name=MODEL_NAME)
+    app.state.model_manager = mm
+    await mm.load_async()
+    logger.info("Device: %s | Ready: %s", mm.device, mm.is_ready)
+    yield
+
+
+app = FastAPI(
+    title="FinStream Sentiment API",
+    version="1.0.0",
+    description="GPU-accelerated FinStream sentiment inference on Hugging Face Spaces",
+    lifespan=lifespan,
+)
+
+
+@app.get("/")
+async def root():
+    mm = getattr(app.state, "model_manager", None)
+    return {
+        "service": "FinStream Sentiment API",
+        "version": "1.0.0",
+        "mode": "transformers",
+        "status": "running",
+        "model": MODEL_NAME,
+        "device": mm.device if mm else "unknown",
+        "endpoints": {
+            "predict": "/predict",
+            "analyze_csv": "/analyze-csv",
+            "health": "/health",
+        },
+    }
+
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=False,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(api_router)
+
+
+@app.exception_handler(RequestValidationError)
+async def validation_exception_handler(request: Request, exc: RequestValidationError):
+    logger.warning("Validation error on %s: %s", request.url.path, exc.errors())
+    return JSONResponse(
+        status_code=422,
+        content={"detail": "Invalid request payload", "errors": exc.errors()},
+    )
+
+
+@app.exception_handler(Exception)
+async def unhandled_exception_handler(request: Request, exc: Exception):
+    logger.exception("Unhandled error on %s", request.url.path)
+    return JSONResponse(
+        status_code=500,
+        content={"detail": "Internal server error"},
+    )

requirements.txt

@@ -0,0 +1,10 @@
+fastapi>=0.115.0
+uvicorn[standard]>=0.30.0
+transformers>=4.42.0
+torch>=2.2.0
+pydantic-settings>=2.4.0
+pandas>=2.2.2
+numpy>=1.26.4
+python-multipart>=0.0.9
+reportlab>=4.2.2
+anyio>=4.4.0