myTestFreqAI/freqtrade/templates/v7.py

from freqtrade.strategy import IStrategy
from pandas import DataFrame
import talib.abstract as ta
from typing import Optional, Union
from freqtrade.persistence import Trade
from datetime import datetime

class TheForceV7(IStrategy):
    # 基础参数
    timeframe = '5m'
    stoploss = -0.14 # 全局止损
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    @property
    def protections(self):
        return [
            {
                "method": "CooldownPeriod",
                "stop_duration_candles": 5  # 卖出后禁止再次买入的K线数
            },
            {
                "method": "MaxDrawdown",
                "lookback_period_candles": 48,
                "trade_limit": 20,
                "stop_duration_candles": 4,
                "max_allowed_drawdown": 0.2  # 最大允许回撤 20%
            },
            {
                "method": "StoplossGuard",
                "lookback_period_candles": 24,
                "trade_limit": 4,  # 在 lookback 内最多触发几次止损
                "stop_duration_candles": 2,  # 锁定多少根 K 线
                "only_per_pair": False  # 是否按币种统计
            },
            {
                "method": "LowProfitPairs",
                "lookback_period_candles": 6,
                "trade_limit": 2,
                "stop_duration_candles": 60,
                "required_profit": 0.02  # 最低平均收益 2%
            },
            {
                "method": "LowProfitPairs",
                "lookback_period_candles": 24,
                "trade_limit": 4,
                "stop_duration_candles": 2,
                "required_profit": 0.01  # 最低平均收益 1%
            }
        ]    

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe['ema200c'] = ta.EMA(dataframe['close'], timeperiod=200)
        dataframe['ema50c'] = ta.EMA(dataframe['close'], timeperiod=50)
        dataframe['ema20c'] = ta.EMA(dataframe['close'], timeperiod=20)
        dataframe['rsi7'] = ta.RSI(dataframe['close'], timeperiod=7)
        macd = ta.MACD(dataframe['close'])
        dataframe['macd'] = macd[0]
        dataframe['macdsignal'] = macd[1]
        stoch = ta.STOCH(dataframe['high'], dataframe['low'], dataframe['close'], fastk_period=14, slowk_period=3, slowd_period=3)
        dataframe['slowk'] = stoch[0]
        dataframe['slowd'] = stoch[1]
        dataframe['adx'] = ta.ADX(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14)
        dataframe['volvar'] = dataframe['volume'].rolling(window=20).mean()

        return dataframe

    def crossover(self, series1: DataFrame, series2: DataFrame) -> DataFrame:
        """Detects when series1 crosses above series2."""
        return (series1 > series2) & (series1.shift(1) <= series2.shift(1))

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[
            (dataframe['close'] > dataframe['ema200c']) &  # Relaxed trend
            (dataframe['close'] > dataframe['ema50c']) &  # Short-term trend
            (dataframe['rsi7'] < 50) &  # Relaxed RSI
            (dataframe['macd'] > 0) &  # Relaxed MACD
            (dataframe['volume'] > dataframe['volvar'] * 0.5) &  # Relaxed volume
            (dataframe['adx'] > 20),  # Trend strength
            'enter_long'
        ] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[
            (
                (dataframe['slowk'] > 65) | (dataframe['slowd'] > 65) |  # Relaxed STOCH
                (dataframe['rsi7'] > 70)  # Overbought
            ) &
            (dataframe['close'] < dataframe['ema20c']) &  # Trend reversal
            (self.crossover(dataframe['macdsignal'], dataframe['macd'])) &  # Custom crossover
            (dataframe['macd'] < 0),  # Downtrend
            'exit_long'
        ] = 1

        return dataframe

    def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
                    current_profit: float, **kwargs) -> Optional[Union[str, bool]]:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty or dataframe['date'].iloc[-1] < current_time:  # Fixed: Use date column
            return None

        last_candle = dataframe.iloc[-1]
        atr = ta.ATR(dataframe, timeperiod=14).iloc[-1]
        duration = (current_time - trade.open_date).total_seconds() / 60  # Minutes

        # Dynamic Take-Profit
        take_profit = current_rate + 1.0 * atr  # Lowered ATR
        if current_rate >= take_profit:
            return "take_profit"

        # Partial Take-Profit at 2%
        if current_profit >= 0.02:
            return "partial_take_profit"

        # Dynamic Stop-Loss
        stop_loss = current_rate - 2.0 * atr  # Relaxed ATR
        if current_rate <= stop_loss:
            return "dynamic_stop_loss"

        # Trailing Stop
        if current_profit > 0.005:  # Lowered threshold
            self.trailing_stop = True
            self.trailing_stop_positive = 0.003  # 0.3% retracement
            self.trailing_stop_positive_offset = 0.008  # 0.8% offset
            if current_profit < trade.max_profit - self.trailing_stop_positive:
                return "trailing_stop"

        # Time Stop
        if duration > 60:  # 1 hour
            return "time_stop"

        return None

    def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
                           proposed_stake: float, min_stake: float, max_stake: float,
                           leverage: float, entry_tag: Optional[str], side: str,
                           **kwargs) -> float:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty:
            return proposed_stake

        atr = ta.ATR(dataframe, timeperiod=14).iloc[-1]
        price_std = dataframe['close'].std()
        combined_volatility = atr + price_std

        base_stake = self.wallets.get_total_stake_amount() * 0.0333  # 3.33% risk
        base_stake = min(base_stake, 50.0)  # Cap at 50 USDT

        risk_factor = 1.0
        if combined_volatility > current_rate * 0.03:  # High volatility
            risk_factor = 0.3 if pair in ['SOL/USDT', 'OKB/USDT'] else 0.5
        elif combined_volatility < current_rate * 0.01:  # Low volatility
            risk_factor = 1.2 if pair in ['BTC/USDT', 'ETH/USDT'] else 1.0

        return base_stake * risk_factor