# source: https://raw.githubusercontent.com/xielk/freqtrade-test/925e067f080172a4581bc71ae9ee64be247606f4/user_data/strategies/FibonacciStrategy.py # 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 Dict, Optional, Union, Tuple 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, AnnotationType, ) # -------------------------------- # Add your lib to import here import talib.abstract as ta from technical import qtpylib class Github_xielk_freqtrade_test__FibonacciStrategy__20250910_084214(IStrategy): """ 基于斐波那契回归的15分钟交易策略 - 简化版本 策略逻辑： 1. 计算最近N个周期的最高价和最低价 2. 基于价格区间计算斐波那契回归水平 3. 在关键回归位附近寻找买入和卖出机会 4. 结合RSI确认信号 """ # Strategy interface version INTERFACE_VERSION = 3 # 策略时间框架 - 15分钟 timeframe = "15m" # 是否支持做空 can_short: bool = False # 最小ROI设置 minimal_roi = { "120": 0.01, # 2小时后1%收益 "60": 0.02, # 1小时后2%收益 "30": 0.03, # 30分钟后3%收益 "0": 0.05 # 立即5%收益 } # 止损设置 stoploss = -0.06 # 6%止损 # 追踪止损 trailing_stop = True trailing_stop_positive = 0.02 trailing_stop_positive_offset = 0.03 trailing_only_offset_is_reached = True # 只处理新K线 process_only_new_candles = True # 策略参数 use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # 策略启动所需的K线数量 startup_candle_count: int = 50 # 斐波那契策略参数 fib_period = IntParameter(20, 100, default=30, space="buy") # 计算斐波那契的周期数 rsi_period = IntParameter(10, 30, default=14, space="buy") # RSI周期 rsi_oversold = IntParameter(25, 45, default=35, space="buy") # RSI超卖线 rsi_overbought = IntParameter(55, 75, default=65, space="sell") # RSI超买线 # 订单类型 order_types = { "entry": "limit", "exit": "limit", "stoploss": "market", "stoploss_on_exchange": False } # 订单时间 order_time_in_force = { "entry": "GTC", "exit": "GTC" } @property def plot_config(self): return { "main_plot": { "fib_0.236": {"color": "red", "type": "line"}, "fib_0.382": {"color": "orange", "type": "line"}, "fib_0.5": {"color": "yellow", "type": "line"}, "fib_0.618": {"color": "green", "type": "line"}, "fib_0.786": {"color": "blue", "type": "line"}, }, "subplots": { "RSI": { "rsi": {"color": "purple"}, } } } def informative_pairs(self): """ 定义额外的信息性交易对 """ return [] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 计算技术指标 """ # 计算RSI dataframe["rsi"] = ta.RSI(dataframe, timeperiod=self.rsi_period.value) # 计算斐波那契回归水平 # 使用滚动窗口计算最高价和最低价 period = self.fib_period.value dataframe["highest"] = dataframe["high"].rolling(window=period).max() dataframe["lowest"] = dataframe["low"].rolling(window=period).min() # 计算价格区间 dataframe["price_range"] = dataframe["highest"] - dataframe["lowest"] # 计算斐波那契回归水平 dataframe["fib_0.236"] = dataframe["highest"] - 0.236 * dataframe["price_range"] dataframe["fib_0.382"] = dataframe["highest"] - 0.382 * dataframe["price_range"] dataframe["fib_0.5"] = dataframe["highest"] - 0.5 * dataframe["price_range"] dataframe["fib_0.618"] = dataframe["highest"] - 0.618 * dataframe["price_range"] dataframe["fib_0.786"] = dataframe["highest"] - 0.786 * dataframe["price_range"] # 计算EMA作为趋势确认 dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20) dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50) # 计算MACD macd = ta.MACD(dataframe) dataframe["macd"] = macd["macd"] dataframe["macdsignal"] = macd["macdsignal"] dataframe["macdhist"] = macd["macdhist"] return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 基于斐波那契回归的买入信号 - 简化版本 """ dataframe.loc[ ( # 价格在0.618斐波那契回归位附近（关键支撑位） (dataframe["close"] <= dataframe["fib_0.618"] * 1.02) & # 允许2%的误差 (dataframe["close"] >= dataframe["fib_0.618"] * 0.98) & # RSI显示超卖状态 (dataframe["rsi"] <= self.rsi_oversold.value) & # 价格在EMA20之上（趋势确认） (dataframe["close"] > dataframe["ema_20"]) & # 确保有足够的数据 (dataframe["volume"] > 0) & (dataframe["price_range"] > 0) & (dataframe["highest"].notna()) & (dataframe["lowest"].notna()) ), "enter_long" ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 基于斐波那契回归的卖出信号 - 简化版本 """ dataframe.loc[ ( # 价格在0.382斐波那契回归位附近（关键阻力位） (dataframe["close"] >= dataframe["fib_0.382"] * 0.98) & # 允许2%的误差 (dataframe["close"] <= dataframe["fib_0.382"] * 1.02) & # RSI显示超买状态 (dataframe["rsi"] >= self.rsi_overbought.value) & # 确保有足够的数据 (dataframe["volume"] > 0) & (dataframe["price_range"] > 0) & (dataframe["highest"].notna()) & (dataframe["lowest"].notna()) ), "exit_long" ] = 1 return dataframe def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ 自定义止损逻辑 当价格突破关键斐波那契水平时调整止损 """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) == 0: return self.stoploss last_candle = dataframe.iloc[-1].squeeze() # 如果价格跌破0.786斐波那契水平，收紧止损 if current_rate <= last_candle["fib_0.786"]: return -0.04 # 4%止损 # 如果价格突破0.5斐波那契水平，设置保本止损 if current_rate >= last_candle["fib_0.5"]: return -0.01 # 1%止损（接近保本） # 默认止损 return self.stoploss