providers/base.py

"""Normalized provider primitives for the Short Hunter datasource gateway.

The gateway must be truthful: if a provider cannot supply a capability we return
structured degraded/unavailable state instead of fabricating data.
"""

from __future__ import annotations

import asyncio
import os
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, List, Optional, Tuple

import httpx

SOURCE_LIVE = "LIVE"
SOURCE_DEGRADED = "DEGRADED"
SOURCE_CACHED = "CACHED"
SOURCE_UNAVAILABLE = "UNAVAILABLE"
SOURCE_DISABLED = "DISABLED"

DATA_REAL = "REAL"
DATA_PARTIAL = "PARTIAL"
DATA_CACHED = "CACHED"
DATA_UNAVAILABLE = "UNAVAILABLE"

DEFAULT_HEADERS = {
    "User-Agent": "Mozilla/5.0 (compatible; ShortHunterDatasourceGateway/1.0; +https://huggingface.co/spaces)",
    "Accept": "application/json",
}


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def normalize_spot_symbol(symbol: str) -> str:
    raw = (symbol or "BTCUSDT").upper().strip()
    raw = raw.replace("/", "").replace("-", "").replace("_", "")
    if raw.endswith("USDT"):
        return raw
    if raw.endswith("USD") and not raw.endswith("USDT"):
        return f"{raw[:-3]}USDT"
    return f"{raw}USDT"


def base_asset(symbol: str) -> str:
    s = normalize_spot_symbol(symbol)
    if s.endswith("USDT"):
        return s[:-4]
    if s.endswith("USD"):
        return s[:-3]
    return s


def kucoin_futures_symbol(symbol: str) -> str:
    """Best-effort KuCoin Futures USDT-M contract symbol.

    KuCoin Futures commonly uses XBTUSDTM for BTC and <BASE>USDTM for most other
    USDT-margined contracts. The universe endpoint later confirms the exact
    symbol and aliases.
    """
    base = base_asset(symbol)
    if base == "BTC":
        return "XBTUSDTM"
    return f"{base}USDTM"


def to_float(value: Any, default: Optional[float] = None) -> Optional[float]:
    try:
        if value is None or value == "":
            return default
        return float(value)
    except Exception:
        return default


def to_int(value: Any, default: Optional[int] = None) -> Optional[int]:
    try:
        if value is None or value == "":
            return default
        return int(float(value))
    except Exception:
        return default


@dataclass
class ProviderResult:
    success: bool
    source: str
    capability: str
    sourceMode: str
    dataState: str
    data: Any = None
    errors: List[str] = field(default_factory=list)
    warnings: List[str] = field(default_factory=list)
    missingCapabilities: List[str] = field(default_factory=list)
    latencyMs: Optional[int] = None
    timestamp: str = field(default_factory=utc_now)
    providerStatus: Optional[int] = None
    cached: bool = False
    noTradeGuard: bool = False
    noTradeGuardReason: Optional[str] = None
    providerUsed: Optional[str] = None
    fallbackUsed: bool = False
    providersTried: List[Dict[str, Any]] = field(default_factory=list)
    cacheUsed: bool = False
    degradedReason: Optional[str] = None
    freshnessMs: Optional[int] = None
    healthScore: Optional[int] = None
    networkErrorType: Optional[str] = None
    providerStatusCode: Optional[int] = None
    isGeoBlockedPossible: bool = False
    isRateLimited: bool = False
    shouldTryProxy: bool = False
    suggestedRemediation: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        return {
            "success": self.success,
            "source": self.source,
            "provider": self.source,
            "providerUsed": self.providerUsed or self.source,
            "capability": self.capability,
            "sourceMode": self.sourceMode,
            "dataState": self.dataState,
            "latencyMs": self.latencyMs,
            "timestamp": self.timestamp,
            "data": self.data,
            "errors": self.errors,
            "warnings": self.warnings,
            "missingCapabilities": self.missingCapabilities,
            "providerStatus": self.providerStatus,
            "cached": self.cached,
            "noTradeGuard": self.noTradeGuard,
            "noTradeGuardReason": self.noTradeGuardReason,
            "fallbackUsed": self.fallbackUsed,
            "providersTried": self.providersTried,
            "cacheUsed": self.cacheUsed or self.cached,
            "degradedReason": self.degradedReason,
            "freshnessMs": self.freshnessMs,
            "healthScore": self.healthScore,
            "networkErrorType": self.networkErrorType,
            "providerStatusCode": self.providerStatusCode,
            "isGeoBlockedPossible": self.isGeoBlockedPossible,
            "isRateLimited": self.isRateLimited,
            "shouldTryProxy": self.shouldTryProxy,
            "suggestedRemediation": self.suggestedRemediation,
        }


def ok(source: str, capability: str, data: Any, latency_ms: Optional[int] = None, warnings: Optional[List[str]] = None) -> ProviderResult:
    return ProviderResult(
        success=True,
        source=source,
        capability=capability,
        sourceMode=SOURCE_LIVE,
        dataState=DATA_REAL,
        data=data,
        latencyMs=latency_ms,
        warnings=warnings or [],
    )


def fail(source: str, capability: str, error: str, latency_ms: Optional[int] = None, status: Optional[int] = None) -> ProviderResult:
    return ProviderResult(
        success=False,
        source=source,
        capability=capability,
        sourceMode=SOURCE_UNAVAILABLE,
        dataState=DATA_UNAVAILABLE,
        data=None,
        errors=[str(error)],
        missingCapabilities=[capability],
        latencyMs=latency_ms,
        providerStatus=status,
        noTradeGuard=True,
        noTradeGuardReason=f"{capability} unavailable from {source}",
    )


def disabled(source: str, capability: str, reason: str) -> ProviderResult:
    return ProviderResult(
        success=False,
        source=source,
        capability=capability,
        sourceMode=SOURCE_DISABLED,
        dataState=DATA_UNAVAILABLE,
        data=None,
        errors=[str(reason)],
        missingCapabilities=[capability],
        noTradeGuard=False,
        noTradeGuardReason=str(reason),
    )


def degraded(source: str, capability: str, data: Any, missing: List[str], warnings: List[str], latency_ms: Optional[int] = None) -> ProviderResult:
    return ProviderResult(
        success=True,
        source=source,
        capability=capability,
        sourceMode=SOURCE_DEGRADED,
        dataState=DATA_PARTIAL,
        data=data,
        warnings=warnings,
        missingCapabilities=missing,
        latencyMs=latency_ms,
        noTradeGuard=bool(missing),
        noTradeGuardReason=("; ".join(missing) + " missing") if missing else None,
    )


class CircuitState:
    """Lightweight in-memory health/cooldown tracker per provider capability."""

    def __init__(self, max_failures: int = 3, cooldown_seconds: int = 90) -> None:
        self.max_failures = int(os.getenv("SH_PROVIDER_MAX_FAILURES", str(max_failures)))
        self.cooldown_seconds = int(os.getenv("SH_PROVIDER_COOLDOWN_SECONDS", str(cooldown_seconds)))
        self._state: Dict[str, Dict[str, Any]] = {}

    def key(self, provider: str, capability: str) -> str:
        return f"{provider}:{capability}"

    def can_call(self, provider: str, capability: str) -> Tuple[bool, Optional[str]]:
        entry = self._state.get(self.key(provider, capability))
        if not entry:
            return True, None
        cooldown_until = entry.get("cooldownUntil")
        if cooldown_until and time.time() < cooldown_until:
            return False, f"provider in cooldown for {int(cooldown_until - time.time())}s"
        return True, None

    def record_success(self, provider: str, capability: str, latency_ms: Optional[int]) -> None:
        self._state[self.key(provider, capability)] = {
            "provider": provider,
            "capability": capability,
            "status": "healthy",
            "failures": 0,
            "lastSuccess": utc_now(),
            "lastLatencyMs": latency_ms,
            "cooldownUntil": None,
        }

    def record_failure(self, provider: str, capability: str, error: str) -> None:
        key = self.key(provider, capability)
        entry = self._state.get(key, {"provider": provider, "capability": capability, "failures": 0})
        failures = int(entry.get("failures", 0)) + 1
        status = "degraded" if failures < self.max_failures else "unavailable"
        cooldown_until = time.time() + self.cooldown_seconds if failures >= self.max_failures else None
        entry.update({
            "status": status,
            "failures": failures,
            "lastError": utc_now(),
            "lastErrorMessage": str(error)[:400],
            "cooldownUntil": cooldown_until,
        })
        self._state[key] = entry

    def snapshot(self) -> Dict[str, Any]:
        now = time.time()
        values = []
        for entry in self._state.values():
            cloned = dict(entry)
            if cloned.get("cooldownUntil"):
                cloned["cooldownRemainingSeconds"] = max(0, int(cloned["cooldownUntil"] - now))
            values.append(cloned)
        total = len(values)
        return {
            "total": total,
            "healthy": sum(1 for v in values if v.get("status") == "healthy"),
            "degraded": sum(1 for v in values if v.get("status") == "degraded"),
            "unavailable": sum(1 for v in values if v.get("status") == "unavailable"),
            "providers": values,
        }


circuit = CircuitState()


async def fetch_json(
    source: str,
    capability: str,
    url: str,
    *,
    params: Optional[Dict[str, Any]] = None,
    timeout: float = 10.0,
    headers: Optional[Dict[str, str]] = None,
) -> Tuple[Optional[Any], Optional[str], Optional[int], Optional[int]]:
    """Fetch JSON with latency and structured failure details."""
    allowed, reason = circuit.can_call(source, capability)
    if not allowed:
        return None, reason or "provider in cooldown", None, None
    started = time.perf_counter()
    try:
        async with httpx.AsyncClient(
            timeout=httpx.Timeout(min(timeout, 12.0), connect=4.0),
            headers={**DEFAULT_HEADERS, **(headers or {})},
            follow_redirects=True,
            trust_env=True,
        ) as client:
            response = await client.get(url, params=params)
            latency = int((time.perf_counter() - started) * 1000)
            if response.status_code < 200 or response.status_code >= 300:
                message = f"HTTP {response.status_code} {response.text[:160]}"
                circuit.record_failure(source, capability, message)
                return None, message, response.status_code, latency
            try:
                payload = response.json()
            except Exception as exc:
                message = f"invalid JSON: {exc}"
                circuit.record_failure(source, capability, message)
                return None, message, response.status_code, latency
            circuit.record_success(source, capability, latency)
            return payload, None, response.status_code, latency
    except Exception as exc:
        latency = int((time.perf_counter() - started) * 1000)
        message = f"{type(exc).__name__}: {exc}"
        circuit.record_failure(source, capability, message)
        return None, message, None, latency


async def first_success(tasks: Iterable, capability: str) -> ProviderResult:
    """Run provider calls sequentially by priority until one succeeds.

    Sequential rotation is intentional to respect free rate limits and keep
    provenance clear. Provider functions must return ProviderResult.
    """
    errors: List[str] = []
    missing: List[str] = []
    for task in tasks:
        result = await task()
        if result.success and result.data not in (None, [], {}):
            if errors:
                result.warnings.extend(errors)
            return result
        errors.extend([f"{result.source}: {e}" for e in result.errors])
        missing.extend(result.missingCapabilities)
    return ProviderResult(
        success=False,
        source="rotation",
        capability=capability,
        sourceMode=SOURCE_UNAVAILABLE,
        dataState=DATA_UNAVAILABLE,
        data=None,
        errors=errors or [f"no provider supplied {capability}"],
        missingCapabilities=sorted(set(missing or [capability])),
        noTradeGuard=True,
        noTradeGuardReason=f"{capability} unavailable from all configured providers",
    )


async def gather_capabilities(calls: Dict[str, Any]) -> Dict[str, ProviderResult]:
    """Run independent capability calls concurrently while preserving results."""
    keys = list(calls.keys())
    results = await asyncio.gather(*[calls[k]() for k in keys], return_exceptions=True)
    output: Dict[str, ProviderResult] = {}
    for key, value in zip(keys, results):
        if isinstance(value, ProviderResult):
            output[key] = value
        else:
            output[key] = fail("internal", key, str(value))
    return output