fix(qav3): filter noisy reversal identification in zigzag()

Signed-off-by: Jérôme Benoit <jerome.benoit@piment-noir.org>

diff --git a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
index 831045951eb12535026be07fd51d20aaf2a9f572..db24a64f68b25bbd0e026279ee4b1fef1ac09fd8 100644 (file)
--- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
+++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py
@@ -915,12 +915,52 @@ def zigzag(
         if len(next_closes) == 0 or len(previous_closes) == 0:
             return False
 
+        next_slope_ok = True
+        if len(next_closes) >= 2:
+            next_slope = np.polyfit(range(len(next_closes)), next_closes, 1)[0]
+            if direction == TrendDirection.DOWN:
+                next_slope_ok = next_slope < 0
+            elif direction == TrendDirection.UP:
+                next_slope_ok = next_slope > 0
+        previous_slope_ok = True
+        if len(previous_closes) >= 2:
+            previous_slope = np.polyfit(
+                range(len(previous_closes)), previous_closes, 1
+            )[0]
+            if direction == TrendDirection.DOWN:
+                previous_slope_ok = previous_slope > 0
+            elif direction == TrendDirection.UP:
+                previous_slope_ok = previous_slope < 0
+
+        previous_timing_ok = True
+        if direction == TrendDirection.DOWN:
+            if len(previous_highs) >= 1:
+                previous_timing_ok = (
+                    highs[previous_slice].argmax() >= len(previous_highs) // 2
+                )
+        elif direction == TrendDirection.UP:
+            if len(previous_lows) >= 1:
+                previous_timing_ok = (
+                    lows[previous_slice].argmin() <= len(previous_lows) // 2
+                )
+        next_timing_ok = True
+        if direction == TrendDirection.DOWN:
+            if len(next_lows) >= 1:
+                next_timing_ok = lows[next_slice].argmin() >= len(next_lows) // 2
+        elif direction == TrendDirection.UP:
+            if len(next_highs) >= 1:
+                next_timing_ok = highs[next_slice].argmax() >= len(next_highs) // 2
+
         if direction == TrendDirection.DOWN:
             return (
                 np.all(next_closes < highs[pos])
                 and np.all(previous_closes < highs[pos])
                 and np.max(next_highs) <= highs[pos]
                 and np.max(previous_highs) <= highs[pos]
+                and next_timing_ok
+                and previous_timing_ok
+                and next_slope_ok
+                and previous_slope_ok
             )
         elif direction == TrendDirection.UP:
             return (
@@ -928,6 +968,10 @@ def zigzag(
                 and np.all(previous_closes > lows[pos])
                 and np.min(next_lows) >= lows[pos]
                 and np.min(previous_lows) >= lows[pos]
+                and next_timing_ok
+                and previous_timing_ok
+                and next_slope_ok
+                and previous_slope_ok
             )
         return False
 

diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py
index 05eaf7d30a339492b83946b4b01dfd82ebc0fdbf..36bcc89dbae108835d5748310e48b88ba777889d 100644 (file)
--- a/quickadapter/user_data/strategies/QuickAdapterV3.py
+++ b/quickadapter/user_data/strategies/QuickAdapterV3.py
@@ -58,7 +58,7 @@ class QuickAdapterV3(IStrategy):
     INTERFACE_VERSION = 3
 
     def version(self) -> str:
-        return "3.3.25"
+        return "3.3.26"
 
     timeframe = "5m"
 

diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py
index c19316cb4a6a49f710d47aa2da87300cdd1015a9..e05bd060b654e5ae78c85be2bc6857e288e601d8 100644 (file)
--- a/quickadapter/user_data/strategies/Utils.py
+++ b/quickadapter/user_data/strategies/Utils.py
@@ -409,12 +409,52 @@ def zigzag(
         if len(next_closes) == 0 or len(previous_closes) == 0:
             return False
 
+        next_slope_ok = True
+        if len(next_closes) >= 2:
+            next_slope = np.polyfit(range(len(next_closes)), next_closes, 1)[0]
+            if direction == TrendDirection.DOWN:
+                next_slope_ok = next_slope < 0
+            elif direction == TrendDirection.UP:
+                next_slope_ok = next_slope > 0
+        previous_slope_ok = True
+        if len(previous_closes) >= 2:
+            previous_slope = np.polyfit(
+                range(len(previous_closes)), previous_closes, 1
+            )[0]
+            if direction == TrendDirection.DOWN:
+                previous_slope_ok = previous_slope > 0
+            elif direction == TrendDirection.UP:
+                previous_slope_ok = previous_slope < 0
+
+        previous_timing_ok = True
+        if direction == TrendDirection.DOWN:
+            if len(previous_highs) >= 1:
+                previous_timing_ok = (
+                    highs[previous_slice].argmax() >= len(previous_highs) // 2
+                )
+        elif direction == TrendDirection.UP:
+            if len(previous_lows) >= 1:
+                previous_timing_ok = (
+                    lows[previous_slice].argmin() <= len(previous_lows) // 2
+                )
+        next_timing_ok = True
+        if direction == TrendDirection.DOWN:
+            if len(next_lows) >= 1:
+                next_timing_ok = lows[next_slice].argmin() >= len(next_lows) // 2
+        elif direction == TrendDirection.UP:
+            if len(next_highs) >= 1:
+                next_timing_ok = highs[next_slice].argmax() >= len(next_highs) // 2
+
         if direction == TrendDirection.DOWN:
             return (
                 np.all(next_closes < highs[pos])
                 and np.all(previous_closes < highs[pos])
                 and np.max(next_highs) <= highs[pos]
                 and np.max(previous_highs) <= highs[pos]
+                and next_timing_ok
+                and previous_timing_ok
+                and next_slope_ok
+                and previous_slope_ok
             )
         elif direction == TrendDirection.UP:
             return (
@@ -422,6 +462,10 @@ def zigzag(
                 and np.all(previous_closes > lows[pos])
                 and np.min(next_lows) >= lows[pos]
                 and np.min(previous_lows) >= lows[pos]
+                and next_timing_ok
+                and previous_timing_ok
+                and next_slope_ok
+                and previous_slope_ok
             )
         return False