indicators/indicators.py

import numpy as np
import pandas as pd

def sma(series: pd.Series, window: int) -> pd.Series:
    return series.rolling(window=window).mean()

def ema(series: pd.Series, window: int) -> pd.Series:
    return series.ewm(span=window, adjust=False).mean()

def _wilder_rma(values: pd.Series, window: int) -> pd.Series:
    if isinstance(window, bool) or not isinstance(window, (int, np.integer)) or window < 1:
        raise ValueError('Wilder period must be a positive integer.')
    arr = values.to_numpy(dtype=float)
    n = len(arr)
    out = np.full(n, np.nan)
    seed, prev = [], None
    for i, value in enumerate(arr):
        if not np.isfinite(value):
            seed, prev = [], None
            continue
        if prev is None:
            seed.append(value)
            if len(seed) == window:
                prev = float(np.mean(seed))
                out[i] = prev
        else:
            prev = (prev * (window - 1) + value) / window
            out[i] = prev
    return pd.Series(out, index=values.index)

def rsi(series: pd.Series, window: int=14) -> pd.Series:
    delta = series.diff()
    gain = delta.clip(lower=0)
    loss = -delta.clip(upper=0)
    avg_gain = _wilder_rma(gain, window)
    avg_loss = _wilder_rma(loss, window)
    rs = avg_gain / avg_loss.replace(0, np.nan)
    out = 100 - 100 / (1 + rs)
    out = out.mask((avg_loss == 0) & (avg_gain > 0), 100.0)
    return out.mask((avg_loss == 0) & (avg_gain == 0), 50.0)

def macd(series: pd.Series, fast: int=12, slow: int=26, signal: int=9):
    fast_ema = ema(series, fast)
    slow_ema = ema(series, slow)
    macd_line = fast_ema - slow_ema
    signal_line = ema(macd_line, signal)
    histogram = macd_line - signal_line
    return (macd_line, signal_line, histogram)

def bollinger_bands(series: pd.Series, window: int=20, num_std: float=2.0):
    mid = sma(series, window)
    std = series.rolling(window=window).std(ddof=0)
    return (mid + num_std * std, mid, mid - num_std * std)

def stochastic_oscillator(high: pd.Series, low: pd.Series, close: pd.Series, k_window: int=14, d_window: int=3):
    lowest_low = low.rolling(window=k_window).min()
    highest_high = high.rolling(window=k_window).max()
    span = highest_high - lowest_low
    percent_k = 100 * (close - lowest_low) / span.replace(0, np.nan)
    percent_k = percent_k.mask(span == 0, 50.0)
    percent_d = percent_k.rolling(window=d_window).mean()
    return (percent_k, percent_d)

def add_all_indicators(df: pd.DataFrame) -> pd.DataFrame:
    out = df.copy()
    out['SMA_20'] = sma(out['Close'], 20)
    out['EMA_20'] = ema(out['Close'], 20)
    out['RSI_14'] = rsi(out['Close'], 14)
    macd_line, signal_line, hist = macd(out['Close'])
    out['MACD'] = macd_line
    out['MACD_Signal'] = signal_line
    out['MACD_Hist'] = hist
    upper, mid, lower = bollinger_bands(out['Close'])
    out['BB_Upper'] = upper
    out['BB_Mid'] = mid
    out['BB_Lower'] = lower
    k, d = stochastic_oscillator(out['High'], out['Low'], out['Close'])
    out['Stoch_%K'] = k
    out['Stoch_%D'] = d
    return out

def _round(value, digits):
    return round(float(value), digits) if pd.notna(value) else 'n/a'

def latest_signals(df_with_indicators: pd.DataFrame) -> dict:
    if df_with_indicators.empty:
        raise ValueError('No candles available for latest indicators.')
    last = df_with_indicators.iloc[-1]
    signals = {}
    rsi_value = last['RSI_14']
    signals['RSI (14)'] = {'value': _round(rsi_value, 2), 'signal': 'Insufficient data' if pd.isna(rsi_value) else 'Overbought' if rsi_value > 70 else 'Oversold' if rsi_value < 30 else 'Neutral'}
    macd_ready = pd.notna(last['MACD']) and pd.notna(last['MACD_Signal'])
    signals['MACD'] = {'value': _round(last['MACD'], 5), 'signal': 'Insufficient data' if not macd_ready else 'Bullish (above signal)' if last['MACD'] > last['MACD_Signal'] else 'Bearish (below signal)' if last['MACD'] < last['MACD_Signal'] else 'Neutral'}
    bb_ready = pd.notna(last['BB_Upper']) and pd.notna(last['BB_Lower'])
    signals['Bollinger Bands'] = {'value': _round(last['Close'], 5), 'signal': 'Insufficient data' if not bb_ready else 'Above Upper Band' if last['Close'] > last['BB_Upper'] else 'Below Lower Band' if last['Close'] < last['BB_Lower'] else 'Inside Bands'}
    sma_ready = pd.notna(last['SMA_20'])
    signals['Moving Averages'] = {'value': f"SMA20={_round(last['SMA_20'], 5)}, EMA20={_round(last['EMA_20'], 5)}", 'signal': 'Insufficient data for SMA' if not sma_ready else 'Price Above MA (Uptrend)' if last['Close'] > last['SMA_20'] else 'Price Below MA (Downtrend)' if last['Close'] < last['SMA_20'] else 'Neutral'}
    stoch_value = last['Stoch_%K']
    signals['Stochastic (14,3)'] = {'value': _round(stoch_value, 2), 'signal': 'Insufficient data' if pd.isna(stoch_value) else 'Overbought' if stoch_value > 80 else 'Oversold' if stoch_value < 20 else 'Neutral'}
    return signals