122 lines
4.7 KiB
Python
122 lines
4.7 KiB
Python
import logging
|
||
from datetime import datetime, timedelta
|
||
from typing import Any
|
||
import logging
|
||
|
||
from freqtrade.constants import Config, LongShort
|
||
from freqtrade.persistence import Trade
|
||
from freqtrade.plugins.protections import IProtection, ProtectionReturn
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ConsecutiveLossProtection(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._max_consecutive_losses = protection_config.get("max_consecutive_losses", 3)
|
||
self._stop_duration = protection_config.get("stop_duration", 30)
|
||
self._lookback_period = protection_config.get("lookback_period", 120)
|
||
|
||
# 设置默认值，对应20根K线内3次亏损，锁5根K线（新参数名，基于candles）
|
||
# 优先使用基于candles的配置参数
|
||
self._loss_limit = protection_config.get("loss_limit", self._max_consecutive_losses)
|
||
|
||
# 确保lookback_period是基于candles计算的
|
||
from freqtrade.exchange import timeframe_to_minutes
|
||
tf_in_min = timeframe_to_minutes(config["timeframe"])
|
||
|
||
# 计算基于candles的参数值
|
||
if "lookback_period_candles" in protection_config:
|
||
self._lookback_period_candles = protection_config["lookback_period_candles"]
|
||
self._lookback_period = tf_in_min * self._lookback_period_candles
|
||
else:
|
||
# 从分钟转换为candles数
|
||
self._lookback_period_candles = int(self._lookback_period / tf_in_min)
|
||
|
||
if "stop_duration_candles" in protection_config:
|
||
self._stop_duration_candles = protection_config["stop_duration_candles"]
|
||
else:
|
||
# 从分钟转换为candles数
|
||
self._stop_duration_candles = int(self._stop_duration / tf_in_min)
|
||
|
||
def short_desc(self) -> str:
|
||
"""
|
||
Short method description - used for startup-messages
|
||
"""
|
||
return (
|
||
f"{self.name} - Loss Count Protection, locks pairs with "
|
||
f"{self._loss_limit} losses within {self._lookback_period_candles} candles, "
|
||
f"locking for {self._stop_duration_candles} candles."
|
||
)
|
||
|
||
def _reason(self, loss_count: int) -> str:
|
||
"""
|
||
LockReason to use
|
||
"""
|
||
return (
|
||
f"{loss_count} losses within {self._lookback_period_candles} candles, "
|
||
f"locking for {self._stop_duration_candles} candles."
|
||
)
|
||
|
||
def _loss_count_check(
|
||
self, date_now: datetime, pair: str, side: LongShort
|
||
) -> ProtectionReturn | None:
|
||
"""
|
||
Evaluate recent trades for pair to check for number of losses
|
||
"""
|
||
look_back_until = date_now - timedelta(minutes=self._lookback_period)
|
||
|
||
# Get closed trades for the pair within the lookback period
|
||
trades = Trade.get_trades_proxy(
|
||
pair=pair, is_open=False, close_date=look_back_until
|
||
)
|
||
|
||
if not trades:
|
||
return None
|
||
|
||
# Count the number of losing trades in the lookback period
|
||
loss_count = sum(1 for trade in trades if trade.close_profit and trade.close_profit < 0)
|
||
|
||
if loss_count >= self._loss_limit:
|
||
self.log_once(
|
||
f"Trading for {pair} stopped due to {loss_count} losses within "
|
||
f"{self._lookback_period_candles} candles.",
|
||
logger.info,
|
||
)
|
||
until = self.calculate_lock_end(trades)
|
||
|
||
return ProtectionReturn(
|
||
lock=True,
|
||
until=until,
|
||
reason=self._reason(loss_count),
|
||
lock_side=side,
|
||
)
|
||
|
||
return None
|
||
|
||
def global_stop(self, date_now: datetime, side: LongShort) -> ProtectionReturn | None:
|
||
"""
|
||
Stops trading (position entering) for all pairs
|
||
This must evaluate to true for the whole period of the "cooldown period".
|
||
:return: Tuple of [bool, until, reason].
|
||
If true, all pairs will be locked with <reason> until <until>
|
||
"""
|
||
return None
|
||
|
||
def stop_per_pair(
|
||
self, pair: str, date_now: datetime, side: LongShort
|
||
) -> ProtectionReturn | None:
|
||
"""
|
||
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>
|
||
"""
|
||
return self._loss_count_check(date_now, pair=pair, side=side) |