calculPlateaux

Zeus_8_3_2_B_4_2.py

@@ -733,6 +733,8 @@ class Zeus_8_3_2_B_4_2(IStrategy):
         #     print("##################")
         #     self.calculateStats(informative, 'sma5_deriv1', 'futur_percent_3')
 
+        self.calculePlateaux(informative, 24, 0.01)
+
         dataframe = merge_informative_pair(dataframe, informative, self.timeframe, "1h", ffill=True)
 
         ################### INFORMATIVE 1d
@@ -785,6 +787,7 @@ class Zeus_8_3_2_B_4_2(IStrategy):
 
         # self.calculateProbabilite2Index(informative, futur_cols, indic_1, indic_2)
 
+        self.calculePlateaux(informative, 14, 0.01)
         dataframe = merge_informative_pair(dataframe, informative, self.timeframe, "1d", ffill=True)
 
         dataframe['last_price'] = dataframe['close']
@@ -2141,3 +2144,50 @@ class Zeus_8_3_2_B_4_2(IStrategy):
             return max_pair == pair or pct_max < - 0.25 or (pct_max_max < - 0.15 and max_pair != pair and days_since_open > 60)
         else:
             return True
+
+    def calculePlateaux(self, informative: pd.DataFrame, plateau_duration, plateau_tolerance) -> pd.DataFrame:
+
+        # 1. Détection plateau
+        informative['rolling_min'] = informative['close'].rolling(plateau_duration).min()
+        informative['rolling_max'] = informative['close'].rolling(plateau_duration).max()
+        informative['plateau_amplitude'] = (informative['rolling_max'] - informative['rolling_min']) / informative['rolling_min']
+        informative['plateau'] = informative['plateau_amplitude'] < plateau_tolerance
+
+        # 2. Détection "fin de plateau"
+        #informative['plateau_end'] = (informative['plateau'] & ~informative['plateau'].shift(-1).fillna(False).astype(bool))
+        next_plateau = informative['plateau'].shift(-1)
+        next_plateau = next_plateau.fillna(False).astype(bool)
+        informative['plateau_end'] = informative['plateau'] & ~next_plateau
+
+
+        # 3. Enregistrer dernier plateau (min/max)
+        last_min = None
+        last_max = None
+        last_status = []
+
+        for i, row in informative.iterrows():
+            if row['plateau_end']:
+                last_min = row['rolling_min']
+                last_max = row['rolling_max']
+
+            if last_min is not None and last_max is not None:
+                if row['close'] > last_max:
+                    breakout = "up"
+                    distance = (row['close'] - last_max) / last_max
+                elif row['close'] < last_min:
+                    breakout = "down"
+                    distance = (last_min - row['close']) / last_min
+                else:
+                    breakout = "inside"
+                    distance = 0
+            else:
+                breakout = None
+                distance = None
+
+            last_status.append((breakout, distance))
+
+        informative['breakout_status'] = [s[0] for s in last_status]
+        informative['breakout_distance'] = [s[1] for s in last_status]
+
+        return informative
+