Add multi df dataframe divergence support

Author: MDUYN

pyindicators/__init__.py

@@ -5,7 +5,8 @@
     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, stochastic_oscillator
+    bearish_divergence, bullish_divergence, stochastic_oscillator, \
+    bearish_divergence_multi_dataframe
 from .exceptions import PyIndicatorException
 from .date_range import DateRange
 
@@ -42,4 +43,5 @@
     'bullish_divergence',
     'is_divergence',
     'stochastic_oscillator',
+    'bearish_divergence_multi_dataframe'
 ]

pyindicators/indicators/__init__.py

@@ -16,7 +16,7 @@
 from .is_up_trend import is_up_trend
 from .up_and_down_trends import up_and_downtrends
 from .divergence import detect_peaks, bearish_divergence, \
-    bullish_divergence
+    bullish_divergence, bearish_divergence_multi_dataframe
 from .stochastic_oscillator import stochastic_oscillator
 
 __all__ = [
@@ -50,4 +50,5 @@
     'bullish_divergence',
     'is_divergence',
     'stochastic_oscillator',
+    'bearish_divergence_multi_dataframe'
 ]

pyindicators/indicators/divergence.py

@@ -462,3 +462,122 @@ def bearish_divergence(
 
     df[result_column] = result
     return pl.DataFrame(df) if is_polars else df
+
+
+def bearish_divergence_multi_dataframe(
+    first_df: Union[pd.DataFrame, pl.DataFrame],
+    second_df: Union[pd.DataFrame, pl.DataFrame],
+    result_df: Union[pd.DataFrame, pl.DataFrame],
+    first_column: str,
+    second_column: str,
+    window_size: int = 1,
+    result_column: str = "bearish_divergence",
+    number_of_neighbors_to_compare: int = 5,
+    min_consecutive: int = 2
+) -> Union[pd.DataFrame, pl.DataFrame]:
+    """
+    Detect bearish divergence between two different DataFrames.
+
+    Args:
+        first_df: DataFrame containing the indicator data (e.g., RSI).
+        second_df: DataFrame containing the price data.
+        result_df: DataFrame used to store results. Must be aligned in time.
+        first_column: Column in first_df (e.g., RSI).
+        second_column: Column in second_df (e.g., price).
+        window_size: Number of bars to consider for pattern.
+        result_column: Output column name.
+        number_of_neighbors_to_compare: For peak detection.
+        min_consecutive: Minimum consecutive peaks required.
+
+    Returns:
+        A DataFrame with a new column indicating bearish divergence.
+    """
+    is_polars = isinstance(first_df, pl.DataFrame)
+
+    if is_polars:
+        first_df = first_df.to_pandas()
+        second_df = second_df.to_pandas()
+        result_df = result_df.to_pandas()
+
+    # Validate columns
+    for df, col, label in [
+        (first_df, first_column, "first_df"),
+        (second_df, second_column, "second_df")
+    ]:
+        if col not in df.columns:
+            raise PyIndicatorException(f"{col} column is missing in {label}")
+
+    # Determine which df has more granular datetime index
+    first_freq = first_df.index.to_series().diff().median()
+    second_freq = second_df.index.to_series().diff().median()
+
+    if first_freq < second_freq:
+        align_index = first_df.index
+    else:
+        align_index = second_df.index
+
+    if len(result_df) != len(align_index):
+        raise PyIndicatorException(
+            "result_df must have the same length as the aligned index"
+        )
+
+    # Reindex all DataFrames to the most granular one
+    first_df = first_df.reindex(align_index)
+    second_df = second_df.reindex(align_index)
+
+    # Peak detection
+    first_highs_col = f"{first_column}_highs"
+    second_highs_col = f"{second_column}_highs"
+
+    if first_highs_col not in first_df.columns:
+        first_df = detect_peaks(
+            first_df,
+            source_column=first_column,
+            number_of_neighbors_to_compare=number_of_neighbors_to_compare,
+            min_consecutive=min_consecutive
+        )
+
+    if second_highs_col not in second_df.columns:
+        second_df = detect_peaks(
+            second_df,
+            source_column=second_column,
+            number_of_neighbors_to_compare=number_of_neighbors_to_compare,
+            min_consecutive=min_consecutive
+        )
+
+    # Now align and merge
+    merged_df = pd.concat([
+        first_df[[first_highs_col]],
+        second_df[[second_highs_col]],
+        result_df.copy()
+    ], axis=1, join='inner')
+
+    # Validate enough data
+    if len(merged_df) < window_size:
+        raise PyIndicatorException(
+            f"Not enough data points (need at least {window_size}, "
+            f"got {len(merged_df)})"
+        )
+
+    # Apply divergence detection
+    indicator_highs = merged_df[first_highs_col].values
+    price_highs = merged_df[second_highs_col].values
+    result = [False] * len(merged_df)
+
+    i = window_size - 1
+    while i < len(merged_df):
+        win_a = indicator_highs[i - window_size + 1:i + 1]
+        win_b = price_highs[i - window_size + 1:i + 1]
+
+        if check_divergence_pattern(win_a, win_b):
+            result[i] = True
+            i += window_size  # Skip forward to avoid overlap
+        else:
+            i += 1
+
+    merged_df[result_column] = result
+
+    # Merge back result column to result_df using the original index
+    result_df[result_column] = merged_df[result_column]
+
+    return pl.DataFrame(result_df) if is_polars else result_df

tests/indicators/test_divergence.py

@@ -1,5 +1,7 @@
+import numpy as np
 from unittest import TestCase
-from pyindicators import is_divergence
+from pyindicators import is_divergence, bearish_divergence_multi_dataframe, \
+    PyIndicatorException
 
 import pandas as pd
 
@@ -100,3 +102,162 @@ def test_detect_bearish_sequence_with_wrong_params_pandas(self):
             "DateTime": pd.date_range("2021-01-01", periods=10, freq="D")
         })
         self.assertFalse(is_divergence(df, window_size=1, number_of_data_points=10, column_one="RSI_highs", column_two="Close_highs"))
+
+
+class TestBearishDivergenceMultiDataFrame(TestCase):
+
+    def test_bearish_divergence_detected(self):
+        # Setup indicator (e.g., RSI) and price (e.g., Close) with divergence
+        indicator = pd.DataFrame({
+            "RSI": [50, 60, 70, 65, 60, 58, 55],
+        }, index=pd.date_range("2022-01-01", periods=7))
+        price = pd.DataFrame({
+            "Close": [100, 102, 105, 108, 110, 112, 115],  # Higher highs
+        }, index=pd.date_range("2022-01-01", periods=7))
+
+        result = pd.DataFrame(index=indicator.index)
+
+        # Force peaks manually for deterministic test
+        indicator["RSI_highs"] = [0, 0, 0, 0, -1, 0, 0]  # Two indicator highs
+        price["Close_highs"] = [0, 0, 0, 0, 1, 0, 0]    # Two price highs
+
+        out = bearish_divergence_multi_dataframe(
+            first_df=indicator,
+            second_df=price,
+            result_df=result,
+            first_column="RSI",
+            second_column="Close",
+            window_size=2,
+            result_column="bearish_divergence"
+        )
+
+        self.assertIn("bearish_divergence", out.columns)
+        self.assertTrue(any(out["bearish_divergence"]))
+
+        indicator = pd.DataFrame({
+            "RSI": [50, 60, 70, 65, 60, 58, 55],
+        }, index=pd.date_range("2022-01-01", periods=7))
+        price = pd.DataFrame({
+            "Close": [100, 102, 105, 108, 110, 112, 115],  # Higher highs
+        }, index=pd.date_range("2022-01-01", periods=7))
+
+        result = pd.DataFrame(index=indicator.index)
+
+        # Force peaks manually for deterministic test
+        indicator["RSI_highs"] = [0, 0, 0, 0, 1, 0, 0]  # Two indicator highs
+        price["Close_highs"] = [0, 0, 0, 0, 1, 0, 0]  # Two price highs
+
+        out = bearish_divergence_multi_dataframe(
+            first_df=indicator,
+            second_df=price,
+            result_df=result,
+            first_column="RSI",
+            second_column="Close",
+            window_size=2,
+            result_column="bearish_divergence"
+        )
+
+        self.assertIn("bearish_divergence", out.columns)
+        self.assertFalse(any(out["bearish_divergence"]))
+
+        indicator = pd.DataFrame({
+            "RSI": [50, 60, 70, 65, 60, 58, 55],
+        }, index=pd.date_range("2022-01-01", periods=7))
+        price = pd.DataFrame({
+            "Close": [100, 102, 105, 108, 110, 112, 115],  # Higher highs
+        }, index=pd.date_range("2022-01-01", periods=7))
+
+        result = pd.DataFrame(index=indicator.index)
+
+        # Force peaks manually for deterministic test
+        indicator["RSI_highs"] = [0, 0, 0, -1, 0, 0, 0]  # Two indicator highs
+        price["Close_highs"] = [0, 0, 0, 0, 0, 1, 0]  # Two price highs
+
+        out = bearish_divergence_multi_dataframe(
+            first_df=indicator,
+            second_df=price,
+            result_df=result,
+            first_column="RSI",
+            second_column="Close",
+            window_size=2,
+            result_column="bearish_divergence"
+        )
+
+        self.assertIn("bearish_divergence", out.columns)
+        self.assertFalse(any(out["bearish_divergence"]))
+
+        out = bearish_divergence_multi_dataframe(
+            first_df=indicator,
+            second_df=price,
+            result_df=result,
+            first_column="RSI",
+            second_column="Close",
+            window_size=3,
+            result_column="bearish_divergence"
+        )
+
+        self.assertIn("bearish_divergence", out.columns)
+        self.assertTrue(any(out["bearish_divergence"]))
+
+    def test_missing_column_exception(self):
+        df1 = pd.DataFrame({"RSI": [50, 60]}, index=pd.date_range("2022-01-01", periods=2))
+        df2 = pd.DataFrame({"Close": [100, 110]}, index=pd.date_range("2022-01-01", periods=2))
+        result = pd.DataFrame(index=df1.index)
+
+        with self.assertRaises(PyIndicatorException):
+            bearish_divergence_multi_dataframe(
+                first_df=df1.drop("RSI", axis=1),
+                second_df=df2,
+                result_df=result,
+                first_column="RSI",
+                second_column="Close"
+            )
+
+    def test_not_enough_data_exception(self):
+        df1 = pd.DataFrame({"RSI": [50]}, index=pd.date_range("2022-01-01", periods=1))
+        df2 = pd.DataFrame({"Close": [100]}, index=pd.date_range("2022-01-01", periods=1))
+        result = pd.DataFrame(index=df1.index)
+
+        # Assume detect_peaks adds _highs column
+        df1["RSI_highs"] = [1]
+        df2["Close_highs"] = [1]
+
+        with self.assertRaises(PyIndicatorException):
+            bearish_divergence_multi_dataframe(
+                first_df=df1,
+                second_df=df2,
+                result_df=result,
+                first_column="RSI",
+                second_column="Close",
+                window_size=3
+            )
+
+    def test_different_timeframes_align_correctly(self):
+        daily_index = pd.date_range("2022-01-01", periods=2, freq="D")
+        indicator_df = pd.DataFrame({
+            "RSI": [65, 60],
+        }, index=daily_index)
+
+        # 2-hour close prices — only some times will match the daily timestamps
+        two_hour_index = pd.date_range("2022-01-01", periods=12, freq="2h")
+        price_df = pd.DataFrame({
+            "Close": [100, 102, 105, 108, 110, 112, 115, 117, 120, 122, 125, 130]
+        }, index=two_hour_index)
+
+        result_df = pd.DataFrame(index=price_df.index)
+
+        # Inject fake peaks
+        indicator_df["RSI_highs"] = [-1, 0]
+        price_df["Close_highs"] = [0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
+
+        result = bearish_divergence_multi_dataframe(
+            first_df=indicator_df,
+            second_df=price_df,
+            result_df=result_df,
+            first_column="RSI",
+            second_column="Close",
+            window_size=2,
+            result_column="bearish_divergence"
+        )
+        self.assertIn("bearish_divergence", result.columns)
+        self.assertTrue(any(result["bearish_divergence"]))