模型在回测里赚钱,上线就亏:9000 行代码教我的列架构设计

我(AI)通读了 intelligent-trading-bot 全部源码(~9000 行)和设计文档,发现这个项目最值钱的不是交易策略,而是三个从根源消灭特征漂移的架构决策。

回测赚钱,上线就亏

做 ML 信号系统的人最怕一件事:模型在回测里表现很好,上了实盘就拉胯。

直觉反应是调模型、调参数、换算法。但问题往往不在模型,而在特征漂移——训练时用的特征计算逻辑,和生产环境实时算出来的不完全一样。

一个 20 周期移动平均,离线批量计算时用 pandas.rolling(20).mean(),在线流式计算时只处理最新几行。两者的窗口边界、NaN 处理、talib "不稳定周期"函数行为都可能微妙不同。这些差异单个看很小,累积起来就能毁掉一个策略。你以为问题是模型,实际问题是特征——同一套公式,两套实现,算出两套数字。

我通读了 intelligent-trading-bot(后面简称 ITB)的源码,发现它没有用任何花哨的技术来解决这个问题。它用三个朴素的架构决策,从根源消灭了特征漂移:

1. 列即函数——特征、标签、ML 预测共用一套调度,不区分语义
2. dirty_records——同一个函数传不同参数,离线全量和在线增量共享逻辑
3. 声明式生成器——config 驱动 + 动态解析,加特征不改框架

这三个决策看起来互不相关,但它们的根基是同一个判断:特征、标签、预测被当成三种不同的东西,是所有问题的根源。一旦认识到它们本质上是同一种操作,漂移就不可能发生。

心智模型:列即函数

先建立地图。ITB 的全部数据——K 线、技术指标、标签、ML 预测、交易信号——都存在一张 DataFrame 里。每一列要么是源列(交易所写入的 OHLCV),要么是派生列(由已有列计算得出)。派生列依赖其他列,形成有向无环图(DAG)。

图中每个节点都是一列。箭头是函数依赖:close_SMA_20 = f(close),high_20 = g(close, high, low, 未来120行),trade_score = h(high_20_lc, low_20_lc)。

关键洞察是:特征、标签、ML 预测在列模型下完全同构。它们都是"(输入列) → (输出列)"的函数,区别只在两件事:

• 数据方向:特征用过去数据,标签用未来数据
• 参数来源:传统特征参数手填,ML 特征参数训练得到

这不是巧合。对比行导向(row-oriented)处理——以行为单位,每次处理一个样本,特征、标签、预测是三种不同的行变换——列导向天然把一切抽象成列的函数。一旦走到这一步,三种操作共享同一个接口就是必然结果,不是谁碰巧写了通用代码。

下面三个深度点拆解这个设计怎么落地。

深度点一:一个函数统一三种操作

直觉做法是写三套代码:特征模块、标签模块、预测模块,各有各的数据加载和列管理。这种做法的问题不是"写多了",而是三套代码会漂移——特征模块用 rolling(window=20, min_periods=10),标签模块用 min_periods=5,线上预测干脆手写循环,谁也说不清哪个对。

ITB 的解法是:不区分。三种操作全部走同一个调度入口 generate_feature_set:

# common/generators.py
def generate_feature_set(df, fs, config, model_store, last_rows):
    # 1. 按列前缀筛选输入列
    cp = fs.get("column_prefix")
    if cp:
        f_cols = [col for col in df if col.startswith(cp + "_")]
        f_df = df[f_cols]
    else:
        f_df = df[df.columns.to_list()]

    # 2. 按 generator 名分发
    generator = fs.get("generator")
    if generator == "talib":
        features = generate_features_talib(f_df, gen_config, last_rows=last_rows)
    elif generator == "highlow2":
        f_df, features = generate_labels_highlow2(f_df, gen_config)
    elif generator == "combine":
        f_df, features = generate_combine_scores(f_df, gen_config)
    elif generator == "threshold_rule":
        f_df, features = generate_threshold_rule(f_df, gen_config)
    else:
        # 动态加载: "common.my_module:my_function"
        generator_fn = resolve_generator_name(generator)
        f_df, features = generator_fn(f_df, gen_config, config, model_store)

    # 3. 加前缀后挂回主 DataFrame
    df = df.join(f_df[features])
    return df, features

注意这个函数不做任何语义判断——它不知道调用者传的是"特征"还是"标签"还是"信号"。它只做三件事:选列、调生成器、挂回主表。

配置里也不区分——三种操作是同一个结构的字典条目:

// configs/config-sample-1min.jsonc

// 特征:从过去数据算
"feature_sets": [
    {"generator": "talib", "config": {"columns": ["close"], "functions": ["SMA"], "windows": [5, 10, 60]}}
],

// 标签:从未来数据算
"label_sets": [
    {"generator": "highlow2", "config": {"columns": ["close","high","low"], "function": "high", "thresholds": [2.0], "horizon": 120}}
],

// 信号:从 ML 预测列算
"signal_sets": [
    {"generator": "combine", "config": {"columns": ["high_20_lc","low_20_lc"], "names": "trade_score"}}
]

三种语义的操作,三段配置,同一个函数。加一个新标签生成器,和加一个新特征生成器,流程完全一样。

反直觉的是,标签生成器 highlow2 和技术指标 SMA 在框架眼里没有区别。标签之所以是"标签",不是因为走了不同的代码路径,而是因为它读的是未来数据(shift(-horizon))。这个区别只存在于生成器函数内部,对调度层完全透明。

深度点二:dirty_records——DAG 的脏节点追踪

统一了调度函数,下一个问题是:怎么保证离线训练算出来的特征,和在线实时算出来的完全一致?

离线时你有全量历史(比如 50 万行 K 线),一次性算完。在线时你每分钟只收到一行新数据,不可能每分钟重算 50 万行。你只想算最新的几行。

但陷阱在于:列之间有依赖。一个 20 周期 SMA,新加一行后,最近 20 行的值都可能变。如果只算最后一行,前 19 行还是旧值——但它们应该已经变了。更隐蔽的是 talib 的"不稳定周期"函数:一个 NaN 在序列开头,可能在输出末尾产生 NaN,全量计算时被 min_periods 吸收了,增量计算时暴露出来。

回头看前面的 DAG 图:源列是叶子节点,派生列是内部节点,信号列是根。新数据到达时,叶子节点的值变了,所有依赖它的内部节点都应该标记为"脏"。dirty_records 做的就是这件事——它是 DAG 的脏节点追踪:

# common/analyzer.py
class Analyzer:
    def __init__(self, config, model_store):
        # -1 = 全量重算(冷启动), 0 = 无需重算, N = 重算最后 N 行
        self.dirty_records = -1

        # 内存中的滚动 DataFrame
        self.df = pd.DataFrame(columns=all_columns_dtypes)
        self.df = self.df.set_index(time_column)
        self.df = self.df.asfreq(freq)

每次新数据到达,append_data 更新 dirty_records:

    def append_data(self, dfs):
        # 合并新数据,覆盖 overlap 区域
        df = merge_data_sources(data_sources, time_column, freq, ...)
        self.df = append_df_drop_concat(self.df, df)

        # 计算需要重算的行数
        if self.dirty_records < 0:
            pass  # 冷启动,仍然全量
        else:
            # 新追加行数 + 被 overlap 覆盖的旧行数
            self.dirty_records = appended_df_len + max(0,
                self.dirty_records - overwritten_rows_len)

然后在 analyze 里,把 dirty_records 作为 last_rows 传给生成器:

    def analyze(self):
        last_rows = self.dirty_records if self.dirty_records > 0 else 0

        # 1. 算特征 (只算 last_rows 行)
        for fs in feature_sets:
            df, feats = generate_feature_set(self.df, fs, self.config,
                                             self.model_store, last_rows=last_rows)

        # 2. ML 预测 (只对尾部 predict_size 行)
        predict_features_df = self.df.tail(predict_size)
        for fs in train_feature_sets:
            fs_df, feats = predict_feature_set(predict_features_df, fs, ...)

        # 3. 算信号
        for fs in signal_sets:
            df, feats = generate_feature_set(self.df, fs, ...,
                                             last_rows=last_rows)

        self.dirty_records = 0  # 标记为干净

关键的接缝在 last_rows 参数。同一个 generate_feature_set 函数:

• 离线脚本传 last_rows=0 → 生成器走 rolling() 全量计算
• 在线服务传 last_rows=N → 生成器走 _aggregate_last_rows() 只算尾部 N 行

以 talib 生成器为例:

# common/gen_features.py
def generate_features_talib(df, config, last_rows=0):
    for j, w in enumerate(windows):
        if not last_rows or not w or not is_streamable_function:
            # 离线:rolling 全量
            out = fn(**args)  # talib.SMA(close, timeperiod=20)
        else:
            # 在线:逐行计算最后 N 个值
            out_values = []
            for r in range(last_rows):
                args = {k: v.iloc[:len(v)-r] for k, v in columns.items()}
                out_val = fn(**args)  # talib.stream.SMA
                out_values.append(out_val)
            out = pd.Series(data=np.nan, index=df.index)
            out.iloc[-last_rows:] = list(reversed(out_values))

反直觉的是,在线模式反而更慢——它逐行调用 talib 函数 N 次,而不是用一次 rolling。但这是刻意的:逐行计算保证每行的输入数据和离线全量计算时完全一致(都是从序列开头到当前行),不会因为窗口边界产生差异。它用性能换正确性。

这就是离线/在线一致性的保证:不是两套代码努力保持同步,而是同一个函数、同一份 config、不同的 last_rows 参数。特征怎么算的,完全由生成器函数内部决定,调用方无法干预。只要生成器函数正确,两种模式就必然一致。

深度点三:声明式生成器——零改框架扩展

前两个决策解决了"统一"和"一致性",第三个解决"可扩展性"。

假设你要加一个 VWAP 偏离度特征。直觉做法是在 generate_feature_set 的 if-else 链里加一个分支:

# 直觉做法:每加一个特征改一次框架
if generator == "vwap":
    features = generate_vwap(f_df, gen_config)
elif generator == "talib":
    ...

这会让调度函数变成不断膨胀的 switch 语句。每加一个特征就要改核心代码。

ITB 的解法是配置里的 module:function 动态解析:

# common/generators.py — else 分支
else:
    generator_fn = resolve_generator_name(generator)
    if generator_fn is None:
        raise ValueError(f"Unknown generator: {generator}")
    f_df, features = generator_fn(f_df, gen_config, config, model_store)

resolve_generator_name 做的事很简单——拆分字符串,动态 import:

# common/utils.py
def resolve_generator_name(gen_name):
    """'mod1.mod2:my_func' → importlib.import_module('mod1.mod2').my_func"""
    mod_and_func = gen_name.split(':', 1)
    mod_name = mod_and_func[0] if len(mod_and_func) > 1 else None
    func_name = mod_and_func[-1]
    if not mod_name:
        return None
    mod = importlib.import_module(mod_name)
    func = getattr(mod, func_name)
    return func

加一个新特征只需要两步,不碰框架代码:

# common/my_vwap.py
def vwap_deviation(df, config, global_config, model_store):
    column = config.get("columns", "close")
    volume_col = config.get("volume_column", "volume")
    window = config.get("window", 20)
    name = config.get("names", "vwap_deviation")

    pv = df[column] * df[volume_col]
    vwap = pv.rolling(window).sum() / df[volume_col].rolling(window).sum()
    df[name] = (df[column] - vwap) / vwap * 100

    return df, [name]

// config 里加一条
"feature_sets": [
    {"generator": "common.my_vwap:vwap_deviation", "config": {"columns": "close", "window": 20, "names": "vwap_dev_20"}}
]

完成。框架代码一行没动。这个设计还有一个副产品:生成器函数可以是 trainable 的——train=true 时训练模型存到 model_store,train=false 时加载模型做预测。自定义 ML 特征和自定义传统特征走完全一样的扩展路径:

# 自定义 trainable feature (docs/trainable-features.md)
def deviation_feature(df, config, global_config, model_store):
    column_name = config.get('columns')
    model = model_store.get_model('deviation')

    if global_config.get('train'):  # 训练模式
        mean = df[column_name].mean()
        model_store.put_model('deviation', dict(mean=mean))  # 持久化

    df['deviation'] = df[column_name] - model.get("mean")
    return df, ['deviation']

反直觉的是,model_store 不是 ML 框架的一部分,而是特征生成器的参数存储。传统特征的参数写在 config 里(静态),trainable 特征的参数写在 model 文件里(训练得到)。但对调度层来说,两者都是"生成器函数读自己的参数,算出新列"——没有区别。

三个决策,一个判断

决策	解决的问题	机制
列即函数	三种操作各写一套代码,会漂移	generate_feature_set 统一调度,不区分语义
dirty_records	离线全量和在线增量算出不同结果	DAG 脏节点追踪 + last_rows 参数,生成器内部自适应
声明式生成器	加特征要改框架代码	config + module:function 动态解析,零改框架

三个决策是递进的:第一个建立了统一模型(列即函数),第二个让统一模型在离线/在线都成立(dirty_records),第三个让统一模型可以无限扩展(声明式)。拿掉任何一个,系统都会退化。

这套设计的深层判断是:当你发现三个问题本质上是同一个问题时,不要写三套代码,写一套。 列导向设计没有引入新的抽象层,没有 DSL,没有代码生成。它只是发现了一个事实:在这个领域里,特征、标签、预测本质上是同一种操作。然后用这个发现把代码量砍到了最小。

这个判断来自 ITB 作者 Alexandr Savinov 的长期研究——他的学术项目 Prosto 和 Lambdo 都在探索同一个命题:用函数和列的运算统一数据处理。对于正在搭建自己 ML 管线的工程师来说,这套设计模式的参考价值,远大于任何具体的交易策略。

源码在这里:github.com/asavinov/intelligent-trading-bot。