diff --git a/freqtrade/plugins/protectionmanager.py b/freqtrade/plugins/protectionmanager.py index 6e55ade11..4f60ae0e0 100644 --- a/freqtrade/plugins/protectionmanager.py +++ b/freqtrade/plugins/protectionmanager.py @@ -1,6 +1,7 @@ """ Protection manager class """ + import logging from datetime import datetime, timezone from typing import Dict, List, Optional @@ -16,14 +17,13 @@ logger = logging.getLogger(__name__) class ProtectionManager: - def __init__(self, config: Config, protections: List) -> None: self._config = config self._protection_handlers: List[IProtection] = [] for protection_handler_config in protections: protection_handler = ProtectionResolver.load_protection( - protection_handler_config['method'], + protection_handler_config["method"], config=config, protection_config=protection_handler_config, ) @@ -45,8 +45,9 @@ class ProtectionManager: """ return [{p.name: p.short_desc()} for p in self._protection_handlers] - def global_stop(self, now: Optional[datetime] = None, - side: LongShort = 'long') -> Optional[PairLock]: + def global_stop( + self, now: Optional[datetime] = None, side: LongShort = "long" + ) -> Optional[PairLock]: if not now: now = datetime.now(timezone.utc) result = None @@ -56,20 +57,22 @@ class ProtectionManager: if lock and lock.until: if not PairLocks.is_global_lock(lock.until, side=lock.lock_side): result = PairLocks.lock_pair( - '*', lock.until, lock.reason, now=now, side=lock.lock_side) + "*", lock.until, lock.reason, now=now, side=lock.lock_side + ) return result - def stop_per_pair(self, pair, now: Optional[datetime] = None, - side: LongShort = 'long') -> Optional[PairLock]: + def stop_per_pair( + self, pair, now: Optional[datetime] = None, side: LongShort = "long" + ) -> Optional[PairLock]: if not now: now = datetime.now(timezone.utc) result = None for protection_handler in self._protection_handlers: if protection_handler.has_local_stop: - lock = protection_handler.stop_per_pair( - pair=pair, date_now=now, side=side) + lock = protection_handler.stop_per_pair(pair=pair, date_now=now, side=side) if lock and lock.until: if not PairLocks.is_pair_locked(pair, lock.until, lock.lock_side): result = PairLocks.lock_pair( - pair, lock.until, lock.reason, now=now, side=lock.lock_side) + pair, lock.until, lock.reason, now=now, side=lock.lock_side + ) return result diff --git a/freqtrade/plugins/protections/cooldown_period.py b/freqtrade/plugins/protections/cooldown_period.py index 426b8f1b6..2948d17d0 100644 --- a/freqtrade/plugins/protections/cooldown_period.py +++ b/freqtrade/plugins/protections/cooldown_period.py @@ -1,4 +1,3 @@ - import logging from datetime import datetime, timedelta from typing import Optional @@ -12,7 +11,6 @@ logger = logging.getLogger(__name__) class CooldownPeriod(IProtection): - has_global_stop: bool = False has_local_stop: bool = True @@ -20,13 +18,13 @@ class CooldownPeriod(IProtection): """ LockReason to use """ - return (f'Cooldown period for {self.stop_duration_str}.') + return f"Cooldown period for {self.stop_duration_str}." def short_desc(self) -> str: """ Short method description - used for startup-messages """ - return (f"{self.name} - Cooldown period of {self.stop_duration_str}.") + return f"{self.name} - Cooldown period of {self.stop_duration_str}." def _cooldown_period(self, pair: str, date_now: datetime) -> Optional[ProtectionReturn]: """ @@ -66,7 +64,8 @@ class CooldownPeriod(IProtection): return None def stop_per_pair( - self, pair: str, date_now: datetime, side: LongShort) -> Optional[ProtectionReturn]: + 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". diff --git a/freqtrade/plugins/protections/iprotection.py b/freqtrade/plugins/protections/iprotection.py index 378eccfef..91d591c48 100644 --- a/freqtrade/plugins/protections/iprotection.py +++ b/freqtrade/plugins/protections/iprotection.py @@ -1,4 +1,3 @@ - import logging from abc import ABC, abstractmethod from dataclasses import dataclass @@ -20,11 +19,10 @@ class ProtectionReturn: lock: bool until: datetime reason: Optional[str] - lock_side: str = '*' + lock_side: str = "*" class IProtection(LoggingMixin, ABC): - # Can globally stop the bot has_global_stop: bool = False # Can stop trading for one pair @@ -36,19 +34,19 @@ class IProtection(LoggingMixin, ABC): 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) + 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 = int(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._stop_duration = int(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)) + self._lookback_period = int(protection_config.get("lookback_period", 60)) LoggingMixin.__init__(self, logger) @@ -62,11 +60,12 @@ class IProtection(LoggingMixin, ABC): 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')}") + 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')}") + return f"{self._stop_duration} " f"{plural(self._stop_duration, 'minute', 'minutes')}" @property def lookback_period_str(self) -> str: @@ -74,11 +73,14 @@ class IProtection(LoggingMixin, ABC): 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')}") + 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')}") + return ( + f"{self._lookback_period} " f"{plural(self._lookback_period, 'minute', 'minutes')}" + ) @abstractmethod def short_desc(self) -> str: @@ -96,7 +98,8 @@ class IProtection(LoggingMixin, ABC): @abstractmethod def stop_per_pair( - self, pair: str, date_now: datetime, side: LongShort) -> Optional[ProtectionReturn]: + 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". diff --git a/freqtrade/plugins/protections/low_profit_pairs.py b/freqtrade/plugins/protections/low_profit_pairs.py index f638673fa..360f6721c 100644 --- a/freqtrade/plugins/protections/low_profit_pairs.py +++ b/freqtrade/plugins/protections/low_profit_pairs.py @@ -1,4 +1,3 @@ - import logging from datetime import datetime, timedelta from typing import Any, Dict, Optional @@ -12,33 +11,37 @@ logger = logging.getLogger(__name__) class LowProfitPairs(IProtection): - has_global_stop: bool = False has_local_stop: bool = True def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None: super().__init__(config, protection_config) - self._trade_limit = protection_config.get('trade_limit', 1) - self._required_profit = protection_config.get('required_profit', 0.0) - self._only_per_side = protection_config.get('only_per_side', False) + self._trade_limit = protection_config.get("trade_limit", 1) + self._required_profit = protection_config.get("required_profit", 0.0) + self._only_per_side = protection_config.get("only_per_side", False) def short_desc(self) -> str: """ Short method description - used for startup-messages """ - return (f"{self.name} - Low Profit Protection, locks pairs with " - f"profit < {self._required_profit} within {self.lookback_period_str}.") + return ( + f"{self.name} - Low Profit Protection, locks pairs with " + f"profit < {self._required_profit} within {self.lookback_period_str}." + ) def _reason(self, profit: float) -> str: """ LockReason to use """ - return (f'{profit} < {self._required_profit} in {self.lookback_period_str}, ' - f'locking for {self.stop_duration_str}.') + return ( + f"{profit} < {self._required_profit} in {self.lookback_period_str}, " + f"locking for {self.stop_duration_str}." + ) def _low_profit( - self, date_now: datetime, pair: str, side: LongShort) -> Optional[ProtectionReturn]: + self, date_now: datetime, pair: str, side: LongShort + ) -> Optional[ProtectionReturn]: """ Evaluate recent trades for pair """ @@ -57,20 +60,23 @@ class LowProfitPairs(IProtection): return None profit = sum( - trade.close_profit for trade in trades if trade.close_profit - and (not self._only_per_side or trade.trade_direction == side) - ) + trade.close_profit + for trade in trades + if trade.close_profit and (not self._only_per_side or trade.trade_direction == side) + ) if profit < self._required_profit: self.log_once( f"Trading for {pair} stopped due to {profit:.2f} < {self._required_profit} " - f"within {self._lookback_period} minutes.", logger.info) + f"within {self._lookback_period} minutes.", + logger.info, + ) until = self.calculate_lock_end(trades, self._stop_duration) return ProtectionReturn( lock=True, until=until, reason=self._reason(profit), - lock_side=(side if self._only_per_side else '*') + lock_side=(side if self._only_per_side else "*"), ) return None @@ -85,7 +91,8 @@ class LowProfitPairs(IProtection): return None def stop_per_pair( - self, pair: str, date_now: datetime, side: LongShort) -> Optional[ProtectionReturn]: + 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". diff --git a/freqtrade/plugins/protections/max_drawdown_protection.py b/freqtrade/plugins/protections/max_drawdown_protection.py index 8193dc7e4..3e252185f 100644 --- a/freqtrade/plugins/protections/max_drawdown_protection.py +++ b/freqtrade/plugins/protections/max_drawdown_protection.py @@ -1,4 +1,3 @@ - import logging from datetime import datetime, timedelta from typing import Any, Dict, Optional @@ -15,30 +14,33 @@ logger = logging.getLogger(__name__) class MaxDrawdown(IProtection): - has_global_stop: bool = True has_local_stop: bool = False def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None: super().__init__(config, protection_config) - self._trade_limit = protection_config.get('trade_limit', 1) - self._max_allowed_drawdown = protection_config.get('max_allowed_drawdown', 0.0) + self._trade_limit = protection_config.get("trade_limit", 1) + self._max_allowed_drawdown = protection_config.get("max_allowed_drawdown", 0.0) # TODO: Implement checks to limit max_drawdown to sensible values def short_desc(self) -> str: """ Short method description - used for startup-messages """ - return (f"{self.name} - Max drawdown protection, stop trading if drawdown is > " - f"{self._max_allowed_drawdown} within {self.lookback_period_str}.") + return ( + f"{self.name} - Max drawdown protection, stop trading if drawdown is > " + f"{self._max_allowed_drawdown} within {self.lookback_period_str}." + ) def _reason(self, drawdown: float) -> str: """ LockReason to use """ - return (f'{drawdown} passed {self._max_allowed_drawdown} in {self.lookback_period_str}, ' - f'locking for {self.stop_duration_str}.') + return ( + f"{drawdown} passed {self._max_allowed_drawdown} in {self.lookback_period_str}, " + f"locking for {self.stop_duration_str}." + ) def _max_drawdown(self, date_now: datetime) -> Optional[ProtectionReturn]: """ @@ -57,14 +59,16 @@ class MaxDrawdown(IProtection): # Drawdown is always positive try: # TODO: This should use absolute profit calculation, considering account balance. - drawdown, _, _, _, _, _ = calculate_max_drawdown(trades_df, value_col='close_profit') + drawdown, _, _, _, _, _ = calculate_max_drawdown(trades_df, value_col="close_profit") except ValueError: return None if drawdown > self._max_allowed_drawdown: self.log_once( f"Trading stopped due to Max Drawdown {drawdown:.2f} > {self._max_allowed_drawdown}" - f" within {self.lookback_period_str}.", logger.info) + f" within {self.lookback_period_str}.", + logger.info, + ) until = self.calculate_lock_end(trades, self._stop_duration) return ProtectionReturn( @@ -85,7 +89,8 @@ class MaxDrawdown(IProtection): return self._max_drawdown(date_now) def stop_per_pair( - self, pair: str, date_now: datetime, side: LongShort) -> Optional[ProtectionReturn]: + 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". diff --git a/freqtrade/plugins/protections/stoploss_guard.py b/freqtrade/plugins/protections/stoploss_guard.py index 23ceebbc9..a9aca20b4 100644 --- a/freqtrade/plugins/protections/stoploss_guard.py +++ b/freqtrade/plugins/protections/stoploss_guard.py @@ -1,4 +1,3 @@ - import logging from datetime import datetime, timedelta from typing import Any, Dict, Optional @@ -13,44 +12,59 @@ logger = logging.getLogger(__name__) class StoplossGuard(IProtection): - has_global_stop: bool = True has_local_stop: bool = True def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None: super().__init__(config, protection_config) - self._trade_limit = protection_config.get('trade_limit', 10) - self._disable_global_stop = protection_config.get('only_per_pair', False) - self._only_per_side = protection_config.get('only_per_side', False) - self._profit_limit = protection_config.get('required_profit', 0.0) + self._trade_limit = protection_config.get("trade_limit", 10) + self._disable_global_stop = protection_config.get("only_per_pair", False) + self._only_per_side = protection_config.get("only_per_side", False) + self._profit_limit = protection_config.get("required_profit", 0.0) def short_desc(self) -> str: """ Short method description - used for startup-messages """ - return (f"{self.name} - Frequent Stoploss Guard, {self._trade_limit} stoplosses " - f"with profit < {self._profit_limit:.2%} within {self.lookback_period_str}.") + return ( + f"{self.name} - Frequent Stoploss Guard, {self._trade_limit} stoplosses " + f"with profit < {self._profit_limit:.2%} within {self.lookback_period_str}." + ) def _reason(self) -> str: """ LockReason to use """ - return (f'{self._trade_limit} stoplosses in {self._lookback_period} min, ' - f'locking for {self._stop_duration} min.') + return ( + f"{self._trade_limit} stoplosses in {self._lookback_period} min, " + f"locking for {self._stop_duration} min." + ) - def _stoploss_guard(self, date_now: datetime, pair: Optional[str], - side: LongShort) -> Optional[ProtectionReturn]: + def _stoploss_guard( + self, date_now: datetime, pair: Optional[str], side: LongShort + ) -> Optional[ProtectionReturn]: """ Evaluate recent trades """ look_back_until = date_now - timedelta(minutes=self._lookback_period) trades1 = Trade.get_trades_proxy(pair=pair, is_open=False, close_date=look_back_until) - trades = [trade for trade in trades1 if (str(trade.exit_reason) in ( - ExitType.TRAILING_STOP_LOSS.value, ExitType.STOP_LOSS.value, - ExitType.STOPLOSS_ON_EXCHANGE.value, ExitType.LIQUIDATION.value) - and trade.close_profit and trade.close_profit < self._profit_limit)] + trades = [ + trade + for trade in trades1 + if ( + str(trade.exit_reason) + in ( + ExitType.TRAILING_STOP_LOSS.value, + ExitType.STOP_LOSS.value, + ExitType.STOPLOSS_ON_EXCHANGE.value, + ExitType.LIQUIDATION.value, + ) + and trade.close_profit + and trade.close_profit < self._profit_limit + ) + ] if self._only_per_side: # Long or short trades only @@ -59,15 +73,18 @@ class StoplossGuard(IProtection): if len(trades) < self._trade_limit: return None - self.log_once(f"Trading stopped due to {self._trade_limit} " - f"stoplosses within {self._lookback_period} minutes.", logger.info) + self.log_once( + f"Trading stopped due to {self._trade_limit} " + f"stoplosses within {self._lookback_period} minutes.", + logger.info, + ) until = self.calculate_lock_end(trades, self._stop_duration) return ProtectionReturn( lock=True, until=until, reason=self._reason(), - lock_side=(side if self._only_per_side else '*') - ) + lock_side=(side if self._only_per_side else "*"), + ) def global_stop(self, date_now: datetime, side: LongShort) -> Optional[ProtectionReturn]: """ @@ -81,7 +98,8 @@ class StoplossGuard(IProtection): return self._stoploss_guard(date_now, None, side) def stop_per_pair( - self, pair: str, date_now: datetime, side: LongShort) -> Optional[ProtectionReturn]: + 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".