StatistaAmeri/backend/live_stream.py

import math, random

def _generate_sparkline(current_price: float, pct_change: float = 0.0, n_points: int = 40) -> list:
    if not current_price or current_price <= 0:
        return [0.0] * n_points
    pct = (pct_change or 0.0) / 100.0
    start_price = current_price / (1.0 + pct) if (1.0 + pct) > 0 else current_price
    points = []
    trend_step = (current_price - start_price) / max(1, n_points - 1)
    volatility = max(abs(pct) * 0.4, 0.007) * current_price
    rand_seed = int(abs(current_price) * 1000) % 100000
    r = random.Random(rand_seed)
    for i in range(n_points - 1):
        progress = i / float(n_points - 1)
        base = start_price + trend_step * i
        wave = math.sin(progress * math.pi * 3.5) * (volatility * 0.6) + math.cos(progress * math.pi * 5.2) * (volatility * 0.4)
        noise = (r.random() - 0.5) * volatility * 0.45
        val = base + wave + noise
        points.append(round(max(val, current_price * 0.2), 2 if current_price > 100 else 4))
    points.append(round(current_price, 2 if current_price > 100 else 4))
    return points

"""
Polls Yahoo Finance every 3 seconds for live commodity prices.
Broadcasts to all connected WebSocket clients.
"""
import asyncio
import json
import yfinance as yf
from fastapi import WebSocket

SYMBOLS = {
    "CL=F":  "Crude Oil WTI",
    "BZ=F":  "Brent Crude",
    "NG=F":  "Natural Gas",
    "RB=F":  "Gasoline",
    "HO=F":  "Heating Oil",
    "GC=F":  "Gold",
    "SI=F":  "Silver",
    "HG=F":  "Copper",
    "PL=F":  "Platinum",
    "PA=F":  "Palladium",
    "ALI=F": "Aluminum",
    "NI=F":  "Nickel",
    "ZC=F":  "Corn",
    "ZW=F":  "Wheat",
    "ZS=F":  "Soybeans",
    "KC=F":  "Coffee",
    "CT=F":  "Cotton",
    "CC=F":  "Cocoa",
    "SB=F":  "Sugar",
    "OJ=F":  "Orange Juice",
}

_clients: set[WebSocket] = set()
_latest: list[dict] = []
_MAX_CLIENTS = 500  # ponytail: global cap — the board is identical for everyone, so no per-IP fairness needed
_SYM_LIST = " ".join(SYMBOLS.keys())

def _fetch_sync() -> list[dict]:
    try:
        # Single batch HTTP request — much faster than one-by-one
        raw = yf.download(
            _SYM_LIST,
            period="2d",
            progress=False,
            group_by="ticker",
            auto_adjust=True,
            threads=True,
        )
        out = []
        for sym, name in SYMBOLS.items():
            try:
                if sym in raw.columns.get_level_values(0):
                    df = raw[sym].dropna()
                else:
                    df = raw.dropna()
                if df.empty:
                    continue
                price = float(df["Close"].iloc[-1])
                prev  = float(df["Close"].iloc[-2]) if len(df) > 1 else price
                change = price - prev
                pct    = (change / prev * 100) if prev else 0
                out.append({
                    "symbol": sym,
                    "name":   name,
                    "price":  round(price, 4),
                    "change": round(change, 4),
                    "pct":    round(pct, 3),
                    "currency": "USD",
                    "up":     change >= 0,
                    "sparkline": _generate_sparkline(price, pct),
                })
            except Exception:
                pass
        return out
    except Exception as e:
        print(f"[live] batch fetch error: {e}")
        # fallback: individual fetch
        out = []
        tickers = yf.Tickers(_SYM_LIST)
        for sym, name in SYMBOLS.items():
            try:
                fi = tickers.tickers[sym].fast_info
                price = fi.last_price or 0
                prev  = fi.previous_close or price
                change = price - prev
                pct    = (change / prev * 100) if prev else 0
                out.append({"symbol": sym, "name": name, "price": round(price,4),
                             "change": round(change,4), "pct": round(pct,3),
                             "currency": "USD", "up": change >= 0,
                             "sparkline": _generate_sparkline(price, pct)})
            except Exception:
                pass
        return out

async def fetch_prices() -> list[dict]:
    return await asyncio.to_thread(_fetch_sync)

def get_latest() -> list[dict]:
    """Latest commodity snapshot — served over plain HTTP (tunnel/proxy friendly)."""
    return _latest

async def broadcast_loop():
    global _latest, _clients  # _clients is reassigned below (-= dead), so declare it global
    while True:
        try:
            data = await fetch_prices()
            if data:
                _latest = data
                msg = json.dumps(data)
                dead = set()
                for ws in _clients.copy():
                    try:
                        await ws.send_text(msg)
                    except Exception:
                        dead.add(ws)
                _clients -= dead
        except Exception as e:
            print(f"[live] error: {e}")
        await asyncio.sleep(3)

async def connect(ws: WebSocket):
    if len(_clients) >= _MAX_CLIENTS:
        await ws.close(code=1013)  # 1013 = "try again later"
        return
    await ws.accept()
    _clients.add(ws)
    if _latest:
        await ws.send_text(json.dumps(_latest))
    try:
        while True:
            await ws.receive_text()
    except Exception:
        pass
    finally:
        _clients.discard(ws)