Added unlock_at field for protection config

pull/10400/head
simwai 2 years ago
parent 1d3ca5743b
commit 2b456cbdeb

@ -1,6 +1,7 @@
import logging
from collections import Counter
from copy import deepcopy
from datetime import datetime
from typing import Any, Dict
from jsonschema import Draft4Validator, validators
@ -192,18 +193,40 @@ def _validate_protections(conf: Dict[str, Any]) -> None:
"""
for prot in conf.get("protections", []):
parsed_unlock_at = _validate_unlock_at(prot)
if "stop_duration" in prot and "stop_duration_candles" in prot:
raise ConfigurationError(
"Protections must specify either `stop_duration` or `stop_duration_candles`.\n"
f"Please fix the protection {prot.get('method')}"
f"Please fix the protection {prot.get('method')}."
)
if "lookback_period" in prot and "lookback_period_candles" in prot:
raise ConfigurationError(
"Protections must specify either `lookback_period` or `lookback_period_candles`.\n"
f"Please fix the protection {prot.get('method')}"
f"Please fix the protection {prot.get('method')}."
)
if parsed_unlock_at is not None and "stop_duration" in prot:
raise ConfigurationError(
"Protections must specify either `unlock_at` or `stop_duration`.\n"
f"Please fix the protection {prot.get('method')}."
)
if parsed_unlock_at is not None and "stop_duration_candles" in prot:
raise ConfigurationError(
"Protections must specify either `unlock_at` or `stop_duration_candles`.\n"
f"Please fix the protection {prot.get('method')}."
)
def _validate_unlock_at(config_unlock_at: str) -> datetime:
if config_unlock_at is not None and isinstance(config_unlock_at, str):
try:
return datetime.strptime(config_unlock_at, "%H:%M")
except ValueError:
raise ConfigurationError(f"Invalid date format for unlock_at: {config_unlock_at}.")
def _validate_ask_orderbook(conf: Dict[str, Any]) -> None:
ask_strategy = conf.get("exit_pricing", {})

@ -356,6 +356,7 @@ CONF_SCHEMA = {
"properties": {
"method": {"type": "string", "enum": AVAILABLE_PROTECTIONS},
"stop_duration": {"type": "number", "minimum": 0.0},
"unlock_at": {"type": "string"},
"stop_duration_candles": {"type": "number", "minimum": 0},
"trade_limit": {"type": "number", "minimum": 1},
"lookback_period": {"type": "number", "minimum": 1},

@ -18,7 +18,10 @@ class CooldownPeriod(IProtection):
"""
LockReason to use
"""
return f"Cooldown period for {self.stop_duration_str}."
reason = f"Cooldown period for {self.stop_duration_str}."
if self.unlock_at_str is not None:
reason += f" Unlocking trading at {self.unlock_at_str}."
return reason
def short_desc(self) -> str:
"""

@ -2,7 +2,7 @@ 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 typing import Any, Dict, List, Optional, Union
from freqtrade.constants import Config, LongShort
from freqtrade.exchange import timeframe_to_minutes
@ -33,21 +33,33 @@ class IProtection(LoggingMixin, ABC):
self._protection_config = protection_config
self._stop_duration_candles: Optional[int] = None
self._lookback_period_candles: Optional[int] = None
self.unlock_at: Optional[datetime] = 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 = 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))
if "unlock_at" in protection_config:
now_time = datetime.now(timezone.utc)
unlock_at = datetime.strptime(protection_config["unlock_at"], "%H:%M").replace(
day=now_time.day, year=now_time.year, month=now_time.month
)
if unlock_at.time() < now_time.time():
unlock_at = unlock_at.replace(day=now_time.day + 1)
unlock_at = unlock_at.replace(tzinfo=timezone.utc)
self._stop_duration = self.calculate_timespan(now_time, unlock_at)
self.unlock_at = unlock_at
LoggingMixin.__init__(self, logger)
@property
@ -80,6 +92,15 @@ class IProtection(LoggingMixin, ABC):
else:
return f"{self._lookback_period} {plural(self._lookback_period, 'minute', 'minutes')}"
@property
def unlock_at_str(self) -> Union[str, None]:
"""
Output configured unlock time
"""
if self.unlock_at:
return self.unlock_at.strftime("%H:%M")
return None
@abstractmethod
def short_desc(self) -> str:
"""
@ -118,3 +139,14 @@ class IProtection(LoggingMixin, ABC):
until = max_date + timedelta(minutes=stop_minutes)
return until
@staticmethod
def calculate_timespan(start_time: datetime, end_time: datetime) -> int:
"""
Calculate the timespan between two datetime objects in minutes.
:param start_time: The start datetime.
:param end_time: The end datetime.
:return: The difference between the two datetimes in minutes.
"""
return int((end_time - start_time).total_seconds() / 60)

@ -34,10 +34,13 @@ class LowProfitPairs(IProtection):
"""
LockReason to use
"""
return (
reason = (
f"{profit} < {self._required_profit} in {self.lookback_period_str}, "
f"locking for {self.stop_duration_str}."
)
if self.unlock_at_str is not None:
reason += f" Unlocking trading at {self.unlock_at_str}."
return reason
def _low_profit(
self, date_now: datetime, pair: str, side: LongShort

@ -37,10 +37,13 @@ class MaxDrawdown(IProtection):
"""
LockReason to use
"""
return (
reason = (
f"{drawdown} passed {self._max_allowed_drawdown} in {self.lookback_period_str}, "
f"locking for {self.stop_duration_str}."
)
if self.unlock_at_str is not None:
reason += f" Unlocking trading at {self.unlock_at_str}."
return reason
def _max_drawdown(self, date_now: datetime) -> Optional[ProtectionReturn]:
"""

@ -36,10 +36,13 @@ class StoplossGuard(IProtection):
"""
LockReason to use
"""
return (
reason = (
f"{self._trade_limit} stoplosses in {self._lookback_period} min, "
f"locking for {self._stop_duration} min."
)
if self.unlock_at_str is not None:
reason += f" Unlocking trading at {self.unlock_at_str}."
return reason
def _stoploss_guard(
self, date_now: datetime, pair: Optional[str], side: LongShort

@ -102,56 +102,94 @@ def test_protectionmanager(mocker, default_conf):
@pytest.mark.parametrize(
"timeframe,expected,protconf",
"timeframe,expected_lookback,expected_stop,protconf",
[
(
"1m",
[20, 10],
20,
10,
[{"method": "StoplossGuard", "lookback_period_candles": 20, "stop_duration": 10}],
),
(
"5m",
[100, 15],
100,
15,
[{"method": "StoplossGuard", "lookback_period_candles": 20, "stop_duration": 15}],
),
(
"1h",
[1200, 40],
1200,
40,
[{"method": "StoplossGuard", "lookback_period_candles": 20, "stop_duration": 40}],
),
(
"1d",
[1440, 5],
1440,
5,
[{"method": "StoplossGuard", "lookback_period_candles": 1, "stop_duration": 5}],
),
(
"1m",
[20, 5],
20,
5,
[{"method": "StoplossGuard", "lookback_period": 20, "stop_duration_candles": 5}],
),
(
"5m",
[15, 25],
15,
25,
[{"method": "StoplossGuard", "lookback_period": 15, "stop_duration_candles": 5}],
),
(
"1h",
[50, 600],
50,
600,
[{"method": "StoplossGuard", "lookback_period": 50, "stop_duration_candles": 10}],
),
(
"1h",
[60, 540],
60,
540,
[{"method": "StoplossGuard", "lookback_period_candles": 1, "stop_duration_candles": 9}],
),
(
"1m",
20,
"01:00",
[{"method": "StoplossGuard", "lookback_period_candles": 20, "unlock_at": "01:00"}],
),
(
"5m",
100,
"02:00",
[{"method": "StoplossGuard", "lookback_period_candles": 20, "unlock_at": "02:00"}],
),
(
"1h",
1200,
"03:00",
[{"method": "StoplossGuard", "lookback_period_candles": 20, "unlock_at": "03:00"}],
),
(
"1d",
1440,
"04:00",
[{"method": "StoplossGuard", "lookback_period_candles": 1, "unlock_at": "04:00"}],
),
],
)
def test_protections_init(default_conf, timeframe, expected, protconf):
def test_protections_init(default_conf, timeframe, expected_lookback, expected_stop, protconf):
"""
Test the initialization of protections with different configurations, including unlock_at.
"""
default_conf["timeframe"] = timeframe
man = ProtectionManager(default_conf, protconf)
assert len(man._protection_handlers) == len(protconf)
assert man._protection_handlers[0]._lookback_period == expected[0]
assert man._protection_handlers[0]._stop_duration == expected[1]
assert man._protection_handlers[0]._lookback_period == expected_lookback
if isinstance(expected_stop, int):
assert man._protection_handlers[0]._stop_duration == expected_stop
else:
assert man._protection_handlers[0].unlock_at.strftime("%H:%M") == expected_stop
@pytest.mark.parametrize("is_short", [False, True])
@ -654,6 +692,30 @@ def test_MaxDrawdown(mocker, default_conf, fee, caplog):
"if drawdown is > 0.0 within 20 candles.'}]",
None,
),
(
{
"method": "StoplossGuard",
"lookback_period_candles": 12,
"trade_limit": 2,
"required_profit": -0.05,
"unlock_at": "01:00",
},
"[{'StoplossGuard': 'StoplossGuard - Frequent Stoploss Guard, "
"2 stoplosses with profit < -5.00% within 12 candles. Unlocking trading at 01:00.'}]",
None,
),
(
{"method": "LowProfitPairs", "lookback_period_candles": 11, "unlock_at": "03:00"},
"[{'LowProfitPairs': 'LowProfitPairs - Low Profit Protection, locks pairs with "
"profit < 0.0 within 11 candles. Unlocking trading at 03:00.'}]",
None,
),
(
{"method": "MaxDrawdown", "lookback_period_candles": 20, "unlock_at": "04:00"},
"[{'MaxDrawdown': 'MaxDrawdown - Max drawdown protection, stop trading "
"if drawdown is > 0.0 within 20 candles. Unlocking trading at 04:00.'}]",
None,
),
],
)
def test_protection_manager_desc(

Loading…
Cancel
Save