121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
import logging
|
||
from datetime import datetime, timedelta
|
||
from typing import Any
|
||
|
||
from freqtrade.constants import Config, LongShort
|
||
from freqtrade.persistence import Trade
|
||
from freqtrade.plugins.protections.iprotection import IProtection, ProtectionReturn
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class TotalLossProtection(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)
|
||
|
||
# 配置参数：10根半小时K线的亏损阈值（10美元）
|
||
self._threshold_10_candles = protection_config.get("threshold_10_candles", 10.0)
|
||
# 配置参数：20根半小时K线的亏损阈值（15美元）
|
||
self._threshold_20_candles = protection_config.get("threshold_20_candles", 15.0)
|
||
|
||
def short_desc(self) -> str:
|
||
"""
|
||
Short method description - used for startup-messages
|
||
"""
|
||
return (
|
||
f"{self.name} - Total Loss Protection, stop trading if loss > {self._threshold_10_candles} within last 10 30min candles or > {self._threshold_20_candles} within last 20 30min candles."
|
||
)
|
||
|
||
def _reason(self, lookback_period: int, total_loss: float, threshold: float) -> str:
|
||
"""
|
||
Lock reason text
|
||
"""
|
||
return (
|
||
f"Total Loss {total_loss:.2f} > {threshold} in {lookback_period} 30min candles, "
|
||
f"locking {self.unlock_reason_time_element}."
|
||
)
|
||
|
||
def _total_loss_check(
|
||
self, date_now: datetime, side: LongShort
|
||
) -> ProtectionReturn | None:
|
||
"""
|
||
检查最近10根和20根半小时K线内的总亏损
|
||
1根半小时K线 = 30分钟
|
||
10根半小时K线 = 5小时
|
||
20根半小时K线 = 10小时
|
||
"""
|
||
# 计算10根半小时K线的回溯时间（5小时）
|
||
look_back_until_10_candles = date_now - timedelta(hours=5)
|
||
# 计算20根半小时K线的回溯时间（10小时）
|
||
look_back_until_20_candles = date_now - timedelta(hours=10)
|
||
|
||
# 获取10根K线时间窗口内的所有已关闭交易
|
||
trades_10_candles = Trade.get_trades_proxy(
|
||
is_open=False,
|
||
close_date=look_back_until_10_candles
|
||
)
|
||
|
||
# 获取20根K线时间窗口内的所有已关闭交易
|
||
trades_20_candles = Trade.get_trades_proxy(
|
||
is_open=False,
|
||
close_date=look_back_until_20_candles
|
||
)
|
||
|
||
# 计算10根K线时间窗口内的总亏损
|
||
total_loss_10_candles = sum(
|
||
min(0, trade.close_profit) # 只计算亏损部分
|
||
for trade in trades_10_candles
|
||
if trade.close_profit is not None
|
||
) * -1 # 转换为正数表示亏损金额
|
||
|
||
# 计算20根K线时间窗口内的总亏损
|
||
total_loss_20_candles = sum(
|
||
min(0, trade.close_profit) # 只计算亏损部分
|
||
for trade in trades_20_candles
|
||
if trade.close_profit is not None
|
||
) * -1 # 转换为正数表示亏损金额
|
||
|
||
# 检查是否触发任一条件
|
||
if total_loss_10_candles > self._threshold_10_candles and trades_10_candles:
|
||
self.log_once(
|
||
f"Trading stopped due to Total Loss Protection: {total_loss_10_candles:.2f} > {self._threshold_10_candles} within last 10 30min candles.",
|
||
logger.info,
|
||
)
|
||
until = self.calculate_lock_end(trades_10_candles)
|
||
|
||
return ProtectionReturn(
|
||
lock=True,
|
||
until=until,
|
||
reason=self._reason(10, total_loss_10_candles, self._threshold_10_candles),
|
||
)
|
||
elif total_loss_20_candles > self._threshold_20_candles and trades_20_candles:
|
||
self.log_once(
|
||
f"Trading stopped due to Total Loss Protection: {total_loss_20_candles:.2f} > {self._threshold_20_candles} within last 20 30min candles.",
|
||
logger.info,
|
||
)
|
||
until = self.calculate_lock_end(trades_20_candles)
|
||
|
||
return ProtectionReturn(
|
||
lock=True,
|
||
until=until,
|
||
reason=self._reason(20, total_loss_20_candles, self._threshold_20_candles),
|
||
)
|
||
|
||
return None
|
||
|
||
def global_stop(self, date_now: datetime, side: LongShort) -> ProtectionReturn | None:
|
||
"""
|
||
停止所有交易对的交易（开仓）
|
||
"""
|
||
return self._total_loss_check(date_now, side)
|
||
|
||
def stop_per_pair(
|
||
self, pair: str, date_now: datetime, side: LongShort
|
||
) -> ProtectionReturn | None:
|
||
"""
|
||
本保护机制不支持单个交易对的停止
|
||
"""
|
||
return None |