模型本地化训练指南

由hxgre创建，最终由hxgre更新于2026-07-16 08:18 被浏览 221 用户

背景

端到端（End-to-End, E2E）模型直接以分钟级行情和盘口快照为输入，自行学习特征表达与预测目标，相比传统因子方式对算力和数据量的要求高出很多。为降低本地训练的门槛，主办方开放一条本地化训练通道：把端到端模型所需的训练集数据打包成压缩文件供直接下载，在本地完成模型训练，再把训练好的模型与训练代码一并提交到云端进行预测评测。

早期方案是通过 BigQuant SDK 把数据逐表拉到本地，但分钟级数据流量消耗巨大、且受 quota 限制。现改为提供压缩包直链下载，一次下载、本地复用，不再依赖 SDK 拉数。

采用这一方案需要特别注意以下三点：

1. 本地数据做了压缩存储。分钟数据量较大，本地数据表把绝大多数价格/金额字段 ×100 后转换为整数存储（单位由“元”变为“分”），instrument（字符串股票代码）替换为整数 instrument_id，盘口数据从原始 5 档裁剪为 3 档。读取本地数据时务必按约定还原。
2. 训练代码必须随模型一起提交。本地训练出的模型，需把训练代码一并提交，便于复现与审核。
3. 本地与云端数据格式不一致，需在特征构建时对齐。代码提交到云端后，预测阶段使用的是云端未经压缩处理的原始数据。这意味着本地训练时的输入（“分”为单位的 int、3 档盘口、instrument_id）与云端预测时（“元”为单位的 float、5 档盘口、字符串 instrument）并不相同。务必在特征构建层做好两边的一致性处理，否则会出现“本地训练分数高、云端预测对不上”的问题。

一、下载与解压

1. 下载链接

按频率分成 4 个压缩包，按需下载（不必全下）：

• bigalpha_2026_e2e_bar1m
• bigalpha_2026_e2e_bar5m
• bigalpha_2026_e2e_bar15m
• bigalpha_2026_e2e_bar30m

下载链接：【密码通已发过站内信，请查看】

分享链接： https://bigquant.com/codesharev3/03e15f36-7805-49ef-a6ce-ed9b533ebe8a

\

# 示例：下载并解压 5 分钟数据到本地 e2e_data 目录
mkdir -p e2e_data && cd e2e_data
curl -L -O https://bigquant-public-download.tos-cn-shanghai.volces.com/bigalpha2026/bigalpha_2026_e2e_bar30m.zip
unzip bigalpha_2026_e2e_bar30m.zip

2. 数据组织结构

解压后每个频率是一个独立目录（目录名即数据表名 datasource_id），内部按月份分区存为 feather 文件，另有一个 bmeta.bin（BigQuant bdb 元数据，纯 feather 读取用不到，忽略即可）：

e2e_data/
├── bigalpha_2026_e2e_bar1m/
│   ├── 201901.0.feather      # 2019 年 1 月
│   ├── 201902.0.feather
│   ├── ...
│   ├── 202412.0.feather      # 2024 年 12 月，共 72 个月
│   └── bmeta.bin             # bdb 元数据，读 feather 时忽略
├── bigalpha_2026_e2e_bar5m/
│   └── ...（同上结构）
├── bigalpha_2026_e2e_bar15m/
└── bigalpha_2026_e2e_bar30m/

• 时间覆盖：2019-01 ~ 2024-12，共 72 个月分区，每个频率一致。
• 股票池：中证 1000 指数在历史相应时间点上的成分股（约 1000 只）。
• 文件名：YYYYMM.0.feather，可直接按文件名做月份粗筛，只读区间覆盖到的月份，避免全量 IO。
• 读取方式：pandas.read_feather(path, columns=[...]) 即可，只读需要的列进一步省内存。

二、本地表 vs 云端表（关键差异）

端到端模型的本地表与云端表一一对应，仅频率不同、以及本地做了压缩：

字段差异对照

本地完整字段（28 列）

date, instrument_id, adjust_factor, high, open, low, close, deal_number, volume, amount, ask_price1~3, bid_price1~3, ask_volume1~3, bid_volume1~3, ask_num_orders1~3, bid_num_orders1~3

云端 stock 表额外含 instrument（字符串）、pre_close，且盘口为 5 档。

为什么这样压缩

• 价格/金额 ×100 转 int：整数的 delta 压缩远优于 float，且无浮点误差；amount 用 int64 是为了避免 float32 累计到亿元量级后精度退化（float32 仅约 7 位有效数字）。
• instrument 替换为 int16 的 instrument_id：字符串列存储与比较成本高，整数 ID 更省空间、更快。
• 盘口裁剪到 3 档：在体积与信息量之间取平衡。

三、数据一致性处理（最重要）

把“原始字段 → 模型输入特征”的逻辑收敛到同一个函数，并在入口先把两边的数据归一到同一份 canonical 表示，再做后续特征。canonical 约定：价格/金额单位为“元”（float）、OHLC 缺失为 NaN、盘口只保留 3 档、用一列 key 作为标的分组键。

import numpy as np
import pandas as pd

PRICE_SCALE = 100.0
SCALE_FIELDS = ["open", "high", "low", "close", "amount",
                "ask_price1", "ask_price2", "ask_price3",
                "bid_price1", "bid_price2", "bid_price3"]
OHLC_COLS = ["open", "high", "low", "close"]

def to_canonical(df: pd.DataFrame, *, is_local: bool) -> pd.DataFrame:
    """本地 e2e 表 / 云端 stock 表 -> 同一份 canonical 表示，供特征构建复用。"""
    df = df.copy()
    if is_local:
        # 1) OHLC 的 -1（停牌/无成交）视为缺失
        for c in OHLC_COLS:
            df.loc[df[c] == -1, c] = np.nan
        # 2) "分" -> "元"
        for c in SCALE_FIELDS:
            if c in df.columns:
                df[c] = df[c].astype("float64") / PRICE_SCALE
        # 3) 标的键：本地只有整数 instrument_id
        df["key"] = df["instrument_id"]
    else:
        # 云端原始表：价格已是"元"、缺失已是 NaN；只需丢掉 4/5 档使列与本地对齐
        drop_cols = [c for c in df.columns
                     if any(c.startswith(p) and c[-1] in "45"
                            for p in ("ask_price", "bid_price", "ask_volume",
                                      "bid_volume", "ask_num_orders", "bid_num_orders"))]
        df = df.drop(columns=drop_cols, errors="ignore")
        df["key"] = df["instrument"]
    return df

对齐检查清单

• [ ] 本地价格/金额是否已 /100？（int 的 608 应还原为 6.08 元）
• [ ] OHLC 的 -1 是否已转 NaN？（否则会被当成 -0.01 元的真实价）
• [ ] 特征是否只依赖前 3 档盘口？（云端 4/5 档会被丢弃）
• [ ] 分组键在本地用 instrument_id、云端用 instrument，是否已统一为 key？
• [ ] 训练集算好的标准化统计（mean/std）是否随权重存盘、推理时复用？

四、完整可运行示例（本地训练）

一份可直接复制运行的本地训练示例：读 30 分钟频率、2024 全年数据，训练一个小 Transformer，存成 transformer_model.json。只是跑通用的 demo，换频率/区间改开头几个常量即可。完整训练/推理脚本另见同目录 transformer_train_local.py。

准备两件事：

• 装依赖：pip install torch pandas pyarrow numpy。注意装到你实际运行脚本的那个解释器里——比如用 python3.11 a.py 跑，就 python3.11 -m pip install ...，否则会报 ModuleNotFoundError: No module named 'torch'。
• 改读权限：解压出来的 feather 若权限是 000（ls -la 显示 ----------）会读不了，先 chmod u+r /path/to/e2e_data/*/*.feather。

import os, glob, json, time, warnings
import numpy as np, pandas as pd
import torch, torch.nn as nn
from torch.utils.data import TensorDataset, DataLoader

warnings.filterwarnings("ignore", category=FutureWarning)  # 忽略 pyarrow read_feather 弃用提示

# ========== 改这几个常量即可 ==========
LOCAL_DATA_ROOT = "/path/to/e2e_data"              # 解压后的根目录
LOCAL_TABLE     = "bigalpha_2026_e2e_bar30m"       # 频率（目录名）
TRAIN_START, TRAIN_END = "2024-01-01", "2024-12-31 23:59:59"
SEQ_LEN, EPOCHS, BATCH = 40, 8, 512                # 30分钟每天8根bar，回看40根≈5天
MAX_INSTRUMENTS = 300                              # demo 限量控时长；全量设 None
# =====================================

MODEL_PATH   = os.path.join(LOCAL_DATA_ROOT, "transformer_model.json")
PRICE_COLS   = ["open", "high", "low", "close", "bid_price1", "ask_price1"]
VOL_COLS     = ["volume", "amount", "bid_volume1", "ask_volume1"]
FEATURE_COLS = PRICE_COLS + VOL_COLS
N_FEAT       = len(FEATURE_COLS)
SCALE_FIELDS = ["open", "high", "low", "close", "amount", "bid_price1", "ask_price1"]
OHLC_COLS    = ["open", "high", "low", "close"]
MODEL_CFG    = dict(n_feat=N_FEAT, d_model=64, nhead=4, nlayers=2, dim_ff=128, seq_len=SEQ_LEN)
np.random.seed(42); torch.manual_seed(42)

# ---- 1. 设备：CUDA -> Apple MPS -> CPU ----
if torch.cuda.is_available():
    device = torch.device("cuda")
elif getattr(torch.backends, "mps", None) and torch.backends.mps.is_available():
    device = torch.device("mps")
else:
    device = torch.device("cpu")
print("设备:", device)

# ---- 2. 模型 ----
class StockTransformer(nn.Module):
    def __init__(self, n_feat, d_model, nhead, nlayers, dim_ff, seq_len):
        super().__init__()
        self.proj = nn.Linear(n_feat, d_model)
        self.pos  = nn.Parameter(torch.zeros(1, seq_len, d_model))
        layer = nn.TransformerEncoderLayer(d_model, nhead, dim_ff, 0.1,
                                           batch_first=True, activation="gelu")
        self.encoder = nn.TransformerEncoder(layer, nlayers)
        self.head = nn.Sequential(nn.LayerNorm(d_model), nn.Linear(d_model, 1))
    def forward(self, x):                                  # (B, L, N_FEAT) -> (B,)
        return self.head(self.encoder(self.proj(x) + self.pos).mean(1)).squeeze(-1)

# ---- 3. 读本地 feather，还原压缩：分->元、-1->NaN、instrument_id 作分组键 ----
root = os.path.join(LOCAL_DATA_ROOT, LOCAL_TABLE)
buf  = (pd.to_datetime(TRAIN_START) - pd.Timedelta(days=20)).strftime("%Y%m")  # 缓冲凑窗口
lo, hi = int(buf), int(pd.Timestamp(TRAIN_END).strftime("%Y%m"))
need = set(FEATURE_COLS) | {"date", "instrument_id"}
parts = [pd.read_feather(fp, columns=list(need))
         for fp in sorted(glob.glob(os.path.join(root, "*.feather")))
         if os.path.basename(fp).split(".")[0].isdigit()
         and lo <= int(os.path.basename(fp).split(".")[0]) <= hi]
df = pd.concat(parts, ignore_index=True)
df["date"] = pd.to_datetime(df["date"])
for c in OHLC_COLS:      df.loc[df[c] == -1, c] = np.nan       # OHLC 缺失
for c in SCALE_FIELDS:   df[c] = df[c] / 100.0                 # 分 -> 元
for c in VOL_COLS:       df[c] = np.log1p(df[c].clip(lower=0)) # 量纲大先 log1p
df["key"] = df["instrument_id"]
if MAX_INSTRUMENTS:                                            # demo 限量
    df = df[df["key"].isin(df["key"].value_counts().index[:MAX_INSTRUMENTS])]
df = df.sort_values(["key", "date"])
for c in OHLC_COLS:      df[c] = df.groupby("key")[c].ffill()  # 停牌前向填充
print("标的数:", df["key"].nunique(), "| 行数:", len(df))

# ---- 4. 切滑动窗口，标签=每日末根 bar 的未来 1 日收益 ----
sd, ed = pd.to_datetime(TRAIN_START), pd.to_datetime(TRAIN_END)
wins, ys = [], []
for k, sub in df.groupby("key", sort=False):
    if len(sub) <= SEQ_LEN: continue
    feats = sub[FEATURE_COLS].to_numpy(np.float32)
    day   = sub["date"].dt.normalize().to_numpy()
    eod   = np.flatnonzero(np.append(day[1:] != day[:-1], True))   # 每日最后一根 bar
    cpx   = sub["close"].to_numpy(np.float64)[eod]
    for j, p in enumerate(eod):
        d = pd.Timestamp(day[eod][j])
        if p + 1 < SEQ_LEN or d < sd or d > ed: continue
        win = feats[p - SEQ_LEN + 1: p + 1]
        if j + 1 < len(eod) and cpx[j] > 0 and np.isfinite(win).all():
            r = cpx[j + 1] / cpx[j] - 1.0
            if np.isfinite(r):
                wins.append(win); ys.append(np.float32(r))
X = np.stack(wins).astype(np.float32)
y = np.array(ys, np.float32)
mean = X.reshape(-1, N_FEAT).mean(0).astype(np.float32)
std  = X.reshape(-1, N_FEAT).std(0).astype(np.float32) + 1e-6
X = ((X - mean) / std).astype(np.float32)
p1, p99 = np.percentile(y, [1, 99])
y = np.clip(y, p1, p99).astype(np.float32)   # winsorize；.astype 必留，否则 MPS 不吃 float64
print("样本数:", len(y))

# ---- 5. 训练 ----
model = StockTransformer(**MODEL_CFG).to(device)
loader = DataLoader(TensorDataset(torch.from_numpy(X), torch.from_numpy(y)),
                    batch_size=BATCH, shuffle=True)
opt = torch.optim.Adam(model.parameters(), lr=1e-3)
loss_fn = nn.MSELoss()
model.train()
for ep in range(EPOCHS):
    t, tot = time.time(), 0.0
    for xb, yb in loader:
        xb, yb = xb.to(device), yb.to(device)
        opt.zero_grad()
        loss = loss_fn(model(xb), yb)
        loss.backward()
        opt.step()
        tot += loss.item()
    print(f"epoch {ep+1}/{EPOCHS}  mse={tot/len(loader):.6f}  {time.time()-t:.1f}s")

# ---- 6. 存成纯文本 JSON（权重 + 标准化统计 + 结构超参） ----
sd_json = {k: {"dtype": str(v.dtype).replace("torch.", ""),
               "shape": list(v.shape), "data": v.cpu().reshape(-1).tolist()}
           for k, v in model.state_dict().items()}
with open(MODEL_PATH, "w", encoding="utf-8") as f:
    json.dump({"state_dict": sd_json, "model_cfg": MODEL_CFG, "feature_cols": FEATURE_COLS,
               "seq_len": SEQ_LEN, "mean": mean.tolist(), "std": std.tolist()}, f)
print("模型已保存:", MODEL_PATH)

提醒：这是最小示例，只处理本地压缩表。提交云端时，价格还原/缺失/分组键那几步要收敛进第三节带 is_local 参数的 to_canonical，本地传 is_local=True、云端传 is_local=False，两阶段共用同一预处理才不会 train/infer 漂移。y 那行的 .astype(np.float32) 别删——Apple MPS 不支持 float64 张量。

云端推理（main，读原始数据）

下面是配套的云端推理骨架（阶段二，平台调用）。本地训练用哪个频率，这里的 datasources.get(...) 键就取哪个——上面示例用了 30 分钟，就改成 datasources.get("bar30m")。注意 to_canonical 要用第三节的双分支版本并传 is_local=False。

def main(datasources, start_date, end_date):
    import dai                                              # 云端才有，本地训练不依赖
    ckpt  = load_model(MODEL_PATH)                          # 加载本地训练的权重
    stats = (np.asarray(ckpt["mean"]), np.asarray(ckpt["std"]))
    model = StockTransformer(**ckpt["model_cfg"]); model.load_state_dict(ckpt["state_dict"])

    table = datasources.get("bar5m") or next(iter(datasources.values()))
    raw   = dai.query(f"SELECT date, instrument, {','.join(FEATURE_COLS)} FROM {table} "
                      f"ORDER BY instrument, date",
                      filters={"date": [buf, str(end_date)]}).df()
    canon = to_canonical(raw, is_local=False)               # 云端 -> 同一 canonical
    Xte, _, idx_df, _ = build_windows(canon, start_date, end_date, "infer", stats)  # 复用 stats
    ...                                                     # 打分
    return idx_df.rename(columns={"key": "instrument"})[["date", "instrument", "score"]]

要点：训练与推理共用 to_canonical + build_windows，且标准化 mean/std 在训练集上算好、随权重存盘、推理时直接复用——这样两阶段预处理严格一致，杜绝 train/infer 漂移与数据泄漏。

五、提交清单

提交到云端的目录需包含：

1. 推理 notebook（定义供平台调用的 main(datasources, start_date, end_date)）；
2. 训练脚本 transformer_train_local.py（含 to_canonical / build_windows / train_and_save，便于复现与审核）；
3. 训练产物 transformer_model.json（本地训练一次生成，随代码一起上传）。

公榜阶段平台不会重训，直接加载 transformer_model.json 推理；私榜阶段平台可能在隔离环境用训练脚本从零重训，故训练逻辑需保持可运行、结果可复现（已固定随机种子）。

\