Skip to article frontmatterSkip to article content
Freqtrade 策略开发

Freqtrade 高级策略指南

高级策略开发概念和方法

高级策略

本页解释了一些可用于策略的高级概念。 如果你是初学者，请先熟悉 Freqtrade 基础 和 策略定制 中描述的方法。

这里描述的方法的调用顺序在 机器人执行逻辑 中有详细说明。这些文档也有助于决定哪种方法最适合你的定制需求。

存储信息（持久化）

Freqtrade 允许在数据库中存储/检索与特定交易相关的用户自定义信息。

使用交易对象，可以使用 trade.set_custom_data(key='my_key', value=my_value) 存储信息，使用 trade.get_custom_data(key='my_key') 检索信息。每个数据条目都与一个交易和一个用户提供的键（类型为 string）相关联。这意味着这只能在也提供交易对象的回调中使用。

为了使数据能够存储在数据库中，freqtrade 必须序列化数据。这是通过将数据转换为 JSON 格式的字符串来完成的。 Freqtrade 将在检索时尝试反转此操作，因此从策略的角度来看，这应该无关紧要。

from freqtrade.persistence import Trade
from datetime import timedelta

class AwesomeStrategy(IStrategy):

    def bot_loop_start(self, **kwargs) -> None:
        for trade in Trade.get_open_order_trades():
            fills = trade.select_filled_orders(trade.entry_side)
            if trade.pair == 'ETH/USDT':
                trade_entry_type = trade.get_custom_data(key='entry_type')
                if trade_entry_type is None:
                    trade_entry_type = 'breakout' if 'entry_1' in trade.enter_tag else 'dip'
                elif fills > 1:
                    trade_entry_type = 'buy_up'
                trade.set_custom_data(key='entry_type', value=trade_entry_type)
        return super().bot_loop_start(**kwargs)

    def adjust_entry_price(self, trade: Trade, order: Order | None, pair: str,
                           current_time: datetime, proposed_rate: float, current_order_rate: float,
                           entry_tag: str | None, side: str, **kwargs) -> float:
        # 对于 BTC/USDT 交易对，在入场触发后的前 10 分钟内使用限价单并跟随 SMA200 作为价格目标。
        if (
            pair == 'BTC/USDT' 
            and entry_tag == 'long_sma200' 
            and side == 'long' 
            and (current_time - timedelta(minutes=10)) > trade.open_date_utc 
            and order.filled == 0.0
        ):
            dataframe, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe)
            current_candle = dataframe.iloc[-1].squeeze()
            # 存储入场调整信息
            existing_count = trade.get_custom_data('num_entry_adjustments', default=0)
            if not existing_count:
                existing_count = 1
            else:
                existing_count += 1
            trade.set_custom_data(key='num_entry_adjustments', value=existing_count)

            # 调整订单价格
            return current_candle['sma_200']

        # 默认：维持现有订单
        return current_order_rate

    def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs):

        entry_adjustment_count = trade.get_custom_data(key='num_entry_adjustments')
        trade_entry_type = trade.get_custom_data(key='entry_type')
        if entry_adjustment_count is None:
            if current_profit > 0.01 and (current_time - timedelta(minutes=100) > trade.open_date_utc):
                return True, 'exit_1'
        else
            if entry_adjustment_count > 0 and if current_profit > 0.05:
                return True, 'exit_2'
            if trade_entry_type == 'breakout' and current_profit > 0.1:
                return True, 'exit_3

        return False, None

上面是一个简单的例子 - 有更简单的方法来检索交易数据，如入场调整。

存储信息（非持久化）

数据框访问

你可以通过从数据提供者查询来在各种策略函数中访问数据框。

from freqtrade.exchange import timeframe_to_prev_date

class AwesomeStrategy(IStrategy):
    def confirm_trade_exit(self, pair: str, trade: 'Trade', order_type: str, amount: float,
                           rate: float, time_in_force: str, exit_reason: str,
                           current_time: 'datetime', **kwargs) -> bool:
        # 获取交易对数据框。
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

        # 获取最后一个可用的蜡烛图。不要使用 current_time 查找最新的蜡烛图，因为
        # current_time 指向当前未完成的蜡烛图，其数据不可用。
        last_candle = dataframe.iloc[-1].squeeze()
        # <...>

        # 在模拟/实盘运行中，交易开始日期不会匹配蜡烛图开始日期，因此必须
        # 进行四舍五入。
        trade_date = timeframe_to_prev_date(self.timeframe, trade.open_date_utc)
        # 查找交易蜡烛图。
        trade_candle = dataframe.loc[dataframe['date'] == trade_date]
        # 对于刚刚开始的交易，trade_candle 可能为空，因为它仍然未完成。
        if not trade_candle.empty:
            trade_candle = trade_candle.squeeze()
            # <...>

入场标签

当你的策略有多个买入信号时，你可以命名触发的信号。 然后你可以在 custom_exit 中访问你的买入信号

def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    dataframe.loc[
        (
            (dataframe['rsi'] < 35) &
            (dataframe['volume'] > 0)
        ),
        ['enter_long', 'enter_tag']] = (1, 'buy_signal_rsi')

    return dataframe

def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
                current_profit: float, **kwargs):
    dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
    last_candle = dataframe.iloc[-1].squeeze()
    if trade.enter_tag == 'buy_signal_rsi' and last_candle['rsi'] > 80:
        return 'sell_signal_rsi'
    return None

出场标签

类似于入场标签，你也可以指定出场标签。

def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    dataframe.loc[
        (
            (dataframe['rsi'] > 70) &
            (dataframe['volume'] > 0)
        ),
        ['exit_long', 'exit_tag']] = (1, 'exit_rsi')

    return dataframe

提供的出场标签然后用作卖出原因 - 并在回测结果中显示为如此。

策略版本

你可以通过使用 “version” 方法来实现自定义策略版本控制，并返回你希望此策略具有的版本。

def version(self) -> str:
    """
    返回策略的版本。
    """
    return "1.1"

派生策略

策略可以从其他策略派生。这避免了自定义策略代码的重复。你可以使用这种技术来覆盖主策略的小部分，保持其余部分不变：

class MyAwesomeStrategy(IStrategy):
    ...
    stoploss = 0.13
    trailing_stop = False
    # 所有其他属性和方法都在这里，就像
    # 在任何自定义策略中一样...
    ...

from myawesomestrategy import MyAwesomeStrategy
class MyAwesomeStrategy2(MyAwesomeStrategy):
    # 覆盖某些内容
    stoploss = 0.08
    trailing_stop = True

属性和方法都可以被覆盖，以你需要的方式改变原始策略的行为。

虽然在技术上可以在同一文件中保持子类，但这可能会导致超优化参数文件的一些问题，因此我们建议使用单独的策略文件，并如上所示导入父策略。

嵌入策略

Freqtrade 为你提供了一种简单的方法来将策略嵌入到配置文件中。 这是通过利用 BASE64 编码并在你选择的配置文件的策略配置字段中提供此字符串来完成的。

将字符串编码为 BASE64

这是一个快速示例，如何在 python 中生成 BASE64 字符串

from base64 import urlsafe_b64encode

with open(file, 'r') as f:
    content = f.read()
content = urlsafe_b64encode(content.encode('utf-8'))

变量 ‘content’ 将包含 BASE64 编码形式的策略文件。现在可以在配置文件中设置如下

"strategy": "NameOfStrategy:BASE64String"

请确保 ‘NameOfStrategy’ 与策略名称完全相同！

性能警告

在执行策略时，有时会在日志中看到以下内容

PerformanceWarning: DataFrame is highly fragmented.

这是来自 pandas 的警告，正如警告继续说的： 使用 pd.concat(axis=1)。 这可能会有轻微的性能影响，通常只在超优化期间可见（在优化指标时）。

例如：

for val in self.buy_ema_short.range:
    dataframe[f'ema_short_{val}'] = ta.EMA(dataframe, timeperiod=val)

应该重写为

frames = [dataframe]
for val in self.buy_ema_short.range:
    frames.append(DataFrame({
        f'ema_short_{val}': ta.EMA(dataframe, timeperiod=val)
    }))

# 组合所有数据框，并重新分配原始数据框列
dataframe = pd.concat(frames, axis=1)

然而，Freqtrade 通过在 populate_indicators() 方法之后立即运行 dataframe.copy() 来抵消这一点 - 因此这的性能影响应该很低或不存在。

高级话题
递归分析
高级话题
高级超参数优化