Added numba implementation for smc.ob method named smc.ob_numba

setup.py

@@ -18,8 +18,21 @@
     long_description_content_type="text/markdown",
     long_description=LONG_DESCRIPTION,
     packages=["smartmoneyconcepts"],
-    install_requires=["pandas==2.0.2", "numpy==1.24.3"],
-    keywords=['smart', 'money', 'concepts', 'ict', 'indicators', 'trading', 'forex', 'stocks', 'crypto', 'order', 'blocks', 'liquidity'],
+    install_requires=["pandas==2.0.2", "numpy==1.24.3", "numba==0.59.1"],
+    keywords=[
+        "smart",
+        "money",
+        "concepts",
+        "ict",
+        "indicators",
+        "trading",
+        "forex",
+        "stocks",
+        "crypto",
+        "order",
+        "blocks",
+        "liquidity",
+    ],
     url="https://github.com/joshyattridge/smartmoneyconcepts",
     classifiers=[
         "Development Status :: 1 - Planning",
@@ -28,5 +41,5 @@
         "Operating System :: Unix",
         "Operating System :: MacOS :: MacOS X",
         "Operating System :: Microsoft :: Windows",
-    ]
+    ],
 )

smartmoneyconcepts/smc.py

@@ -3,7 +3,7 @@
 import numpy as np
 from pandas import DataFrame, Series
 from datetime import datetime
-
+import numba as nb
 
 def inputvalidator(input_="ohlc"):
     def dfcheck(func):
@@ -49,6 +49,170 @@ def decorate(cls):
     return decorate
 
 
+@nb.jit(nopython=True)
+def calculate_ob_with_numba(ohlc_arr, swing_highs_lows_arr, close_mitigation):
+
+    ohlc_len = len(ohlc_arr)
+    swing_highs_lows_len = len(swing_highs_lows_arr)
+    _open = 0
+    _high = 1
+    _low = 2
+    _close = 3
+    _volume = 4
+    _highlow = 0
+
+    crossed = np.full(ohlc_len, False, dtype=np.bool_)
+    ob = np.zeros(ohlc_len, dtype=np.int32)
+    top = np.zeros(ohlc_len, dtype=np.float32)
+    bottom = np.zeros(ohlc_len, dtype=np.float32)
+    obVolume = np.zeros(ohlc_len, dtype=np.float32)
+    lowVolume = np.zeros(ohlc_len, dtype=np.float32)
+    highVolume = np.zeros(ohlc_len, dtype=np.float32)
+    percentage = np.zeros(ohlc_len, dtype=np.float32)
+    mitigated_index = np.zeros(ohlc_len, dtype=np.int32)
+    breaker = np.full(ohlc_len, False, dtype=np.bool_)
+
+    for i in range(ohlc_len):
+        close_index = i
+        close_price = ohlc_arr[i, _close]
+
+        # Bullish Order Block
+        if np.any(ob == 1):
+            for j in range(ohlc_len - 1, -1, -1):
+                if ob[j] == 1:
+                    currentOB = j
+                    if breaker[currentOB]:
+                        if ohlc_arr[i, _high] > top[currentOB]:
+                            ob[j] = top[j] = bottom[j] = obVolume[j] = lowVolume[j] = (
+                                highVolume[j]
+                            ) = mitigated_index[j] = percentage[j] = 0.0
+
+                    elif (
+                        not close_mitigation and ohlc_arr[i, _low] < bottom[currentOB]
+                    ) or (
+                        close_mitigation
+                        and np.min(np.array([ohlc_arr[i, _open], ohlc_arr[i, _close]]))
+                        < bottom[currentOB]
+                    ):
+                        breaker[currentOB] = True
+                        mitigated_index[currentOB] = close_index - 1
+
+        last_top_index = -1
+        for j in range(swing_highs_lows_len):
+            if swing_highs_lows_arr[j, _highlow] == 1 and j < close_index:
+                last_top_index = j
+
+        if last_top_index != -1:
+            swing_top_price = ohlc_arr[last_top_index, _high]
+
+            if close_price > swing_top_price and not crossed[last_top_index]:
+                crossed[last_top_index] = True
+                obBtm = ohlc_arr[close_index - 1, _high]
+                obTop = ohlc_arr[close_index - 1, _low]
+                obIndex = close_index - 1
+                for j in range(1, close_index - last_top_index):
+                    obBtm = np.min(
+                        np.array([ohlc_arr[last_top_index + j, _low], obBtm])
+                    )
+                    if obBtm == ohlc_arr[last_top_index + j, _low]:
+                        obTop = ohlc_arr[last_top_index + j, _high]
+                    obIndex = (
+                        last_top_index + j
+                        if obBtm == ohlc_arr[last_top_index + j, _low]
+                        else obIndex
+                    )
+
+                ob[obIndex] = 1
+                top[obIndex] = obTop
+                bottom[obIndex] = obBtm
+                obVolume[obIndex] = np.sum(ohlc_arr[i - 2 : i + 1, 4])
+                lowVolume[obIndex] = ohlc_arr[close_index - 2, _volume]
+                highVolume[obIndex] = np.sum(ohlc_arr[i - 1 : i + 1, 4])
+                percentage[obIndex] = (
+                    np.min(np.array([highVolume[obIndex], lowVolume[obIndex]]))
+                    / np.max(np.array([highVolume[obIndex], lowVolume[obIndex]]))
+                    if np.max(np.array([highVolume[obIndex], lowVolume[obIndex]])) != 0
+                    else 1
+                ) * 100.0
+
+    for i in range(ohlc_len):
+        close_index = i
+        close_price = ohlc_arr[i, _close]
+
+        # Bearish Order Block
+        if np.any(ob == -1):
+            for j in range(ohlc_len - 1, -1, -1):
+                if ob[j] == -1:
+                    currentOB = j
+                    if breaker[currentOB]:
+                        if ohlc_arr[i, _low] < bottom[currentOB]:
+                            ob[j] = top[j] = bottom[j] = obVolume[j] = lowVolume[j] = (
+                                highVolume[j]
+                            ) = mitigated_index[j] = percentage[j] = 0.0
+
+                        elif (
+                            not close_mitigation and ohlc_arr[i, _high] > top[currentOB]
+                        ) or (
+                            close_mitigation
+                            and np.max(
+                                np.array([ohlc_arr[i, _open], ohlc_arr[i, _close]])
+                            )
+                            > top[currentOB]
+                        ):
+                            breaker[currentOB] = True
+                            mitigated_index[currentOB] = close_index
+
+        last_btm_index = -1
+        for j in range(swing_highs_lows_len):
+            if swing_highs_lows_arr[j, _highlow] == -1 and j < close_index:
+                last_btm_index = j
+
+        if last_btm_index != -1:
+            swing_btm_price = ohlc_arr[last_btm_index, _low]
+
+            if close_price < swing_btm_price and not crossed[last_btm_index]:
+                crossed[last_btm_index] = True
+                obBtm = ohlc_arr[i - 1, _low]
+                obTop = ohlc_arr[i - 1, _high]
+                obIndex = close_index - 1
+                for j in range(1, close_index - last_btm_index):
+                    obTop = np.max(
+                        np.array([ohlc_arr[last_btm_index + j, _high], obTop])
+                    )
+                    obBtm = (
+                        ohlc_arr[last_btm_index + j, _low]
+                        if obTop == ohlc_arr[last_btm_index + j, _high]
+                        else obBtm
+                    )
+                    obIndex = (
+                        last_btm_index + j
+                        if obTop == ohlc_arr[last_btm_index + j, _high]
+                        else obIndex
+                    )
+
+                ob[obIndex] = -1
+                top[obIndex] = obTop
+                bottom[obIndex] = obBtm
+                obVolume[obIndex] = np.sum(ohlc_arr[i - 2 : i + 1, 4])
+                lowVolume[obIndex] = np.sum(ohlc_arr[i - 1 : i + 1, 4])
+                highVolume[obIndex] = ohlc_arr[close_index - 2, _volume]
+                percentage[obIndex] = (
+                    np.min(np.array([highVolume[obIndex], lowVolume[obIndex]]))
+                    / np.max(np.array([highVolume[obIndex], lowVolume[obIndex]]))
+                    if np.max(np.array([highVolume[obIndex], lowVolume[obIndex]])) != 0
+                    else 1
+                ) * 100.0
+
+    ob = np.where(ob != 0, ob, np.nan)
+    top = np.where(~np.isnan(ob), top, np.nan)
+    bottom = np.where(~np.isnan(ob), bottom, np.nan)
+    obVolume = np.where(~np.isnan(ob), obVolume, np.nan)
+    mitigated_index = np.where(~np.isnan(ob), mitigated_index, np.nan)
+    percentage = np.where(~np.isnan(ob), percentage, np.nan)
+
+    return [ob, top, bottom, obVolume, mitigated_index, percentage]
+
+
 @apply(inputvalidator(input_="ohlc"))
 class smc:
     __version__ = "0.0.21"
@@ -576,6 +740,50 @@ def ob(
             axis=1,
         )
 
+    @classmethod
+    def ob_numba(
+        cls,
+        ohlc: DataFrame,
+        swing_highs_lows: DataFrame,
+        close_mitigation: bool = False,
+    ) -> DataFrame:
+        """
+        ob_numba - Order Blocks using numba
+        Numba is an open source JIT compiler that translates a subset of Python and NumPy code into fast machine code.
+        First usage might take more time becuase of initial compilation of python code to machine code, so please use smaller dataset for initial run, in subsequent runs it is very fast in the order of 100x in some cases.
+
+        Just like OB, this method also detects order blocks when there is a high amount of market orders exist on a price range.
+        Results might very slightly than OB.
+
+        parameters:
+        ohlc: DataFrame - OHLCV data
+        swing_highs_lows: DataFrame - provide the dataframe from the swing_highs_lows function
+        close_mitigation: bool - if True then the order block will be mitigated based on the close of the candle otherwise it will be the high/low.
+
+        returns:
+        OB = 1 if bullish order block, -1 if bearish order block
+        Top = top of the order block
+        Bottom = bottom of the order block
+        OBVolume = volume + 2 last volumes amounts
+        Percentage = strength of order block (min(highVolume, lowVolume)/max(highVolume,lowVolume))
+        """
+
+        swing_highs_lows = swing_highs_lows.copy()
+        ohlc_arr = ohlc[["open", "high", "low", "close", "volume"]].values
+        swing_highs_lows_arr = swing_highs_lows.values
+
+        result = calculate_ob_with_numba(
+            ohlc_arr=ohlc_arr,
+            swing_highs_lows_arr=swing_highs_lows_arr,
+            close_mitigation=close_mitigation,
+        )
+
+        df = pd.DataFrame(result)
+        df = df.transpose()
+        df.columns = ["OB", "Top", "Bottom", "OBVolume", "MitigatedIndex", "Percentage"]
+
+        return df
+
     @classmethod
     def liquidity(
         cls, ohlc: DataFrame, swing_highs_lows: DataFrame, range_percent: float = 0.01