从论文到代码：Kronos论文核心观点与GitHub开源实现对照解读

姜闽弋Flora

957人浏览 · 2025-10-21 00:10:29

姜闽弋Flora · 2025-10-21 00:10:29 发布

从论文到代码：Kronos论文核心观点与GitHub开源实现对照解读

【免费下载链接】Kronos Kronos: A Foundation Model for the Language of Financial Markets 项目地址: https://gitcode.com/GitHub_Trending/kronos14/Kronos

Kronos作为首个面向金融市场"语言"的开源基础模型，通过创新性的双层架构解决了金融时间序列预测的核心挑战。本文将系统拆解论文提出的技术方案与GitHub开源代码实现的对应关系，帮助开发者快速理解模型原理并上手实践。

核心架构设计：论文理论与代码实现对照

Kronos论文提出的双层框架（专用Tokenizer+自回归Transformer）在代码中通过模块化设计得到完整实现。论文第3.1节描述的"分层离散化"机制对应model/kronos.py中的KronosTokenizer类，该类通过编码器-解码器结构将连续的K线(OHLCV)数据转换为离散令牌。

# 论文中"BSQ量化器"的代码实现
self.tokenizer = BSQuantizer(self.s1_bits, self.s2_bits, beta, gamma0, gamma, zeta, group_size)

论文图2展示的模型架构在代码中分解为三个核心模块：

分层嵌入模块：model/kronos.py实现了论文提出的s1_bits + s2_bits双精度令牌嵌入
时间感知模块：model/kronos.py的TemporalEmbedding类对应论文3.3节的时间戳编码方案
依赖感知解码：model/kronos.py的DependencyAwareLayer实现了论文提出的"sibling token"条件生成机制

关键技术点解析：从数学公式到代码实现

1. 金融时间序列的"语言化"表示

论文第4.2节提出的"金融数据令牌化"方法在代码中通过两个关键步骤实现：

特征量化：model/kronos.py的BSQuantizer类实现了论文公式(2)的二进制球形量化，将连续特征压缩为s1_bits + s2_bits位的离散令牌
序列建模：model/kronos.py的TransformerBlock序列对应论文公式(5)的自回归建模，通过多层注意力机制捕捉时间依赖关系

2. 动态上下文窗口机制

针对金融数据的长序列依赖问题，论文第3.4节提出的"动态上下文截断"策略在model/kronos.py的get_dynamic_stamp函数中实现：

def get_dynamic_stamp(x_stamp, y_stamp, current_seq_len, pred_step):
    if current_seq_len <= max_context - pred_step:
        return torch.cat([x_stamp, y_stamp[:, :pred_step, :]], dim=1)
    else:
        start_idx = max_context - pred_step
        return torch.cat([x_stamp[:, -start_idx:, :], y_stamp[:, :pred_step, :]], dim=1)

该实现严格遵循论文提出的"保持最近max_context长度序列"原则，确保模型始终关注最新市场动态。

训练与推理流程：从论文实验到代码实践

预训练与微调策略

论文第5.1节描述的两阶段训练流程在代码中通过独立脚本实现：

Tokenizer训练：finetune/train_tokenizer.py实现论文4.3节的预训练过程，使用BSQ损失优化量化器参数
预测器微调：finetune/train_predictor.py对应论文5.2节的下游任务适配，通过finetune/config.py配置不同市场的微调参数

推理过程完整实现

论文算法1描述的自回归推理流程在model/kronos.py的auto_regressive_inference函数中完整实现，关键步骤包括：

序列截断：确保输入不超过max_context长度（论文建议512）
令牌生成：通过sample_from_logits函数实现论文公式(8)的采样策略
温度控制：通过T参数调节预测多样性，对应论文5.3节的实验分析

# 论文算法1的核心实现
for i in range(pred_len):
    current_seq_len = initial_seq_len + i
    input_tokens = x_token if current_seq_len <= max_context else [t[:, -max_context:].contiguous() for t in x_token]
    current_stamp = get_dynamic_stamp(x_stamp, y_stamp, current_seq_len, i)
    
    s1_logits, context = model.decode_s1(input_tokens[0], input_tokens[1], current_stamp)
    sample_pre = sample_from_logits(s1_logits[:, -1, :], temperature=T, top_k=top_k, top_p=top_p)
    
    s2_logits = model.decode_s2(context, sample_pre)
    sample_post = sample_from_logits(s2_logits[:, -1, :], temperature=T, top_k=top_k, top_p=top_p)
    
    x_token[0] = torch.cat([x_token[0], sample_pre], dim=1)
    x_token[1] = torch.cat([x_token[1], sample_post], dim=1)

实验验证：论文结论与代码复现

预测性能可视化

论文图4展示的预测效果可通过examples/prediction_example.py复现，运行后生成类似下图的预测对比结果：

该脚本实现了论文5.4节描述的评价流程，自动对比模型预测与真实市场走势。

回测系统实现

论文第6章讨论的策略应用在finetune/qlib_test.py中实现，通过Qlib框架构建的回测系统验证模型实际交易价值。运行后生成的回测结果如图所示：

代码实现了论文提出的"Top-K选股策略"，并添加了交易成本和流动性约束，使实验结果更接近实盘环境。

快速上手指南：从代码到应用

环境准备

# 克隆仓库
git clone https://gitcode.com/GitHub_Trending/kronos14/Kronos
cd Kronos

# 安装依赖
pip install -r requirements.txt

基础预测流程

完整预测示例参见examples/prediction_example.py，核心步骤包括：

加载模型与令牌器

from model import Kronos, KronosTokenizer, KronosPredictor
tokenizer = KronosTokenizer.from_pretrained("NeoQuasar/Kronos-Tokenizer-base")
model = Kronos.from_pretrained("NeoQuasar/Kronos-small")

准备输入数据

import pandas as pd
df = pd.read_csv("examples/data/XSHG_5min_600977.csv")
df['timestamps'] = pd.to_datetime(df['timestamps'])

生成预测

predictor = KronosPredictor(model, tokenizer, device="cuda:0", max_context=512)
pred_df = predictor.predict(
    df=df,
    x_timestamp=df['timestamps'][:-120],
    y_timestamp=df['timestamps'][-120:],
    pred_len=120,
    T=1.0,
    top_p=0.9
)