TEST Simple

Simple.py

@@ -0,0 +1,265 @@
+# Zeus Strategy: First Generation of GodStra Strategy with maximum
+# AVG/MID profit in USDT
+# Author: @Mablue (Masoud Azizi)
+# github: https://github.com/mablue/
+# IMPORTANT: INSTALL TA BEFOUR RUN(pip install ta)
+# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLoss --spaces buy sell roi --strategy Zeus
+# --- Do not remove these libs ---
+from datetime import timedelta, datetime
+from freqtrade.persistence import Trade
+from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, stoploss_from_open,
+                                IntParameter, IStrategy, merge_informative_pair, informative, stoploss_from_absolute)
+import pandas as pd
+import numpy as np
+from pandas import DataFrame
+from typing import Optional, Union, Tuple
+from typing import List
+
+import logging
+import configparser
+from technical import pivots_points
+# --------------------------------
+
+# Add your lib to import here test git
+import ta
+import talib.abstract as talib
+import freqtrade.vendor.qtpylib.indicators as qtpylib
+import requests
+from datetime import timezone, timedelta
+from scipy.signal import savgol_filter
+from ta.trend import SMAIndicator, EMAIndicator, MACD, ADXIndicator
+from collections import Counter
+
+logger = logging.getLogger(__name__)
+
+from tabulate import tabulate
+
+# Couleurs ANSI de base
+RED = "\033[31m"
+GREEN = "\033[32m"
+YELLOW = "\033[33m"
+BLUE = "\033[34m"
+MAGENTA = "\033[35m"
+CYAN = "\033[36m"
+RESET = "\033[0m"
+
+
+def pprint_df(dframe):
+    print(tabulate(dframe, headers='keys', tablefmt='psql', showindex=False))
+
+
+def normalize(df):
+    df = (df - df.min()) / (df.max() - df.min())
+    return df
+
+"""
+SMA z-score derivative strategy with trailing exit (large).
+- timeframe: 1h
+- sma5, relative derivatives, z-score normalization (rolling z over z_window)
+- smoothing: ewm(span=5) on z-scores
+- entry: z_d1_smooth > entry_z1 AND z_d2_smooth > entry_z2
+- exit: inversion (z_d1_smooth < 0) AND retrace from highest since entry >= trailing_stop
+"""
+class Simple(IStrategy):
+    levels = [1, 2, 2, 4, 6, 8, 10, 12, 14, 16, 18, 20]
+    startup_candle_count = 12 * 24 * 2
+
+    # ROI table:
+    minimal_roi = {
+        "0": 10
+    }
+    stakes = 40
+
+    # Stoploss:
+    stoploss = -1  # 0.256
+    # Custom stoploss
+    use_custom_stoploss = True
+
+    # Buy hypers
+    timeframe = '1h'
+
+    max_open_trades = 5
+    max_amount = 40
+
+    # DCA config
+    position_adjustment_enable = True
+
+    # Parameters (tweakable)
+    z_window = 50  # window for rolling mean/std to compute zscore
+    entry_z1 = 0.1  # threshold on z-score of first derivative
+    entry_z2 = 0.1  # threshold on z-score of second derivative
+    min_volume = 0.0  # minimal volume to accept an entry
+    min_relative_d1 = 1e-6  # clip tiny d1 relative values to reduce noise
+
+    # Trailing parameters for "large" trailing requested
+    trailing_stop_pct = 0.05  # 5% retracement from highest since entry
+
+    # Smoothing for z-scores
+    ewm_span = 5
+
+    # Plot config: price + sma5 + markers + slope/accel subplots
+    plot_config = {
+        "main_plot": {
+            "close": {"color": "blue"},
+            "sma5": {"color": "orange"},
+        },
+        "subplots": {
+            "slope_and_accel": {
+                "z_d1": {"color": "green"},
+                "z_d2": {"color": "red"},
+            }
+        },
+        # Markers (Freqtrade charting supports markers via these keys)
+        "markers": [
+            # buy marker: '^' green when enter_long==1
+            {"type": "buy", "column": "enter_long", "marker": "^", "color": "green", "markersize": 10},
+            # sell marker: 'v' red when exit_long==1
+            {"type": "sell", "column": "exit_long", "marker": "v", "color": "red", "markersize": 10},
+        ],
+    }
+
+    def informative_pairs(self):
+        return []
+
+    def _zscore(self, series: pd.Series, window: int) -> pd.Series:
+        mean = series.rolling(window=window, min_periods=3).mean()
+        std = series.rolling(window=window, min_periods=3).std(ddof=0)
+        z = (series - mean) / std
+        z = z.replace([np.inf, -np.inf], np.nan)
+        return z
+
+    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        df = dataframe
+
+        # SMA(5)
+        df['sma5'] = df['close'].rolling(5, min_periods=1).mean()
+
+        # Absolute derivatives
+        df['sma5_d1'] = df['sma5'].diff().rolling(5).mean()
+        df['sma5_d2'] = df['sma5_d1'].diff().rolling(5).mean()
+
+        # Relative derivatives (percentage-like)
+        eps = 1e-9
+        df['sma5_d1_rel'] = df['sma5_d1'] / (df['sma5'].shift(1).replace(0, np.nan) + eps)
+        df['sma5_d2_rel'] = df['sma5_d2'] / (df['sma5'].shift(2).replace(0, np.nan) + eps)
+
+        # Clip micro-noise
+        df.loc[df['sma5_d1_rel'].abs() < self.min_relative_d1, 'sma5_d1_rel'] = 0.0
+        df.loc[df['sma5_d2_rel'].abs() < self.min_relative_d1, 'sma5_d2_rel'] = 0.0
+
+        # Z-scores on relative derivatives
+        df['z_d1'] = self._zscore(df['sma5_d1_rel'], self.z_window)
+        df['z_d2'] = self._zscore(df['sma5_d2_rel'], self.z_window)
+
+        # Smoothing z-scores with EWM to reduce jitter
+        df['z_d1_smooth'] = df['z_d1'].ewm(span=self.ewm_span, adjust=False).mean()
+        df['z_d2_smooth'] = df['z_d2'].ewm(span=self.ewm_span, adjust=False).mean()
+
+        # Prepare marker columns (for plots). They will be filled in populate_entry_trend/populate_exit_trend
+        df['enter_long'] = 0
+        df['exit_long'] = 0
+
+        return df
+
+    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        df = dataframe.copy()
+
+        # Use last closed candle for signals -> shift(1)
+        cond_entry = (
+                (df['z_d1'] > self.entry_z1) &
+                (df['z_d2'] > self.entry_z2) &
+                (df['volume'].shift(1) > self.min_volume)
+        )
+
+        df.loc[cond_entry, 'enter_long'] = 1
+        # Ensure others are explicitly zero (for clean plotting)
+        df.loc[~cond_entry, 'enter_long'] = 0
+
+        return df
+
+    def custom_exit(self, pair: str, trade: Trade, current_time, current_rate, current_profit, **kwargs) -> \
+    Optional[str]:
+        """
+        Exit policy (mode C - trailing large):
+        - Must detect inversion: z_d1_smooth < 0 on last closed candle
+        - Compute highest close since trade entry (inclusive)
+        - If price has retraced >= trailing_stop_pct from that highest point and we're in inversion -> exit
+        """
+
+        df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
+        last = df.iloc[-1].squeeze()
+
+        z1 = last.get('z_d1_smooth', None)
+        if z1 is None:
+            return None
+
+        # Only consider exits when inversion detected (z1 < 0)
+        inversion = (z1 < 0)
+
+        if not inversion:
+            return None
+
+        # If we don't have profit info, be conservative
+        if current_profit is None:
+            return None
+
+        # Determine highest close since entry
+        highest_since_entry = None
+        try:
+            # trade.open_date_utc is available: find rows after that timestamp
+            # df.index is expected to be pd.DatetimeIndex in UTC or local; convert safely
+            if hasattr(trade, 'open_date_utc') and trade.open_date_utc:
+                # pandas comparison: ensure same tz awareness
+                entry_time = pd.to_datetime(trade.open_date_utc)
+                # select rows with index >= entry_time
+                mask = df.index >= entry_time
+                if mask.any():
+                    highest_since_entry = df.loc[mask, 'close'].max()
+            # fallback: use trade.open_rate or the max over full df
+        except Exception as e:
+            logger.debug(f"Couldn't compute highest_since_entry from open_date_utc: {e}")
+
+        if highest_since_entry is None:
+            # fallback: use the maximum close in the entire provided dataframe slice
+            highest_since_entry = df['close'].max() if not df['close'].empty else current_rate
+
+        # Calculate retracement ratio from the highest
+        if highest_since_entry and highest_since_entry > 0:
+            retrace = 1.0 - (current_rate / highest_since_entry)
+        else:
+            retrace = 0.0
+
+        # Exit if:
+        # - currently in profit AND
+        # - retracement >= trailing_stop_pct (i.e. price has fallen enough from top since entry) AND
+        # - inversion detected (z1 < 0)
+        if (current_profit > 0) and (retrace >= self.trailing_stop_pct):
+            # Mark the dataframe for plotting (if possible)
+            # Note: freqtrade expects strategies to set exit flags in populate_exit_trend,
+            # but we set exit via custom_exit return reason; plotting will read exit_long if populated.
+            return "zscore"
+
+        # Otherwise, do not exit yet
+        return None
+
+    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
+        """
+        For plotting only: mark exit points where our logic would trigger.
+        This is approximate: we mark an exit when z_d1_smooth < 0 and the price has retraced >= trailing_stop_pct
+        based on the available dataframe window (best-effort).
+        """
+        df = dataframe
+        # df['exit_long'] = 0
+        #
+        # # compute highest close since each possible entry (best-effort: use rolling max up to current index)
+        # rolling_max = df['close'].cummax()
+        #
+        # # retracement relative to rolling max
+        # retrace = 1.0 - (df['close'] / rolling_max.replace(0, np.nan))
+        #
+        # # mark exits where inversion and retrace >= threshold
+        # cond_exit = (df['z_d1_smooth'] < 0) & (retrace >= self.trailing_stop_pct)
+        # # shift by 0: we mark the candle where the exit condition appears
+        # df.loc[cond_exit, 'exit_long'] = 1
+
+        return df