Freqtrade/Simple.py

# 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