openbb-invest-api/routes_technical.py

"""Routes for technical analysis indicators."""

from fastapi import APIRouter, Path, Query

from models import ApiResponse
from route_utils import safe, validate_symbol
import technical_service

router = APIRouter(prefix="/api/v1")


@router.get("/stock/{symbol}/technical", response_model=ApiResponse)
@safe
async def stock_technical(symbol: str = Path(..., min_length=1, max_length=20)):
    """Get technical indicators: RSI, MACD, SMA, EMA, Bollinger Bands + signal interpretation."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_technical_indicators(symbol)
    return ApiResponse(data=data)


# --- Individual Technical Indicators (Group I) ---


@router.get("/stock/{symbol}/technical/atr", response_model=ApiResponse)
@safe
async def stock_atr(
    symbol: str = Path(..., min_length=1, max_length=20),
    length: int = Query(default=14, ge=1, le=100),
):
    """Average True Range -- volatility for position sizing and stop-loss."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_atr(symbol, length=length)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/adx", response_model=ApiResponse)
@safe
async def stock_adx(
    symbol: str = Path(..., min_length=1, max_length=20),
    length: int = Query(default=14, ge=1, le=100),
):
    """Average Directional Index -- trend strength (>25 strong, <20 range-bound)."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_adx(symbol, length=length)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/stoch", response_model=ApiResponse)
@safe
async def stock_stoch(
    symbol: str = Path(..., min_length=1, max_length=20),
    fast_k: int = Query(default=14, ge=1, le=100),
    slow_d: int = Query(default=3, ge=1, le=100),
    slow_k: int = Query(default=3, ge=1, le=100),
):
    """Stochastic Oscillator -- overbought/oversold momentum signal."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_stoch(symbol, fast_k=fast_k, slow_d=slow_d, slow_k=slow_k)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/obv", response_model=ApiResponse)
@safe
async def stock_obv(symbol: str = Path(..., min_length=1, max_length=20)):
    """On-Balance Volume -- cumulative volume for divergence detection."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_obv(symbol)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/ichimoku", response_model=ApiResponse)
@safe
async def stock_ichimoku(symbol: str = Path(..., min_length=1, max_length=20)):
    """Ichimoku Cloud -- comprehensive trend system with support/resistance."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_ichimoku(symbol)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/donchian", response_model=ApiResponse)
@safe
async def stock_donchian(
    symbol: str = Path(..., min_length=1, max_length=20),
    length: int = Query(default=20, ge=1, le=100),
):
    """Donchian Channels -- breakout detection system."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_donchian(symbol, length=length)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/aroon", response_model=ApiResponse)
@safe
async def stock_aroon(
    symbol: str = Path(..., min_length=1, max_length=20),
    length: int = Query(default=25, ge=1, le=100),
):
    """Aroon Indicator -- identifies trend direction and potential changes."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_aroon(symbol, length=length)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/cci", response_model=ApiResponse)
@safe
async def stock_cci(
    symbol: str = Path(..., min_length=1, max_length=20),
    length: int = Query(default=14, ge=1, le=100),
):
    """Commodity Channel Index -- cyclical trend identification."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_cci(symbol, length=length)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/kc", response_model=ApiResponse)
@safe
async def stock_kc(
    symbol: str = Path(..., min_length=1, max_length=20),
    length: int = Query(default=20, ge=1, le=100),
):
    """Keltner Channels -- ATR-based volatility bands."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_kc(symbol, length=length)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/fib", response_model=ApiResponse)
@safe
async def stock_fib(
    symbol: str = Path(..., min_length=1, max_length=20),
    days: int = Query(default=120, ge=5, le=365),
):
    """Fibonacci Retracement -- key support/resistance levels."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_fib(symbol, days=days)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/ad", response_model=ApiResponse)
@safe
async def stock_ad(symbol: str = Path(..., min_length=1, max_length=20)):
    """Accumulation/Distribution Line -- volume-based trend indicator."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_ad(symbol)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/cones", response_model=ApiResponse)
@safe
async def stock_cones(symbol: str = Path(..., min_length=1, max_length=20)):
    """Volatility Cones -- realized vol quantiles for options analysis."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_cones(symbol)
    return ApiResponse(data=data)


@router.get("/stock/{symbol}/technical/vwap", response_model=ApiResponse)
@safe
async def stock_vwap(symbol: str = Path(..., min_length=1, max_length=20)):
    """Volume Weighted Average Price -- intraday fair value benchmark."""
    symbol = validate_symbol(symbol)
    data = await technical_service.get_vwap(symbol)
    return ApiResponse(data=data)


@router.get("/technical/relative-rotation", response_model=ApiResponse)
@safe
async def relative_rotation(
    symbols: str = Query(..., min_length=1, max_length=200, description="Comma-separated symbols, e.g. AAPL,MSFT,GOOGL"),
    benchmark: str = Query(default="SPY", min_length=1, max_length=20),
    study: str = Query(default="price", pattern="^(price|volume|volatility)$"),
):
    """Relative Rotation Graph -- compare multiple symbols vs benchmark.

    Returns RS-Ratio and RS-Momentum for each symbol, indicating
    RRG quadrant: Leading, Weakening, Lagging, or Improving.
    """
    symbol_list = [validate_symbol(s.strip()) for s in symbols.split(",") if s.strip()]
    if not symbol_list:
        return ApiResponse(data=[], error="No valid symbols provided")
    benchmark = validate_symbol(benchmark)
    data = await technical_service.get_relative_rotation(
        symbol_list, benchmark=benchmark, study=study,
    )
    return ApiResponse(data=data)