diff options
Diffstat (limited to 'services/strategy-engine/strategies/indicators')
5 files changed, 242 insertions, 0 deletions
diff --git a/services/strategy-engine/strategies/indicators/__init__.py b/services/strategy-engine/strategies/indicators/__init__.py new file mode 100644 index 0000000..1a54d59 --- /dev/null +++ b/services/strategy-engine/strategies/indicators/__init__.py @@ -0,0 +1,12 @@ +"""Reusable technical indicator functions.""" +from strategies.indicators.trend import ema, sma, macd, adx +from strategies.indicators.volatility import atr, bollinger_bands, keltner_channels +from strategies.indicators.momentum import rsi, stochastic +from strategies.indicators.volume import volume_sma, volume_ratio, obv + +__all__ = [ + "ema", "sma", "macd", "adx", + "atr", "bollinger_bands", "keltner_channels", + "rsi", "stochastic", + "volume_sma", "volume_ratio", "obv", +] diff --git a/services/strategy-engine/strategies/indicators/momentum.py b/services/strategy-engine/strategies/indicators/momentum.py new file mode 100644 index 0000000..395c52d --- /dev/null +++ b/services/strategy-engine/strategies/indicators/momentum.py @@ -0,0 +1,36 @@ +"""Momentum indicators: RSI, Stochastic.""" +import pandas as pd +import numpy as np + + +def rsi(closes: pd.Series, period: int = 14) -> pd.Series: + """RSI using Wilder's smoothing (EMA-based).""" + delta = closes.diff() + gain = delta.clip(lower=0) + loss = -delta.clip(upper=0) + avg_gain = gain.ewm(com=period - 1, min_periods=period).mean() + avg_loss = loss.ewm(com=period - 1, min_periods=period).mean() + rs = avg_gain / avg_loss + rsi_vals = pd.Series(np.where(avg_loss == 0, 100.0, 100 - (100 / (1 + rs))), index=closes.index) + # Keep NaN where we don't have enough data + rsi_vals[:period] = np.nan + return rsi_vals + + +def stochastic( + highs: pd.Series, + lows: pd.Series, + closes: pd.Series, + k_period: int = 14, + d_period: int = 3, +) -> tuple[pd.Series, pd.Series]: + """Stochastic Oscillator. + + Returns: (%K, %D) + """ + lowest_low = lows.rolling(window=k_period).min() + highest_high = highs.rolling(window=k_period).max() + denom = highest_high - lowest_low + k = 100 * (closes - lowest_low) / denom.replace(0, float("nan")) + d = k.rolling(window=d_period).mean() + return k, d diff --git a/services/strategy-engine/strategies/indicators/trend.py b/services/strategy-engine/strategies/indicators/trend.py new file mode 100644 index 0000000..10b69fa --- /dev/null +++ b/services/strategy-engine/strategies/indicators/trend.py @@ -0,0 +1,104 @@ +"""Trend indicators: EMA, SMA, MACD, ADX.""" +import pandas as pd +import numpy as np + + +def sma(series: pd.Series, period: int) -> pd.Series: + """Simple Moving Average.""" + return series.rolling(window=period).mean() + + +def ema(series: pd.Series, period: int) -> pd.Series: + """Exponential Moving Average.""" + return series.ewm(span=period, adjust=False).mean() + + +def macd( + closes: pd.Series, + fast: int = 12, + slow: int = 26, + signal: int = 9, +) -> tuple[pd.Series, pd.Series, pd.Series]: + """MACD indicator. + + Returns: (macd_line, signal_line, histogram) + """ + fast_ema = ema(closes, fast) + slow_ema = ema(closes, 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 adx( + highs: pd.Series, + lows: pd.Series, + closes: pd.Series, + period: int = 14, +) -> pd.Series: + """Average Directional Index (ADX). + + Returns ADX series. Values > 25 indicate strong trend, < 20 indicate ranging. + """ + high = highs.values + low = lows.values + close = closes.values + n = len(close) + + if n < period + 1: + return pd.Series([np.nan] * n) + + # True Range + tr = np.zeros(n) + tr[0] = high[0] - low[0] + for i in range(1, n): + tr[i] = max( + high[i] - low[i], + abs(high[i] - close[i - 1]), + abs(low[i] - close[i - 1]), + ) + + # Directional Movement + plus_dm = np.zeros(n) + minus_dm = np.zeros(n) + for i in range(1, n): + up = high[i] - high[i - 1] + down = low[i - 1] - low[i] + plus_dm[i] = up if up > down and up > 0 else 0.0 + minus_dm[i] = down if down > up and down > 0 else 0.0 + + # Smoothed with Wilder's method + atr_vals = np.zeros(n) + plus_di = np.zeros(n) + minus_di = np.zeros(n) + + atr_vals[period] = np.mean(tr[1 : period + 1]) + plus_smooth = np.mean(plus_dm[1 : period + 1]) + minus_smooth = np.mean(minus_dm[1 : period + 1]) + + if atr_vals[period] > 0: + plus_di[period] = 100 * plus_smooth / atr_vals[period] + minus_di[period] = 100 * minus_smooth / atr_vals[period] + + for i in range(period + 1, n): + atr_vals[i] = (atr_vals[i - 1] * (period - 1) + tr[i]) / period + plus_smooth = (plus_smooth * (period - 1) + plus_dm[i]) / period + minus_smooth = (minus_smooth * (period - 1) + minus_dm[i]) / period + if atr_vals[i] > 0: + plus_di[i] = 100 * plus_smooth / atr_vals[i] + minus_di[i] = 100 * minus_smooth / atr_vals[i] + + # DX and ADX + dx = np.zeros(n) + for i in range(period, n): + di_sum = plus_di[i] + minus_di[i] + dx[i] = 100 * abs(plus_di[i] - minus_di[i]) / di_sum if di_sum > 0 else 0.0 + + adx_vals = np.full(n, np.nan) + if 2 * period < n: + adx_vals[2 * period] = np.mean(dx[period : 2 * period + 1]) + for i in range(2 * period + 1, n): + adx_vals[i] = (adx_vals[i - 1] * (period - 1) + dx[i]) / period + + return pd.Series(adx_vals, index=closes.index if hasattr(closes, 'index') else None) diff --git a/services/strategy-engine/strategies/indicators/volatility.py b/services/strategy-engine/strategies/indicators/volatility.py new file mode 100644 index 0000000..d47eb86 --- /dev/null +++ b/services/strategy-engine/strategies/indicators/volatility.py @@ -0,0 +1,69 @@ +"""Volatility indicators: ATR, Bollinger Bands, Keltner Channels.""" +import pandas as pd +import numpy as np + + +def atr( + highs: pd.Series, + lows: pd.Series, + closes: pd.Series, + period: int = 14, +) -> pd.Series: + """Average True Range using Wilder's smoothing.""" + high = highs.values + low = lows.values + close = closes.values + n = len(close) + + tr = np.zeros(n) + tr[0] = high[0] - low[0] + for i in range(1, n): + tr[i] = max( + high[i] - low[i], + abs(high[i] - close[i - 1]), + abs(low[i] - close[i - 1]), + ) + + atr_vals = np.full(n, np.nan) + if n >= period: + atr_vals[period - 1] = np.mean(tr[:period]) + for i in range(period, n): + atr_vals[i] = (atr_vals[i - 1] * (period - 1) + tr[i]) / period + + return pd.Series(atr_vals, index=closes.index if hasattr(closes, 'index') else None) + + +def bollinger_bands( + closes: pd.Series, + period: int = 20, + num_std: float = 2.0, +) -> tuple[pd.Series, pd.Series, pd.Series]: + """Bollinger Bands. + + Returns: (upper_band, middle_band, lower_band) + """ + middle = closes.rolling(window=period).mean() + std = closes.rolling(window=period).std() + upper = middle + num_std * std + lower = middle - num_std * std + return upper, middle, lower + + +def keltner_channels( + highs: pd.Series, + lows: pd.Series, + closes: pd.Series, + ema_period: int = 20, + atr_period: int = 14, + atr_multiplier: float = 2.0, +) -> tuple[pd.Series, pd.Series, pd.Series]: + """Keltner Channels. + + Returns: (upper_channel, middle_ema, lower_channel) + """ + from strategies.indicators.trend import ema as calc_ema + middle = calc_ema(closes, ema_period) + atr_vals = atr(highs, lows, closes, atr_period) + upper = middle + atr_multiplier * atr_vals + lower = middle - atr_multiplier * atr_vals + return upper, middle, lower diff --git a/services/strategy-engine/strategies/indicators/volume.py b/services/strategy-engine/strategies/indicators/volume.py new file mode 100644 index 0000000..323d427 --- /dev/null +++ b/services/strategy-engine/strategies/indicators/volume.py @@ -0,0 +1,21 @@ +"""Volume indicators: Volume SMA, Volume Ratio, OBV.""" +import pandas as pd +import numpy as np + + +def volume_sma(volumes: pd.Series, period: int = 20) -> pd.Series: + """Simple moving average of volume.""" + return volumes.rolling(window=period).mean() + + +def volume_ratio(volumes: pd.Series, period: int = 20) -> pd.Series: + """Current volume as ratio of average volume. >1 = above average.""" + avg = volume_sma(volumes, period) + return volumes / avg.replace(0, float("nan")) + + +def obv(closes: pd.Series, volumes: pd.Series) -> pd.Series: + """On-Balance Volume.""" + direction = np.sign(closes.diff()) + direction.iloc[0] = 0 + return (direction * volumes).cumsum() |