DolphinDB 深度学习案例实践：Kronos 金融时序模型

金融 K 线数据（ Open , High , Low , Close , Volume , Amount，简称 OHLCVA）具有低信噪比、非平稳、厚尾分布等特点。直接使用连续值回归模型容易过拟合或对异常值敏感。Kronos 借鉴自然语言处理中的成功经验，将连续的市场观测转化为离散的“词汇”（tokens），然后学习这些词汇的序列规律，实现更鲁棒的预测。

整个 Kronos 模型分为两大模块：

• K 线分词器（K‑line Tokenizer）：将每个时间点的 6 维 K 线数据压缩并量化为一个离散 token 。每个 token 又进一步分解为 粗粒度子 token（ coarse subtoken ） 和 细粒度子 token（ fine subtoken ），分别编码主要价格结构和局部波动细节。
• 分层自回归预测器（ Hierarchical Autoregressive Transformer ）：一个 decoder‑only 的 Transformer ，它先根据历史 token 序列预测下一个粗子 token ，再以该粗子 token 为条件预测细子 token 。通过这种“先粗后细”的生成方式，模型能够显式模拟市场认知的过程：先判断大趋势，再补充具体波动。

分词器的目标是将原始 K 线序列 x_1:T（每个 x_t∈ℝ^D）映射为一个离散 token 序列 b_1:T 。在 Kronos 项目中，使用的是 6 个维度的 K 线数据，所以 D = 6。

（1）特征提取与量化

• 一个轻量 Transformer 编码器 E_enc 将每个 x_t 转换为连续潜变量 ξ_t∈ℝ^D；

• 随后采用 Binary Spherical Quantization（ BSQ ） 技术进行量化：将 ξ_t​ 投影到单位球面，然后通过 k 个可学习的超平面进行二值化，得到一个 k 位二进制码 b_t∈{-1,1}^k 。论文中 k = 20，因此理论词汇表大小为 2²⁰≈ 1M。

（2）粗‑细分层 token

为了降低后续自回归模型的计算复杂度，同时引入层次结构，Kronos 将 20 位二进制码等分成两个 10 位子码：

• 粗子 token b_t^c：前 10 位，词汇表大小 1024。

• 细子 token b_t^f：后 10 位，词汇表大小 1024。

为了确保粗子 token 学习主要的价格结构、细子 token 学习残差细节，训练了复合的 BSQ 分词器：

ℒ_tokenizer = ℒ_coarse + ℒ_fine + λℒ_quant

• ℒ_coarse：只用粗 token b^c 经过解码器重建原始 K 线，使粗 token 学习主要的价格形态。

• ℒ_fine：用完整 token 重建原始 K 线，使完整 token 补充细节。

• ℒ_quant ​：BSQ 的量化正则项，保证编码器输出与二进制码对齐。

通过这种设计，token 内部自动形成了“先粗后细”的信息层级，为后续的条件生成奠定基础。

给定 token 序列 b_1:T，预测器的目标是学习下一个 token 的概率分布，并自回归生成未来 token。

（1）输入融合

每个时刻 i 的粗、细子 token 分别通过两个独立的嵌入层映射为向量 e_c(b_i^c) 和 e_f(b_i^f) ，拼接后经一个可学习的线性融合层 W_fuse​ 得到最终的输入向量 v_i。同时，为了引入时间周期性信息（如分钟、小时、星期等），还会加上对应的时间特征嵌入。

（2）Transformer 主干

采用 decoder‑only Transformer ，使用因果自注意力（ causal mask ）确保每个位置只能看到历史信息。位置编码采用 RoPE（ Rotary Position Embedding ），能够很好地处理相对位置关系。层归一化使用 RMSNorm ，提高训练稳定性。

（3）分层预测（先粗后细）

对于下一个时刻 t ，预测过程分为两步：

• 粗子 token 预测：由 Transformer 输出的上下文向量 h_t 通过一个线性头 W_c 得到粗子 token 的 logits。

• 细子 token 预测：从预测分布中采样得到 b_t^c，嵌入作为查询（ query ），将历史上下文 h_t 作为键和值（key/value），通过交叉注意力得到更新后的向量 h_t^update，再经另一个线性头 W_f 预测细子 token 的 logits。

  

这种设计使得细子 token 的预测显式依赖于同一步的粗子 token，从而实现了 coarse‑to‑fine 的条件生成。

（4）训练目标

最大化观测 token 序列的对数似然，其中，D 表示数据的分散。

在推理阶段，Kronos 采用自回归方式逐时刻生成未来的 token。解码过程引入温度采样（ temperature scaling ）和 top‑p（ nucleus ）采样来控制生成结果的随机性。对于需要高稳定性的任务（如价格预测），可以通过生成多条不同的未来轨迹（ Monte Carlo 采样），分别解码为连续值后取平均，以获得更精确的点预测。实验表明，随着采样轨迹数量的增加，预测精度持续提升。

Kronos 模型能够实现三种目标：

• 价格序列预测：将生成的多步 token 解码回连续的价格序列，可用于构建投资信号。

• 波动率预测：基于生成的价格序列计算已实现波动率（realized volatility），用于风险控制。

• 合成 K 线生成：通过随机采样得到全新的、符合原始数据分布的 K 线序列，可用于数据增强或市场模拟。

DolphinDB server 版本：3.00.5

Python 版本：3.12

cuda 版本：13.2

torch 版本：2.11

其余 Python 包的特殊版本要求详见代码中的 requirements 文件。

参考文档部署 Kronos 项目。

执行语句：

pip install -r requirements.txt

下载必要的包，为了从 DolphinDB 中获取数据，需要额外下载 dolphindb 包，Python API 教程链接见：Python API。

pip install dolphindb

本章节演示直接使用已训练好的模型全流程，这里使用的是项目中已有的 Kronos-small 模型：

导入所需要的包：

import pandas as pd
from model import Kronos, KronosTokenizer, KronosPredictor
import matplotlib.pyplot as plt
import dolphindb as ddb

从 hugging face 获取配置文件如下。后续如果想要使用自己训练的模型，只需要将 tokenizer 和 model 替换为训练后的模型存储路径即可。

tokenizer = KronosTokenizer.from_pretrained("NeoQuasar/Kronos-Tokenizer-base")
model = Kronos.from_pretrained("NeoQuasar/Kronos-small")
predictor = KronosPredictor(model, tokenizer, max_context=512)

原项目的代码是读取 csv 文件。从 DolphinDB 获取数据，可以将代码中数据框获取的部分改为如下代码：

s = ddb.session()
s.connect("localhost", 8848, "admin", "123456")  --改为实际IP
script = '''
    select trade_time as timestamps, open, high, low, close, vol as volume, amount from loadTable("dfs://tushare_minute_db", "stock_5min_k") where trade_date >= 2025.01.01 and code = `000514.SZ
'''
df = s.run(script)
df['timestamps'] = pd.to_datetime(df['timestamps'])

后续可以使用模型进行预测，绘制图片并保存，完整代码如下：

# Define context window and prediction length
lookback = 400
pred_len = 100

# Prepare inputs for the predictor
x_df = df.loc[:lookback-1, ['open', 'high', 'low', 'close', 'volume', 'amount']]
x_timestamp = df.loc[:lookback-1, 'timestamps']
y_timestamp = df.loc[lookback:lookback+pred_len-1, 'timestamps']


def plot_prediction(kline_df, pred_df):
    pred_df.index = kline_df.index[-pred_df.shape[0]:]
    sr_close = kline_df['close']
    sr_pred_close = pred_df['close']
    sr_close.name = 'Ground Truth'
    sr_pred_close.name = "Prediction"

    sr_volume = kline_df['volume']
    sr_pred_volume = pred_df['volume']
    sr_volume.name = 'Ground Truth'
    sr_pred_volume.name = "Prediction"

    close_df = pd.concat([sr_close, sr_pred_close], axis=1)
    volume_df = pd.concat([sr_volume, sr_pred_volume], axis=1)

    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(8, 6), sharex=True)

    ax1.plot(close_df['Ground Truth'], label='Ground Truth', color='blue', linewidth=1.5)
    ax1.plot(close_df['Prediction'], label='Prediction', color='red', linewidth=1.5)
    ax1.set_ylabel('Close Price', fontsize=14)
    ax1.legend(loc='lower left', fontsize=12)
    ax1.grid(True)

    ax2.plot(volume_df['Ground Truth'], label='Ground Truth', color='blue', linewidth=1.5)
    ax2.plot(volume_df['Prediction'], label='Prediction', color='red', linewidth=1.5)
    ax2.set_ylabel('Volume', fontsize=14)
    ax2.legend(loc='upper left', fontsize=12)
    ax2.grid(True)

    plt.tight_layout()
    plt.show()
    plt.savefig('test260422.png')


# Generate predictions
pred_df = predictor.predict(
    df=x_df,
    x_timestamp=x_timestamp,
    y_timestamp=y_timestamp,
    pred_len=pred_len,
    T=1.0,
    top_p=0.9,
    sample_count=1
)

print("Forecasted Data Head:")
print(pred_df.head())

kline_df = df.loc[:lookback+pred_len-1]
plot_prediction(kline_df, pred_df)

上述提供的脚本仅限于数据量较小可以一次取完的情况。当数据量较大时，一次性获取会导致 DolphinDB 数据库提示内存超限，此时需要按照分区来分 batch 来取数，最后在内存中合并。下为按天取数模板：

batch_list = []
start_time = pd.Timestamp('2025-01-01 00:00:00')
end_time = pd.Timestamp('2025-03-01 00:00:00')
current_time = start_time

while current_time < end_time:
    script = f'''
        select trade_date, trade_time, open, high, low, close, vol as volume, amount
        from loadTable("dfs://tushare_minute_db", "stock_5min_k")
        where trade_date = {current_time.strftime('%Y.%m.%d')}
          and code = `000514.SZ
    '''
    batch_df = s.run(script)
    batch_list.append(batch_df)
    current_time = current_time + pd.Timedelta(days=1)

df = pd.concat(batch_list, ignore_index=True)
df['timestamps'] = pd.to_datetime(df['trade_date'].astype(str) + ' ' + df['trade_time'].astype(str))

若内存足够，或数据量不大，不使用 DDP 进行训练的情况下，可以按股票代码从 DolphinDB 中分批次获取数据，一次性加载进内存后进行模型训练。初始化数据集：

class FactorDataset(Dataset):
    def __init__(self, df):
        self.factor_array = df[FACTOR_COLUMNS].values.astype(np.float32)
        self.stamp_array = df[TIMEFAC_COLUMNS].values.astype(np.float32)
        self.n_samples = len(df) - WINDOW + 1

    def __len__(self):
        return self.n_samples

    def __getitem__(self, idx):
        x = self.factor_array[idx:idx + WINDOW].copy()
        x_stamp = self.stamp_array[idx:idx + WINDOW].copy()

        x_mean = np.mean(x, axis=0)
        x_std = np.std(x, axis=0)
        x = (x - x_mean) / (x_std + 1e-5)
        x = np.clip(x, -CLIP, CLIP)

        return torch.from_numpy(x), torch.from_numpy(x_stamp)

按照股票进行取数，这里需要确保取出的列名中包含参数设置的列和股票代码、时间列。本例中时间分为 date 和 time 两列，若数据有 timestamp 格式的列可以直接用一列代替：

for symbol in SYMBOL:
    script = f'''
        select code, trade_date, trade_time, open, high, low, close, vol, amount
        from {DDB_TABLE}
        where trade_date >= {START_TIME.strftime('%Y.%m.%d')} 
        and trade_date <= {END_TIME.strftime('%Y.%m.%d')} 
        and code = `{symbol}
    '''
    batch_df = s.run(script)
    if len(batch_df) > 0:
        batch_list.append(batch_df)
        print(symbol, len(batch_df))
    current_time = current_time + pd.Timedelta(days=1)

数据获取后，需要进行格式上的处理和清洗，然后划分训练集和测试集。

填充 NaN 值，处理时间列，用函数将 timestamp 或其他格式的时间转为参数设置的时间：

df = pd.concat(batch_list, ignore_index=True)
df[FACTOR_COLUMNS] = df[FACTOR_COLUMNS].ffill().fillna(0.0)

df['minute'] = df['trade_time'].dt.minute
df['hour'] = df['trade_time'].dt.hour
df['weekday'] = df['trade_date'].dt.weekday
df['day'] = df['trade_date'].dt.day
df['month'] = df['trade_date'].dt.month

划分测试集和训练集：此处划分方式是直接随机划分，在实际训练中，可以根据需要，按照时间等方式划分。

train_end = int(len(df) * TRAIN_RATIO)
val_end = int(len(df) * (TRAIN_RATIO + VAL_RATIO))

train_df = df.iloc[:train_end].copy()
val_df = df.iloc[train_end:val_end].copy()

print('train rows:', len(train_df))
print('val rows:', len(val_df))

train_dataset = FactorDataset(train_df)
val_dataset = FactorDataset(val_df)

首先训练分词器，对分词器进行初始化。可以直接加载 hugging face 中原项目的分词器作为初始化状态，若想要训练的 K 线非六维，项目已有的分词器无法匹配，则可以自己设置初始化状态，参考代码如下：

tokenizer = KronosTokenizer(
    d_in=len(FACTOR_COLUMNS),
    d_model=256,
    n_heads=4,
    ff_dim=512,
    n_enc_layers=4,
    n_dec_layers=4,
    ffn_dropout_p=0.0,
    attn_dropout_p=0.0,
    resid_dropout_p=0.0,
    s1_bits=10,
    s2_bits=10,
    beta=0.05,
    gamma0=1.0,
    gamma=1.1,
    zeta=0.05,
    group_size=4,
).to(device)

初始化模型训练相关参数：

• 损失函数：粗粒度 token 重建损失 + 完整粒度 token 重建损失 + BSQ 量化损失

• 优化器：AdamW

• 学习率调度器：OneCycleLR，初始最大学习率：2e-4 ，weight decay：0.1

tokenizer_optimizer = torch.optim.AdamW(
    tokenizer.parameters(),
    lr=TOKENIZER_LR,
    weight_decay=0.1,
)

tokenizer_scheduler = torch.optim.lr_scheduler.OneCycleLR(
    tokenizer_optimizer,
    max_lr=TOKENIZER_LR,
    steps_per_epoch=len(train_loader),
    epochs=TOKENIZER_EPOCHS,
    pct_start=0.03,
    div_factor=10,
)
best_tokenizer_val_loss = float('inf')

遍历 trainLoader ，从其内部队列中迭代获取批量数据，计算并优化模型的损失函数。主要训练循环部分如下，注意由于使用的 BSQ 模型，损失应分布在 [ -1, 1 ] 之间，出现负值是正常的训练结果。

for epoch in range(TOKENIZER_EPOCHS):
    tokenizer.train()
    train_loss_sum = 0.0

    for batch_x, _ in train_loader:
        batch_x = batch_x.to(device, non_blocking=True)

        zs, bsq_loss, _, _ = tokenizer(batch_x)
        z_pre, z = zs

        recon_loss_pre = F.mse_loss(z_pre, batch_x)
        recon_loss_all = F.mse_loss(z, batch_x)
        recon_loss = recon_loss_pre + recon_loss_all
        loss = (recon_loss + bsq_loss) / 2

        tokenizer_optimizer.zero_grad()
        loss.backward()
        torch.nn.utils.clip_grad_norm_(tokenizer.parameters(), max_norm=2.0)
        tokenizer_optimizer.step()
        tokenizer_scheduler.step()

        train_loss_sum += loss.item()

    tokenizer.eval()
    val_loss_sum = 0.0

    with torch.no_grad():
        for batch_x, _ in val_loader:
            batch_x = batch_x.to(device, non_blocking=True)
            zs, _, _, _ = tokenizer(batch_x)
            _, z = zs
            val_loss_sum += F.mse_loss(z, batch_x).item()

    avg_train_loss = train_loss_sum / len(train_loader)
    avg_val_loss = val_loss_sum / len(val_loader)
    print(f'tokenizer epoch {epoch + 1}/{TOKENIZER_EPOCHS} train_loss={avg_train_loss:.6f} val_loss={avg_val_loss:.6f}')

    if avg_val_loss < best_tokenizer_val_loss:
        best_tokenizer_val_loss = avg_val_loss
        tokenizer.save_pretrained(TOKENIZER_SAVE_DIR)
        print('save tokenizer:', TOKENIZER_SAVE_DIR)

首先加载 3.3 中训练好的分词器。如果只想训练预测器，可以在这里加载项目已有的，存于 hugging face 中的分词器。

tokenizer = KronosTokenizer.from_pretrained(TOKENIZER_SAVE_DIR).to(device)
tokenizer.eval()

初始化预测器训练的相关参数。预测器不受训练变量维度的影响，可以直接加载项目原有的预测器模型。相关参数说明可以参考分词器的部分。

predictor = Kronos.from_pretrained(PRETRAINED_PREDICTOR_PATH).to(device)
predictor_optimizer = torch.optim.AdamW(
    predictor.parameters(),
    lr=PREDICTOR_LR,
    betas=(0.9, 0.95),
    weight_decay=0.1,
)

predictor_scheduler = torch.optim.lr_scheduler.OneCycleLR(
    predictor_optimizer,
    max_lr=PREDICTOR_LR,
    steps_per_epoch=len(train_loader),
    epochs=PREDICTOR_EPOCHS,
    pct_start=0.03,
    div_factor=10,
)

best_predictor_val_loss = float('inf')

训练循环和分词器的架构相似，完整代码可以在附件中查看。

本章提供了从 DolphinDB 取数到本地进行训练的方式，上述方式会将数据全都缓存到内存中，但股票 K 线数据量极大，尤其是 1 分钟 K 线数据，许多情况下内存无法支持一次性读取。这种情况则使用后续章节提供的数据读取方式。

取数时间范围：2025.01.01-2025.12.31

股票样本量：100支

股票数据统计：可以自行选择需要预测的股票代码，也可以直接从库表中得到全部股票代码。这里以取出截止日的前 100 支股票为例，后续每次取数从中选择 5 支股票的代码进行查询。

code_list_script = f"""
      exec distinct code
      from {DDB_TABLE}
      where trade_date = {END_TIME.strftime('%Y.%m.%d')}
"""
code_list = session.run(code_list_script)
code_list = sorted(list(set(code_list.tolist())))[: args.code_count]

在 DolphinDB 上获取股票 K 线数据，DolphinDB 语句参考代码如下：

query_script = f"""
    select code, trade_date, trade_time, open, high, low, close, vol, amount
    from loadTable("dfs://tushare_minute_db", "stock_1min_k")
    where code in {code_str}
      and trade_date >= {train_start.strftime('%Y.%m.%d')}
      and trade_date < {train_end.strftime('%Y.%m.%d')}
"""

流程：每次获取一个 chunk 数据，每个 chunk 的数据量为5支股票的因子数据，获取 20 次。每取 4 次 chunk（即 20 支股票）保存为一个 pkl 文件，便于后续传入 Dataloader 训练。

注：按照股票代码分 chunk 取建议在 DolphinDB 中股票代码进行分区，如果只有日期分区，按照日期分批取数。

在 Python 端拼接 trade_date 与 trade_time 生成 timestamps 列（chunk_df["trade_date"] + (chunk_df["trade_time"] - chunk_df["trade_time"].dt.normalize())），相比在 DolphinDB 中使用 concatDateTime 函数直接返回时间戳，能够将计算压力从数据库端转移到客户端，利用 Pandas 的向量化操作降低数据库查询开销。

使用 pickle.HIGHEST_PROTOCOL 可启用最高版本的 pickle 协议，从而减少序列化数据的体积并提升 I/O 吞吐量。

将数据按股票代码存储为字典结构，通过字典的 update 方法增量更新，相比将股票代码作为列名存入 DataFrame 再使用 concat 拼接，具有更低的合并开销和更高的写入效率。

PKLdataset:

使用 dataset.py 加载保存过的 pkl 文件,传入后续训练。

以 config.py 中的 lookback_window、predict_window、window_stride 作为参数取数。以 2025 年一整年的数据为例，按照训练集：验证集=8:2 的比例划分训练测试集。总划分窗口量：634,200，训练集样本量：510,700，验证集样本量：123,500。

提取股票数据时间特征和 K 线特征，对特征做清洗和标准化：

def __getitem__(self, idx: int) -> tuple[torch.Tensor, torch.Tensor]:
        part_path, symbol, start_idx = self.indices[idx]
        self._load_part(part_path)

        df = self.current_part_data[symbol]
        end_idx = start_idx + self.window
        win_df = df.iloc[start_idx:end_idx]

        factor_df = win_df[self.feature_list].copy()
        factor_df = factor_df.replace([np.inf, -np.inf], np.nan)
        factor_df = factor_df.ffill().fillna(0.0)
        x = factor_df.values.astype(np.float64)
        x_stamp = win_df[self.time_feature_list].values.astype(np.float32)

        x_mean = np.mean(x, axis=0)
        x_std = np.std(x, axis=0)
        x = (x - x_mean) / (x_std + 1e-5)
        x = np.clip(x, -self.config.clip, self.config.clip)

        x_tensor = torch.from_numpy(x.astype(np.float32))
        x_stamp_tensor = torch.from_numpy(x_stamp)
        return x_tensor, x_stamp_tensor

将 inf/-inf 变成 nan ，用前一个非空值填充当前的空值。

一般来说，K 线数值都在 float32 的范围内，但如果传入的参数是计算或处理过的，在 DolphinDB 中数值较大的因子，需要先转成 float64 格式，否则按 float32 格式转换会转成 inf 。然后按当前窗口、按列计算均值和标准差。再按 clip 截断后，转回 float32 格式。

配置文件： config 文件放在 finetune 文件夹下。包括从 DolphinDB 获取数据、tokenizer 训练、predictor训练的配置。

在项目目录下，运行 torchrun --standalone --nproc_per_node=NUM_GPU finetune/train_tokenizer.py。主要训练循环代码如下：

for epoch_idx in range(config['epochs']):
    epoch_start_time = time.time()
    model.train()
    train_loader.sampler.set_epoch(epoch_idx)

    train_dataset.set_epoch_seed(epoch_idx * 10000 + rank)
    valid_dataset.set_epoch_seed(0)

    for step_idx, (batch_x, _) in enumerate(train_loader):
        batch_x = batch_x.to(device, non_blocking=True)

        zs, bsq_loss, _, _ = model(batch_x)
        z_pre, z = zs

        recon_loss_pre = F.mse_loss(z_pre, batch_x)
        recon_loss_all = F.mse_loss(z, batch_x)
        recon_loss = recon_loss_pre + recon_loss_all
        loss = (recon_loss + bsq_loss) / 2

        optimizer.zero_grad()
        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=2.0)
        optimizer.step()
        scheduler.step()

保存最佳模型：

if avg_val_loss < best_val_loss:
      best_val_loss = avg_val_loss
      save_path = f"{save_dir}/checkpoints/best_model"
      model.module.save_pretrained(save_path)
      print(f"Best model saved to {save_path} (Val Loss: {best_val_loss:.4f})")

保存的内容包括 config.json：初始化模型参数和 model.safetensors：模型权重。

在项目目录下，运行: torchrun --standalone --nproc_per_node=NUM_GPU finetune/train_predictor.py。

模型初始化：

tokenizer = KronosTokenizer.from_pretrained(config['pretrained_tokenizer_path'])
tokenizer.eval().to(device)

model = Kronos.from_pretrained(config['pretrained_predictor_path'])
model.to(device)
model = DDP(model, device_ids=[local_rank], find_unused_parameters=False)

先加载分词器，再加载预训练 predictor ，其中pretrained_tokenizer_path 是上一步分词器训练得到的，也可以从 hugging face 下载 Kronos 提供的预训练权重，然后用自己的数据做 finetune 。

损失函数：粗粒度 token 预测交叉熵损失+细粒度 token 预测交叉熵损失

优化器：AdamW

学习率调度器：OneCycleLR，初始最大学习率：2e-4，weight decay：0.1

参考文档：DDBDataLoader 详细介绍，安装并导入所需要的包，使用 DDBDataLoader 接口将 DolphinDB 数据对接到 PyTorch 中。

环境准备：

需要安装 dolphindb_tools 包，并通过以下方式导入 DDBDataLoader：

from dolphindb_tools.dataloader import DDBDataLoader

为了进行 DDP 训练，需要传入 torchrun 的相关环境变量：

# 获取分布式参数（兼容 torchrun 环境变量）
        if rank is None:
            rank = int(os.environ.get('LOCAL_RANK', 0))
        if world_size is None:
            world_size = int(os.environ.get('WORLD_SIZE', 1))
        self.rank = rank
        self.world_size = world_size

由于 DDBDataloader 本身没有 distributesampler 的实现，在代码中将进行分区拆分与样本分配，以适应 DDP 的训练方式。

symbol 格式用 exec 取出 tolist 后，需要用 self.stock_list = [f"'{code}'" for code in stock_codes] 的方式加上引号。

stock_sql = f"""
            exec distinct code
            from {config['ddb_table']}
            where trade_date = 2026.04.20
        """
        all_stocks = self.sess.run(stock_sql).tolist()[:100]
        print(f"[Rank {rank}] Total stocks in {data_type} set: {len(all_stocks)}")

        # 4. 按 rank 均匀分片（丢弃余数，保证每个进程股票数量相同）
        per_rank = len(all_stocks) // world_size
        if per_rank == 0:
            raise RuntimeError(f"world_size ({world_size}) > number of stocks ({len(all_stocks)}).")
        start_idx = rank * per_rank
        end_idx = start_idx + per_rank
        rank_stocks = all_stocks[start_idx:end_idx]
        self.stock_list = [f"'{code}'" for code in rank_stocks]
        print(f"[Rank {rank}] Assigned {len(self.stock_list)} stocks (indices {start_idx}-{end_idx-1})")

分区随机抽取+按照股票进行分组。需要注意，这里取数的数据表要按照股票代码分区，否则取数性能很差。一般来说，K 线数据按照股票代码逐个取不会 OOM ，如果内存较小，时间跨度较大，可以调整 DDBDataloader 的参数，设置 repartitionCol 与 repartitionScheme 进行额外的分区。

此外，Dataloader 会输出 batch_x , batch_y 的一组张量，分别对应 inputCol 和 targetCol 参数。需要注意的是，这些向量的顺序会按照 SQL 语句筛选的顺序排列，不会按照传参的顺序排列，注意与配置中的字段顺序相对应。

在代码中导入 5.1 中写好的 DDBDataloader 。

def create_dataloaders(config: dict, rank: int, world_size: int):
    print(f"[Rank {rank}] Creating distributed dataloaders...")
    train_loader = DolphinDBStreamingDataset('train', config, rank, world_size).ddb_loader
    val_loader = DolphinDBStreamingDataset('val', config, rank, world_size).ddb_loader

    print(f"[Rank {rank}] Dataloaders created.")
    return train_loader, val_loader

需要注意的是，DDBDataloader 没有 len 的实现，也没有数据预处理的部分，所以在训练中涉及到的 steps_per_epoch 参数，项目原代码中是用 len 方法来进行计算，修改后通过 steps_per 进行传参。这里的 steps_per 是指一轮循环里涉及到多少 batch ，可以按照数据量大小和 DDBDataloader 里的 batch_size 参数进行估算，估算时 steps_per 的设置可以略高于，但不要小于实际的 batch 数量，否则训练会报错。

steps_per = 5000
    if rank == 0:
        effective_bs = config['batch_size'] * world_size * config['accumulation_steps']
        print(f"[Rank {rank}] BATCHSIZE (per GPU): {config['batch_size']}")
        print(f"[Rank {rank}] Effective total batch size: {effective_bs}")

    train_loader, val_loader = create_dataloaders(config, rank, world_size)

    optimizer = torch.optim.AdamW(
        model.parameters(),
        lr=config['tokenizer_learning_rate'],
        weight_decay=config['adam_weight_decay']
    )

    scheduler = torch.optim.lr_scheduler.OneCycleLR(
        optimizer=optimizer,
        max_lr=config['tokenizer_learning_rate'],
        steps_per_epoch=steps_per,
        epochs=config['epochs'],
        pct_start=0.03,
        div_factor=10
    )

主要循环代码和原项目基本相同，需要注意的是，DDBDataLoader 中没有数据处理的步骤，所以要在训练中处理。数据处理代码如下：

for i, (ori_batch_x, _) in enumerate(train_loader):
            ori_batch_x = torch.nan_to_num(ori_batch_x, nan=0.0).float()
            lookback = config['lookback_window']
            past = ori_batch_x[:,:lookback,:]
            mean = past.mean(dim=1, keepdim=True)
            std = past.std(dim=1, keepdim=True) + 1e-5
            ori_batch_x = (ori_batch_x - mean) / std
            ori_batch_x = torch.clamp(ori_batch_x, -config['clip'], config['clip'])

            ori_batch_x = ori_batch_x.to(device, non_blocking=True)

主要内容和分词器相同，首先加载 DDBDataLoader：

def create_dataloaders(config: dict, rank: int, world_size: int):
    print(f"[Rank {rank}] Creating distributed dataloaders...")
    train_loader = DolphinDBStreamingDataset('train', config, rank, world_size).ddb_loader
    val_loader = DolphinDBStreamingDataset('val', config, rank, world_size).ddb_loader
    return train_loader, val_loader

在训练过程中进行数据处理：

for i, (batch_x, batch_x_stamp) in enumerate(train_loader):
            batch_x = torch.nan_to_num(batch_x, nan=0.0).float()

            # 2. 局部归一化（基于过去窗口）
            lookback = config['lookback_window']
            past = batch_x[:,:lookback,:]
            mean = past.mean(dim=1, keepdim=True)
            std = past.std(dim=1, keepdim=True) + 1e-5
            batch_x = (batch_x - mean) / std
            batch_x = torch.clamp(batch_x, -config['clip'], config['clip'])

            batch_x = batch_x.to(device, non_blocking=True)
            batch_x_stamp = batch_x_stamp.to(device, non_blocking=True)

完整代码见附件。各参数含义在第四章中已有介绍。