api/app/indicators.py

import pandas as pd
import numpy as np
import ta
from typing import List, Dict, Any

def calculate_indicators(df: pd.DataFrame, indicators: List[str]) -> pd.DataFrame:
    """
    Calculate technical indicators for the given OHLCV data
    
    Args:
        df: DataFrame with OHLCV data
        indicators: List of indicators to calculate
    
    Returns:
        DataFrame with calculated indicators
    """
    # Make a copy to avoid modifying the original DataFrame
    result = df.copy()
    
    # Ensure we have required columns
    required_columns = ['timestamp', 'datetime', 'open', 'high', 'low', 'close', 'volume']
    for col in required_columns:
        if col not in result.columns:
            if col == 'datetime' and 'timestamp' in result.columns:
                # We can calculate datetime from timestamp
                continue
            else:
                raise ValueError(f"Required column {col} not found in DataFrame")
    
    # Calculate RSI
    if 'rsi' in indicators:
        result['rsi'] = ta.momentum.RSIIndicator(
            close=result['close'], 
            window=14
        ).rsi()
    
    # Calculate MACD
    if 'macd' in indicators:
        macd = ta.trend.MACD(
            close=result['close'],
            window_fast=12,
            window_slow=26,
            window_sign=9
        )
        result['macd'] = macd.macd()
        result['macd_signal'] = macd.macd_signal()
        result['macd_histogram'] = macd.macd_diff()
    
    # Calculate Bollinger Bands
    if 'bollinger' in indicators:
        bollinger = ta.volatility.BollingerBands(
            close=result['close'],
            window=20,
            window_dev=2
        )
        result['bb_upper'] = bollinger.bollinger_hband()
        result['bb_middle'] = bollinger.bollinger_mavg()
        result['bb_lower'] = bollinger.bollinger_lband()
    
    # Calculate Exponential Moving Averages
    if 'ema' in indicators:
        result['ema_9'] = ta.trend.EMAIndicator(
            close=result['close'], window=9
        ).ema_indicator()
        
        result['ema_21'] = ta.trend.EMAIndicator(
            close=result['close'], window=21
        ).ema_indicator()
        
        result['ema_50'] = ta.trend.EMAIndicator(
            close=result['close'], window=50
        ).ema_indicator()
        
        result['ema_200'] = ta.trend.EMAIndicator(
            close=result['close'], window=200
        ).ema_indicator()
    
    # Calculate Average True Range (ATR)
    if 'atr' in indicators:
        result['atr'] = ta.volatility.AverageTrueRange(
            high=result['high'],
            low=result['low'],
            close=result['close'],
            window=14
        ).average_true_range()
    
    # Calculate Stochastic Oscillator
    if 'stoch' in indicators:
        stoch = ta.momentum.StochasticOscillator(
            high=result['high'],
            low=result['low'],
            close=result['close'],
            window=14,
            smooth_window=3
        )
        result['stoch_k'] = stoch.stoch()
        result['stoch_d'] = stoch.stoch_signal()
    
    return result