# 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, Series from typing import Dict, Optional, Union, Tuple import logging 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 import pandas_ta as pta from technical import qtpylib # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[logging.StreamHandler()] ) logger = logging.getLogger(__name__) class MomentumCCITrendStrategy(IStrategy): """ 基于商品通道指数(CCI)的动量趋势跟踪策略 策略说明： - 使用长期CCI(50周期)判断整体趋势方向 - 使用短期CCI(5周期)的零轴穿越作为入场信号 - 通过状态追踪确保同一趋势周期内只交易一次 - 当长期CCI反向穿越零轴时平仓 """ # Strategy interface version INTERFACE_VERSION = 3 # 策略基础配置 can_short: bool = True timeframe = "1d" # 策略参数 cci_long_period = IntParameter(30, 100, default=50, space="buy") cci_short_period = IntParameter(3, 15, default=5, space="buy") enable_short_trading = BooleanParameter(default=True, space="buy") # 风险管理 stoploss = -0.10 # 10% 止损 # ROI (取利) 设置 minimal_roi = { "0": 0.15, # 15% 收益立即卖出 "60": 0.10, # 1小时后10%收益卖出 "120": 0.05, # 2小时后5%收益卖出 "240": 0.02 # 4小时后2%收益卖出 } # 订单类型配置 order_types = { "entry": "limit", "exit": "limit", "stoploss": "market", "stoploss_on_exchange": False } order_time_in_force = { "entry": "GTC", "exit": "GTC" } # 存储完整数据帧和状态信息 _full_dataframes = {} _pair_states = {} @property def plot_config(self): return { "main_plot": {}, "subplots": { "CCI": { "cci_long": {"color": "green", "type": "line"}, "cci_short": {"color": "yellow", "type": "line"}, "cci_zero": {"color": "gray", "type": "line", "value": 0} } } } def informative_pairs(self): """ 定义需要额外缓存的交易对/时间间隔组合 """ return [] def populate_indicators(self, dataframe: DataFrame, metadata: dict): """ 计算策略所需的技术指标 """ pair = metadata['pair'] # 计算长期和短期 CCI dataframe['cci_long'] = ta.CCI( dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=self.cci_long_period.value ) dataframe['cci_short'] = ta.CCI( dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=self.cci_short_period.value ) # 计算CCI穿越零轴的信号 dataframe['cci_long_cross_below_zero'] = qtpylib.crossed_below(dataframe['cci_long'], 0) dataframe['cci_short_cross_below_zero'] = qtpylib.crossed_below(dataframe['cci_short'], 0) # 初始化状态追踪列 dataframe['first_crossover_occurred'] = False dataframe['first_crossunder_occurred'] = False dataframe['buy_signal'] = False dataframe['sell_signal'] = False dataframe['exit_long_signal'] = False dataframe['exit_short_signal'] = False # 实现状态追踪逻辑 if pair not in self._pair_states: self._pair_states[pair] = { 'first_crossover_occurred': False, 'first_crossunder_occurred': False, 'last_cci_long_positive': False, 'last_cci_long_negative': False } state = self._pair_states[pair] for i in range(len(dataframe)): current_cci_long = dataframe.iloc[i]['cci_long'] current_cci_short = dataframe.iloc[i]['cci_short'] prev_cci_long = dataframe.iloc[i-1]['cci_long'] if i > 0 else 0 # 重置状态当CCI Long改变符号时 if i > 0: if prev_cci_long <= 0 and current_cci_long > 0: # CCI Long从负转正，重置crossunder状态 state['first_crossunder_occurred'] = False state['last_cci_long_positive'] = True state['last_cci_long_negative'] = False elif prev_cci_long >= 0 and current_cci_long < 0: # CCI Long从正转负，重置crossover状态 state['first_crossover_occurred'] = False state['last_cci_long_positive'] = False state['last_cci_long_negative'] = True # 更新dataframe中的状态 dataframe.iloc[i, dataframe.columns.get_loc('first_crossover_occurred')] = state['first_crossover_occurred'] dataframe.iloc[i, dataframe.columns.get_loc('first_crossunder_occurred')] = state['first_crossunder_occurred'] # 买入信号逻辑 (做多) buy_signal = ( current_cci_long > 0 and prev_cci_long > 0 and dataframe.iloc[i]['cci_short_cross_above_zero'] and not state['first_crossover_occurred'] ) if buy_signal: state['first_crossover_occurred'] = True dataframe.iloc[i, dataframe.columns.get_loc('buy_signal')] = True # 卖出信号逻辑 (做空) sell_signal = ( current_cci_long < 0 and prev_cci_long < 0 and dataframe.iloc[i]['cci_short_cross_below_zero'] and not state['first_crossunder_occurred'] ) if sell_signal: state['first_crossunder_occurred'] = True dataframe.iloc[i, dataframe.columns.get_loc('sell_signal')] = True # 平多仓信号 if dataframe.iloc[i]['cci_long_cross_below_zero']: dataframe.iloc[i, dataframe.columns.get_loc('exit_long_signal')] = True # 平空仓信号 if dataframe.iloc[i]['cci_long_cross_above_zero']: dataframe.iloc[i, dataframe.columns.get_loc('exit_short_signal')] = True # 保存完整数据帧 self._full_dataframes[pair] = dataframe.copy() logger.info(f"{pair}: CCI指标计算完成，长期CCI周期={self.cci_long_period.value}，短期CCI周期={self.cci_short_period.value}") return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict): """ 基于技术指标填充入场信号 """ # 做多入场条件 dataframe.loc[ (dataframe['buy_signal'] == True), 'enter_long' ] = 1 # 做空入场条件（如果启用） if self.enable_short_trading.value: dataframe.loc[ (dataframe['sell_signal'] == True), 'enter_short' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict): """ 基于技术指标填充出场信号 """ # 做多出场条件 dataframe.loc[ (dataframe['exit_long_signal'] == True), 'exit_long' ] = 1 # 做空出场条件（如果启用） if self.enable_short_trading.value: dataframe.loc[ (dataframe['exit_short_signal'] == True), 'exit_short' ] = 1 return dataframe def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: str | None, side: str, **kwargs): """ 确认交易入场 """ try: logger.info(f"{pair}: 确认{side}交易入场 - 价格: {rate}, 数量: {amount}, 时间: {current_time}") return True except Exception as e: logger.error(f"{pair}: 确认交易入场时出错: {e}") return False def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs): """ 自定义退出逻辑 """ try: # 获取当前数据 dataframe = self._full_dataframes.get(pair) if dataframe is None: return None # 找到当前时间对应的数据行 current_data = dataframe[dataframe['date'] <= current_time] if current_data.empty: return None current_row = current_data.iloc[-1] # 检查是否触发退出条件 if trade.is_short: # 空头仓位检查 if current_row['exit_short_signal']: logger.info(f"{pair}: CCI信号触发空头退出") return "CCI Exit Short" else: # 多头仓位检查 if current_row['exit_long_signal']: logger.info(f"{pair}: CCI信号触发多头退出") return "CCI Exit Long" return None except Exception as e: logger.error(f"{pair}: 自定义退出逻辑出错: {e}") return None def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str, side: str, **kwargs): """ 调整杠杆率 """ return 3 # 使用3倍杠杆