Add MACD and WMA indicators

Author: MDUYN

README.md

@@ -8,7 +8,7 @@ PyIndicators is a powerful and user-friendly Python library for technical analys
     <picture>
     <source media="(prefers-color-scheme: dark)" srcset="static/sponsors/finterion-dark.png">
     <source media="(prefers-color-scheme: light)" srcset="static/sponsors/finterion-light.png">
-    <img src="static/sponsors/finterion-light.svg" alt="Finterion Logo" style="height: 55px;">
+    <img src="static/sponsors/finterion-light.svg" alt="Finterion Logo" style="height: 40px;">
     </picture>
 </a>
 
@@ -26,8 +26,10 @@ pip install pyindicators
 * Dataframe first approach, with support for both pandas dataframes and polars dataframes
 * Supports python version 3.9 and above.
 * [Trend indicators](#trend-indicators)
+  * [Weighted Moving Average (WMA)](#weighted-moving-average-wma)
   * [Simple Moving Average (SMA)](#simple-moving-average-sma)
   * [Exponential Moving Average (EMA)](#exponential-moving-average-ema)
+  * [Moving Average Convergence Divergence (MACD)](#moving-average-convergence-divergence-macd)
 * [Momentum indicators](#momentum-indicators)
   * [Relative Strength Index (RSI)](#relative-strength-index-rsi)
   * [Relative Strength Index Wilders method (Wilders RSI)](#wilders-relative-strength-index-wilders-rsi)
@@ -39,6 +41,31 @@ pip install pyindicators
 
 ### Trend Indicators
 
+#### Weighted Moving Average (WMA)
+
+```python
+from investing_algorithm_framework import CSVOHLCVMarketDataSource
+
+from pyindicators import wma
+
+# For this example the investing algorithm framework is used for dataframe creation,
+csv_path = "./tests/test_data/OHLCV_BTC-EUR_BINANCE_15m_2023-12-01:00:00_2023-12-25:00:00.csv"
+data_source = CSVOHLCVMarketDataSource(csv_file_path=csv_path)
+
+pl_df = data_source.get_data()
+pd_df = data_source.get_data(pandas=True)
+
+# Calculate SMA for Polars DataFrame
+pl_df = wma(pl_df, source_column="Close", period=200, result_column="SMA_200")
+pl_df.show(10)
+
+# Calculate SMA for Pandas DataFrame
+pd_df = wma(pd_df, source_column="Close", period=200, result_column="SMA_200")
+pd_df.tail(10)
+```
+
+![WMA](https://github.com/coding-kitties/PyIndicators/blob/main/static/images/indicators/wma.png)
+
 #### Simple Moving Average (SMA)
 
 Smooth out price data to identify trend direction.
@@ -93,6 +120,33 @@ pd_df.tail(10)
 
 ![EMA](https://github.com/coding-kitties/PyIndicators/blob/main/static/images/indicators/ema.png)
 
+#### Moving Average Convergence Divergence (MACD)
+
+```python
+from investing_algorithm_framework import CSVOHLCVMarketDataSource
+
+from pyindicators import macd
+
+# For this example the investing algorithm framework is used for dataframe creation,
+csv_path = "./tests/test_data/OHLCV_BTC-EUR_BINANCE_15m_2023-12-01:00:00_2023-12-25:00:00.csv"
+data_source = CSVOHLCVMarketDataSource(csv_file_path=csv_path)
+
+pl_df = data_source.get_data()
+pd_df = data_source.get_data(pandas=True)
+
+# Calculate MACD for Polars DataFrame
+pl_df = macd(pl_df, source_column="Close", short_period=12, long_period=26, signal_period=9)
+
+# Calculate MACD for Pandas DataFrame
+pd_df = macd(pd_df, source_column="Close", short_period=12, long_period=26, signal_period=9)
+
+pl_df.show(10)
+pd_df.tail(10)
+```
+
+![EMA](https://github.com/coding-kitties/PyIndicators/blob/main/static/images/indicators/macd.png)
+
+
 ### Momentum Indicators
 
 #### Relative Strength Index (RSI)

pyindicators/__init__.py

@@ -1,13 +1,15 @@
 from .indicators import sma, rsi, is_crossover, crossunder, ema, wilders_rsi, \
-    crossover, is_crossover
+    crossover, is_crossover, wma, macd
 
 __all__ = [
     'sma',
+    'wma',
     'is_crossover',
     'crossunder',
     'crossover',
     'is_crossover',
     'ema',
     'rsi',
-    "wilders_rsi"
+    "wilders_rsi",
+    'macd'
 ]

pyindicators/indicators/__init__.py

@@ -1,15 +1,19 @@
 from .simple_moving_average import sma
+from .weighted_moving_average import wma
 from .crossover import is_crossover, crossover
 from .crossunder import crossunder
 from .exponential_moving_average import ema
 from .rsi import rsi, wilders_rsi
+from .macd import macd
 
 __all__ = [
     'sma',
+    "wma",
     'is_crossover',
     "crossover",
     'crossunder',
     'ema',
     'rsi',
     'wilders_rsi',
+    'macd'
 ]

pyindicators/indicators/macd.py

@@ -0,0 +1,64 @@
+from typing import Union
+
+import numpy as np
+from pandas import DataFrame as PdDataFrame
+from polars import DataFrame as PlDataFrame
+import pandas as pd
+import polars as pl
+
+from pyindicators.exceptions import PyIndicatorException
+from pyindicators.indicators import ema
+
+
+def macd(
+    data: Union[PdDataFrame, PlDataFrame],
+    source_column: str,
+    short_period: int = 12,
+    long_period: int = 26,
+    signal_period: int = 9,
+    macd_column: str = "macd",
+    signal_column: str = "macd_signal",
+    histogram_column: str = "macd_histogram"
+) -> Union[PdDataFrame, PlDataFrame]:
+    """
+    Calculate the MACD (Moving Average Convergence Divergence) for a given DataFrame.
+
+    Args:
+        data (Union[pd.DataFrame, pl.DataFrame]): Input data containing the price series.
+        source_column (str): Column name for the price series.
+        short_period (int, optional): Period for the short-term EMA (default: 12).
+        long_period (int, optional): Period for the long-term EMA (default: 26).
+        signal_period (int, optional): Period for the Signal Line EMA (default: 9).
+        macd_column (str, optional): Column name to store the MACD line.
+        signal_column (str, optional): Column name to store the Signal line.
+        histogram_column (str, optional): Column name to store the MACD histogram.
+
+    Returns:
+        Union[pd.DataFrame, pl.DataFrame]: DataFrame with MACD, Signal Line, and Histogram.
+    """
+    if source_column not in data.columns:
+        raise PyIndicatorException(
+            f"Column '{source_column}' not found in DataFrame"
+        )
+
+    if isinstance(data, PdDataFrame):
+        # Calculate the short-term and long-term EMAs
+        data = ema(data, source_column, short_period, f"EMA_{short_period}")
+        data = ema(data, source_column, long_period, f"EMA_{long_period}")
+
+        # Calculate the MACD line
+        data[macd_column] = \
+            data[f"EMA_{short_period}"] - data[f"EMA_{long_period}"]
+
+        # Calculate the Signal Line
+        data = ema(data, macd_column, signal_period, signal_column)
+
+        # Calculate the MACD Histogram
+        data[histogram_column] = data[macd_column] - data[signal_column]
+        return data
+    elif isinstance(data, pl.DataFrame):
+        return None
+    else:
+        raise PyIndicatorException(
+            "Unsupported DataFrame type. Use Pandas or Polars."
+        )

pyindicators/indicators/weighted_moving_average.py

@@ -0,0 +1,73 @@
+from typing import Union
+
+import numpy as np
+from pandas import DataFrame as PdDataFrame
+from polars import DataFrame as PlDataFrame
+import pandas as pd
+import polars as pl
+
+from pyindicators.exceptions import PyIndicatorException
+
+
+def wma(
+    data: Union[PdDataFrame, PlDataFrame],
+    source_column: str,
+    period: int,
+    result_column: str = None,
+) -> Union[PdDataFrame, PlDataFrame]:
+    """
+    Function to calculate the weighted moving average of a series.
+
+    Args:
+        data (Union[PdDataFrame, PlDataFrame]): The input data.
+        source_column (str): The name of the series.
+        period (int): The period for the simple moving average.
+        result_column (str, optional): The name of the column to store the
+            simple moving average. Defaults to None.
+
+    Returns:
+        Union[PdDataFrame, PlDataFrame]: Returns a DataFrame
+            with the weighted moving average of the series.
+    """
+    if len(data) < period:
+        raise PyIndicatorException(
+            "The data must be larger than the period " +
+            f"{period} to calculate the WMA. The data " +
+            f"only contains {len(data)} data points."
+        )
+    if result_column is None:
+        result_column = f"WMA_{period}"
+
+    weights = np.arange(1, period + 1)
+
+    if isinstance(data, pd.DataFrame):
+        if source_column not in data.columns:
+            raise PyIndicatorException(
+                f"Column '{source_column}' not found in DataFrame"
+            )
+
+        data[result_column] = (
+            data[source_column]
+            .rolling(window=period)
+            .apply(lambda x: np.dot(x, weights) / weights.sum(), raw=True)
+        )
+        return data
+
+    elif isinstance(data, pl.DataFrame):
+        if source_column not in data.columns:
+            raise PyIndicatorException(
+                f"Column '{source_column}' not found in DataFrame"
+            )
+
+        wma_values = (
+            data[source_column]
+            .rolling_mean(window_size=period, weights=weights.tolist())
+        )
+
+        data = data.with_columns(pl.Series(result_column, wma_values))
+        return data
+
+    else:
+        raise PyIndicatorException(
+            "Unsupported DataFrame type. Use Pandas or Polars."
+        )

static/images/indicators/macd.png

@@ -0,0 +1,79 @@
+import pandas as pd
+import polars as pl
+import pandas.testing as pdt
+from polars.testing import assert_frame_equal
+
+from tests.resources import TestBaseline
+from pyindicators import wma
+
+
+class Test(TestBaseline):
+    correct_output_csv_filename = \
+        "WMA_200_BTC-EUR_BINANCE_15m_2023-12-01-00-00_2023-12-25-00-00.csv"
+
+    def generate_pandas_df(self, polars_source_df):
+        polars_source_df = wma(
+            data=polars_source_df,
+            period=200,
+            result_column="WMA_200",
+            source_column="Close"
+        )
+        return polars_source_df
+
+    def generate_polars_df(self, pandas_source_df):
+        pandas_source_df = wma(
+            data=pandas_source_df,
+            period=200,
+            result_column="WMA_200",
+            source_column="Close"
+        )
+        return pandas_source_df
+
+    def test_comparison_pandas(self):
+
+        # Load the correct output in a pandas dataframe
+        correct_output_pd = pd.read_csv(self.get_correct_output_csv_path())
+
+        # Load the source in a pandas dataframe
+        source = pd.read_csv(self.get_source_csv_path())
+
+        # Generate the pandas dataframe
+        output = self.generate_pandas_df(source)
+        output = output[correct_output_pd.columns]
+        output["Datetime"] = \
+            pd.to_datetime(output["Datetime"]).dt.tz_localize(None)
+        correct_output_pd["Datetime"] = \
+            pd.to_datetime(correct_output_pd["Datetime"]).dt.tz_localize(None)
+
+        pdt.assert_frame_equal(correct_output_pd, output)
+
+    def test_comparison_polars(self):
+
+        # Load the correct output in a polars dataframe
+        correct_output_pl = pl.read_csv(
+            self.get_correct_output_csv_path(),
+            schema_overrides={"WMA_200": pl.Float64}
+        )
+
+        # Load the source in a polars dataframe
+        source = pl.read_csv(self.get_source_csv_path())
+
+        # Generate the polars dataframe
+        output = self.generate_polars_df(source)
+
+        # Convert the datetime columns to datetime
+        # Convert the 'Datetime' column in both DataFrames to datetime
+        output = output.with_columns(
+            pl.col("Datetime").str.strptime(pl.Datetime).alias("Datetime")
+        )
+
+        correct_output_pl = correct_output_pl.with_columns(
+            pl.col("Datetime").str.strptime(pl.Datetime).alias("Datetime")
+        )
+        output = output[correct_output_pl.columns]
+        output = self.make_polars_column_datetime_naive(output, "Datetime")
+        correct_output_pl = self.make_polars_column_datetime_naive(
+            correct_output_pl, "Datetime"
+        )
+
+        assert_frame_equal(correct_output_pl, output)