高级策略 (Advanced Strategies)

本页解释了策略中可用的一些高级概念。 如果您刚开始接触，请先熟悉 Freqtrade 基础 以及 策略自定义 中描述的方法。

此处描述的方法调用顺序包含在 机器人执行逻辑 中。这些文档也有助于决定哪种方法最适合您的自定义需求。

INFO

回调方法应 仅 在策略使用它们时才被实现。

提示

通过运行 freqtrade new-strategy --strategy MyAwesomeStrategy --template advanced，可以从包含所有可用回调方法的策略模板开始。

存储信息 (持久化)

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 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

上述是一个简单的示例 —— 还有更简单的方法来检索交易数据，如进场调整（entry-adjustments）。

INFO

建议使用简单的数据类型 [bool, int, float, str]，以确保在序列化需要存储的数据时不会出现问题。存储大块数据可能会导致意外的副作用，例如数据库变得过大（从而变慢）。

不可序列化的数据

如果提供的数据无法序列化，将记录警告日志，并且指定 key 的条目将包含 None 作为数据。

所有属性

自定义数据（custom-data）通过 Trade 对象（下文假设为 trade）具有以下访问器：

• trade.get_custom_data(key='something', default=0) - 以提供的类型返回实际值。
• trade.get_custom_data_entry(key='something') - 返回条目 —— 包括元数据。该值可通过 .value 属性访问。
• trade.set_custom_data(key='something', value={'some': 'value'}) - 设置或更新此交易的相应键。值必须是可序列化的 —— 并且我们建议保持存储的数据相对较小。

"value" 可以是任何类型（设置和接收时都是） —— 但必须是 JSON 可序列化的。

存储信息 (非持久化)

已弃用

这种存储信息的方法已被弃用，我们建议不要使用非持久化存储。请改用 持久化存储。以下内容已折叠。

存储信息

存储信息可以通过在策略类中创建一个新字典来实现。变量名可以随意选取，但应加前缀 custom_ 以避免与预定义的策略变量发生命名冲突。

class AwesomeStrategy(IStrategy):
    # 创建自定义字典
    custom_info = {}

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 检查条目是否已存在
        if not metadata["pair"] in self.custom_info:
            # 为该交易对创建空条目
            self.custom_info[metadata["pair"]] = {}

        if "crosstime" in self.custom_info[metadata["pair"]]:
            self.custom_info[metadata["pair"]]["crosstime"] += 1
        else:
            self.custom_info[metadata["pair"]]["crosstime"] = 1

WARNING

在机器人重启（或配置重新加载）后，数据不会被持久化。此外，数据量应保持较小（不要存放 DataFrame 等），否则机器人将开始消耗大量内存，最终可能因内存不足而崩溃。

INFO

如果数据是特定于交易对的，请确保使用交易对作为字典中的键之一。

数据帧 (Dataframe) 访问

您可以通过从数据提供者（dataprovider）查询，在各种策略函数中访问数据帧。

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)

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

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

使用 .iloc[-1]

您可以在这里使用 .iloc[-1]，因为 get_analyzed_dataframe() 仅返回回测允许看到的 K 线。这在 populate_* 方法中不起作用，因此请确保不要在那个区域使用 .iloc[]。此外，这仅在 2021.5 及更高版本中有效。

---

进场标签 (Enter Tag)

当您的策略有多个进场信号时，您可以为触发的信号命名。然后您可以在 custom_exit 中访问您的进场信号。

def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    dataframe["enter_tag"] = ""
    signal_rsi = (qtpylib.crossed_above(dataframe["rsi"], 35))
    signal_bblower = (dataframe["bb_lowerband"] < dataframe["close"])
    # 其他条件
    dataframe.loc[
        (
            signal_rsi
            | signal_bblower
            # ... 入场做多的其他信号
        )
        & (dataframe["volume"] > 0)
            , "enter_long"
        ] = 1
    # 拼接标签，以便保留所有信号
    dataframe.loc[signal_rsi, "enter_tag"] += "long_signal_rsi "
    dataframe.loc[signal_bblower, "enter_tag"] += "long_signal_bblower "

    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 "long_signal_rsi" in trade.enter_tag and last_candle["rsi"] > 80:
        return "exit_signal_rsi"
    if "long_signal_bblower" in trade.enter_tag and last_candle["high"] > last_candle["bb_upperband"]:
        return "exit_signal_bblower"
    # ...
    return None

INFO

enter_tag 限制为 255 个字符，超出的数据将被截断。

WARNING

只有一个 enter_tag 列，多空交易共用该列。因此，该列必须被视为“以最后一次写入为准”（毕竟它只是一个数据帧列）。在一些极端情况下，如果多个信号发生碰撞（或基于不同条件再次取消激活信号），可能会导致错误的结果，即错误的标签被应用。这些结果是策略覆盖先前标签的结果 —— 最后一个写入的标签将“固定下来”，并成为 Freqtrade 所使用的标签。

离场标签 (Exit tag)

与 进场标签 类似，您也可以指定离场标签。

def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    dataframe["exit_tag"] = ""
    rsi_exit_signal = (dataframe["rsi"] > 70)
    ema_exit_signal  = (dataframe["ema20"] < dataframe["ema50"])
    # 其他条件
    dataframe.loc[
        (
            rsi_exit_signal
            | ema_exit_signal
            # ... 离场做多的其他信号
        ) &
        (dataframe["volume"] > 0)
        ,
    "exit_long"] = 1
    # 拼接标签，以便保留所有信号
    dataframe.loc[rsi_exit_signal, "exit_tag"] += "exit_signal_rsi "
    dataframe.loc[ema_exit_signal, "exit_tag"] += "exit_signal_ema "

    return dataframe

提供的离场标签随后被用作离场原因 —— 并作为此类原因显示在回测结果中。

INFO

exit_reason 限制为 100 个字符，超出的数据将被截断。

策略版本 (Strategy version)

您可以利用 version 方法并返回您希望该策略拥有的版本，来实现自定义策略版本控制。

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

INFO

您应该确保除此之外还采用了适当的版本控制（如 Git 仓库），因为 Freqtrade 不会保留策略的历史版本，因此最终回滚到先前版本的策略取决于用户自己。

派生策略 (Derived strategies)

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

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

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

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

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

嵌入式策略 (Embedding Strategies)

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 与策略名称完全一致！

性能警告 (Performance warning)

执行策略时，日志中偶尔会出现以下内容：

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() 来应对这一问题 —— 因此这方面的性能影响应该是微乎其微甚至不存在的。