lychiyu/freqtrade_origin
1
0
Fork 0
mirror of https://github.com/freqtrade/freqtrade.git synced 2024-11-10 10:21:59 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

ruff format: plugins/protections

Browse Source
This commit is contained in:
Matthias 2024-05-12 16:38:10 +02:00
parent 700b7acb6f
commit 6bfe7aa72d
6 changed files with 118 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
freqtrade/plugins/protectionmanager.py
View File

@ -1,6 +1,7 @@
""" """
Protection manager class Protection manager class
""" """
import logging import logging
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Dict, List, Optional from typing import Dict, List, Optional
@ -16,14 +17,13 @@ logger = logging.getLogger(__name__)
class ProtectionManager: class ProtectionManager:
def __init__(self, config: Config, protections: List) -> None: def __init__(self, config: Config, protections: List) -> None:
self._config = config self._config = config
self._protection_handlers: List[IProtection] = [] self._protection_handlers: List[IProtection] = []
for protection_handler_config in protections: for protection_handler_config in protections:
protection_handler = ProtectionResolver.load_protection( protection_handler = ProtectionResolver.load_protection(
protection_handler_config['method'], protection_handler_config["method"],
config=config, config=config,
protection_config=protection_handler_config, protection_config=protection_handler_config,
) )
@ -45,8 +45,9 @@ class ProtectionManager:
""" """
return [{p.name: p.short_desc()} for p in self._protection_handlers] return [{p.name: p.short_desc()} for p in self._protection_handlers]
def global_stop(self, now: Optional[datetime] = None, def global_stop(
side: LongShort = 'long') -> Optional[PairLock]: self, now: Optional[datetime] = None, side: LongShort = "long"
) -> Optional[PairLock]:
if not now: if not now:
now = datetime.now(timezone.utc) now = datetime.now(timezone.utc)
result = None result = None
@ -56,20 +57,22 @@ class ProtectionManager:
if lock and lock.until: if lock and lock.until:
if not PairLocks.is_global_lock(lock.until, side=lock.lock_side): if not PairLocks.is_global_lock(lock.until, side=lock.lock_side):
result = PairLocks.lock_pair( 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 return result
def stop_per_pair(self, pair, now: Optional[datetime] = None, def stop_per_pair(
side: LongShort = 'long') -> Optional[PairLock]: self, pair, now: Optional[datetime] = None, side: LongShort = "long"
) -> Optional[PairLock]:
if not now: if not now:
now = datetime.now(timezone.utc) now = datetime.now(timezone.utc)
result = None result = None
for protection_handler in self._protection_handlers: for protection_handler in self._protection_handlers:
if protection_handler.has_local_stop: if protection_handler.has_local_stop:
lock = protection_handler.stop_per_pair( lock = protection_handler.stop_per_pair(pair=pair, date_now=now, side=side)
pair=pair, date_now=now, side=side)
if lock and lock.until: if lock and lock.until:
if not PairLocks.is_pair_locked(pair, lock.until, lock.lock_side): if not PairLocks.is_pair_locked(pair, lock.until, lock.lock_side):
result = PairLocks.lock_pair( 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 return result

9
freqtrade/plugins/protections/cooldown_period.py
View File

@ -1,4 +1,3 @@
import logging import logging
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Optional from typing import Optional
@ -12,7 +11,6 @@ logger = logging.getLogger(__name__)
class CooldownPeriod(IProtection): class CooldownPeriod(IProtection):
has_global_stop: bool = False has_global_stop: bool = False
has_local_stop: bool = True has_local_stop: bool = True
@ -20,13 +18,13 @@ class CooldownPeriod(IProtection):
""" """
LockReason to use 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: def short_desc(self) -> str:
""" """
Short method description - used for startup-messages 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]: def _cooldown_period(self, pair: str, date_now: datetime) -> Optional[ProtectionReturn]:
""" """
@ -66,7 +64,8 @@ class CooldownPeriod(IProtection):
return None return None
def stop_per_pair( 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 Stops trading (position entering) for this pair
This must evaluate to true for the whole period of the "cooldown period". This must evaluate to true for the whole period of the "cooldown period".

43
freqtrade/plugins/protections/iprotection.py
View File

@ -1,4 +1,3 @@
import logging import logging
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from dataclasses import dataclass from dataclasses import dataclass
@ -20,11 +19,10 @@ class ProtectionReturn:
lock: bool lock: bool
until: datetime until: datetime
reason: Optional[str] reason: Optional[str]
lock_side: str = '*' lock_side: str = "*"
class IProtection(LoggingMixin, ABC): class IProtection(LoggingMixin, ABC):
# Can globally stop the bot # Can globally stop the bot
has_global_stop: bool = False has_global_stop: bool = False
# Can stop trading for one pair # Can stop trading for one pair
@ -36,19 +34,19 @@ class IProtection(LoggingMixin, ABC):
self._stop_duration_candles: Optional[int] = None self._stop_duration_candles: Optional[int] = None
self._lookback_period_candles: Optional[int] = None self._lookback_period_candles: Optional[int] = None
tf_in_min = timeframe_to_minutes(config['timeframe']) tf_in_min = timeframe_to_minutes(config["timeframe"])
if 'stop_duration_candles' in protection_config: if "stop_duration_candles" in protection_config:
self._stop_duration_candles = int(protection_config.get('stop_duration_candles', 1)) self._stop_duration_candles = int(protection_config.get("stop_duration_candles", 1))
self._stop_duration = (tf_in_min * self._stop_duration_candles) self._stop_duration = tf_in_min * self._stop_duration_candles
else: else:
self._stop_duration_candles = None self._stop_duration_candles = None
self._stop_duration = int(protection_config.get('stop_duration', 60)) self._stop_duration = int(protection_config.get("stop_duration", 60))
if 'lookback_period_candles' in protection_config: if "lookback_period_candles" in protection_config:
self._lookback_period_candles = int(protection_config.get('lookback_period_candles', 1)) self._lookback_period_candles = int(protection_config.get("lookback_period_candles", 1))
self._lookback_period = tf_in_min * self._lookback_period_candles self._lookback_period = tf_in_min * self._lookback_period_candles
else: else:
self._lookback_period_candles = None 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) LoggingMixin.__init__(self, logger)
@ -62,11 +60,12 @@ class IProtection(LoggingMixin, ABC):
Output configured stop duration in either candles or minutes Output configured stop duration in either candles or minutes
""" """
if self._stop_duration_candles: if self._stop_duration_candles:
return (f"{self._stop_duration_candles} " return (
f"{plural(self._stop_duration_candles, 'candle', 'candles')}") f"{self._stop_duration_candles} "
f"{plural(self._stop_duration_candles, 'candle', 'candles')}"
)
else: else:
return (f"{self._stop_duration} " return f"{self._stop_duration} " f"{plural(self._stop_duration, 'minute', 'minutes')}"
f"{plural(self._stop_duration, 'minute', 'minutes')}")
@property @property
def lookback_period_str(self) -> str: def lookback_period_str(self) -> str:
@ -74,11 +73,14 @@ class IProtection(LoggingMixin, ABC):
Output configured lookback period in either candles or minutes Output configured lookback period in either candles or minutes
""" """
if self._lookback_period_candles: if self._lookback_period_candles:
return (f"{self._lookback_period_candles} " return (
f"{plural(self._lookback_period_candles, 'candle', 'candles')}") f"{self._lookback_period_candles} "
f"{plural(self._lookback_period_candles, 'candle', 'candles')}"
)
else: else:
return (f"{self._lookback_period} " return (
f"{plural(self._lookback_period, 'minute', 'minutes')}") f"{self._lookback_period} " f"{plural(self._lookback_period, 'minute', 'minutes')}"
)
@abstractmethod @abstractmethod
def short_desc(self) -> str: def short_desc(self) -> str:
@ -96,7 +98,8 @@ class IProtection(LoggingMixin, ABC):
@abstractmethod @abstractmethod
def stop_per_pair( 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 Stops trading (position entering) for this pair
This must evaluate to true for the whole period of the "cooldown period". This must evaluate to true for the whole period of the "cooldown period".

37
freqtrade/plugins/protections/low_profit_pairs.py
View File

@ -1,4 +1,3 @@
import logging import logging
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
@ -12,33 +11,37 @@ logger = logging.getLogger(__name__)
class LowProfitPairs(IProtection): class LowProfitPairs(IProtection):
has_global_stop: bool = False has_global_stop: bool = False
has_local_stop: bool = True has_local_stop: bool = True
def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None: def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None:
super().__init__(config, protection_config) super().__init__(config, protection_config)
self._trade_limit = protection_config.get('trade_limit', 1) self._trade_limit = protection_config.get("trade_limit", 1)
self._required_profit = protection_config.get('required_profit', 0.0) self._required_profit = protection_config.get("required_profit", 0.0)
self._only_per_side = protection_config.get('only_per_side', False) self._only_per_side = protection_config.get("only_per_side", False)
def short_desc(self) -> str: def short_desc(self) -> str:
""" """
Short method description - used for startup-messages Short method description - used for startup-messages
""" """
return (f"{self.name} - Low Profit Protection, locks pairs with " return (
f"profit < {self._required_profit} within {self.lookback_period_str}.") 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: def _reason(self, profit: float) -> str:
""" """
LockReason to use LockReason to use
""" """
return (f'{profit} < {self._required_profit} in {self.lookback_period_str}, ' return (
f'locking for {self.stop_duration_str}.') f"{profit} < {self._required_profit} in {self.lookback_period_str}, "
f"locking for {self.stop_duration_str}."
)
def _low_profit( 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 Evaluate recent trades for pair
""" """
@ -57,20 +60,23 @@ class LowProfitPairs(IProtection):
return None return None
profit = sum( profit = sum(
trade.close_profit for trade in trades if trade.close_profit trade.close_profit
and (not self._only_per_side or trade.trade_direction == side) for trade in trades
if trade.close_profit and (not self._only_per_side or trade.trade_direction == side)
) )
if profit < self._required_profit: if profit < self._required_profit:
self.log_once( self.log_once(
f"Trading for {pair} stopped due to {profit:.2f} < {self._required_profit} " 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) until = self.calculate_lock_end(trades, self._stop_duration)
return ProtectionReturn( return ProtectionReturn(
lock=True, lock=True,
until=until, until=until,
reason=self._reason(profit), reason=self._reason(profit),
lock_side=(side if self._only_per_side else '*') lock_side=(side if self._only_per_side else "*"),
) )
return None return None
@ -85,7 +91,8 @@ class LowProfitPairs(IProtection):
return None return None
def stop_per_pair( 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 Stops trading (position entering) for this pair
This must evaluate to true for the whole period of the "cooldown period". This must evaluate to true for the whole period of the "cooldown period".

27
freqtrade/plugins/protections/max_drawdown_protection.py
View File

@ -1,4 +1,3 @@
import logging import logging
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
@ -15,30 +14,33 @@ logger = logging.getLogger(__name__)
class MaxDrawdown(IProtection): class MaxDrawdown(IProtection):
has_global_stop: bool = True has_global_stop: bool = True
has_local_stop: bool = False has_local_stop: bool = False
def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None: def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None:
super().__init__(config, protection_config) super().__init__(config, protection_config)
self._trade_limit = protection_config.get('trade_limit', 1) self._trade_limit = protection_config.get("trade_limit", 1)
self._max_allowed_drawdown = protection_config.get('max_allowed_drawdown', 0.0) self._max_allowed_drawdown = protection_config.get("max_allowed_drawdown", 0.0)
# TODO: Implement checks to limit max_drawdown to sensible values # TODO: Implement checks to limit max_drawdown to sensible values
def short_desc(self) -> str: def short_desc(self) -> str:
""" """
Short method description - used for startup-messages Short method description - used for startup-messages
""" """
return (f"{self.name} - Max drawdown protection, stop trading if drawdown is > " return (
f"{self._max_allowed_drawdown} within {self.lookback_period_str}.") 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: def _reason(self, drawdown: float) -> str:
""" """
LockReason to use LockReason to use
""" """
return (f'{drawdown} passed {self._max_allowed_drawdown} in {self.lookback_period_str}, ' return (
f'locking for {self.stop_duration_str}.') 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]: def _max_drawdown(self, date_now: datetime) -> Optional[ProtectionReturn]:
""" """
@ -57,14 +59,16 @@ class MaxDrawdown(IProtection):
# Drawdown is always positive # Drawdown is always positive
try: try:
# TODO: This should use absolute profit calculation, considering account balance. # 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: except ValueError:
return None return None
if drawdown > self._max_allowed_drawdown: if drawdown > self._max_allowed_drawdown:
self.log_once( self.log_once(
f"Trading stopped due to Max Drawdown {drawdown:.2f} > {self._max_allowed_drawdown}" 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) until = self.calculate_lock_end(trades, self._stop_duration)
return ProtectionReturn( return ProtectionReturn(
@ -85,7 +89,8 @@ class MaxDrawdown(IProtection):
return self._max_drawdown(date_now) return self._max_drawdown(date_now)
def stop_per_pair( 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 Stops trading (position entering) for this pair
This must evaluate to true for the whole period of the "cooldown period". This must evaluate to true for the whole period of the "cooldown period".

58
freqtrade/plugins/protections/stoploss_guard.py
View File

@ -1,4 +1,3 @@
import logging import logging
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
@ -13,44 +12,59 @@ logger = logging.getLogger(__name__)
class StoplossGuard(IProtection): class StoplossGuard(IProtection):
has_global_stop: bool = True has_global_stop: bool = True
has_local_stop: bool = True has_local_stop: bool = True
def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None: def __init__(self, config: Config, protection_config: Dict[str, Any]) -> None:
super().__init__(config, protection_config) super().__init__(config, protection_config)
self._trade_limit = protection_config.get('trade_limit', 10) self._trade_limit = protection_config.get("trade_limit", 10)
self._disable_global_stop = protection_config.get('only_per_pair', False) self._disable_global_stop = protection_config.get("only_per_pair", False)
self._only_per_side = protection_config.get('only_per_side', False) self._only_per_side = protection_config.get("only_per_side", False)
self._profit_limit = protection_config.get('required_profit', 0.0) self._profit_limit = protection_config.get("required_profit", 0.0)
def short_desc(self) -> str: def short_desc(self) -> str:
""" """
Short method description - used for startup-messages Short method description - used for startup-messages
""" """
return (f"{self.name} - Frequent Stoploss Guard, {self._trade_limit} stoplosses " return (
f"with profit < {self._profit_limit:.2%} within {self.lookback_period_str}.") 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: def _reason(self) -> str:
""" """
LockReason to use LockReason to use
""" """
return (f'{self._trade_limit} stoplosses in {self._lookback_period} min, ' return (
f'locking for {self._stop_duration} min.') 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], def _stoploss_guard(
side: LongShort) -> Optional[ProtectionReturn]: self, date_now: datetime, pair: Optional[str], side: LongShort
) -> Optional[ProtectionReturn]:
""" """
Evaluate recent trades Evaluate recent trades
""" """
look_back_until = date_now - timedelta(minutes=self._lookback_period) 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) 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 ( trades = [
ExitType.TRAILING_STOP_LOSS.value, ExitType.STOP_LOSS.value, trade
ExitType.STOPLOSS_ON_EXCHANGE.value, ExitType.LIQUIDATION.value) for trade in trades1
and trade.close_profit and trade.close_profit < self._profit_limit)] 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: if self._only_per_side:
# Long or short trades only # Long or short trades only
@ -59,14 +73,17 @@ class StoplossGuard(IProtection):
if len(trades) < self._trade_limit: if len(trades) < self._trade_limit:
return None return None
self.log_once(f"Trading stopped due to {self._trade_limit} " self.log_once(
f"stoplosses within {self._lookback_period} minutes.", logger.info) 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) until = self.calculate_lock_end(trades, self._stop_duration)
return ProtectionReturn( return ProtectionReturn(
lock=True, lock=True,
until=until, until=until,
reason=self._reason(), 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]: 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) return self._stoploss_guard(date_now, None, side)
def stop_per_pair( 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 Stops trading (position entering) for this pair
This must evaluate to true for the whole period of the "cooldown period". This must evaluate to true for the whole period of the "cooldown period".