1. 从零搭建AI期货交易Agent的整体思路1.1 为什么选择Agent架构而不是传统量化策略做期货交易这些年我见过太多人一上来就写死规则均线金叉买入、死叉卖出参数调了几百遍回测曲线漂亮得不行一上实盘就拉胯。问题出在哪市场结构会变波动率会变而你写死的规则不会变。Agent架构的核心区别在于它不是一套固定规则而是一个能感知市场状态、调用工具、做出决策、并根据结果调整行为的闭环系统。你可以把它理解成一个交易员而不是一个计算器。交易员会看盘、会分析、会犹豫、会止损、会复盘Agent做的事情本质上是一样的。我选择用Agent来做期货交易主要基于三个判断期货市场的多因子特性影响螺纹钢价格的因素包括铁矿石成本、钢厂开工率、库存数据、宏观情绪、季节性规律等这些因子之间的权重关系是动态变化的。传统多因子模型需要人工设定权重而Agent可以通过工具调用动态获取这些数据并让大模型做综合判断。决策链的可解释性Agent的每一步决策都有推理过程记录这对于复盘和迭代至关重要。传统黑盒模型告诉你“买”但不告诉你为什么出了问题你都不知道从哪改。工具生态的复用性Agent可以调用行情API、新闻接口、计算工具、风控模块这些工具是独立开发的可以单独测试、单独替换不会牵一发动全身。1.2 整体架构分层设计我把整个系统分成四层从下到上依次是层级职责核心组件数据层行情获取、数据清洗、特征计算行情API、数据管道、指标计算模块工具层封装可被Agent调用的原子能力下单接口、持仓查询、风控检查、新闻摘要决策层大模型推理、策略生成、信号输出LLM推理引擎、Prompt模板、记忆模块执行层订单管理、仓位控制、日志记录订单管理器、风控网关、审计日志这个分层的好处是每一层都可以独立替换。比如你今天用某个大模型做决策明天想换一个只需要改决策层的接口适配下面的数据层和工具层完全不用动。1.3 技术选型背后的取舍选型这件事我的原则是能用成熟方案就不自己造轮子但核心风控必须自己写。大模型选择我测试过多个模型最终选择了一个在中文金融文本理解上表现稳定的模型。原因很简单期货交易中很多信息来自中文新闻、研报、公告模型对中文语境的理解能力直接决定了决策质量。另外推理延迟也是硬指标期货行情变化快如果一次推理要等十几秒那基本没法用。Agent框架我用了轻量级的编排方案没有上重型框架。原因是我需要精确控制每一步的输入输出重型框架的抽象层太多出了问题排查成本高。轻量方案虽然代码量大一些但每一行逻辑都是透明的。数据源行情数据用了一家国内期货数据服务商的API延迟在可接受范围内。新闻数据用了RSS聚合加人工筛选的方式没有直接对接付费终端因为初期验证阶段没必要花那个钱。注意不要一上来就追求全自动交易。我的建议是先把Agent做成“决策辅助”模式它给出信号你手动确认执行。跑一两个月确认信号质量稳定了再逐步放开自动执行。2. 核心模块拆解与实操要点2.1 行情数据管道的搭建细节数据是Agent的粮食粮食不干净后面全白搭。我在数据管道上踩过的坑比在策略本身上踩的还多。第一个坑是时间对齐。期货不同品种的交易时段不一样夜盘和日盘的衔接处容易出现数据断层。我的处理方式是统一用交易所时间戳在数据入库时做一次对齐检查如果发现某根K线的时间戳与前后间隔超过正常周期的1.5倍就标记为异常后续计算指标时跳过这根。第二个坑是主力合约切换。螺纹钢有01、05、10三个主力合约切换时价格会跳空。如果不做处理均线、ATR这些指标全乱。我的做法是维护一个主力合约映射表每天收盘后更新在计算连续价格时用价差调整法拼接。第三个坑是数据频率的选择。我试过1分钟、5分钟、15分钟和日线四个频率。最终选择以5分钟为主、日线为辅的组合。1分钟噪音太大Agent容易被假信号带偏15分钟以上反应太慢止损来不及。5分钟是一个平衡点既能捕捉日内波动又不会过于敏感。数据管道的核心代码逻辑大概是这样def align_timestamps(df, freq5min): df df.set_index(timestamp) full_range pd.date_range( startdf.index.min(), enddf.index.max(), freqfreq ) df df.reindex(full_range) # 标记缺失值不填充让Agent知道这里没数据 df[is_gap] df[close].isna() return df注意这里我没有做前向填充。很多人习惯用ffill把缺失值补上但在交易场景里这等于伪造数据。Agent如果基于伪造的数据做决策后果比不决策更严重。我选择保留缺失标记让Agent自己决定“数据不足时是否放弃这次机会”。2.2 工具层的接口设计原则工具层是Agent的手和脚。设计工具接口时我遵循三个原则原则一每个工具只做一件事。比如“查询当前持仓”和“计算持仓盈亏”是两个独立工具不要合并成一个“获取持仓信息”。合并之后Agent调用时无法区分自己到底需要哪个信息容易产生冗余调用。原则二工具返回值必须结构化。不要返回一段自然语言让Agent去解析直接返回JSON。Agent的推理能力应该用在决策上不是用在解析文本上。原则三所有写操作必须幂等。下单接口要支持幂等键防止Agent因为重试逻辑重复下单。这个坑我踩过当时Agent在超时后自动重试结果开了两倍仓位幸好是模拟盘。工具清单大概长这样工具名称功能输入输出get_market_data获取指定品种K线品种代码、周期、数量K线数组get_position查询当前持仓账户ID持仓明细calc_indicators计算技术指标K线数据、指标列表指标值字典check_risk风控检查拟下单信息通过/拒绝原因place_order下单方向、手数、价格订单IDget_news_summary获取新闻摘要品种关键词摘要文本2.3 决策层的Prompt工程实战Prompt是Agent的大脑皮层写得好不好直接决定决策质量。我前后改了十几版说几个关键心得。第一角色设定要具体。不要写“你是一个交易助手”要写“你是一个有十年期货交易经验的分析师擅长螺纹钢和铁矿石的日内交易风格偏保守宁可错过也不做错”。角色越具体模型的输出风格越稳定。第二输入信息要分层。我把输入分成三块市场状态当前价格、涨跌幅、成交量、技术指标均线、MACD、ATR、上下文当前持仓、今日盈亏、剩余可用资金。每块之间用明确的分隔符隔开避免模型混淆。第三输出格式要强制约束。我要求模型必须输出JSON格式包含actionbuy/sell/hold、confidence0-1、reason推理过程、stop_loss止损价、take_profit止盈价五个字段。如果格式不对直接丢弃这次决策不执行。一个实际的Prompt模板片段你是一个保守型期货交易分析师。当前时间{time} 品种{symbol}当前价格{price}今日涨跌{change}% 技术指标 - MA5: {ma5}, MA20: {ma20} - MACD: {macd}, 信号线: {signal} - ATR(14): {atr} 当前持仓{position} 今日盈亏{pnl} 可用资金{available} 请基于以上信息做出交易决策。要求 1. 如果ATR超过过去20日均值的1.5倍视为高波动建议观望 2. 如果当前持仓已有浮盈且MACD出现死叉建议止盈 3. 输出必须是JSON格式包含action、confidence、reason、stop_loss、take_profit实操心得Prompt里的规则不要写太多超过7条模型就开始顾此失彼。我的做法是把硬性风控规则放在代码层做Prompt里只放软性判断逻辑。比如“单笔亏损不超过总资金2%”这条直接在check_risk工具里拦截不依赖模型自觉。2.4 记忆模块的设计与实现Agent如果没有记忆每次决策都是从零开始那就退化成了一个普通的分类器。记忆模块让Agent能记住“上次在这个位置做多亏了”从而调整行为。我的记忆模块分三层短期记忆最近20次决策的记录包括输入、输出、实际结果。这部分直接放在上下文里每次推理都带上。中期记忆按品种和交易时段聚合的统计信息比如“螺纹钢夜盘开盘后30分钟内Agent的做多胜率是35%”。这部分定期更新以摘要形式注入Prompt。长期记忆向量化存储的历史决策案例当当前市场状态与某个历史案例相似时检索出来作为参考。短期记忆的实现最简单就是一个固定长度的队列。中期记忆需要写一个聚合脚本每天收盘后跑一次。长期记忆用了一个轻量级向量库把每次决策的市场状态编码成向量存进去。这里有个细节记忆检索的相似度阈值要调。阈值太高检索不到有用案例阈值太低检索出一堆不相关的反而干扰决策。我实测下来余弦相似度0.75左右比较合适但这个值因品种而异需要根据回测调整。3. 完整实操流程与关键环节实现3.1 环境准备与依赖安装先把基础环境搭起来。我用的Python 3.10太新的版本有些库兼容性不好太老的版本类型提示支持不够。python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install pandas numpy requests apscheduler pip install openai # 或其他模型SDK pip install chromadb # 向量存储 pip install loguru # 日志数据库我用的是SQLite初期验证阶段够用了。等策略稳定了再考虑上PostgreSQL或TimescaleDB。目录结构这样组织ai_futures_agent/ ├── config/ │ ├── settings.yaml # API密钥、品种配置 │ └── prompts/ # Prompt模板 ├── data/ │ ├── market.db # 行情数据 │ └── decisions.db # 决策记录 ├── src/ │ ├── data_pipeline/ # 数据管道 │ ├── tools/ # 工具层 │ ├── agent/ # 决策层 │ ├── execution/ # 执行层 │ └── utils/ # 通用工具 ├── scripts/ │ ├── backtest.py # 回测脚本 │ └── daily_report.py # 日报生成 └── main.py # 主入口3.2 行情数据接入与清洗以螺纹钢主力合约为例数据接入的核心逻辑import requests import pandas as pd from datetime import datetime, timedelta def fetch_kline(symbol, period5m, limit500): 获取K线数据 url f{BASE_URL}/kline params { symbol: symbol, period: period, limit: limit } resp requests.get(url, paramsparams, timeout10) data resp.json() df pd.DataFrame(data[data]) df[timestamp] pd.to_datetime(df[timestamp], unitms) df df.sort_values(timestamp).reset_index(dropTrue) # 数据质量检查 df check_data_quality(df, period) return df def check_data_quality(df, period): 检查数据质量标记异常 freq_map {1m: 60, 5m: 300, 15m: 900, 1d: 86400} expected_gap freq_map.get(period, 300) df[gap] df[timestamp].diff().dt.total_seconds() df[is_abnormal] df[gap] expected_gap * 1.5 # 检查价格异常 df[price_change] df[close].pct_change().abs() df[is_spike] df[price_change] 0.05 # 单根K线涨跌超5%标记 return df清洗完之后计算技术指标。我用的是TA-Lib的替代方案自己写计算函数避免安装TA-Lib的麻烦def calc_ma(series, window): return series.rolling(windowwindow).mean() def calc_atr(df, window14): high, low, close df[high], df[low], df[close] tr1 high - low tr2 (high - close.shift()).abs() tr3 (low - close.shift()).abs() tr pd.concat([tr1, tr2, tr3], axis1).max(axis1) return tr.rolling(windowwindow).mean() def calc_macd(series, fast12, slow26, signal9): ema_fast series.ewm(spanfast).mean() ema_slow series.ewm(spanslow).mean() macd_line ema_fast - ema_slow signal_line macd_line.ewm(spansignal).mean() histogram macd_line - signal_line return macd_line, signal_line, histogram3.3 Agent决策循环的完整实现决策循环是整个系统的心脏。我用APScheduler做定时触发每5分钟跑一次与K线周期对齐。from apscheduler.schedulers.blocking import BlockingScheduler from loguru import logger def decision_loop(): 一次完整的决策循环 try: # 1. 获取行情 df fetch_kline(RB2410, 5m, 200) if df[is_abnormal].iloc[-1]: logger.warning(最新K线异常跳过本次决策) return # 2. 计算指标 indicators { ma5: calc_ma(df[close], 5).iloc[-1], ma20: calc_ma(df[close], 20).iloc[-1], atr: calc_atr(df).iloc[-1], macd: calc_macd(df[close])[0].iloc[-1], signal: calc_macd(df[close])[1].iloc[-1], } # 3. 获取持仓和账户信息 position get_position(account_001) account get_account_info(account_001) # 4. 检索相关记忆 memory retrieve_memory(df, indicators) # 5. 构建Prompt prompt build_prompt( symbolRB2410, pricedf[close].iloc[-1], indicatorsindicators, positionposition, accountaccount, memorymemory ) # 6. 调用模型 response call_llm(prompt) decision parse_decision(response) # 7. 风控检查 if decision[action] ! hold: risk_result check_risk(decision, account, position) if not risk_result[passed]: logger.info(f风控拦截: {risk_result[reason]}) save_decision(decision, executedFalse, reasonrisk_result[reason]) return # 8. 执行 if decision[action] ! hold: order_id place_order(decision) logger.info(f下单成功: {order_id}) # 9. 记录 save_decision(decision, executedTrue) except Exception as e: logger.error(f决策循环异常: {e}) send_alert(fAgent异常: {e}) scheduler BlockingScheduler() scheduler.add_job(decision_loop, cron, minute*/5, hour9-11,13-15,21-23) scheduler.start()3.4 风控网关的硬性拦截逻辑风控是保命的东西必须写在代码里不能交给模型判断。我的风控网关包含以下检查def check_risk(decision, account, position): 风控检查返回是否通过 checks [] # 1. 单笔亏损上限 if decision[action] in [buy, sell]: entry decision.get(price, 0) stop decision.get(stop_loss, 0) if entry and stop: risk_per_lot abs(entry - stop) * 10 # 螺纹钢10吨/手 max_risk account[total] * 0.02 # 总资金2% max_lots int(max_risk / risk_per_lot) if decision.get(lots, 1) max_lots: checks.append(f单笔风险超限建议手数{max_lots}) # 2. 总仓位上限 current_exposure position[market_value] new_exposure decision.get(lots, 0) * decision.get(price, 0) * 10 if (current_exposure new_exposure) account[total] * 0.5: checks.append(总仓位超过50%上限) # 3. 日内亏损熔断 if account[today_pnl] -account[total] * 0.05: checks.append(日内亏损达5%触发熔断) # 4. 频繁交易限制 recent_trades count_recent_trades(minutes30) if recent_trades 3: checks.append(30分钟内交易超过3次限制开仓) return { passed: len(checks) 0, reason: ; .join(checks) if checks else 通过 }注意风控参数不是拍脑袋定的。2%的单笔风险上限来自凯利公式的保守估计50%的总仓位上限是为了留足保证金应对波动5%的日内熔断是防止情绪化连续交易。这些数字你可以根据自己的风险偏好调整但一定要有。3.5 回测验证与参数调优在实盘之前必须做回测。我的回测框架是自己写的因为要模拟Agent的决策过程现成的回测框架不太适配。回测的核心逻辑是用历史数据逐根K线喂给Agent记录每次决策和结果最后统计胜率、盈亏比、最大回撤等指标。def backtest(symbol, start_date, end_date, initial_capital100000): 回测主函数 df load_historical_data(symbol, start_date, end_date) capital initial_capital position 0 trades [] for i in range(50, len(df)): # 前50根用于计算指标 window df.iloc[:i1] indicators calc_all_indicators(window) # 模拟Agent决策 decision simulate_agent_decision(window, indicators, position, capital) if decision[action] buy and position 0: position decision[lots] entry_price window[close].iloc[-1] trades.append({type: buy, price: entry_price, time: window[timestamp].iloc[-1]}) elif decision[action] sell and position 0: exit_price window[close].iloc[-1] pnl (exit_price - entry_price) * position * 10 capital pnl trades.append({type: sell, price: exit_price, pnl: pnl, time: window[timestamp].iloc[-1]}) position 0 return analyze_trades(trades, initial_capital)回测结果的分析指标指标计算方式目标值胜率盈利交易数/总交易数45%盈亏比平均盈利/平均亏损1.5最大回撤峰值到谷底的最大跌幅15%夏普比率超额收益/收益标准差1.0交易频率日均交易次数1-3次我第一版回测跑出来胜率只有38%盈亏比1.2最大回撤22%。问题出在Agent太容易被短期波动触发交易。后来在Prompt里加了“如果价格在MA20附近震荡且ATR低于均值建议观望”这条规则胜率提升到47%回撤降到13%。4. 常见问题与排查技巧实录4.1 模型输出格式不稳定的处理这是最常见的问题。你要求输出JSON模型有时候会加一段解释文字在前面有时候字段名拼错有时候confidence给个“高”而不是数字。我的处理方案是三层防护第一层Prompt约束。在Prompt末尾加一句“只输出JSON不要有任何其他文字”。这句话能解决80%的格式问题。第二层正则提取。用正则表达式从输出中提取JSON部分忽略前后的多余文字。第三层Schema校验。用Pydantic定义输出结构校验失败就重试重试两次还失败就放弃本次决策。from pydantic import BaseModel, validator import re, json class Decision(BaseModel): action: str confidence: float reason: str stop_loss: float 0 take_profit: float 0 validator(action) def action_must_be_valid(cls, v): if v not in [buy, sell, hold]: raise ValueError(fInvalid action: {v}) return v validator(confidence) def confidence_range(cls, v): if not 0 v 1: raise ValueError(fConfidence out of range: {v}) return v def parse_decision(response_text): # 尝试直接解析 try: return Decision(**json.loads(response_text)) except: pass # 正则提取JSON match re.search(r\{.*\}, response_text, re.DOTALL) if match: try: return Decision(**json.loads(match.group())) except Exception as e: logger.warning(f解析失败: {e}) return None4.2 行情延迟导致决策滞后的应对期货行情变化快如果数据延迟超过30秒决策基本就失效了。我遇到过几次因为API响应慢导致错过止损点的情况。应对措施本地缓存最新价格每次获取行情后把最新价格写入内存缓存。决策时先检查缓存数据的时间戳如果超过60秒直接跳过本次决策。设置API超时所有行情请求设置5秒超时超时后使用上一次的有效数据并在Prompt中标注“数据可能延迟”。关键时段降频开盘前5分钟和收盘前5分钟行情波动剧烈我把决策频率从5分钟调到1分钟但只做平仓决策不开新仓。4.3 连续亏损后的策略调整Agent连续亏损是必然会遇到的。我的处理不是简单停机而是分级响应连续亏损次数响应措施2次降低仓位至正常的50%3次暂停开仓只允许平仓4次全面暂停发送告警等待人工介入5次自动切换到保守模式只做趋势跟随这个分级机制写在执行层不依赖模型判断。触发后Agent的决策权限被限制直到人工确认恢复。4.4 常见问题速查表问题现象可能原因排查步骤解决方案Agent不输出决策Prompt过长超token限制检查Prompt长度精简上下文只保留关键信息决策频繁反转短期记忆干扰查看最近决策记录降低短期记忆权重增加趋势确认下单失败保证金不足或非交易时段检查账户和交易时间增加保证金检查非交易时段跳过回测与实盘差异大滑点和手续费未计入对比回测和实盘成交价回测中加入滑点模型和手续费模型响应慢网络或模型负载测试API延迟设置超时准备备用模型记忆检索不准相似度阈值不当检查检索结果调整阈值增加品种过滤4.5 实盘上线前的检查清单在把Agent接入实盘之前我列了一个检查清单每次上线新版本都过一遍模拟盘连续运行至少5个交易日无异常中断回测最大回撤在可接受范围内风控网关的所有检查项都经过单元测试日志记录完整能追溯每一次决策的输入和输出告警机制正常异常时能及时通知手动干预接口可用能随时暂停Agent资金账户设置了独立的子账户与手动交易隔离实操心得我第一次实盘上线时忘了设置子账户Agent和我的手动仓位混在一起导致风控计算总仓位时把手动仓位也算进去了结果Agent一直觉得仓位过重不敢开仓。排查了半天才发现这个问题。所以隔离账户这件事一定要在第一天就做好。5. 后续迭代方向与扩展思路5.1 多品种扩展的注意事项目前我只跑了螺纹钢一个品种后续想扩展到铁矿石、焦炭等黑色系品种。扩展时需要注意品种间的相关性螺纹钢和铁矿石高度相关如果同时做多两个品种实际风险敞口是叠加的。风控模块需要加入相关性检查。交易时段的差异不同品种的夜盘收盘时间不同调度器需要按品种配置。Prompt的品种适配每个品种的基本面逻辑不同Prompt模板需要参数化不能一套模板打天下。5.2 从决策辅助到半自动执行现在的模式是Agent给信号我手动确认。下一步想做成半自动Agent直接下单但每笔订单有30秒的缓冲期我可以在缓冲期内取消。这个模式的技术难点在于订单状态管理。需要维护一个“待确认订单”队列缓冲期结束后自动提交。同时要处理撤单、部分成交等边界情况。5.3 模型微调的可能性通用大模型在期货领域的知识是有限的。我考虑过用历史决策数据做微调让模型更懂期货交易的语境。但微调需要大量标注数据而且微调后的模型可能失去通用推理能力。目前的替代方案是RAG把期货交易的专业知识、历史案例、品种手册做成知识库决策时检索相关片段注入Prompt。这样既保留了通用推理能力又补充了领域知识。5.4 风险控制的持续优化风控不是一次写完就完事的。市场结构变化时风控参数也需要调整。我计划加入一个自适应风控模块根据近期波动率自动调整仓位上限和止损距离。比如当ATR连续5天高于历史均值时自动把单笔风险上限从2%降到1.5%。这个逻辑不复杂但需要仔细回测验证避免过度拟合。这个项目我还在持续迭代中目前实盘跑了三个月整体收益曲线还算稳定。最大的体会是Agent不是万能的它只是一个工具核心还是你对市场的理解和风控的执行力。模型可以帮你处理信息、生成信号但最终为每一笔交易负责的还是你自己。