从回测到实盘：基于大语言模型的量化交易AI智能体架构与实战

1. 项目概述：当“龙虾”遇上真实行情

最近在量化圈子里，“龙虾”这个词的热度有点高。不是餐桌上的那个，而是一个代号，指的是一类新兴的、试图将大语言模型（LLM）能力与量化交易逻辑结合的开源项目或工具集。我最初也是抱着“这玩意儿到底是不是噱头”的心态，从GitHub上拉了个类似“OpenClaw”的代码仓库，想看看所谓的“AI交易员”到底有几斤几两。

在本地用历史数据回测时，它的表现中规中矩，能生成一些基于新闻情绪或技术指标的逻辑描述，但总感觉隔靴搔痒，像是一个在纸上谈兵的参谋。直到我下决心，把它接入了一个实时的、带Tick数据的模拟交易API，整个项目的画风才彻底变了。当虚拟账户的盈亏数字随着真实市场的每一次跳动而闪烁，当“龙虾”需要实时解析行情、计算指标、并立刻做出“买”、“卖”或“等待”的决策时，我才真正体会到，一个量化系统从“玩具”到“工具”的蜕变，关键就在于这“最后一公里”的接入。

这个项目，就是记录我如何将一个基于Python的、代号“龙虾”的量化AI智能体，从离线回测环境，一步步接入真实行情与模拟交易接口，并观察其行为模式发生质变的全过程。它不再只是分析过去，而是开始尝试“干活”了——虽然这“活”干得可能还很笨拙，甚至会闯祸，但这个过程本身，对于理解AI在金融时序决策中的应用边界和挑战，价值巨大。

2. 核心设计：从回测沙箱到实时战场的架构演进

2.1 原有“龙虾”系统的局限性分析

我手头这个“龙虾”原型，其核心是一个基于Transformer架构微调过的中型语言模型，配合一套传统的量化分析库（比如TA-Lib,pandas）。它的工作流是典型的离线模式：

1. 数据输入：加载CSV格式的日K线或分钟K线历史数据。
2. 信息处理：模型接收当前时间点及之前一段窗口期的数据（价格、成交量、指标），并结合可能爬取的文本新闻（需额外模块），生成一段自然语言描述，例如：“当前价格突破20日均线，但RSI处于超买区域，市场情绪偏多但需警惕短期回调。”
3. 信号生成：一个固定的、硬编码的规则解析器（rule parser）会尝试从这段描述中提取关键词（如“突破”、“超买”、“警惕”），映射成预定义的信号：“强多”、“弱多”、“中性”、“弱空”、“强空”。
4. 策略回测：根据这些信号，在历史数据上模拟交易，计算夏普比率、最大回撤等绩效指标。

问题立刻暴露了：

• 延迟幻觉：模型处理的是已经静止的历史切片，它“知道”所有后续数据，但必须假装不知道。这种训练方式容易让模型产生“后见之明”的偏差。
• 信号粗糙：从自然语言到交易指令的映射损失了大量信息。“警惕回调”到底该不该平仓？何时平仓？规则解析器无法处理这种灰度。
• 无状态、无成本：它没有“持仓”概念，没有“交易成本”（佣金、滑点），更不考虑订单能否以理想价格成交。这就像在玩一个无限金币的单机游戏。

2.2 接入实时行情的核心架构设计

要让“龙虾”干活，必须将其嵌入一个实时事件驱动的系统中。我的设计目标很明确：低延迟、可观测、风控优先。整体架构演变为下图所示的一个异步工作流：

整个系统围绕一个主事件循环运行，由行情API推送驱动。我选择了asyncio来构建这个异步框架，因为行情和交易指令都是高IO密集型的操作，异步能有效避免阻塞。

核心模块拆解：

1. 行情网关：这是系统的感官。我接入了提供模拟交易的券商API（如一些券商提供的量化仿真接口）或开源项目vn.py支持的接口。它的职责是订阅标的（如000001.SZ平安银行）的实时Tick（逐笔成交）或快照（3秒/笔）数据，并以统一格式发布到内部事件总线。
2. 数据总线和缓存：我使用redis作为高速缓存和轻量级消息队列。每一个新的行情Tick到来，都会更新redis中该标的的最新价、买卖盘、成交量等字段。同时，也会触发一个行情更新事件。
3. 策略引擎（“龙虾”核心）：这是系统的大脑。它监听行情更新事件。一旦触发，引擎会：
   • 获取上下文：从redis中读取最近N个周期的K线（由原始Tick实时合成），计算技术指标。
   • 组装Prompt：将“时间、价格、指标、当前持仓状态、账户余额”等信息，结构化地填充到一个预设的Prompt模板中。Prompt的设计至关重要，例如：

     “当前时间：{time}，标的：{symbol}，最新价：{price}，持仓：{position}股，可用资金：{cash}元。近10分钟K线数据如下：[...]。技术指标：RSI={rsi}, MACD={macd}...。请分析当前市场状况，并给出具体的交易操作建议。必须严格按照以下格式输出：操作：买入/卖出/持有；理由：...；数量：...；限价：...（可选）”

   • 调用模型：将组装好的Prompt发送给本地部署的“龙虾”大模型（我用了FastChat部署的Vicuna-7B量化版，对16G内存的机器比较友好）。这里的关键是响应速度，模型推理必须在几百毫秒内完成，否则信号就过时了。因此，模型量化（如使用bitsandbytes库进行INT8量化）和硬件加速（CUDA）是必须的。
   • 解析与验证：对模型输出的文本进行严格的格式和逻辑解析。不仅要用正则表达式提取“操作、数量、价格”等字段，还要进行业务逻辑校验：买入数量是否超过资金允许？卖出数量是否超过持仓？价格是否偏离市价过远（避免异常指令）？
4. 订单管理与风控：这是系统的手和刹车。通过校验的指令会被转化为具体的订单请求（订单类型、价格、数量），发送给交易API。同时，这里部署了硬风控：单笔最大下单量、日内累计亏损限额、最大持仓比例等。任何订单执行前必须通过风控检查。
5. 监控与日志：这是系统的黑匣子。所有事件——行情数据、模型Prompt、模型输出、解析结果、订单状态、账户变动——都以高密度写入日志文件（如structlog）和数据库（如InfluxDB用于时间序列数据）。一个简单的Flask面板用来实时展示账户权益曲线、持仓、以及最新的模型决策理由。

注意：在Prompt中强制规定输出格式，是保证程序可解析的关键。直接让模型生成自由文本，再试图用另一个模型去理解，在实时系统中是灾难性的。结构化输出（JSON或固定格式文本）是工业级应用的前提。

2.3 技术栈选型与考量

• Python 3.9+：量化生态最成熟的语言。
• 异步框架：asyncio+aiohttp，处理高并发行情和网络IO。
• 缓存与消息：Redis，性能极高，用作数据缓存和简易消息队列足够。
• 模型服务：FastChat或vLLM，用于高效部署和推理开源LLM。考虑到资源，我选择了7B参数的模型，并使用bitsandbytes进行INT8量化，在消费级显卡（如RTX 4060 Ti 16G）上也能获得不错的推理速度（~100-300ms）。
• 量化分析：pandas（数据处理）、TA-Lib（技术指标）、numpy（数值计算）。
• 交易接口：根据选择的券商或模拟平台使用其Python SDK，或使用vn.py这类开源框架进行封装。
• 监控：Loguru或Structlog用于日志，InfluxDB+Grafana用于可视化监控。

选择这个技术栈的核心考量是平衡性能、开发效率与资源消耗。全量LLM（如70B参数）实时推理对个人开发者不现实，量化后的7B-13B模型是可行的起点。Redis的引入避免了每次计算都从数据库读取大量历史数据，极大降低了延迟。

3. 关键实现：打通数据流与决策流的魔鬼细节

3.1 实时行情接入与K线合成

接入行情API后，拿到的是源源不断的Tick数据。但大多数技术指标是基于K线（OHLC：开、高、低、收、量）计算的。因此，第一个关键环节是实时K线合成。

我创建了一个BarGenerator类，它维护一个字典，以标的和周期（如1分钟、5分钟）为键，缓存最新的Tick数据流。

import asyncio from collections import defaultdict from datetime import datetime, timedelta from typing import Dict, List import pandas as pd class BarGenerator: def __init__(self): # 存储每个symbol-period对应的未完结的Tick列表和当前K线 self.ticks_buffer: Dict[str, Dict[str, List]] = defaultdict(lambda: defaultdict(list)) self.current_bar: Dict[str, Dict] = defaultdict(dict) async def on_tick(self, symbol: str, tick: dict): """处理新的Tick数据""" # tick 结构: {'price': float, 'volume': int, 'datetime': datetime, ...} current_time = tick['datetime'] for period in ['1min', '5min']: # 支持多周期 period_seconds = int(period.replace('min', '')) * 60 # 计算该Tick所属的K线起始时间 bar_start = self._align_time(current_time, period_seconds) bar_key = f"{symbol}_{period}" if bar_key not in self.current_bar or self.current_bar[bar_key].get('datetime') != bar_start: # 新K线开始，推送旧K线（如果存在），并初始化新K线 if bar_key in self.current_bar: await self._push_bar(bar_key, self.current_bar[bar_key]) self.current_bar[bar_key] = { 'symbol': symbol, 'period': period, 'datetime': bar_start, 'open': tick['price'], 'high': tick['price'], 'low': tick['price'], 'close': tick['price'], 'volume': tick['volume'], 'ticks': 1 } else: # 更新当前K线 bar = self.current_bar[bar_key] bar['high'] = max(bar['high'], tick['price']) bar['low'] = min(bar['low'], tick['price']) bar['close'] = tick['price'] bar['volume'] += tick['volume'] bar['ticks'] += 1 def _align_time(self, dt: datetime, period_seconds: int) -> datetime: """将时间对齐到K线周期起始点""" epoch_seconds = int(dt.timestamp()) aligned_seconds = (epoch_seconds // period_seconds) * period_seconds return datetime.fromtimestamp(aligned_seconds) async def _push_bar(self, bar_key: str, bar: dict): """将完整的K线数据发布出去，例如写入Redis或触发事件""" # 这里可以计算技术指标 bar_df = pd.DataFrame([bar]) # 计算RSI, MACD等 (这里简化) # ... 计算指标逻辑 ... # 将带指标的K线数据存入Redis，供策略引擎读取 # await redis_client.set(f"bar:{bar_key}", json.dumps(bar)) print(f"Bar Generated: {bar}")

这个类确保了无论Tick多么频繁，策略引擎总是能获取到最新、已计算好指标的固定周期K线数据，这是后续模型分析的基石。

3.2 模型Prompt工程与结构化输出解析

这是“龙虾”能否产出有效指令的核心。糟糕的Prompt会让模型胡言乱语，好的Prompt则能引导它进行专业思考。

我的Prompt模板经历了数次迭代：

V1（失败）：“请分析一下平安银行的股票。” 结果：模型输出了一篇关于银行股基本面的短文，毫无操作性。

V2（改进）：“给定最新价格xx，20日均线yy，请给出交易建议。” 结果：模型可能会说“建议关注”、“可以考虑买入”，依然无法解析。

V3（最终版）：

你是一个专业的量化交易员。请严格根据以下信息做出交易决策。 【交易上下文】 - 时间：{current_time} - 标的代码：{symbol} - 当前持仓：{position}股（成本价{avg_cost}） - 账户可用资金：{cash}元 - 最新价：{last_price}元 【近期市场数据（最近10根1分钟K线）】 {df.to_string(index=False)} # 这里插入一个格式化的DataFrame字符串 【计算出的技术指标】 - RSI(14): {rsi_value:.2f} - MACD(12,26,9): DIF={dif:.4f}, DEA={dea:.4f}, MACD柱={macd_bar:.4f} - 当前价格相对于20周期均线的位置：{price_vs_ma}% 【指令】 请综合以上信息，判断当前应该执行何种操作。你的输出必须且只能包含以下三个部分，用换行分隔： 操作：[买入|卖出|持有] 数量：[正整数，若为持有则填0] 理由：[简要说明你的决策逻辑，不超过50字] 示例： 操作：买入 数量：100 理由：价格回调至20日均线附近获得支撑，RSI脱离超卖区，MACD金叉在即，短期有反弹动能。

这个Prompt的成功之处在于：

1. 明确角色：让模型进入“交易员”角色。
2. 提供结构化上下文：持仓、资金、价格、数据、指标，所有必要信息一目了然。
3. 强制结构化输出：严格限定输出格式，极大降低了后续解析的复杂度。

解析代码示例：

import re def parse_model_output(output_text: str) -> dict: """ 解析模型输出的结构化文本 """ pattern = r"操作：(\w+)\s*数量：(\d+)\s*理由：(.*)" match = re.search(pattern, output_text, re.DOTALL) if not match: raise ValueError(f"无法解析模型输出: {output_text}") action, amount_str, reason = match.groups() action = action.strip() amount = int(amount_str.strip()) reason = reason.strip() # 验证操作类型 if action not in ["买入", "卖出", "持有"]: raise ValueError(f"非法操作类型: {action}") # 验证数量（持有时为0） if action == "持有" and amount != 0: raise ValueError(f"持有操作时数量必须为0，得到: {amount}") if action != "持有" and amount <= 0: raise ValueError(f"买卖操作数量必须为正整数，得到: {amount}") return { "action": action, "amount": amount, "reason": reason }

3.3 风险控制模块的实现

没有风控的交易系统是自杀系统。我实现了三层风控：

1. 指令级风控：在解析模型指令后立即执行。

   def pre_trade_risk_check(signal: dict, portfolio: dict) -> bool: """订单执行前风控""" symbol = signal.get('symbol') action = signal.get('action') amount = signal.get('amount') price = signal.get('price', portfolio['last_price']) # 默认用最新价估算 # 1. 单笔最大下单量 if amount > MAX_ORDER_PER_TRADE: log.warning(f"风控拦截：单笔下单数量{amount}超过限制{MAX_ORDER_PER_TRADE}") return False # 2. 买入资金检查 if action == "买入": needed_cash = amount * price * (1 + COMMISSION_RATE) if needed_cash > portfolio['available_cash']: log.warning(f"风控拦截：所需资金{needed_cash:.2f}大于可用资金{portfolio['available_cash']:.2f}") return False # 3. 卖出持仓检查 elif action == "卖出": if amount > portfolio['positions'].get(symbol, 0): log.warning(f"风控拦截：卖出数量{amount}超过持仓{portfolio['positions'].get(symbol, 0)}") return False # 4. 价格偏离检查（避免异常价格订单） if price < portfolio['last_price'] * 0.9 or price > portfolio['last_price'] * 1.1: log.warning(f"风控拦截：订单价格{price}偏离市价{portfolio['last_price']}超过10%") return False return True

2. 账户级风控：独立进程定时检查。

   • 日内最大亏损：如果当日浮动亏损超过总资产的2%，则暂停所有新开仓。
   • 最大持仓比例：单标的持仓市值不超过总资产的20%。
   • 连续止损：如果连续3笔交易亏损，强制进入“冷却期”1小时。

3. 系统级风控：

   • 心跳监测：如果行情中断超过10秒，或模型响应超时（如5秒），系统自动暂停交易，转为待机状态。
   • 异常指令熔断：如果单位时间内（如1分钟）收到超过N次（如5次）被风控拒绝的指令，可能模型已“发疯”，触发熔断，暂停该策略引擎一段时间。

4. 实战观察：接入真实行情后的行为质变

系统跑起来后，我让它在模拟账户上运行了整整一周，观察它与之前回测版本的巨大差异。

4.1 从“分析者”到“决策者”的思维转变

在回测中，“龙虾”更像一个评论员，它的输出是开放式的分析。而在实时系统中，它被逼成了一个必须下注的决策者。最明显的变化是：

• 输出确定性增强：在Prompt的压力下，它很少再输出“可能”、“或许”、“建议关注”这类模糊词汇。它被迫在“买入”、“卖出”、“持有”中三选一，并给出一个明确的理由。这暴露了模型在不确定性下做决策的“性格”，有的版本偏向激进（频繁交易），有的则偏向保守（长期持有）。
• 开始考虑“状态”：因为它能接收到当前的持仓和资金信息，它的决策出现了连贯性。例如，在一次盈利买入后，当价格小幅回落时，它给出的理由是“属于正常技术性回调，多头趋势未改，继续持有”，而不是像回测中每个时间点独立判断那样可能给出“卖出”信号。它开始有了“交易记忆”的雏形。
• 对市场噪音的反应：Tick级别的数据充满了噪音。我观察到模型在行情剧烈波动时（例如快速拉升又砸下），有时会在几分钟内给出相反的信号。这促使我改进了Prompt，加入了“请过滤短期噪音，关注主要趋势”的指令，并增大了数据观察窗口（从10根K线增加到30根），有效减少了“追涨杀跌”的无效交易。

4.2 暴露出的新问题与挑战

1. 延迟与滑点的真实伤害：回测中假设订单立即以当前价成交。现实中，从模型推理（~200ms）到指令解析、风控、发单，再到交易所撮合，可能有500ms-1秒的延迟。在快速变动的市场中，成交价可能与预期价相差甚远（滑点）。我亲眼看到一次模型发出“限价买入”指令，但因价格快速上涨，订单一直未能成交，错过了整个波段。教训：必须对模型进行“延迟与滑点”的感知训练，或者在Prompt中强调“仅在流动性充裕、趋势明确时操作”，或者直接使用市价单并接受更高的成本。
2. 模型的不稳定性：同样的市场情况，模型有时会给出略有不同的决策。这源于LLM本身的概率生成特性。虽然通过设置较低的temperature参数（如0.1）可以增加确定性，但无法完全根除。对策：引入“投票机制”或“多轮思考链（Chain-of-Thought）”。例如，让模型连续推理三次，取多数票作为最终决策；或者在Prompt中要求它先逐步推理，再给出结论，提高决策过程的可靠性。
3. 对极端行情的误判：在一次市场突然暴跌时，模型基于“RSI超卖”和“价格偏离均线过远”的指标，给出了“买入”信号。这从纯技术分析上看似乎合理，但它完全忽略了引发暴跌的潜在宏观消息（系统当时未接入实时新闻）。结果买入后价格继续阴跌，造成了较大回撤。启示：纯技术分析的AI在黑天鹅事件面前是脆弱的。一个更健壮的系统必须融合多模态信息，包括新闻舆情、板块资金流等。

4.3 性能优化与迭代

为了让系统更“可用”，我做了以下关键优化：

• 模型推理加速：
  • 量化：将FP16模型转换为INT8，推理速度提升近一倍，精度损失在可接受范围内。
  • 批处理：如果需要监控多个标的，将多个标的的Prompt组成一个Batch一次性送入模型，能极大提升GPU利用率。
  • 缓存：对于相似的市场状态（如价格、指标变化不大），可以缓存上一次的模型输出，避免重复计算，直到市场状态发生显著变化。
• 事件驱动架构优化：将不同的策略逻辑（如趋势跟踪、均值回归）拆分成独立的“策略工人”，通过消息队列接收行情事件，并行处理，提高系统吞吐量。
• 引入简单的强化学习反馈环：我设计了一个简单的奖励函数：每笔交易平仓后，根据盈亏计算一个奖励值。将这个奖励值连同当时的市场状态（State）和模型采取的动作（Action）一起存储下来。定期用这些数据对模型进行微调（P-tuning或LoRA），目标是让模型学会向盈利动作靠拢。这是一个初步的尝试，但为系统从“基于规则模仿”向“基于结果学习”演进提供了可能。

5. 常见问题与排查实录

在部署和运行过程中，我踩过不少坑，这里记录下最典型的几个问题和解决方法。

5.1 模型响应超时或崩溃

• 现象：策略引擎长时间收不到模型回复，或模型服务进程突然退出。
• 排查：
  1. 检查GPU内存是否溢出。使用nvidia-smi命令监控。LLM推理很吃显存，特别是处理长序列时。
  2. 检查Prompt长度。过长的Prompt（如包含大量历史K线数据）会导致推理时间指数级增长。需要优化数据表示，例如只传递指标值而非原始K线。
  3. 查看模型服务日志。可能是遇到了模型无法处理的特殊字符或格式，导致内部错误。
• 解决：
  • 为模型服务设置显存监控和自动重启机制（如使用supervisor）。
  • 精简Prompt，只传递核心信息。将历史数据摘要为几个关键特征值。
  • 在代码中设置严格的超时（如asyncio.wait_for），超时后触发降级策略（如使用简单规则库生成信号）。

5.2 解析模型输出失败

• 现象：parse_model_output函数频繁抛出ValueError，提示“无法解析模型输出”。
• 排查：查看日志中模型输出的原始文本。常见问题有：
  1. 模型没有严格遵守格式，在“操作：”前加了一些废话。
  2. 模型生成了中文标点或全角字符，导致正则匹配失败。
  3. 模型输出了“建议买入”而不是“买入”。
• 解决：
  • 强化Prompt：在Prompt开头和结尾反复强调“必须严格按照格式输出”。
  • 后处理清洗：在解析前，对输出文本进行清洗，移除多余的空行、首尾空白，并将全角字符转换为半角。
  • 使用更鲁棒的解析器：从正则表达式升级到基于大语言模型的“格式校验与修正”小模型，或者使用langchain的OutputParser组件。对于简单格式，可以尝试用多行正则或分步提取。

5.3 交易指令的“毛刺”与过度交易

• 现象：账户频繁发出小额买卖指令，产生大量手续费，侵蚀利润。
• 排查：分析模型日志，发现当价格在某个关键点位（如均线）附近小幅震荡时，模型会随着价格上下穿均线而频繁改变观点。
• 解决：
  • 在策略层增加过滤器：引入“信号确认”机制。例如，只有当买入信号连续出现2个周期以上，才真正发单。或者在发出反向信号前，必须满足一定幅度的价格变动或时间间隔。
  • 修改Prompt：加入明确的交易频率限制指令，如“避免在窄幅震荡区间内进行交易”、“除非趋势发生明确反转，否则应保持现有持仓”。
  • 在风控层增加频率限制：设置同一标的的最小交易时间间隔（例如，至少间隔5分钟才能发出新指令）。

5.4 模拟环境与实盘的心理差异

• 现象：在模拟盘表现稳定的策略，一旦想到要投入真金白银，就会对模型的每一个决策产生怀疑，总想手动干预。
• 反思：这其实不是技术问题，而是人机协作和心理问题。量化交易的核心纪律之一就是排除情绪干扰，严格执行系统信号。如果无法信任自己的系统，说明系统本身可能存在未发现的缺陷，或者你对它的理解不够深入。
• 建议：
  1. 充分回测与模拟：在模拟盘运行的时间要足够长，经历各种市场行情（趋势、震荡、暴跌、暴涨）。
  2. 小额实盘试水：用你完全能承受损失的金额开始实盘，目的是验证整个流程（开户、API、风控、结算）是否通畅，并感受真实盈亏带来的心理压力。
  3. 建立干预规则：明确界定在什么情况下可以手动干预（如系统出现明显bug、市场出现极端流动性危机），什么情况下绝对不可以。将规则写下来，避免临时起意。

把“龙虾”接入真实行情，就像给一个聪明的学生安排了毕业实习。在学校的模拟考试（回测）中，它可能成绩优异，但面对真实职场（市场）的复杂、动态和压力，它的不足和潜力才被真正激发出来。这个过程让我深刻认识到，AI在量化领域的应用，绝不是用一个模型替代所有，而是构建一个以AI为核心决策组件、同时被严密的风险管理、高效的数据工程和稳健的系统架构所包裹的复杂系统。这个项目远未结束，它只是一个起点。下一步，我计划引入更多维度的数据（如订单簿、新闻情感），尝试多模型集成投票，并深化强化学习的反馈循环。这条路很长，但看着自己搭建的系统开始自主地“呼吸”市场空气并做出反应，这种成就感，远超任何一次离线回测的漂亮曲线。