StatistaAmeri/backend/crypto_market.py

"""
Crypto market data from CoinGecko's official free public API (no key needed).

HTML scraping of coingecko.com is prohibited by their ToS and the page is JS-
rendered behind Cloudflare; the free API returns the same data (prices, 1h/24h/7d
changes, market cap, volume, and a 7-day sparkline array) reliably as JSON.

Keeps a current snapshot in memory (served by /api/crypto) and dedup-persists
changed coin prices into crypto_prices, exactly like the other data modules.
"""
import asyncio
from datetime import datetime, timezone

import requests

from db import get_conn

API_BASE = "https://api.coingecko.com/api/v3"
UA = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
)
HEADERS = {"User-Agent": UA, "Accept": "application/json"}
INTERVAL_SECONDS = 60          # 3 calls/cycle, well under the 100 calls/min free limit
PER_PAGE = 100                 # top 100 coins by market cap
SPARK_POINTS = 40              # downsample the 7d (~168pt) sparkline to keep payload small

_latest: dict = {"coins": [], "global": {}, "trending": [], "fetched_at": None}
_last_saved: dict[str, float] = {}


def create_table() -> None:
    with get_conn() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS crypto_prices (
                id         INTEGER PRIMARY KEY AUTOINCREMENT,
                symbol     TEXT NOT NULL,
                name       TEXT,
                price      REAL,
                pct_24h    REAL,
                market_cap REAL,
                volume_24h REAL,
                fetched_at TEXT NOT NULL
            )
        """)
        conn.execute("CREATE INDEX IF NOT EXISTS idx_crypto_symbol ON crypto_prices(symbol)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_crypto_fetched ON crypto_prices(fetched_at)")


def _downsample(arr: list, n: int = SPARK_POINTS) -> list:
    """Reduce a long price series to ~n points (keeps the last point)."""
    if not arr or len(arr) <= n:
        return [round(float(x), 6) for x in arr] if arr else []
    step = len(arr) / n
    out = [arr[int(i * step)] for i in range(n)]
    out.append(arr[-1])
    return [round(float(x), 6) for x in out]


def _dir(pct) -> str:
    if pct is None:
        return ""
    return "up" if pct >= 0 else "down"


def _fetch_markets() -> list[dict]:
    url = (
        f"{API_BASE}/coins/markets?vs_currency=usd&order=market_cap_desc"
        f"&per_page={PER_PAGE}&page=1&sparkline=true"
        f"&price_change_percentage=1h,24h,7d"
    )
    data = requests.get(url, headers=HEADERS, timeout=25).json()
    if not isinstance(data, list):
        return []
    coins = []
    for c in data:
        pct24 = c.get("price_change_percentage_24h_in_currency")
        coins.append({
            "id": c.get("id"),
            "symbol": (c.get("symbol") or "").upper(),
            "name": c.get("name"),
            "image": c.get("image"),
            "rank": c.get("market_cap_rank"),
            "price": c.get("current_price"),
            "pct_1h": c.get("price_change_percentage_1h_in_currency"),
            "pct_24h": pct24,
            "pct_7d": c.get("price_change_percentage_7d_in_currency"),
            "market_cap": c.get("market_cap"),
            "volume_24h": c.get("total_volume"),
            "high_24h": c.get("high_24h"),
            "low_24h": c.get("low_24h"),
            "circ_supply": c.get("circulating_supply"),
            "ath": c.get("ath"),
            "sparkline": _downsample((c.get("sparkline_in_7d") or {}).get("price") or []),
            "dir": _dir(pct24),
        })
    return coins


def _fetch_global() -> dict:
    data = requests.get(f"{API_BASE}/global", headers=HEADERS, timeout=20).json()
    g = (data or {}).get("data") or {}
    pct = g.get("market_cap_percentage") or {}
    return {
        "market_cap_usd": (g.get("total_market_cap") or {}).get("usd"),
        "volume_usd": (g.get("total_volume") or {}).get("usd"),
        "btc_dominance": pct.get("btc"),
        "eth_dominance": pct.get("eth"),
        "market_cap_change_24h": g.get("market_cap_change_percentage_24h_usd"),
        "active": g.get("active_cryptocurrencies"),
    }


def _fetch_trending() -> list[dict]:
    data = requests.get(f"{API_BASE}/search/trending", headers=HEADERS, timeout=20).json()
    out = []
    for c in (data or {}).get("coins", []):
        item = c.get("item") or {}
        d = item.get("data") or {}
        pct = d.get("price_change_percentage_24h") or {}
        out.append({
            "id": item.get("id"),
            "symbol": (item.get("symbol") or "").upper(),
            "name": item.get("name"),
            "rank": item.get("market_cap_rank"),
            "thumb": item.get("small") or item.get("thumb"),
            "price": d.get("price"),
            "pct_24h": pct.get("usd") if isinstance(pct, dict) else None,
        })
    return out


def _fetch_sync() -> dict:
    """Each section is fetched independently so one failure doesn't lose the rest."""
    out = {"coins": [], "global": {}, "trending": []}
    try:
        out["coins"] = _fetch_markets()
    except Exception as e:
        print(f"[crypto] markets fetch failed: {e}")
    try:
        out["global"] = _fetch_global()
    except Exception as e:
        print(f"[crypto] global fetch failed: {e}")
    try:
        out["trending"] = _fetch_trending()
    except Exception as e:
        print(f"[crypto] trending fetch failed: {e}")
    return out


def _save(coins: list[dict], now: str) -> None:
    """Persist only coins whose price changed since last save (dedup)."""
    changed = [c for c in coins if c["symbol"] and _last_saved.get(c["symbol"]) != c["price"]]
    if not changed:
        return
    with get_conn() as conn:
        conn.executemany(
            """INSERT INTO crypto_prices
               (symbol, name, price, pct_24h, market_cap, volume_24h, fetched_at)
               VALUES(?,?,?,?,?,?,?)""",
            [(c["symbol"], c["name"], c["price"], c["pct_24h"],
              c["market_cap"], c["volume_24h"], now) for c in changed],
        )
    for c in changed:
        _last_saved[c["symbol"]] = c["price"]


def _heartbeat(now: str) -> None:
    with get_conn() as conn:
        conn.execute(
            """INSERT INTO app_config(key, value) VALUES('crypto_last_scrape', ?)
               ON CONFLICT(key) DO UPDATE SET value=excluded.value""",
            (now,),
        )


def _load_last_saved() -> None:
    try:
        with get_conn() as conn:
            rows = conn.execute(
                """SELECT symbol, price FROM crypto_prices
                   WHERE id IN (SELECT MAX(id) FROM crypto_prices GROUP BY symbol)"""
            ).fetchall()
        for r in rows:
            _last_saved[r["symbol"]] = r["price"]
    except Exception as e:
        print(f"[crypto] could not seed dedup from DB: {e}")


async def scrape_loop() -> None:
    global _latest
    create_table()
    _load_last_saved()
    while True:
        try:
            payload = await asyncio.to_thread(_fetch_sync)
            if payload["coins"]:
                now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
                await asyncio.to_thread(_save, payload["coins"], now)
                await asyncio.to_thread(_heartbeat, now)
                _latest = {**payload, "fetched_at": now}
                print(f"[crypto] {len(payload['coins'])} coins at {now}")
        except Exception as e:
            print(f"[crypto] error: {e}")
        await asyncio.sleep(INTERVAL_SECONDS)


def get_latest() -> dict:
    return _latest