将您自己的 LLM 集成到 EA 中（第 3 部分）：使用 CPU 训练自己的 LLM·进阶篇
（2/3）· 没有独显也能跑通训练：从金融数据集构造到 CPU 上微调轻量语言模型的实操分支
用 MT5 批量抓 micro 品种收盘价切片
这段 Python 脚本通过 MetaTrader5 模块直连终端，把带 *micro* 组的交易品种 M5 收盘价按 60 根一组切出来，存成二维表供后续训练或统计。实盘前先确认终端已登录且「允许 DLL 导入」和「允许算法交易」已开，否则 initialize() 会直接返回 False。 脚本里 mt_data_len 写死 2500、sr_len 写死 60，意味着每个品种最多取出 2500-60+1=2441 个不重叠滑动窗口；外层 for j in sbs 配 ct>0 即 break，实际只处理符号列表里第一个 micro 品种，想扩样本得把 ct 限制去掉。 copy_rates_from_pos 从 0 偏移抓最新 2500 根，转 DataFrame 后只留 close 列；while k+1 循环里每轮取 cl_d[k:k+60]，首行建表、后续用 xy.loc[len(xy)]=cl_ds 追加。跑完 mt.shutdown() 断连、xy.to_csv 落盘，外汇与贵金属 micro 品种点差大、跳空多，切片统计结论只反映历史形态，实盘有高风险。
mt_data_len=class="num">2500 sr_len=class="num">60 if not mt.initialize(): print("mt initialize failed!") else: sbs=mt.symbols_get(group=&class="macro">#x27;*micro*&class="macro">#x27;) if sbs is not None: for i in [mt.TIMEFRAME_M5,]: xy=None ct=class="num">0 for j in sbs: if ct>class="num">0: class="kw">break print(j.name) d_=mt.copy_rates_from_pos(j.name,i,class="num">0,mt_data_len) df_d=pd.DataFrame(d_) cl_d=df_d[&class="macro">#x27;close&class="macro">#x27;] k=class="num">0 while k+class="num">1: if mt_data_len-k>=sr_len: cl_ds=cl_d[k:k+sr_len].tolist() if xy is None: xy=pd.DataFrame([cl_ds]) else: xy.loc[len(xy)]=cl_ds k+=class="num">1 else: class="kw">break ct+=class="num">1 mt.shutdown() xy.to_csv(data_file)
「循环收尾与数据落盘的细节」
上面这段是 Python 侧采集脚本的收口逻辑，和 MQL5 终端的联动靠 mt.shutdown() 切断会话。注意 ct+=1 放在 break 之后、shutdown 之前，意味着只有正常走完分支才会计数，异常跳出不会污染样本量。 xy.to_csv(data_file) 把拼好的特征表直接落盘，注释里留了 to_json 的写法但没启用，说明当前流水线优先喂 CSV 给后续训练。外汇与贵金属行情采样本就高波动，这类脚本跑出来的样本若含跳空缺口，需在读表阶段自行过滤，否则模型容易学到噪声。 验证方式很直接：把 data_file 路径指到 MT5 导出的 tick 目录，跑一遍看 xy 行数是否等于你设定的 ct 上限，差一行就说明 break 提前触发了。
else: class="kw">break ct+=class="num">1 mt.shutdown() # print(len(xy)," ",len(xy_list)) xy.to_csv(data_file) # xy.to_json(f&class="macro">#x27;llm_data.json&class="macro">#x27;) class="kw">return xy
◍ 把 MT5 行情序列喂进 GPT-2 分词器
做本地小模型训练时，第一步是把 MT5 拉到的数字序列变成大模型能读的整数流。示例里直接调用 get_data() 拿返回值，而不去读 csv——后者会在存读过程中多出一列和一行表头，得手动 iloc[1:,1:] 切掉，能省一步就省一步。
| 分词器选的是 tiktoken 的 gpt2 预训练编码，特殊标记用 < | endoftext | > 同时当序列头尾。验证集切分靠 val_cut=len(data)//10，也就是总数据的前 10% 进 val_tokens，剩下 90% 进 train_tokens；想调比例就改那个 10。 |
|---|
写 bin 文件时 header[0]=20240520 这个魔数别手滑。后续加载训练数据时框架会校验它，对不上直接报错，排查起来很隐蔽。 原数据本就是数字，为什么还要过一遍分词器？取决于你给模型定的任务规划——如果模型最终要吃类 GPT 的 token 流，统一编码通道比直接塞数组更稳。示例数据集每条长度固定且总量小，没做 padding 和 mask；真要搞复杂数据集，清洗和填充这一步极大影响模型质量。 第一次跑 tiktoken 会联网去 huggingface 下模型文件，MT5 终端机若不通外网要提前备好本地 encoder，否则卡在编码阶段。
data=get_data() # data=pd.read_csv(data_file) # data=data.iloc[class="num">1:,class="num">1:] enc = tiktoken.get_encoding("gpt2") encode = lambda s: enc.encode_ordinary(s) eot = enc._special_tokens[&class="macro">#x27;<|endoftext|>&class="macro">#x27;] train_tokens=[] val_tokens=[] val_cut=len(data)class=class="str">"cmt">//class="num">10 def data_to_file(path, tks): header = np.zeros(class="num">256, dtype=np.int32) header[class="num">0] = class="num">20240520 header[class="num">1] = class="num">1 header[class="num">2] = len(tks) toks_np = np.array(tks, dtype=np.uint16) with open(path, "wb") as f: f.write(header.tobytes()) f.write(toks_np.tobytes()) for i,r in data.iterrows(): ser=r.tolist() ser= &class="macro">#x27;&class="macro">#x27;.join(str(elem) for elem in ser) tokens = encode(ser) if i< val_cut: val_tokens.append(eot) val_tokens.extend(tokens) enc_f = os.path.join(DATA_DIR, "val_data.bin") data_to_file(enc_f, val_tokens) else: train_tokens.append(eot) train_tokens.extend(tokens) enc_f = os.path.join(DATA_DIR, "train_data.bin") data_to_file(enc_f, train_tokens) class="kw">import MetaTrader5 as mt class="kw">import pandas as pd class="kw">import numpy as np class="kw">import os class="kw">import tiktoken DATA_DIR = os.path.dirname(__file__) data_file = os.path.join(DATA_DIR, "llm_data.csv") def get_data(): mt_data_len=class="num">2500 sr_len=class="num">60 if not mt.initialize(): print("mt initialize failed!") else: sbs=mt.symbols_get(group=&class="macro">#x27;*micro*&class="macro">#x27;) if sbs is not None: # for i in [mt.TIMEFRAME_M5,mt.TIMEFRAME_M15,mt.TIMEFRAME_H1,mt.TIMEFRAME_D1]: for i in [mt.TIMEFRAME_M5,]: xy=None # xy_list=[] ct=class="num">0 for j in sbs: if ct>class="num">0: class="kw">break print(j.name)
把K线切片喂给小模型的落盘写法
这段脚本干的事很直接：从 MT5 拉完指定品种和周期的收盘价序列后，按 sr_len 窗口长度逐根滑动，把每段收盘价子序列堆进 DataFrame，最后整表写 csv 供后续微调或推理用。
核心循环里 while k+1 实际是靠 mt_data_len-k>=sr_len 判断剩余长度，不够就 break，因此当 mt_data_len=1000、sr_len=64 时，大约能切出 937 个样本行，样本数随数据长度线性递减。
data_to_file 函数则走另一条路：用 256 个 int32 的头块（首字段写死 20240520 作版本标记，次字段置 1，第三字段存品种数），后面紧接 uint16 的 token 流，直接 wb 二进制落盘，比 csv 省解析开销。
主程序里按 len(data)//10 做验证集切分，前 10% 行编码后塞 val_tokens 并补 eot 终止符；tiktoken 用 gpt2 普通编码，不处理特殊 token，意味着价格数字串会被拆成子词，训练前最好先确认词表覆盖度。外汇与贵金属波动剧烈，这类序列用于模型训练仅作概率性辅助，实盘仍需人工把关。
d_=mt.copy_rates_from_pos(j.name,i,class="num">0,mt_data_len) df_d=pd.DataFrame(d_) cl_d=df_d[&class="macro">#x27;close&class="macro">#x27;] k=class="num">0 while k+class="num">1: if mt_data_len-k>=sr_len: cl_ds=cl_d[k:k+sr_len].tolist() if xy is None: xy=pd.DataFrame([cl_ds]) # xy_list=[cl_ds] else: xy.loc[len(xy)]=cl_ds # xy_list.append(cl_ds) k+=class="num">1 else: class="kw">break ct+=class="num">1 mt.shutdown() # print(len(xy)," ",len(xy_list)) xy.to_csv(data_file) # xy.to_json(f&class="macro">#x27;llm_data.json&class="macro">#x27;) class="kw">return xy def data_to_file(path, tks): header = np.zeros(class="num">256, dtype=np.int32) header[class="num">0] = class="num">20240520 header[class="num">1] = class="num">1 header[class="num">2] = len(tks) toks_np = np.array(tks, dtype=np.uint16) with open(path, "wb") as f: f.write(header.tobytes()) f.write(toks_np.tobytes()) if __name__=="__main__": data=get_data() # data=pd.read_csv(data_file) # data=data.iloc[class="num">1:,class="num">1:] enc = tiktoken.get_encoding("gpt2") encode = lambda s: enc.encode_ordinary(s) eot = enc._special_tokens[&class="macro">#x27;<|endoftext|>&class="macro">#x27;] train_tokens=[] val_tokens=[] val_cut=len(data)class=class="str">"cmt">//class="num">10 for i,r in data.iterrows(): ser=r.tolist() ser=&class="macro">#x27;&class="macro">#x27;.join(str(elem) for elem in ser) # ser = ser.strip() tokens = encode(ser) if i< val_cut: val_tokens.append(eot)