summaryrefslogtreecommitdiff
path: root/services/strategy-engine/strategies/indicators/momentum.py
blob: c4794529132fa01f7afa49c6c7723aabd563f562 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
"""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