第 8 章 · 06 自定义模型开发

第 8 章 · 06 自定义模型开发

本节摘要:内置模型不够用时，FreqAI 鼓励你写自己的模型。本节讲清完整流程:选对基类(BaseRegressionModel / BaseClassifierModel / BasePyTorchModel / BaseReinforcementLearner)，只需实现一个 fit() 方法(基类已帮你实现 train() 和 predict() 的繁琐数据流转);需要更精细控制时再覆写 train()/predict()/define_data_pipeline()。写好后把文件放到 user_data/freqaimodels/，用 --freqaimodel 类名 即可加载。最后以 FreqaiExampleStrategy 串起特征→训练→预测→信号的全链路，巩固前几节。

内容来源:原项目文档 docs/freqai-developers.md、docs/freqai-configuration.md，源码 freqtrade/freqai/base_models/，汉化并套用体系化模板。

⚠️ 风险提示:自定义模型意味着你自己负责正确性——数据泄露、标签错配、维度对不齐都会导致隐蔽的 bug。务必用小数据集先验证 fit/predict 的输入输出形状，再跑全量。

学习目标

阅读完本节，你应当能够:

1. 选择正确的基类继承。
2. 实现 fit() 方法完成自定义训练。
3. 把模型注册到 user_data/freqaimodels/ 并用命令行加载。
4. 按需覆写 train()/predict()/define_data_pipeline()。
5. 读懂 FreqaiExampleStrategy 的完整数据链路。

一、继承结构:选对基类省一半事

FreqAI 的模型继承树设计得很有层次，基类帮你处理了大部分脏活:

核心认知:基类已经实现了 train() 和 predict() 里的数据流转(特征过滤、训练/测试切分、流水线 fit/transform、预测、逆变换、DI 值计算)。你只需实现 fit()——「拿到处理好的数据字典，返回一个训练好的模型对象」。

选基类的决策:

二、最小自定义回归模型:只写 fit()

看 BaseRegressionModel 的 train() 做了什么(源码精简):

class BaseRegressionModel(IFreqaiModel):
    def train(self, unfiltered_df, pair, dk, **kwargs):
        # 1. 过滤特征和标签，处理 NaN
        features_filtered, labels_filtered = dk.filter_features(...)
        # 2. 切分训练/测试集
        dd = dk.make_train_test_datasets(features_filtered, labels_filtered)
        # 3. 构建/拟合特征与标签流水线
        dk.feature_pipeline = self.define_data_pipeline(...)
        dd["train_features"], ... = dk.feature_pipeline.fit_transform(...)
        # 4. 调你的 fit()
        model = self.fit(dd, dk)
        return model

    def predict(self, unfiltered_df, dk, **kwargs):
        # 1. 过滤特征 2. 流水线 transform 3. model.predict 4. 逆变换
        ...
        return pred_df, dk.do_predict
​

所以你只需写 fit()。一个用 scikit-learn 的 Ridge 回归的自定义模型:

from sklearn.linear_model import Ridge
from freqtrade.freqai.base_models import BaseRegressionModel

class MyRidgeRegressor(BaseRegressionModel):
    """
    自定义岭回归模型。
    放到 user_data/freqaimodels/MyRidgeRegressor.py
    用 freqtrade trade --freqaimodel MyRidgeRegressor 加载
    """

    def fit(self, data_dictionary: dict, dk, **kwargs):
        X = data_dictionary["train_features"]
        y = data_dictionary["train_labels"]
        model = Ridge(alpha=1.0)
        model.fit(X, y)
        return model
​

就这样。data_dictionary 是基类准备好的字典，含 train_features、train_labels、test_features、test_labels、train_weights 等(都经过流水线处理)。你返回的 model 对象只要有 .predict() 方法，基类的 predict() 就能用。

💡 模型保存格式:FreqAI 用 joblib 序列化模型。只要你的模型对象可 joblib pickle，就能自动存盘和崩溃恢复。

三、分类模型与 class_names

分类模型多一步——声明类别名。在策略的 set_freqai_targets 里设:

def set_freqai_targets(self, dataframe, metadata, **kwargs):
    self.freqai.class_names = ["down", "up"]  # 必须在标签之前设
    dataframe['&s-up_or_down'] = np.where(
        dataframe["close"].shift(-100) > dataframe["close"], 'up', 'down')
    return dataframe
​

自定义分类模型与回归几乎一样，只是基类不同:

from sklearn.ensemble import GradientBoostingClassifier
from freqtrade.freqai.base_models import BaseClassifierModel

class MyGBClassifier(BaseClassifierModel):
    def fit(self, data_dictionary, dk, **kwargs):
        # 分类模型需要把字符串标签转成整数
        class_names = self.get_class_names()
        self.convert_label_column_to_int(data_dictionary, dk, class_names)

        model = GradientBoostingClassifier(n_estimators=200)
        model.fit(data_dictionary["train_features"],
                  data_dictionary["train_labels"])
        return model
​

BaseClassifierModel 的 predict() 会额外调用 predict_proba 返回各类概率，策略里可用概率做置信度过滤。

四、进阶:覆写 train/predict/define_data_pipeline

当基类的 train()/predict() 不满足需求时，可以覆写。常见场景:

自定义数据流水线(上一节提过):

from datasieve.transforms import SKLearnWrapper
from datasieve.pipeline import Pipeline
from sklearn.preprocessing import QuantileTransformer

class MyFreqaiModel(BaseRegressionModel):
    def fit(self, data_dictionary, dk, **kwargs):
        # ... 训练逻辑
        return model

    def define_data_pipeline(self) -> Pipeline:
        # 用分位数变换替代默认的 MinMaxScaler
        return Pipeline([
            ('qt', SKLearnWrapper(QuantileTransformer(output_distribution='normal'))),
        ])

    def define_label_pipeline(self) -> Pipeline:
        return Pipeline([
            ('scaler', SKLearnWrapper(StandardScaler())),
        ])
​

完全自定义 train/predict(继承 IFreqaiModel，最自由也最重):

class MyFullyCustomModel(IFreqaiModel):
    def train(self, unfiltered_df, pair, dk, **kwargs):
        # 你自己负责特征过滤、切分、流水线、调 fit
        ...
        return model

    def fit(self, data_dictionary, dk, **kwargs):
        # 通常从 train 里调用
        ...

    def predict(self, unfiltered_df, dk, **kwargs):
        # 你自己负责特征过滤、流水线 transform、预测、逆变换
        ...
        return pred_df, do_predict
​

⚠️ 除非你确实需要(比如用非标准的输入结构)，否则不要从 IFreqaiModel 起步。基类里的数据流转经过大量实战验证，自己重写容易踩坑。

五、注册与加载

写好模型类后，把 .py 文件放到 user_data/freqaimodels/:

user_data/
└── freqaimodels/
    └── MyRidgeRegressor.py      # 文件名随意，但类名要唯一
​

⚠️ 命名别撞内置模型。如果你的类叫 LightGBMRegressor，会覆盖内置的那个，引发混乱。起个独特的前缀名。

然后正常启动，--freqaimodel 填类名(不是文件名):

freqtrade trade \
    --strategy FreqaiExampleStrategy \
    --freqaimodel MyRidgeRegressor \
    --config config.json
​

FreqAI 的 freqaimodel_resolver 会扫描 user_data/freqaimodels/ 找到匹配的类名并实例化。

六、FreqaiExampleStrategy 全链路串读

用官方示例策略把前几节串起来。它的数据流:

关键代码段(精简):

class FreqaiExampleStrategy(IStrategy):
    startup_candle_count: int = 40   # ≥ indicator_periods_candles 最大值
    can_short = True
    stoploss = -0.05
    minimal_roi = {"0": 0.1, "240": -1}

    # 1. 定义会裂变的基础特征
    def feature_engineering_expand_all(self, dataframe, period, metadata, **kwargs):
        dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=period)
        dataframe["%-mfi-period"] = ta.MFI(dataframe, timeperiod=period)
        dataframe["%-roc-period"] = ta.ROC(dataframe, timeperiod=period)
        return dataframe

    # 2. 定义标签:未来 N 根 K 线平均收益率
    def set_freqai_targets(self, dataframe, metadata, **kwargs):
        n = self.freqai_info["feature_parameters"]["label_period_candles"]
        dataframe["&-s_close"] = (
            dataframe["close"].shift(-n).rolling(n).mean()
            / dataframe["close"] - 1
        )
        return dataframe

    # 3. FreqAI 接管
    def populate_indicators(self, dataframe, metadata):
        dataframe = self.freqai.start(dataframe, metadata, self)
        return dataframe

    # 4. 用预测值做信号
    def populate_entry_trend(self, df, metadata):
        df.loc[(df["do_predict"]==1) & (df["&-s_close"] > 0.01),
               ["enter_long", "enter_tag"]] = (1, "long")
        df.loc[(df["do_predict"]==1) & (df["&-s_close"] < -0.01),
               ["enter_short", "enter_tag"]] = (1, "short")
        return df

    def populate_exit_trend(self, df, metadata):
        df.loc[(df["do_predict"]==1) & (df["&-s_close"] < 0), "exit_long"] = 1
        df.loc[(df["do_predict"]==1) & (df["&-s_close"] > 0), "exit_short"] = 1
        return df
​

注意几个细节:

• do_predict == 1 是过滤条件——只在模型有信心(DI 未超标)时下单。
• &-s_close > 0.01 用 1% 作阈值，可结合 &-s_close_mean/&-s_close_std 做动态阈值。
• confirm_trade_entry 还做了价格滑点保护:下单价比最新收盘价高 0.25% 就拒绝。

💡 从示例到生产:官方示例是功能展示，跑得快但不优化。生产用要:精简特征(去掉无效的)、调标签(选更适合预测的目标)、换模型(按上一节选型)、严控风险(止损、保护、仓位)。

本节要点回顾

1. 选基类:回归用 BaseRegressionModel、分类用 BaseClassifierModel、PyTorch 用对应 Base、RL 用 ReinforcementLearner、完全自定义才直接继承 IFreqaiModel。
2. 最小实现:基类已实现 train()/predict() 的数据流转，你通常只需写 fit() 返回一个带 .predict() 的模型对象。
3. 分类模型:要在策略里设 self.freqai.class_names，模型里用 convert_label_column_to_int。
4. 进阶覆写:需要自定义流水线时覆写 define_data_pipeline/define_label_pipeline;需要完全控制才覆写 train/predict。
5. 注册:文件放 user_data/freqaimodels/，类名要唯一(别撞内置)，用 --freqaimodel 类名 加载。
6. 全链路:特征函数定义 → 配置裂变 → 标签 → freqai.start() → 预测列回填 → entry/exit 用预测值+do_predict 做信号。

至此第 8 章结束。下一章我们进入交易所底层——基于 ccxt 的交易所适配层、合约杠杆、做空交易、生产者-消费者集群。