add cci and roc

Author: MDUYN

pyindicators/__init__.py

@@ -1,4 +1,4 @@
-from .indicators import sma, rsi, ema, wilders_rsi, adx, \
+from .indicators import sma, rsi, ema, wilders_rsi, adx, roc, cci, \
     crossover, is_crossover, wma, macd, willr, is_crossunder, crossunder, \
     is_lower_low_detected, is_divergence, bollinger_width, \
     is_below, is_above, get_slope, has_any_higher_then_threshold, \
@@ -47,5 +47,7 @@
     'bullish_divergence_multi_dataframe',
     'bollinger_bands',
     'bollinger_width',
-    'atr'
+    'atr',
+    'cci',
+    'roc'
 ]

pyindicators/indicators/__init__.py

@@ -21,6 +21,8 @@
 from .stochastic_oscillator import stochastic_oscillator
 from .average_true_range import atr
 from .bollinger_bands import bollinger_bands, bollinger_width
+from .commodity_channel_index import cci
+from .rate_of_change import roc
 
 __all__ = [
     'sma',
@@ -57,5 +59,7 @@
     'bullish_divergence_multi_dataframe',
     'atr',
     'bollinger_bands',
-    'bollinger_width'
+    'bollinger_width',
+    'cci',
+    'roc'
 ]

pyindicators/indicators/bollinger_bands.py

@@ -64,9 +64,9 @@ def bollinger_width(
         source_column=source_column,
         period=period,
         std_dev=std_dev,
-        middle_band_column='BB_middle_temp',
-        upper_band_column='BB_upper_temp',
-        lower_band_column='BB_lower_temp'
+        middle_band_column_result_column='BB_middle_temp',
+        upper_band_column_result_column='BB_upper_temp',
+        lower_band_column_result_column='BB_lower_temp'
     )
 
     if isinstance(data, PdDataFrame):

pyindicators/indicators/commodity_channel_index.py

@@ -0,0 +1,52 @@
+from typing import Union
+from pandas import DataFrame as PdDataFrame
+from polars import DataFrame as PlDataFrame
+import polars as pl
+from pyindicators.exceptions import PyIndicatorException
+
+
+def cci(
+    data: Union[PdDataFrame, PlDataFrame],
+    high_column='High',
+    low_column='Low',
+    close_column='Close',
+    period=20,
+    result_column='CCI'
+) -> Union[PdDataFrame, PlDataFrame]:
+    """
+    Calculate the Commodity Channel Index (CCI) for a price series.
+
+    Args:
+        data: Input DataFrame (pandas or polars).
+        high_column: Name of the column with high prices.
+        low_column: Name of the column with low prices.
+        close_column: Name of the column with close prices.
+        period: Lookback period for CCI calculation.
+        result_column: Name of the result column to store CCI values.
+
+    Returns the original DataFrame with a new column for CCI.
+    """
+    if isinstance(data, PdDataFrame):
+        # Calculate CCI for pandas DataFrame
+        typical_price = (data[high_column] +
+                         data[low_column] + data[close_column]) / 3
+        sma = typical_price.rolling(window=period).mean()
+        mad = (typical_price - sma).abs().rolling(window=period).mean()
+        data[result_column] = (typical_price - sma) / (0.015 * mad)
+        return data
+
+    elif isinstance(data, PlDataFrame):
+        # Calculate CCI for polars DataFrame
+        typical_price = (pl.col(high_column)
+                         + pl.col(low_column)
+                         + pl.col(close_column)) / 3
+        sma = typical_price.rolling_mean(window_size=period)
+        mad = (typical_price - sma).abs().rolling_mean(window_size=period)
+        return data.with_columns(
+            (typical_price - sma) / (0.015 * mad).alias(result_column)
+        )
+
+    else:
+        raise PyIndicatorException(
+            "Input data must be a pandas or polars DataFrame."
+        )

pyindicators/indicators/rate_of_change.py

@@ -0,0 +1,43 @@
+from typing import Union
+from pandas import DataFrame as PdDataFrame
+from polars import DataFrame as PlDataFrame
+import polars as pl
+from pyindicators.exceptions import PyIndicatorException
+
+
+def roc(
+    data: Union[PdDataFrame, PlDataFrame],
+    source_column='Close',
+    period=14,
+    result_column='ROC'
+) -> Union[PdDataFrame, PlDataFrame]:
+    """
+    Calculate the Rate of Change (ROC) for a price series.
+
+    Args:
+        data: Input DataFrame (pandas or polars).
+        source_column: Name of the column with price data.
+        period: Lookback period for ROC calculation.
+        result_column: Name of the result column to store ROC values.
+
+    Returns:
+        DataFrame with a new column for ROC.
+    """
+    if isinstance(data, PdDataFrame):
+        # Calculate ROC for pandas DataFrame
+        data[result_column] = data[source_column]\
+            .pct_change(periods=period) * 100
+        return data
+
+    elif isinstance(data, PlDataFrame):
+        # Calculate ROC for polars DataFrame
+        return data.with_columns(
+            (
+                pl.col(source_column).pct_change(periods=period) * 100
+            ).alias(result_column)
+        )
+
+    else:
+        raise PyIndicatorException(
+            "Input data must be a pandas or polars DataFrame."
+        )