Add stochastic oscillator implementation

Author: MDUYN

README.md

@@ -29,6 +29,7 @@ pip install pyindicators
   * [Weighted Moving Average (WMA)](#weighted-moving-average-wma)
   * [Simple Moving Average (SMA)](#simple-moving-average-sma)
   * [Exponential Moving Average (EMA)](#exponential-moving-average-ema)
+  * [Stochastic Oscillator (STO)](#stochastic-oscillator-sto)
 * [Momentum indicators](#momentum-indicators)
   * [Moving Average Convergence Divergence (MACD)](#moving-average-convergence-divergence-macd)
   * [Relative Strength Index (RSI)](#relative-strength-index-rsi)
@@ -207,6 +208,56 @@ pd_df.tail(10)
 
 ![EMA](https://github.com/coding-kitties/PyIndicators/blob/main/static/images/indicators/ema.png)
 
+#### Stochastic Oscillator (STO)
+The Stochastic Oscillator (STO) is a momentum indicator that compares a particular closing price of an asset to a range of its prices over a certain period. It is used to identify overbought or oversold conditions in a market. The STO consists of two lines: %K and %D, where %K is the main line and %D is the signal line.
+
+```python
+def stochastic_oscillator(
+    data: Union[pd.DataFrame, pl.DataFrame],
+    high_column: str = "High",
+    low_column: str = "Low",
+    close_column: str = "Close",
+    k_period: int = 14,
+    k_slowing: int = 3,
+    d_period: int = 3,
+    result_column: Optional[str] = None
+) -> Union[pd.DataFrame, pl.DataFrame]:
+```
+
+Example
+
+```python
+from investing_algorithm_framework import download
+from pyindicators import stochastic_oscillator
+pl_df = download(
+    symbol="btc/eur",
+    market="binance",
+    time_frame="1d",
+    start_date="2023-12-01",
+    end_date="2023-12-25",
+    save=True,
+    storage_path="./data"
+)
+pd_df = download(
+    symbol="btc/eur",
+    market="binance",
+    time_frame="1d",
+    start_date="2023-12-01",
+    end_date="2023-12-25",
+    pandas=True,
+    save=True,
+    storage_path="./data"
+)
+# Calculate Stochastic Oscillator for Polars DataFrame
+pl_df = stochastic_oscillator(pl_df, high_column="High", low_column="Low", close_column="Close", k_period=14, k_slowing=3, d_period=3, result_column="STO")
+pl_df.show(10)
+# Calculate Stochastic Oscillator for Pandas DataFrame
+pd_df = stochastic_oscillator(pd_df, high_column="High", low_column="Low", close_column="Close", k_period=14, k_slowing=3, d_period=3, result_column="STO")
+pd_df.tail(10)
+```
+
+![STO](https://github.com/coding-kitties/PyIndicators/blob/main/static/images/indicators/sto.png)
+
 ### Momentum Indicators
 
 Indicators that measure the strength and speed of price movements rather than the direction.

pyindicators/__init__.py

@@ -5,7 +5,7 @@
     has_slope_above_threshold, has_any_lower_then_threshold, \
     has_values_above_threshold, has_values_below_threshold, is_down_trend, \
     is_up_trend, up_and_downtrends, detect_peaks, \
-    bearish_divergence, bullish_divergence
+    bearish_divergence, bullish_divergence, stochastic_oscillator
 from .exceptions import PyIndicatorException
 from .date_range import DateRange
 
@@ -41,4 +41,5 @@
     'bearish_divergence',
     'bullish_divergence',
     'is_divergence',
+    'stochastic_oscillator',
 ]

pyindicators/indicators/__init__.py

@@ -17,6 +17,7 @@
 from .up_and_down_trends import up_and_downtrends
 from .divergence import detect_peaks, bearish_divergence, \
     bullish_divergence
+from .stochastic_oscillator import stochastic_oscillator
 
 __all__ = [
     'sma',
@@ -48,4 +49,5 @@
     'bearish_divergence',
     'bullish_divergence',
     'is_divergence',
+    'stochastic_oscillator',
 ]

pyindicators/indicators/stochastic_oscillator.py

@@ -0,0 +1,88 @@
+from typing import Union, Optional
+import pandas as pd
+import polars as pl
+
+
+def stochastic_oscillator(
+    data: Union[pd.DataFrame, pl.DataFrame],
+    high_column: str = "High",
+    low_column: str = "Low",
+    close_column: str = "Close",
+    k_period: int = 14,
+    k_slowing: int = 3,
+    d_period: int = 3,
+    result_column: Optional[str] = None
+) -> Union[pd.DataFrame, pl.DataFrame]:
+    """
+    Calculate the Stochastic Oscillator (%K and %D) for a given DataFrame.
+    The Stochastic Oscillator is a momentum indicator comparing
+    a particular closing price of a security to a range of its
+    prices over a certain period.
+
+    Args:
+        data (Union[pd.DataFrame, pl.DataFrame]): Input DataFrame
+            containing the high, low, and close prices.
+        high_column (str): Name of the column containing high prices.
+        low_column (str): Name of the column containing low prices.
+        close_column (str): Name of the column containing close prices.
+        k_period (int): The period for %K calculation.
+        d_period (int): The period for %D calculation.
+        k_slowing (int): The period for smoothing the %K line.
+        result_column (Optional[str]): Optional prefix for result
+            columns. If None, defaults to "%K" and "%D".
+
+    Returns:
+        Union[pd.DataFrame, pl.DataFrame]: DataFrame
+            with %K and %D columns.
+    """
+
+    k_col = f"{result_column}_%K" if result_column else "%K"
+    d_col = f"{result_column}_%D" if result_column else "%D"
+
+    if isinstance(data, pd.DataFrame):
+        # Fast %K
+        low_min = data[low_column].rolling(
+            window=k_period, min_periods=k_period
+        ).min()
+        high_max = data[high_column].rolling(
+            window=k_period, min_periods=k_period
+        ).max()
+
+        fast_k = 100 * (data[close_column] - low_min) / (high_max - low_min)
+
+        # Slow %K (smoothed Fast %K)
+        slow_k = fast_k.rolling(window=k_slowing, min_periods=k_slowing).mean()
+
+        # %D (smoothed Slow %K)
+        d = slow_k.rolling(window=d_period, min_periods=d_period).mean()
+
+        data[k_col] = slow_k
+        data[d_col] = d
+
+        return data
+
+    elif isinstance(data, pl.DataFrame):
+        # Compute Fast %K
+        low_min = pl.col(low_column).rolling_min(k_period)
+        high_max = pl.col(high_column).rolling_max(k_period)
+
+        fast_k_expr = ((pl.col(close_column) - low_min) /
+                       (high_max - low_min)) * 100
+
+        # Compute Slow %K and %D in steps
+        df = data.with_columns([
+            fast_k_expr.alias("_fast_k")
+        ])
+
+        df = df.with_columns([
+            pl.col("_fast_k").rolling_mean(k_slowing).alias(k_col)
+        ])
+
+        df = df.with_columns([
+            pl.col(k_col).rolling_mean(d_period).alias(d_col)
+        ])
+
+        return df.drop("_fast_k")
+
+    else:
+        raise TypeError("Input data must be a pandas or polars DataFrame.")