backend/routers/indicators_api.py

#!/usr/bin/env python3
"""
Technical Indicators API Router (PRODUCTION SAFE)
Provides API endpoints for calculating technical indicators on cryptocurrency data.
Includes: Bollinger Bands, Stochastic RSI, ATR, SMA, EMA, MACD, RSI

CRITICAL RULES:
- HTTP 400 for insufficient data (NOT HTTP 500)
- Strict minimum candle requirements enforced
- NaN/Infinity values sanitized before response
- Comprehensive logging for all operations
- Never crash - always return valid JSON
"""

from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional
from datetime import datetime
import logging
import math
import httpx

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/indicators", tags=["Technical Indicators"])

# ============================================================================
# MINIMUM CANDLE REQUIREMENTS (MANDATORY)
# ============================================================================
MIN_CANDLES = {
    "SMA": 20,
    "EMA": 20,
    "RSI": 15,
    "ATR": 15,
    "MACD": 35,
    "STOCH_RSI": 50,
    "BOLLINGER_BANDS": 20
}


# ============================================================================
# Pydantic Models
# ============================================================================

class OHLCVData(BaseModel):
    """OHLCV data model"""
    timestamp: int
    open: float
    high: float
    low: float
    close: float
    volume: float


class IndicatorRequest(BaseModel):
    """Request model for indicator calculation"""
    symbol: str = Field(default="BTC", description="Cryptocurrency symbol")
    timeframe: str = Field(default="1h", description="Timeframe (1m, 5m, 15m, 1h, 4h, 1d)")
    ohlcv: Optional[List[OHLCVData]] = Field(default=None, description="OHLCV data array")
    period: int = Field(default=14, description="Indicator period")


class AnalyzeBatchRequest(BaseModel):
    """Main-app batch indicator request"""
    symbol: str = Field(default="BTC")
    timeframe: str = Field(default="1h")
    indicators: Optional[List[str]] = Field(default=None, description="e.g. ['rsi','macd'] or ['comprehensive']")


class BollingerBandsResponse(BaseModel):
    """Bollinger Bands response model"""
    upper: float
    middle: float
    lower: float
    bandwidth: float
    percent_b: float
    signal: str
    description: str


class StochRSIResponse(BaseModel):
    """Stochastic RSI response model"""
    value: float
    k_line: float
    d_line: float
    signal: str
    description: str


class ATRResponse(BaseModel):
    """Average True Range response model"""
    value: float
    percent: float
    volatility_level: str
    signal: str
    description: str


class SMAResponse(BaseModel):
    """Simple Moving Average response model"""
    sma20: float
    sma50: float
    sma200: Optional[float]
    price_vs_sma20: str
    price_vs_sma50: str
    trend: str
    signal: str
    description: str


class EMAResponse(BaseModel):
    """Exponential Moving Average response model"""
    ema12: float
    ema26: float
    ema50: Optional[float]
    trend: str
    signal: str
    description: str


class MACDResponse(BaseModel):
    """MACD response model"""
    macd_line: float
    signal_line: float
    histogram: float
    trend: str
    signal: str
    description: str


class RSIResponse(BaseModel):
    """RSI response model"""
    value: float
    signal: str
    description: str


class ComprehensiveIndicatorsResponse(BaseModel):
    """All indicators combined response"""
    symbol: str
    timeframe: str
    timestamp: str
    current_price: float
    bollinger_bands: BollingerBandsResponse
    stoch_rsi: StochRSIResponse
    atr: ATRResponse
    sma: SMAResponse
    ema: EMAResponse
    macd: MACDResponse
    rsi: RSIResponse
    overall_signal: str
    recommendation: str


# ============================================================================
# Helper Functions - Data Validation & Sanitization
# ============================================================================

def sanitize_value(value: Any) -> Optional[float]:
    """
    Sanitize a numeric value - remove NaN, Infinity, None
    Returns None if value is invalid, otherwise returns the float value
    """
    if value is None:
        return None
    try:
        val = float(value)
        if math.isnan(val) or math.isinf(val):
            return None
        return val
    except (ValueError, TypeError):
        return None


def sanitize_dict(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Sanitize all numeric values in a dictionary
    Replace NaN/Infinity with None or 0 depending on context
    """
    sanitized = {}
    for key, value in data.items():
        if isinstance(value, dict):
            sanitized[key] = sanitize_dict(value)
        elif isinstance(value, (int, float)):
            clean_val = sanitize_value(value)
            sanitized[key] = clean_val if clean_val is not None else 0
        else:
            sanitized[key] = value
    return sanitized


def validate_ohlcv_data(ohlcv: Optional[Dict[str, Any]], min_candles: int, symbol: str, indicator: str) -> tuple[bool, Optional[List[float]], Optional[str]]:
    """
    Validate OHLCV data and extract prices
    
    Returns:
        (is_valid, prices, error_message)
    """
    if not ohlcv:
        logger.warning(f"❌ {indicator} - {symbol}: No OHLCV data received")
        return False, None, "No market data available"
    
    if "prices" not in ohlcv:
        logger.warning(f"❌ {indicator} - {symbol}: OHLCV missing 'prices' key")
        return False, None, "Invalid market data format"
    
    prices = [p[1] for p in ohlcv["prices"] if len(p) >= 2]
    
    if not prices:
        logger.warning(f"❌ {indicator} - {symbol}: Empty price array")
        return False, None, "No price data available"
    
    if len(prices) < min_candles:
        logger.warning(f"❌ {indicator} - {symbol}: Insufficient candles ({len(prices)} < {min_candles} required)")
        return False, None, f"Insufficient market data: need at least {min_candles} candles, got {len(prices)}"
    
    logger.info(f"✅ {indicator} - {symbol}: Validated {len(prices)} candles (required: {min_candles})")
    return True, prices, None


# ============================================================================
# Helper Functions for Calculations
# ============================================================================

def calculate_sma(prices: List[float], period: int) -> float:
    """Calculate Simple Moving Average"""
    if len(prices) < period:
        return prices[-1] if prices else 0
    return sum(prices[-period:]) / period


def calculate_ema(prices: List[float], period: int) -> float:
    """Calculate Exponential Moving Average"""
    if len(prices) < period:
        return prices[-1] if prices else 0
    
    multiplier = 2 / (period + 1)
    ema = sum(prices[:period]) / period  # SMA for first period
    
    for price in prices[period:]:
        ema = (price * multiplier) + (ema * (1 - multiplier))
    
    return ema


def calculate_rsi(prices: List[float], period: int = 14) -> float:
    """Calculate Relative Strength Index"""
    if len(prices) < period + 1:
        return 50.0
    
    deltas = [prices[i] - prices[i-1] for i in range(1, len(prices))]
    gains = [d if d > 0 else 0 for d in deltas[-period:]]
    losses = [-d if d < 0 else 0 for d in deltas[-period:]]
    
    avg_gain = sum(gains) / period
    avg_loss = sum(losses) / period
    
    if avg_loss == 0:
        return 100.0 if avg_gain > 0 else 50.0
    
    rs = avg_gain / avg_loss
    return 100 - (100 / (1 + rs))


def calculate_bollinger_bands(prices: List[float], period: int = 20, std_dev: float = 2) -> Dict[str, float]:
    """Calculate Bollinger Bands"""
    if len(prices) < period:
        current = prices[-1] if prices else 0
        return {
            "upper": current,
            "middle": current,
            "lower": current,
            "bandwidth": 0,
            "percent_b": 50
        }
    
    recent_prices = prices[-period:]
    middle = sum(recent_prices) / period
    
    # Calculate standard deviation
    variance = sum((p - middle) ** 2 for p in recent_prices) / period
    std = variance ** 0.5
    
    upper = middle + (std_dev * std)
    lower = middle - (std_dev * std)
    
    # Bandwidth as percentage
    bandwidth = ((upper - lower) / middle) * 100 if middle > 0 else 0
    
    # Percent B (position within bands)
    current_price = prices[-1]
    if upper != lower:
        percent_b = ((current_price - lower) / (upper - lower)) * 100
    else:
        percent_b = 50
    
    return {
        "upper": round(upper, 8),
        "middle": round(middle, 8),
        "lower": round(lower, 8),
        "bandwidth": round(bandwidth, 2),
        "percent_b": round(percent_b, 2)
    }


def calculate_stoch_rsi(prices: List[float], rsi_period: int = 14, stoch_period: int = 14) -> Dict[str, float]:
    """Calculate Stochastic RSI"""
    if len(prices) < rsi_period + stoch_period:
        return {"value": 50, "k_line": 50, "d_line": 50}
    
    # Calculate RSI values for the stoch period
    rsi_values = []
    for i in range(stoch_period + 3):  # Extra for smoothing
        end_idx = len(prices) - stoch_period + i + 1
        if end_idx > rsi_period:
            slice_prices = prices[:end_idx]
            rsi_values.append(calculate_rsi(slice_prices, rsi_period))
    
    if len(rsi_values) < stoch_period:
        return {"value": 50, "k_line": 50, "d_line": 50}
    
    recent_rsi = rsi_values[-stoch_period:]
    rsi_high = max(recent_rsi)
    rsi_low = min(recent_rsi)
    
    current_rsi = rsi_values[-1]
    
    if rsi_high == rsi_low:
        stoch_rsi = 50
    else:
        stoch_rsi = ((current_rsi - rsi_low) / (rsi_high - rsi_low)) * 100
    
    # K line is the raw Stoch RSI
    k_line = stoch_rsi
    
    # D line is 3-period SMA of K
    if len(rsi_values) >= 3:
        k_values = []
        for i in range(3):
            idx = -3 + i
            window = rsi_values[idx - stoch_period + 1:idx + 1] if idx + 1 <= 0 else recent_rsi
            if not window:
                k_values.append(50)
                continue
            r_high = max(window)
            r_low = min(window)
            curr = rsi_values[idx]
            if r_high != r_low:
                k_values.append(((curr - r_low) / (r_high - r_low)) * 100)
            else:
                k_values.append(50)
        d_line = sum(k_values) / 3
    else:
        d_line = k_line
    
    return {
        "value": round(stoch_rsi, 2),
        "k_line": round(k_line, 2),
        "d_line": round(d_line, 2)
    }


def calculate_atr(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> float:
    """Calculate Average True Range"""
    if len(closes) < period + 1:
        if len(highs) > 0 and len(lows) > 0:
            return highs[-1] - lows[-1]
        return 0
    
    true_ranges = []
    for i in range(1, len(closes)):
        high = highs[i]
        low = lows[i]
        prev_close = closes[i-1]
        
        tr = max(
            high - low,
            abs(high - prev_close),
            abs(low - prev_close)
        )
        true_ranges.append(tr)
    
    # ATR is the average of the last 'period' true ranges
    if len(true_ranges) < period:
        return sum(true_ranges) / len(true_ranges) if true_ranges else 0
    
    return sum(true_ranges[-period:]) / period


def calculate_macd(prices: List[float], fast: int = 12, slow: int = 26, signal: int = 9) -> Dict[str, float]:
    """Calculate MACD"""
    if len(prices) < slow + signal:
        return {"macd_line": 0, "signal_line": 0, "histogram": 0}
    
    ema_fast = calculate_ema(prices, fast)
    ema_slow = calculate_ema(prices, slow)
    macd_line = ema_fast - ema_slow
    
    # Calculate signal line (EMA of MACD)
    # We need MACD values history for signal line
    macd_values = []
    for i in range(signal + 5):
        idx = len(prices) - signal - 5 + i
        if idx > slow:
            slice_prices = prices[:idx+1]
            ef = calculate_ema(slice_prices, fast)
            es = calculate_ema(slice_prices, slow)
            macd_values.append(ef - es)
    
    if len(macd_values) >= signal:
        signal_line = calculate_ema(macd_values, signal)
    else:
        signal_line = macd_line
    
    histogram = macd_line - signal_line
    
    return {
        "macd_line": round(macd_line, 8),
        "signal_line": round(signal_line, 8),
        "histogram": round(histogram, 8)
    }


# ============================================================================
# API Endpoints
# ============================================================================

@router.get("/services")
async def list_indicator_services():
    """List all available technical indicator services"""
    from backend.services.indicator_analysis_service import services_catalog
    return services_catalog()


@router.post("/analyze")
async def analyze_indicators_batch(request: AnalyzeBatchRequest):
    """Batch indicator analysis for main-app integration."""
    from backend.services.indicator_analysis_service import batch_analyze
    result = await batch_analyze(request.symbol, request.timeframe, request.indicators)
    if not result.get("success"):
        return JSONResponse(status_code=400, content=result)
    return result


def _indicator_error(result):
    return JSONResponse(status_code=400, content=result)


@router.get('/bollinger-bands')
async def get_bollinger_bands(
    symbol: str = Query(default='BTC'),
    timeframe: str = Query(default='1h'),
    period: int = Query(default=20),
    std_dev: float = Query(default=2.0),
):
    from backend.services.indicator_analysis_service import single_indicator
    result = await single_indicator('bollinger_bands', symbol, timeframe, period=period, std_dev=std_dev)
    return _indicator_error(result) if not result.get('success') else result


@router.get('/stoch-rsi')
async def get_stoch_rsi(
    symbol: str = Query(default='BTC'),
    timeframe: str = Query(default='1h'),
    rsi_period: int = Query(default=14),
    stoch_period: int = Query(default=14),
):
    from backend.services.indicator_analysis_service import single_indicator
    result = await single_indicator('stoch_rsi', symbol, timeframe, rsi_period=rsi_period, stoch_period=stoch_period)
    return _indicator_error(result) if not result.get('success') else result


@router.get('/atr')
async def get_atr(
    symbol: str = Query(default='BTC'),
    timeframe: str = Query(default='1h'),
    period: int = Query(default=14),
):
    from backend.services.indicator_analysis_service import single_indicator
    result = await single_indicator('atr', symbol, timeframe, period=period)
    return _indicator_error(result) if not result.get('success') else result


@router.get('/sma')
async def get_sma(symbol: str = Query(default='BTC'), timeframe: str = Query(default='1h')):
    from backend.services.indicator_analysis_service import single_indicator
    result = await single_indicator('sma', symbol, timeframe)
    return _indicator_error(result) if not result.get('success') else result


@router.get('/ema')
async def get_ema(symbol: str = Query(default='BTC'), timeframe: str = Query(default='1h')):
    from backend.services.indicator_analysis_service import single_indicator
    result = await single_indicator('ema', symbol, timeframe)
    return _indicator_error(result) if not result.get('success') else result


@router.get('/macd')
async def get_macd(
    symbol: str = Query(default='BTC'),
    timeframe: str = Query(default='1h'),
    fast: int = Query(default=12),
    slow: int = Query(default=26),
    signal: int = Query(default=9),
):
    from backend.services.indicator_analysis_service import single_indicator
    result = await single_indicator('macd', symbol, timeframe, fast=fast, slow=slow, signal_period=signal)
    return _indicator_error(result) if not result.get('success') else result


@router.get('/rsi')
async def get_rsi(
    symbol: str = Query(default='BTC'),
    timeframe: str = Query(default='1h'),
    period: int = Query(default=14),
):
    from backend.services.indicator_analysis_service import single_indicator
    result = await single_indicator('rsi', symbol, timeframe, period=period)
    return _indicator_error(result) if not result.get('success') else result


@router.get('/comprehensive')
async def get_comprehensive_analysis(
    symbol: str = Query(default='BTC'),
    timeframe: str = Query(default='1h'),
):
    from backend.services.indicator_analysis_service import comprehensive_analysis
    result = await comprehensive_analysis(symbol, timeframe)
    return _indicator_error(result) if not result.get('success') else result