stable/freqtrade/plugins/protections/iprotection.py

import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional

from freqtrade.constants import LongShort
from freqtrade.exchange import timeframe_to_minutes
from freqtrade.misc import plural
from freqtrade.mixins import LoggingMixin
from freqtrade.persistence import LocalTrade


logger = logging.getLogger(__name__)


@dataclass
class ProtectionReturn:
    lock: bool
    until: datetime
    reason: Optional[str]
    lock_side: str = '*'


class IProtection(LoggingMixin, ABC):

    # Can globally stop the bot
    has_global_stop: bool = False
    # Can stop trading for one pair
    has_local_stop: bool = False

    def __init__(self, config: Dict[str, Any], protection_config: Dict[str, Any]) -> None:
        self._config = config
        self._protection_config = protection_config
        self._stop_duration_candles: Optional[int] = None
        self._lookback_period_candles: Optional[int] = None

        tf_in_min = timeframe_to_minutes(config['timeframe'])
        if 'stop_duration_candles' in protection_config:
            self._stop_duration_candles = int(protection_config.get('stop_duration_candles', 1))
            self._stop_duration = (tf_in_min * self._stop_duration_candles)
        else:
            self._stop_duration_candles = None
            self._stop_duration = protection_config.get('stop_duration', 60)
        if 'lookback_period_candles' in protection_config:
            self._lookback_period_candles = int(protection_config.get('lookback_period_candles', 1))
            self._lookback_period = tf_in_min * self._lookback_period_candles
        else:
            self._lookback_period_candles = None
            self._lookback_period = int(protection_config.get('lookback_period', 60))

        LoggingMixin.__init__(self, logger)

    @property
    def name(self) -> str:
        return self.__class__.__name__

    @property
    def stop_duration_str(self) -> str:
        """
        Output configured stop duration in either candles or minutes
        """
        if self._stop_duration_candles:
            return (f"{self._stop_duration_candles} "
                    f"{plural(self._stop_duration_candles, 'candle', 'candles')}")
        else:
            return (f"{self._stop_duration} "
                    f"{plural(self._stop_duration, 'minute', 'minutes')}")

    @property
    def lookback_period_str(self) -> str:
        """
        Output configured lookback period in either candles or minutes
        """
        if self._lookback_period_candles:
            return (f"{self._lookback_period_candles} "
                    f"{plural(self._lookback_period_candles, 'candle', 'candles')}")
        else:
            return (f"{self._lookback_period} "
                    f"{plural(self._lookback_period, 'minute', 'minutes')}")

    @abstractmethod
    def short_desc(self) -> str:
        """
        Short method description - used for startup-messages
        -> Please overwrite in subclasses
        """

    @abstractmethod
    def global_stop(self, date_now: datetime, side: LongShort) -> Optional[ProtectionReturn]:
        """
        Stops trading (position entering) for all pairs
        This must evaluate to true for the whole period of the "cooldown period".
        """

    @abstractmethod
    def stop_per_pair(
            self, pair: str, date_now: datetime, side: LongShort) -> Optional[ProtectionReturn]:
        """
        Stops trading (position entering) for this pair
        This must evaluate to true for the whole period of the "cooldown period".
        :return: Tuple of [bool, until, reason].
            If true, this pair will be locked with <reason> until <until>
        """

    @staticmethod
    def calculate_lock_end(trades: List[LocalTrade], stop_minutes: int) -> datetime:
        """
        Get lock end time
        """
        max_date: datetime = max([trade.close_date for trade in trades if trade.close_date])
        # comming from Database, tzinfo is not set.
        if max_date.tzinfo is None:
            max_date = max_date.replace(tzinfo=timezone.utc)

        until = max_date + timedelta(minutes=stop_minutes)

        return until