将您自己的 LLM 集成到 EA 中（第 5 部分）：使用 LLM 开发和测试交易策略（四） 测试交易策略·进阶篇
◍ 适配器层的权重初始化与前向通路
在给 GPT-2 每个 transformer 层挂适配器时，下投影和上投影的权重都按均值 0.0、标准差 0.02 的正态分布初始化，偏置直接置 0；这种近零小方差起手能压住微调初期的扰动，避免适配器一上来就掀翻预训练表征。 Adapter 的 forward 很直白：先 down_project 压缩维度，接 ReLU 再 dropout，然后 up_project 还原维度，最后再 dropout 一次才返回。两层 dropout 夹住瓶颈层，训练时隐状态被随机屏蔽的概率明显上升，过拟合倾向会弱一些。 外层 GPT2LMHeadModelWithAdapters 在 __init__ 里用 nn.ModuleList 按 config.n_layer 数量堆了一组 Adapter，forward 先拿 transformer 原生输出 hidden_states = transformer_outputs[0]，随后逐层过适配器。 想验证这套结构，把 n_embd 和 n_layer 打出来，看 ModuleList 长度是否等于层数；外汇与贵金属行情序列接进来做表征微调时属高风险实验，回测胜率可能随适配器维度收窄而波动。
nn.init.normal_(self.down_project.weight, mean=class="num">0.0, std=class="num">0.02) nn.init.constant_(self.down_project.bias, class="num">0) # Initialize weights for up projection layer nn.init.normal_(self.up_project.weight, mean=class="num">0.0, std=class="num">0.02) nn.init.constant_(self.up_project.bias, class="num">0) def forward(self, hidden_states): # Apply down projection and ReLU activation hidden_states = self.down_project(hidden_states) hidden_states = F.relu(hidden_states) # Apply dropout hidden_states = self.dropout(hidden_states) # Apply up projection hidden_states = self.up_project(hidden_states) # Apply dropout again hidden_states = self.dropout(hidden_states) class="kw">return hidden_states # Define the GPT2LMHeadModelWithAdapters class, which inherits from GPT2LMHeadModel # and adds adapter layers to each transformer layer class GPT2LMHeadModelWithAdapters(GPT2LMHeadModel): def __init__(self, config): super().__init__(config) # Create a list of adapter modules, one for each transformer layer self.adapters = nn.ModuleList([Adapter(config.n_embd) for _ in range(config.n_layer)]) def forward( self, input_ids=None, past_key_values=None, attention_mask=None, token_type_ids=None, position_ids=None, head_mask=None, inputs_embeds=None, encoder_hidden_states=None, encoder_attention_mask=None, labels=None, use_cache=None, output_attentions=None, output_hidden_states=None, return_dict=None, ): # Get the outputs from the transformer transformer_outputs = self.transformer( input_ids, past_key_values=past_key_values, attention_mask=attention_mask, token_type_ids=token_type_ids, position_ids=position_ids, head_mask=head_mask, inputs_embeds=inputs_embeds, encoder_hidden_states=encoder_hidden_states, encoder_attention_mask=encoder_attention_mask, use_cache=use_cache, output_attentions=output_attentions, output_hidden_states=output_hidden_states, return_dict=return_dict, ) hidden_states = transformer_outputs[class="num">0] # Apply each adapter to the hidden states
「适配器前向与ONNX导出的关键代码路径」
这段 Python 实现展示了带适配器的 GPT2 语言模型如何在前向传播中叠加多路 adapter，并在有标签时计算移位交叉熵损失。循环里 hidden_states = hidden_states + adapter(hidden_states) 表示每个 adapter 对主干隐状态做残差增强，而非替换原特征。
损失计算采用标准因果语言建模做法：将 lm_logits 去掉末位、labels 去掉首位后对齐，再用 nn.CrossEntropyLoss() 展平求均值，这一移位操作使下一 token 预测误差可被反向传播。
load_model_and_tokenizer 用 AutoConfig.from_pretrained 与自定义 GPT2LMHeadModelWithAdapters.from_pretrained 载入权重，异常时走 logging.error 并 re-raise，便于在 MT5 外接 Python 服务里定位加载失败。
导出侧 export_model_to_onnx 将 config.model_type 的连字符替换为下划线，固定取 causal-lm-with-past 特征，这意味着导出的 ONNX 支持带 past_key_values 的增量推理，对盯盘机器人的低延迟响应有直接价值。若目录不存在则进入创建分支（代码截断于此），实跑时建议先确认 output_path.parent 权限避免写入异常。
for i, adapter in enumerate(self.adapters): hidden_states = hidden_states + adapter(hidden_states) # Get the logits for the language modeling head lm_logits = self.lm_head(hidden_states) # Compute loss if labels are provided loss = None if labels is not None: # Shift logits and labels for loss computation shift_logits = lm_logits[..., :-class="num">1, :].contiguous() shift_labels = labels[..., class="num">1:].contiguous() # Flatten the logits and labels for cross-entropy loss loss_fct = nn.CrossEntropyLoss() loss = loss_fct(shift_logits.view(-class="num">1, shift_logits.size(-class="num">1)), shift_labels.view(-class="num">1)) # Return the outputs in the appropriate format if not return_dict: output = (lm_logits,) + transformer_outputs[class="num">1:] class="kw">return ((loss,) + output) if loss is not None else output class="kw">return modeling_outputs.CausalLMOutputWithCrossAttentions( loss=loss, logits=lm_logits, past_key_values=transformer_outputs.past_key_values, hidden_states=transformer_outputs.hidden_states, attentions=transformer_outputs.attentions, cross_attentions=transformer_outputs.cross_attentions, ) # Function to load the model and tokenizer def load_model_and_tokenizer(model_id): try: # Load the model configuration config = AutoConfig.from_pretrained(model_id) # Load the model model = GPT2LMHeadModelWithAdapters.from_pretrained(model_id) # Load the tokenizer # tokenizer = AutoTokenizer.from_pretrained(&class="macro">#x27;gpt2&class="macro">#x27;) class="kw">return config, model,tokenizer except Exception as e: # Log any errors that occur during loading logging.error(f"Error loading model and tokenizer: {e}") raise # Function to class="kw">export the model to ONNX format def export_model_to_onnx(model, config, tokenizer, output_path, opset): try: # Get the appropriate feature for the model model_type = config.model_type.replace("-", "_") feature = "causal-lm-with-past" # Get the ONNX configuration onnx_config_constructor = FeaturesManager.get_config(model_type, feature=feature) onnx_config = onnx_config_constructor(config) # Create the output directory if it doesn&class="macro">#x27;t exist if not os.path.exists(output_path.parent):
把适配器模型落盘成 ONNX 并做 INT4 量化
这段 Python 脚本干的事很直接：把名为 gpt2_Adapter-tuning 的带适配器 GPT2 模型导出为 ONNX，再用动态量化压到 INT4 权重。导出时固定用了 opset=14，输出路径落在 ./gpt2_onnx/gpt2_adapter_tuning.onnx，preprocessor 绑定了 tokenizer，保证推理端前后处理一致。 main() 里先 load_model_and_tokenizer 取回 config、model、tokenizer 三件套，再调 export_model_to_onnx 执行转换；转换过程包了 try/except，任何异常都会经 logging.error 打出具体信息后 raise，不会静默失败。 check_onnx() 用 onnx.checker.check_model 加载并校验模型结构，通过会打印 ONNX model check passed!，这是本地验证图合法性的最快一步。 quantization() 从 onnxruntime.quantization 引入 quantize_dynamic，对原 onnx 做权重量化到 QUInt4，产物存为 gpt2_onnx/quantized_gpt2.onnx。外汇与贵金属行情序列若接这套轻量模型做本地推理，须注意 INT4 动态量化可能引入数值偏差，信号仅作概率参考，实盘高风险。
os.makedirs(output_path.parent) # Export the model to ONNX class="kw">export( model=model, config=onnx_config, opset=opset, output=output_path, preprocessor=tokenizer, ) # Log success message logging.info(f"Model successfully converted to ONNX and saved in {output_path}") except Exception as e: # Log any errors that occur during class="kw">export logging.error(f"Error exporting model to ONNX: {e}") raise # Main function to orchestrate the process def main(): # Define the model ID, output path, and ONNX opset version model_id = "gpt2_Adapter-tuning" onnx_path = "./gpt2_onnx" out_path = Path(os.path.join(onnx_path, "gpt2_adapter_tuning.onnx")) opset = class="num">14 # Load the model and tokenizer config, model, tokenizer = load_model_and_tokenizer(model_id) # Export the model to ONNX export_model_to_onnx(model, config, tokenizer, out_path, opset) def check_onnx(): # Check the ONNX model onnx_model = onnx.load("gpt2_onnx/gpt2_adapter_tuning.onnx") onnx.checker.check_model(onnx_model) print("ONNX model check passed!") def quantization(): from onnxruntime.quantization class="kw">import quantize_dynamic, QuantType # load model model_path = "gpt2_onnx/gpt2_adapter_tuning.onnx" onnx_model = onnx.load(model_path) class="macro">#dynamic quantize INT4 quantized_model_path = "gpt2_onnx/quantized_gpt2.onnx" quantize_dynamic(model_path, quantized_model_path, weight_type=QuantType.QUInt4) print(f"Save the quantized model to: {quantized_model_path}") model = GPT2LMHeadModelWithAdapters.from_pretrained(model_id) out_path = Path(os.path.join(onnx_path, "gpt2_adapter_tuning.onnx")) class="kw">export(model=model, config=onnx_config, opset=opset, output=output_path, preprocessor=tokenizer) opset = class="num">14 model_id = "gpt2_Adapter-tuning" onnx_path = "./gpt2_onnx"
◍ EA 与 Python 服务端的分工和信号逻辑
这套架构里，EA 客户端只干两件事：在 MT5 端采集合约数据、把订单发出去；真正的脑子放在 Python 服务器上，它收数据、跑 GPT-2 推理、再把交易信号回传。两边靠 Socket 通信，客户端不碰模型权重，服务端不碰交易账户。 EA 侧策略刻意做得很薄：每 1 分钟抓过去 20 个收盘点，推给服务端等回包，收到信号就下单，不设止损止盈，且任意时刻只持有一个单。这纯粹是验证管道用的示例，外汇和贵金属波动剧烈、杠杆高风险极高，实盘绝不能直接套用。 服务端拿到 20 个价格后，加载并常驻 GPT-2 模型与 tokenizer，推理出预测价均值，拿当前实际价减预测均值：差大于 0 发买、小于 0 发卖、等于 0 静默。它还会先探一遍设备，决定走 CPU 还是 GPU 推理，再出信号。 要验证这套分工，你在 MT5 里建个 1 分钟图 EA，把 20 根收盘价通过 Socket 扔给本地 Python，看回包信号是否和上面三条规则一一对应即可。
「把 GPT-2 微调模型挂到 MT5 推理服务上」
要让 EA 在 MT5 里实时调用你微调好的 GPT-2 适配器模型，核心是把上一篇文章的 server.py 改成支持 GPT-2 推理与断线重连。原脚本的握手和主解析逻辑保留，改动集中在模型加载、eva() 推理函数以及连接状态检测三处。
load_model() 现在同时加载 GPT2LMHeadModelWithAdapters 和原生 gpt2 分词器；eva() 则取客户端发来的最后 20 个数据点（encoder_length=20）拼成提示，用分词器编码后送模型 generate，最多解 200 token，再切出前 10 个预测值（prediction_length=10）算均值。若最后价减预测均值 ≥0 倾向发卖出信号，反之倾向买入信号——外汇与贵金属波动剧烈，这仅是概率倾向，实盘须自担高风险。
服务器监听地址写成 0.0.0.0 而非 127.0.0.1，端口 10055，这样跨主机 EA 也能连，只要客户端填对服务器 IP。新增 is_connected() 探活、handshake() 隔离握手、self.last_action 记录上一次信号，仅在信号变化时打印，避免回测时刷屏；客户端掉线后该字段重置为 None，重连不会误发旧信号。
以下片段展示了库导入与模型加载、推理输入截取的关键代码逻辑，可直接对照你本地的 server.py 改：
class="kw">import socket from time class="kw">import sleep class="kw">import pandas as pd class="kw">import numpy as np class="kw">import warnings class="kw">import base64 class="kw">import hashlib class="kw">import class="kw">struct from torch2onnx class="kw">import GPT2LMHeadModelWithAdapters,Adapter from transformers class="kw">import AutoTokenizer class="kw">import logging class="kw">import torch from statistics class="kw">import mean # Set logging and warning logging.basicConfig(level=logging.INFO) warnings.filterwarnings("ignore") # Set device dvc=&class="macro">#x27;cuda&class="macro">#x27; if torch.cuda.is_available() else &class="macro">#x27;cpu&class="macro">#x27; # Global model_id = "gpt2_Adapter-tuning" encoder_length=class="num">20 prediction_length=class="num">10 info_file="results.json" host="class="num">0.0.class="num">0.0" port=class="num">10055 # Function to loda model def load_model(): try: # Load the model model = GPT2LMHeadModelWithAdapters.from_pretrained(model_id).to(dvc) # Load the tokenizer tokenizer = AutoTokenizer.from_pretrained(&class="macro">#x27;gpt2&class="macro">#x27;) print("Model loaded!") class="kw">return model,tokenizer except Exception as e: # Log any errors that occur during loading logging.error(f"Error loading model and tokenizer: {e}") raise def eva(msg,model,tokenizer): # Get the data msg=np.fromstring(msg, dtype=class="type">float, sep= &class="macro">#x27;,&class="macro">#x27;).tolist() # Parse the data input_data=msg[-encoder_length:] # Create the prompt prompt = &class="macro">#x27; &class="macro">#x27;.join(map(str, input_data))
把模型输出转成下单信号
预测段跑完之后，真正决定发 buy 还是 sell 的其实是几行后处理。模型 generate 时设了 do_sample=True 且 max_length=200，意味着每次推理带随机性，同样 prompt 可能吐出不同价格序列，这点在外盘黄金或欧美盘里会直接影响信号稳定性。 代码先取生成文本的第一行，用 try_float 把每个 token 转成浮点，过滤掉非数字杂质，再截断到 prediction_length 根 K 线。这里如果模型胡写一通，generated_prices 可能为空，实盘前务必在本地 print 一下长度。 信号逻辑很直白：拿 input_data 最后一根收盘价 last_price，和预测均值 prediction_mean 比。若 last_price 大于等于均值，返回 sell；否则返回 buy。注意这是单均值阈值，没有止损和仓位，外汇贵金属高杠杆下直接挂单可能扛不住反向毛刺。 server_ 类负责把模型挂成 TCP 服务，__init__ 里 bind 完就 accept 阻塞等 MT5 侧连接，listen(1) 说明只接一个客户端。msg 方法里 sleep(0.5) 轮询 is_connected，意味着最低 0.5 秒才探一次链路，做秒级 scalping 的别指望它快。
# Generate the predication
token=tokenizer.encode(prompt, return_tensors=&class="macro">#x27;pt&class="macro">#x27;).to(dvc)
attention_mask = torch.ones_like(token).to(dvc)
model.eval()
generated = tokenizer.decode(
model.generate(
token,
attention_mask=attention_mask,
pad_token_id=tokenizer.eos_token_id,
do_sample=True,
max_length=class="num">200)[class="num">0],
skip_special_tokens=True)
generated_prices=generated.split(&class="macro">#x27;\n&class="macro">#x27;)[class="num">0]
# Remove non-numeric formats
def try_float(s):
try:
class="kw">return class="type">float(s)
except ValueError:
class="kw">return None
generated_prices=generated_prices.split()
generated_prices=list(map(try_float,generated_prices))
generated_prices = [f for f in generated_prices if f is not None]
generated_prices=generated_prices[class="num">0:prediction_length]
# Calculate and send the results
last_price=input_data[-class="num">1]
prediction_mean=mean(generated_prices)
if (last_price-prediction_mean) >= class="num">0:
# print(&class="macro">#x27;Send sell.&class="macro">#x27;)
class="kw">return "sell"
else:
# print("Send buy.")
class="kw">return "buy"
class server_:
def __init__(self, host = host, port = port):
self.sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM,)
self.host = host
self.port = port
self.sk.bind((self.host, self.port))
self.re = &class="macro">#x27;&class="macro">#x27;
self.model,self.tokenizer=load_model()
self.stop=None
self.sk.listen(class="num">1)
self.sk_, self.ad_ = self.sk.accept()
self.last_action=None
print(&class="macro">#x27;server running：&class="macro">#x27;,self.sk_, self.ad_)
def msg(self):
self.re = &class="macro">#x27;&class="macro">#x27;
wsk=False
while True:
sleep(class="num">0.5)
if self.is_connected():
try: