# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these imports --- import numpy as np import pandas as pd from datetime import datetime, timedelta, timezone from pandas import DataFrame from typing import Optional, Union from freqtrade.strategy import ( IStrategy, Trade, Order, PairLocks, informative, # @informative decorator # Hyperopt Parameters BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, RealParameter, # timeframe helpers timeframe_to_minutes, timeframe_to_next_date, timeframe_to_prev_date, # Strategy helper functions merge_informative_pair, stoploss_from_absolute, stoploss_from_open, ) # -------------------------------- # Add your lib to import here import talib.abstract as ta from technical import qtpylib # 基于x.py策略的MA60/MA120趋势跟踪策略 class MA60MA120TrendStrategy(IStrategy): """ 基于x.py策略的MA60/MA120趋势跟踪策略 策略逻辑： 1. 买入条件： - 价格在MA60上方且距离MA60不超过10% - MA60呈上升趋势（过去5天中至少3天上升） - 成交量放大（当日成交量大于过去5日平均） - 前5天中下跌天数不超过2天 - 当天收盘价大于5天前的价格 - 价格在MA120上方 - MA120过去5天不是下降趋势 - RSI小于阈值（默认70） 2. 卖出条件： - 价格跌破MA60且跌幅超过10% 更多信息: https://www.freqtrade.io/en/latest/strategy-customization/ """ # Strategy interface version - allow new iterations of the strategy interface. # Check the documentation or the Sample strategy to get the latest version. INTERFACE_VERSION = 3 # Can this strategy go short? can_short: bool = False # Minimal ROI designed for the strategy. # 基于x.py策略，使用更保守的ROI设置 # minimal_roi = { # "30": 0.3, # 20% 目标收益 # # "" # } # Optimal stoploss designed for the strategy. # 基于x.py策略的卖出条件：跌破MA60且跌幅超过10% stoploss = -1 # Trailing stoploss # trailing_stop = True # trailing_only_offset_is_reached = False # trailing_stop_positive = 0.3 # trailing_stop_positive_offset = 0.0 # Disabled / not configured # Optimal timeframe for the strategy. # 使用日线数据，因为x.py策略基于日线 timeframe = "1d" # Run "populate_indicators()" only for new candle. process_only_new_candles = True # These values can be overridden in the config. use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # 移除原有的RSI参数，因为新策略使用固定值 # 可以添加其他可优化的参数 ma60_period = IntParameter(low=30, high=90, default=60, space="buy", optimize=True, load=True) ma120_period = IntParameter(low=60, high=180, default=120, space="buy", optimize=True, load=True) rsi_threshold = IntParameter(low=50, high=80, default=70, space="buy", optimize=True, load=True) volume_surge_multiplier = DecimalParameter(low=0.5, high=2.0, default=1.0, space="buy", optimize=True, load=True) # Number of candles the strategy requires before producing valid signals # 需要更多数据来计算MA120和趋势指标 startup_candle_count: int = 200 # Optional order type mapping. order_types = { "entry": "limit", "exit": "limit", "stoploss": "market", "stoploss_on_exchange": False, } # Optional order time in force. order_time_in_force = {"entry": "GTC", "exit": "GTC"} plot_config = { "main_plot": { "ma60": {"color": "blue", "width": 2}, "ma120": {"color": "orange", "width": 2}, }, "subplots": { "RSI": { "rsi": {"color": "red"}, "rsi_threshold": {"color": "gray", "type": "line", "width": 1}, }, "Volume": { "volume": {"color": "lightblue"}, "volume_5ma": {"color": "blue"}, }, "Signals": { "buy_signal": {"color": "green", "type": "scatter", "marker": "^"}, "sell_signal": {"color": "red", "type": "scatter", "marker": "v"}, }, }, } def informative_pairs(self): """ Define additional, informative pair/interval combinations to be cached from the exchange. These pair/interval combinations are non-tradeable, unless they are part of the whitelist as well. For more information, please consult the documentation :return: List of tuples in the format (pair, interval) Sample: return [("ETH/USDT", "5m"), ("BTC/USDT", "15m"), ] """ return [] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 基于x.py策略的技术指标计算 实现MA60/MA120趋势跟踪策略的所有必要指标 :param dataframe: Dataframe with data from the exchange :param metadata: Additional information, like the currently traded pair :return: a Dataframe with all mandatory indicators for the strategies """ # 基础移动平均线指标 # ------------------------------------ # MA60 - 使用可优化参数 dataframe['ma60'] = ta.SMA(dataframe, timeperiod=60) # MA120 - 使用可优化参数 dataframe['ma120'] = ta.SMA(dataframe, timeperiod=120) # RSI指标 dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) # 成交量相关指标 # ------------------------------------ # 成交量5日移动平均 dataframe['volume_5ma'] = dataframe['volume'].rolling(window=5).mean() # 成交量放大信号：当日成交量大于过去5日平均的倍数 dataframe['volume_surge'] = dataframe['volume'] > (dataframe['volume_5ma'] * 1) # 价格与MA60关系指标 # ------------------------------------ # 价格在MA60上方且距离MA60不超过10% dataframe['price_above_ma60'] = ( (dataframe['close'] > dataframe['ma60']) & ((dataframe['ma60'] - dataframe['close']) / dataframe['close'] < 0.1) ) # 价格在MA120上方 dataframe['price_above_ma120'] = dataframe['close'] > dataframe['ma120'] # 前一天在MA60附近（95%-105%） dataframe['prev_near_ma'] = ( (dataframe['close'].shift(1) >= dataframe['ma60'].shift(1) * 0.95) & (dataframe['close'].shift(1) <= dataframe['ma60'].shift(1) * 1.05) ) # 前一天在MA60下方 dataframe['prev_below_ma'] = dataframe['close'].shift(1) < dataframe['ma60'].shift(1) # 突破信号：前一天在MA60附近或下方，当天在MA60上方 dataframe['breakthrough'] = ( (dataframe['prev_below_ma'] | dataframe['prev_near_ma']) & dataframe['price_above_ma60'] ) # MA60趋势指标 # ------------------------------------ # 计算MA60的上升趋势（过去5天中至少3天上升） trend_days = 5 ma_trend = [] for i in range(len(dataframe)): if i < trend_days - 1: ma_trend.append(False) else: start_idx = i - trend_days + 1 ma_values = dataframe['ma60'].iloc[start_idx:i+1].values if pd.isna(ma_values).any(): ma_trend.append(False) else: # 检查是否呈上升趋势（至少3天上升） ma_trend.append(sum(ma_values[1:] > ma_values[:-1]) > trend_days - 3) dataframe['ma_uptrend'] = ma_trend # MA120趋势指标 # ------------------------------------ # 计算MA120的趋势（过去5天）不能为下降 ma120_trend = [] for i in range(len(dataframe)): if i < 4: ma120_trend.append(True) # 数据不足时不限制 else: window = dataframe['ma120'].iloc[i-4:i+1].values if pd.isna(window).any(): ma120_trend.append(True) else: # 只要有一天不是上升就不是下降趋势 ma120_trend.append((window[1:] >= window[:-1]).all()) dataframe['ma120_not_downtrend'] = ma120_trend # 价格动量指标 # ------------------------------------ # 阳线条件：收盘价 > 开盘价，且阳线幅度至少0.5% dataframe['is_bullish'] = ( (dataframe['close'] > dataframe['open']) & ((dataframe['close'] - dataframe['open']) / dataframe['open'] >= 0.005) ) # 前5天的下跌天数不超过2天 dataframe['is_down_day'] = dataframe['close'] < dataframe['open'] dataframe['down_days_in_5'] = dataframe['is_down_day'].rolling(window=5, min_periods=1).sum() dataframe['not_too_many_downs'] = dataframe['down_days_in_5'] <= 2 # 当天收盘价大于第五天前的价格 dataframe['price_higher_than_5days_ago'] = dataframe['close'] > dataframe['close'].shift(5) # RSI条件 - 使用可优化参数 dataframe['rsi_lt_threshold'] = dataframe['rsi'] < 70 # 综合买入信号 # ------------------------------------ dataframe["buy_signal"] = ( dataframe["price_above_ma60"] & dataframe["ma_uptrend"] & dataframe['ma60'].notna() & dataframe["volume_surge"] & dataframe["not_too_many_downs"] & dataframe["price_higher_than_5days_ago"] & dataframe['price_above_ma120'] & dataframe['ma120_not_downtrend'] # & # dataframe['rsi_lt_threshold'] ) # 卖出信号：价格跌破MA60且跌幅超过10% dataframe['sell_signal'] = ( (~dataframe['price_above_ma60']) & ((dataframe['ma60'] - dataframe['close']) / dataframe['ma60'] > 0.1) & dataframe['ma60'].notna() ) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 基于x.py策略的买入信号逻辑 实现MA60/MA120趋势跟踪策略的买入条件 :param dataframe: DataFrame :param metadata: Additional information, like the currently traded pair :return: DataFrame with entry columns populated """ # 多头买入信号 - 基于x.py的buy_signal逻辑 dataframe.loc[ ( # 价格在MA60上方且距离MA60不超过10% dataframe["price_above_ma60"] & # MA60呈上升趋势（过去5天中至少3天上升） dataframe["ma_uptrend"] & # MA60数据有效 dataframe['ma60'].notna() & # 成交量放大（当日成交量大于过去5日平均） dataframe["volume_surge"] & # 前5天中下跌天数不超过2天 dataframe["not_too_many_downs"] & # 当天收盘价大于5天前的价格 dataframe["price_higher_than_5days_ago"] & # 价格在MA120上方 dataframe['price_above_ma120'] & # MA120过去5天不是下降趋势 dataframe['ma120_not_downtrend'] & # RSI小于阈值 dataframe['rsi_lt_threshold'] & # 确保成交量不为0 (dataframe["volume"] > 0) ), "enter_long", ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 基于x.py策略的卖出信号逻辑 实现MA60/MA120趋势跟踪策略的卖出条件 :param dataframe: DataFrame :param metadata: Additional information, like the currently traded pair :return: DataFrame with exit columns populated """ # 多头卖出信号 - 基于x.py的sell_signal逻辑 # dataframe.loc[ # ( # # 价格跌破MA60且跌幅超过10% # (~dataframe['price_above_ma60']) & # ((dataframe['ma60'] - dataframe['close']) / dataframe['ma60'] > 0.1) & # dataframe['ma60'].notna() & # # 确保成交量不为0 # (dataframe["volume"] > 0) # ), # "exit_long", # ] = 1 dataframe.loc[ ( # 价格跌破MA60且跌幅超过10% (dataframe['close'] < dataframe['ma120']) & dataframe['ma120'].notna() & # 确保成交量不为0 (dataframe["volume"] > 0) ), "exit_long", ] = 1 return dataframe