实际开发 AI 交易代理时最容易被低估的不是模型效果而是“边界控制”。AI agent 可以在用户设定的限制内执行交易从 paper trading 开始验证这个思路的核心在于先保证代理不会乱来再讨论它能赚多少钱。所谓 paper trading就是模拟盘交易代理按照真实市场数据生成订单但资金是虚拟的成交也不会进入真实账户。下面从一个最小可运行的 AI 交易代理出发讲清楚限制limits怎么设计、模拟交易怎么实现、限制未生效时怎么排查以及从模拟盘切到实盘前必须补充哪些保障。这类项目适合正在做量化交易、Agent 应用或金融科技原型的开发者也适合想用 LLM 构建交易决策系统的人。读者只需要掌握 Python 基础能看懂 JSON 和命令行就可以照着本文搭出一套带风控约束的模拟交易代理。文中会用到少量外部依赖但核心逻辑不依赖特定券商或交易接口因此可以把它当作一个独立的工程原型来学习。1. 为什么交易代理必须内置“限制层”1.1 AI 交易代理解决什么问题传统量化交易系统依赖人为编写的策略规则比如“均线金叉买入”“持仓超过 10% 就减仓”。AI 交易代理的不同点在于它把决策生成交给大语言模型模型读取行情、新闻摘要、账户持仓然后输出“买入、卖出或持有”的交易意图。这样的代理可以把非结构化信息纳入交易决策例如财报电话会议的语气、行业新闻的措辞、社区情绪的波动而这些信息用传统指标很难表达。但 LLM 不是交易系统。它不会主动检查账户里有多少现金不会知道自己上一笔交易是否已经触发日亏损限制也不会理解“我可以买入但你的单笔仓位不能超过总资产的 10%”。AI 交易代理的真正工程难点是把模型输出的意图转化为一个可执行、可审计、可拦截的交易指令。这个过程必须由一个独立的限制层完成。1.2 没有限制的代理会产生什么风险如果让 LLM 直接调用下单接口风险是链条式的单笔仓位过大模型看到强烈上涨信号直接生成一个买入全仓的订单。交易频率失控模型在快速行情下反复交易手续费和滑点快速侵蚀本金。单一标的高度集中模型反复看好同一只股票持仓占比不断上升。单日亏损放大当天已经亏损 5%模型为了“摊平成本”继续买入结果越亏越多。交易品种超出范围模型收到错误数据或被提示词误导去交易未授权的资产。指令格式异常模型返回的不是合法 JSON程序解析失败后可能误判为“不交易”也可能抛出未处理异常。限制层的本质是一个白名单机制。代理可以提出任何交易意图但只有通过风控校验的意图才能进入执行层。Limit 不是对模型能力的限制而是对风险的显式表达。1.3 paper trading 在整个链条里的位置Paper trading 是代理的试运行场。数据是真实的行情是真实的模型决策是真实的但资金是虚拟的。在这个环境里可以验证四类问题代理能不能稳定生成结构化的交易信号而不是偶尔输出一段解释文字。限制层能不能在模型意图越界时准确拦截并记录原因。模拟撮合逻辑是否正确处理了手续费、滑点和现金变动。模型返回格式变化、数据源延迟、风险拦截触发后系统是否能继续运行。从 paper trading 开始本质上是把“模型能力问题”和“系统工程问题”分开处理。模型能力可以慢慢调系统稳定性必须先过关。2. 最小可运行的 AI 交易代理设计2.1 技术选型下面示例使用 Python主要考虑是金融数据处理生态成熟。建议使用 Python 3.10 及以上版本。依赖模块包括模块用途yfinance获取模拟行情数据也可替换为自有数据源openai调用 OpenAI 兼容的 LLM 接口生成交易意图PyYAML读取交易限制配置pydantic校验模型返回的交易意图结构pandas / numpy计算持仓市值、收益和回撤如果不想依赖外部行情接口可以先写一个 MockDataFeed返回固定的价格数据。这样整个流程可以在离线环境跑通。下面的示例默认使用 yfinance但在代码中保留了替换入口。2.2 项目结构建议按模块拆分避免把风控逻辑和模型调用写在一起。目录结构如下paper-trading-agent/ ├── config.yaml ├── requirements.txt ├── main.py ├── agent.py ├── risk_manager.py ├── paper_broker.py ├── data_feed.py ├── models.py └── logs/各文件职责如下main.py运行主循环串联行情、决策、风控、执行。agent.py调用 LLM 生成 TradeIntent。risk_manager.py检查所有限制返回“放行或拦截”。paper_broker.py模拟成交、扣减现金、更新持仓。data_feed.py封装行情获取逻辑。models.py定义 TradeIntent、Portfolio 等核心数据模型。logs/存放交易日志和拒绝日志。2.3 依赖安装requirements.txt 内容pydantic2.0.0 openai1.0.0 pyyaml6.0 pandas2.0.0 numpy1.24.0 yfinance0.2.0安装命令pip install -r requirements.txt这里要注意版本号在不同环境中可能有兼容差异。落地前先确认当前依赖版本之间的兼容性尤其是 pydantic 2.x 和 openai SDK 的版本配合。2.4 定义限制配置用 config.yaml 集中管理限制参数。这样后续如果要调整风控规则不需要改 Python 代码。account: initial_cash: 100000 max_positions: 5 max_single_position_pct: 0.2 max_single_order_pct: 0.1 max_daily_loss_pct: 0.03 max_drawdown_pct: 0.1 max_trades_per_day: 20 allowed_symbols: - AAPL - MSFT - BTC-USD banned_symbols: [] execution: slippage_bps: 5 commission_pct: 0.001 min_order_value: 100 allow_short: false参数解释会在后面章节展开。这里先把配置集中写明后续所有模块都从该文件读取。2.5 获取市场数据DataFeed 模块负责给 agent 提供最新行情快照。下面是一个基于 yfinance 的最小实现import yfinance as yf class DataFeed: def __init__(self, symbols, lookback_days5): self.symbols symbols self.lookback_days lookback_days def fetch(self, symbol: str) - dict: ticker yf.Ticker(symbol) hist ticker.history( periodf{self.lookback_days}d, interval1d ) if hist.empty: raise ValueError(fno data for {symbol}) last hist.iloc[-1] return { symbol: symbol, price: float(last[Close]), volume: int(last[Volume]), timestamp: str(last.name), history: hist.tail(5).to_dict(orientrecords), }关键点是无论数据源是什么fetch 方法都返回一个统一结构的 dict。后面 agent 不关心数据来自 yfinance 还是自建行情服务这样替换成本低。如果想要离线运行可以增加一个 MockDataFeedclass MockDataFeed: def __init__(self, initial_prices): self.prices initial_prices def fetch(self, symbol: str) - dict: price self.prices[symbol] return { symbol: symbol, price: price, volume: 1000, timestamp: 2025-01-01, history: [], }2.6 让 LLM 生成交易意图先定义核心数据模型。这里使用 pydantic 的 BaseModel让字段校验更严格from enum import Enum from pydantic import BaseModel from typing import Optional class Action(str, Enum): BUY buy SELL sell HOLD hold class TradeIntent(BaseModel): action: Action symbol: str quantity: int reason: str confidence: float 0.5agent.py 的核心功能是调用 LLM并把模型输出解析为 TradeIntent。示例import json from openai import OpenAI SYSTEM_PROMPT 你是一个交易助手。根据用户提供的行情和账户数据输出JSON格式的交易意图。 JSON必须包含四个字段 - action: buy / sell / hold - symbol: 股票代码 - quantity: 正整数必须大于0 - reason: 简洁的决策理由 - confidence: 0到1之间的小数 只输出JSON不要输出任何解释。 class Agent: def __init__(self): self.client OpenAI() def decide(self, market_data: dict, portfolio: dict) - TradeIntent: user_content json.dumps( {market: market_data, portfolio: portfolio}, ensure_asciiFalse, ) resp self.client.chat.completions.create( modelgpt-4o-mini, response_format{type: json_object}, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_content}, ], ) content resp.choices[0].message.content raw json.loads(content) return TradeIntent(**raw)这里有几个细节需要注意response_format 强制模型返回 JSON但并不能保证结构完全正确仍需 pydantic 校验。quantity 必须设置为正整数否则后续风控和撮合会出现脏数据。不要把 API Key 写死在代码里。通过环境变量注入例如OPENAI_API_KEY。模型名要根据当前可用的接口调整不同服务的模型名可能不同。2.7 使用模拟撮合器执行订单paper_broker.py 负责模拟成交。它维护现金、持仓、日志并根据配置计算手续费和滑点。class PaperBroker: def __init__(self, config): account config[account] execution config[execution] self.cash float(account[initial_cash]) self.positions {} self.commission_pct execution[commission_pct] self.slippage_bps execution[slippage_bps] self.min_order_value execution[min_order_value] self.allow_short execution[allow_short] self.trades [] def execute(self, intent: TradeIntent, market_price: float) - dict: if intent.action hold: return {status: hold, reason: agent decided to hold} slip market_price * self.slippage_bps / 10000 if intent.action buy: exec_price market_price slip else: exec_price market_price - slip value intent.quantity * exec_price commission max(value * self.commission_pct, 0.01) if intent.action buy: if value commission self.cash: raise ValueError(insufficient cash) self.cash - value commission self.positions[intent.symbol] ( self.positions.get(intent.symbol, 0) intent.quantity ) else: held self.positions.get(intent.symbol, 0) if held intent.quantity: raise ValueError(insufficient position to sell) if not self.allow_short and intent.quantity held: raise ValueError(short selling is not allowed) self.cash value - commission self.positions[intent.symbol] held - intent.quantity trade { symbol: intent.symbol, action: intent.action.value, quantity: intent.quantity, exec_price: exec_price, commission: commission, value: value, } self.trades.append(trade) return {status: filled, trade: trade}模拟撮合器的关键不是追求精确而是保持一致。手续费和滑点不能一次算一次不算否则资金曲线会失真。3. 限制机制的四个关键层级限制层必须独立于 agent不能由 agent 自己判断“我该不该买”。下面按账户层、单笔交易层、日亏损层和拒绝策略四个层级展开。3.1 账户层限制账户层限制关注的是整体组合状态。常检查项目包括总持仓数量是否超过 max_positions。单一标的市值占总资产的比例是否超过 max_single_position_pct。目标标的是否在 allowed_symbols 之外。是否被 banned_symbols 禁止。如果是 sell是否持有足够数量。一个最小的账户层检查实现class RiskManager: def __init__(self, config): self.account config[account] self.max_positions self.account[max_positions] self.max_single_position_pct self.account[max_single_position_pct] self.allowed_symbols set(self.account[allowed_symbols]) self.banned_symbols set(self.account[banned_symbols]) def check_account_limits(self, intent, portfolio) - list[str]: errors [] if intent.symbol in self.banned_symbols: errors.append(symbol is banned) if intent.symbol not in self.allowed_symbols: errors.append(symbol is not in allowed list) if intent.action buy: if ( intent.symbol not in portfolio.positions and len(portfolio.positions) self.max_positions ): errors.append(max_positions exceeded) return errors为什么允许 already-held symbol 在持仓数满时买入因为加仓不增加新标的不会进一步分散风险可以在后续单笔限制中控制。这是工程取舍取决于策略偏好。3.2 单笔交易限制单笔交易限制约束单个订单的大小。核心参数是 max_single_order_pct表示单笔订单价值不能超过当前总资产的比例。示例def check_order_limits(self, intent, portfolio, market_price) - list[str]: errors [] order_value intent.quantity * market_price total_equity portfolio.total_equity(self.cash, self.market_prices) if order_value self.min_order_value: errors.append(order value below minimum) if order_value total_equity * self.max_single_order_pct: errors.append(single order size exceeds max_single_order_pct) if intent.quantity 0: errors.append(quantity must be positive) return errors单笔限制的作用是防止模型一次下太大的单。即使模型非常坚信某只股票会涨单笔订单也只能占很小一部分资金。参数速查配置项含义常见默认调大影响调小影响max_single_order_pct单笔订单价值上限0.110%单笔冲击成本高风险集中交易碎片化手续费占比高min_order_value最小订单价值100过滤小额测试单小额订单增多max_single_position_pct单一标的持仓上限0.2组合集中度升高分散度增加潜在收益变低3.3 日亏损限制和风控熔断单笔限制解决单次风险日亏损限制解决连续风险。当日累计亏损超过 max_daily_loss_pct 后系统应进入“只卖不买”的熔断状态。实现方式class RiskManager: def __init__(self, config): self.max_daily_loss_pct self.account[max_daily_loss_pct] self.circuit_open False def update_daily_pnl(self, daily_pnl: float, total_equity: float): loss_ratio -daily_pnl / total_equity if loss_ratio self.max_daily_loss_pct: self.circuit_open True def check_circuit(self, intent) - list[str]: if self.circuit_open and intent.action buy: return [circuit breaker open: buy blocked] return []为什么熔断后只拦截 buy不拦截 sell因为在亏损场景下卖出通常用于降低风险而买入会继续增加风险暴露。这个策略不一定适合所有市场但适合作为默认选项。3.4 拒绝 AI 意图的策略风险管理器不直接下单它只负责返回错误列表。主流程根据错误列表决定是否执行订单def evaluate(self, intent, portfolio, market_price) - tuple[bool, list[str]]: errors [] errors.extend(self.check_account_limits(intent, portfolio)) errors.extend(self.check_order_limits(intent, portfolio, market_price)) errors.extend(self.check_circuit(intent)) if errors: return False, errors return True, []被拒绝的意图必须写入日志不能静默丢弃。原因也要完整记录否则后续无法判断模型是否在频繁越界。日志字段至少包括timestampintent actionsymbolquantitymarket pricereject reasonsportfolio equity4. 运行、验证与绩效评估4.1 启动模拟交易main.py 负责把数据、agent、风险管理器、paper broker 串起来。一个最小循环import yaml def load_config(path): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def run_once(agent, risk_manager, broker, data_feed, symbol): market data_feed.fetch(symbol) portfolio broker.get_portfolio(market) intent agent.decide(market, portfolio) ok, errors risk_manager.evaluate(intent, portfolio, market[price]) if not ok: log_rejection(intent, errors) return {status: rejected, errors: errors} result broker.execute(intent, market[price]) risk_manager.update_daily_pnl(portfolio.calc_daily_pnl(), portfolio.total_equity()) return result if __name__ __main__: config load_config(config.yaml) agent Agent() risk_manager RiskManager(config) broker PaperBroker(config) data_feed DataFeed(config[account][allowed_symbols]) print(run_once(agent, risk_manager, broker, data_feed, AAPL))run_once 每次只处理一个标的和一个决策。真实系统里会循环多个标的并加入时间控制避免过密调用行情接口。4.2 观察限制是否生效启动后不要只看“有没有成交”。更重要的是验证限制层真的会拦截越界意图。推荐构造三个测试场景把 max_trades_per_day 设为 1运行两次第二次应向风险管理器返回拒绝原因。把 max_single_order_pct 设为 0.01让 agent 生成一个较大数量买入单检查订单是否被拦截。手动把 daily_pnl 设置为超过 max_daily_loss_pct再让 agent 生成 buy 意图检查熔断是否触发。测试时可以在风险管理器里临时加入日志输出print(f[risk] {intent.symbol} {intent.action.value} - {errors})如果拒绝日志出现说明限制链路生效。如果任何越界订单都通过了优先检查是否真的调用了 risk_manager.evaluate而不是只调用了 agent.decide。4.3 简单绩效统计模拟交易运行一段时间后统计收益和回撤。用 pandas 可以快速实现import pandas as pd def compute_metrics(equity_series): returns equity_series.pct_change().dropna() total_return equity_series.iloc[-1] / equity_series.iloc[0] - 1 max_drawdown (equity_series / equity_series.cummax() - 1).min() return { total_return: round(total_return, 4), max_drawdown: round(max_drawdown, 4), volatility: round(returns.std() * (252 ** 0.5), 4), }绩效指标不能只看收益率。一个回撤 50% 后涨回 100% 的账户风险承受难度远高于低回撤策略。所以 Total Return、Max Drawdown、Sharpe 三类指标要一起看。4.4 限制配置速查表配置项默认值检查层级拦截表现建议max_positions5账户层拒绝买入新标的结合策略集中度设置max_single_position_pct0.2账户层拒绝加仓超限标的过高会集中风险max_single_order_pct0.1单笔层拒绝订单价值超限单笔不宜过大max_daily_loss_pct0.03熔断层拒绝所有 buy超过后只允许降仓max_trades_per_day20频率层拒绝当天多余交易防止高频磨损allow_shortfalse账户层拒绝卖出开空多数学习环境建议关闭slippage_bps5执行层影响模拟成交价实盘前应参考真实滑点commission_pct0.001执行层影响现金流按券商实际费率调整5. 常见问题排查5.1 市场数据源连接失败现象程序运行时出现HTTPError、ConnectionError或 yfinance 的Exception: No data found。可能原因网络不可达或外部数据接口临时不可用。标的代码不在平台支持范围内。请求频率过高被限流。本地时区或日期处理导致查询区间为空。检查方式curl -I https://query1.finance.yahoo.com再运行一段最小脚本import yfinance as yf print(yf.Ticker(AAPL).history(period5d))处理建议增加重试逻辑指数退避。离线开发时切换到 MockDataFeed。给 fetch 方法增加超时参数。5.2 明明设置了限制订单还是超限现象日志里没有拒绝记录但订单明显超过max_single_order_pct。可能原因主流程没有调用 risk_manager.evaluate。调用顺序错误先执行后校验。修改 config.yaml 后没有重新加载。限制检查的是“订单名义价值”但实际成交价包含滑点和手续费导致超限发生。检查方式在 execute 前打印 intent 和 current equity。确认 risk_manager 初始化时使用的是最新配置。检查是否有多个 RiskManager 实例部分实例使用旧配置。处理建议在主流程统一入口处调用 evaluate。将配置对象的加载做成单例避免重复加载。如果担心滑点导致超限设置一个 buffer例如只允许用到 90% 的限制额度。5.3 AI 策略频繁被限制拦截现象日志中 rejection 数量远大于 filled 数量。可能原因限制过严模型没有足够空间表达策略。模型没有接收到账户余量信息盲目生成大额买单。prompt 中的输出格式约束不够严格。模型反复尝试同一只标的但该标的已被 banned_symbols 排除。处理建议把当前现金、持仓占比、剩余可买数量注入 prompt。在 prompt 中明确说明“只能买入你实际能支付的数量”。如果模型频繁给出同一越界建议要关注是不是配置本身不合理。不要为了让模型“少被拒绝”而放宽真实风险限制。5.4 模拟撮合和真实行情差异现象运行多次后组合资金变化不符合预期。常见原因有手续费重复计算。滑点方向写反买入时用了低价卖出时用了高价。卖出数量超过持仓时没有拦截。现金更新时忽略了 commission。回测中多次使用同一市场快照导致每次成交价都一样。检查方式打印每一笔 trade 的 exec_price、commission、cash。写一个最小测试购买 1 股再卖出 1 股检查现金是否回到初始值减去两次手续费。处理建议模拟撮合器增加单元测试。所有现金变化都通过cash - value commission或cash value - commission不要散落多处计算。日报表中增加“手续费总额”和“滑点总额”两个字段。6. 从 paper trading 到实盘生产环境补齐什么6.1 模拟盘与实盘的关键差异维度Paper Trading实盘资金虚拟资金真实资金订单路由本地模拟成交经券商 API 发往交易所成交价格按收盘价或延迟数据模拟受盘口深度、流动性和滑点影响风控内存中检查需要独立风控服务和人工干预通道审计本地日志即可需要不可篡改的审计日志合规无监管要求需要遵守券商和市场规则故障影响不影响真实资产可能造成实际损失模拟盘跑通并不代表实盘可用。最大的区别在于延迟和不确定性。模拟撮合假设价格是已知的实盘中订单可能部分成交、延迟成交或者被拒绝。6.2 生产环境必须有审计和告警生产环境建议增加以下机制所有 agent 意图和 reject reason 写入 JSONL 日志保留至少 180 天。所有 risk_manager 配置变更都要版本化最好通过 Git 管理。当日亏损接近限制值时通过邮件或 Webhook 告警。连续多次拒绝或连续执行失败时自动暂停策略。风险管理器必须独立于 agent 进程避免 agent 崩溃导致风控失效。推荐把配置做成外部化配置中心而不是写死在代码包中。这样生产环境调整限制时可以回滚也方便审计。6.3 发布前检查清单可以把这个清单打印出来实盘前逐项确认[ ] 数据源稳定性行情接口是否有重试、降级、超时机制。[ ] LLM 输出校验是否强制 JSON 输出是否校验必填字段。[ ] 风险限制所有限制参数是否与实盘账户规模匹配。[ ] 订单金额限制是否校验最小订单、最大订单和下单频率。[ ] 熔断机制日亏损超过阈值后是否能自动停止买入。[ ] 错误处理API 调用失败、数据缺失、模型解析失败是否有兜底。[ ] 日志审计每笔订单是否有完整记录。[ ] 权限控制实盘 API Key 是否经过加密存储是否最小权限。[ ] 回滚方案系统异常时能否一键停止所有订单。[ ] 监控告警是否有关键指标和异常告警。7. 扩展方向当前示例只完成了最小闭环。下一步可以从几个方向扩展第一把 LLM 决策和传统指标结合。模型不只读行情快照还可以读取均线、RSI、MACD 等指标把技术面信号和语义信号一起输入 prompt。第二引入多智能体协同。一个 agent 负责分析情绪一个 agent 负责生成交易计划一个 agent 负责复核风险限制仍然在最后统一执行。第三加入回测框架。让交易代理在历史数据上分批运行统计胜率、盈亏比、最大回撤再根据结果调整 prompt 和限制参数。第四动态限制。根据市场波动率自动调整单笔仓位限制例如波动率上升时自动缩小 max_single_order_pct。动态限制比固定值更接近实盘需求但实现时要注意逻辑可解释性。第五异常检测。监控 LLM 输出置信度、拒绝率、连续亏损次数当这些指标偏离正常范围时自动降级策略。在实际项目里最值得投入的不是让模型“更聪明”而是让限制层“更可靠”。一个能稳定拦截越界意图的 AI 交易代理即使策略平庸也至少不会因为系统缺陷导致灾难性损失。新手可以先从 paper trading 环境反复实验等日志、监控、回滚和审计都完善后再考虑对接真实券商接口。 AI交易代理的边界控制：从Paper Trading到限制作战
实际开发 AI 交易代理时最容易被低估的不是模型效果而是“边界控制”。AI agent 可以在用户设定的限制内执行交易从 paper trading 开始验证这个思路的核心在于先保证代理不会乱来再讨论它能赚多少钱。所谓 paper trading就是模拟盘交易代理按照真实市场数据生成订单但资金是虚拟的成交也不会进入真实账户。下面从一个最小可运行的 AI 交易代理出发讲清楚限制limits怎么设计、模拟交易怎么实现、限制未生效时怎么排查以及从模拟盘切到实盘前必须补充哪些保障。这类项目适合正在做量化交易、Agent 应用或金融科技原型的开发者也适合想用 LLM 构建交易决策系统的人。读者只需要掌握 Python 基础能看懂 JSON 和命令行就可以照着本文搭出一套带风控约束的模拟交易代理。文中会用到少量外部依赖但核心逻辑不依赖特定券商或交易接口因此可以把它当作一个独立的工程原型来学习。1. 为什么交易代理必须内置“限制层”1.1 AI 交易代理解决什么问题传统量化交易系统依赖人为编写的策略规则比如“均线金叉买入”“持仓超过 10% 就减仓”。AI 交易代理的不同点在于它把决策生成交给大语言模型模型读取行情、新闻摘要、账户持仓然后输出“买入、卖出或持有”的交易意图。这样的代理可以把非结构化信息纳入交易决策例如财报电话会议的语气、行业新闻的措辞、社区情绪的波动而这些信息用传统指标很难表达。但 LLM 不是交易系统。它不会主动检查账户里有多少现金不会知道自己上一笔交易是否已经触发日亏损限制也不会理解“我可以买入但你的单笔仓位不能超过总资产的 10%”。AI 交易代理的真正工程难点是把模型输出的意图转化为一个可执行、可审计、可拦截的交易指令。这个过程必须由一个独立的限制层完成。1.2 没有限制的代理会产生什么风险如果让 LLM 直接调用下单接口风险是链条式的单笔仓位过大模型看到强烈上涨信号直接生成一个买入全仓的订单。交易频率失控模型在快速行情下反复交易手续费和滑点快速侵蚀本金。单一标的高度集中模型反复看好同一只股票持仓占比不断上升。单日亏损放大当天已经亏损 5%模型为了“摊平成本”继续买入结果越亏越多。交易品种超出范围模型收到错误数据或被提示词误导去交易未授权的资产。指令格式异常模型返回的不是合法 JSON程序解析失败后可能误判为“不交易”也可能抛出未处理异常。限制层的本质是一个白名单机制。代理可以提出任何交易意图但只有通过风控校验的意图才能进入执行层。Limit 不是对模型能力的限制而是对风险的显式表达。1.3 paper trading 在整个链条里的位置Paper trading 是代理的试运行场。数据是真实的行情是真实的模型决策是真实的但资金是虚拟的。在这个环境里可以验证四类问题代理能不能稳定生成结构化的交易信号而不是偶尔输出一段解释文字。限制层能不能在模型意图越界时准确拦截并记录原因。模拟撮合逻辑是否正确处理了手续费、滑点和现金变动。模型返回格式变化、数据源延迟、风险拦截触发后系统是否能继续运行。从 paper trading 开始本质上是把“模型能力问题”和“系统工程问题”分开处理。模型能力可以慢慢调系统稳定性必须先过关。2. 最小可运行的 AI 交易代理设计2.1 技术选型下面示例使用 Python主要考虑是金融数据处理生态成熟。建议使用 Python 3.10 及以上版本。依赖模块包括模块用途yfinance获取模拟行情数据也可替换为自有数据源openai调用 OpenAI 兼容的 LLM 接口生成交易意图PyYAML读取交易限制配置pydantic校验模型返回的交易意图结构pandas / numpy计算持仓市值、收益和回撤如果不想依赖外部行情接口可以先写一个 MockDataFeed返回固定的价格数据。这样整个流程可以在离线环境跑通。下面的示例默认使用 yfinance但在代码中保留了替换入口。2.2 项目结构建议按模块拆分避免把风控逻辑和模型调用写在一起。目录结构如下paper-trading-agent/ ├── config.yaml ├── requirements.txt ├── main.py ├── agent.py ├── risk_manager.py ├── paper_broker.py ├── data_feed.py ├── models.py └── logs/各文件职责如下main.py运行主循环串联行情、决策、风控、执行。agent.py调用 LLM 生成 TradeIntent。risk_manager.py检查所有限制返回“放行或拦截”。paper_broker.py模拟成交、扣减现金、更新持仓。data_feed.py封装行情获取逻辑。models.py定义 TradeIntent、Portfolio 等核心数据模型。logs/存放交易日志和拒绝日志。2.3 依赖安装requirements.txt 内容pydantic2.0.0 openai1.0.0 pyyaml6.0 pandas2.0.0 numpy1.24.0 yfinance0.2.0安装命令pip install -r requirements.txt这里要注意版本号在不同环境中可能有兼容差异。落地前先确认当前依赖版本之间的兼容性尤其是 pydantic 2.x 和 openai SDK 的版本配合。2.4 定义限制配置用 config.yaml 集中管理限制参数。这样后续如果要调整风控规则不需要改 Python 代码。account: initial_cash: 100000 max_positions: 5 max_single_position_pct: 0.2 max_single_order_pct: 0.1 max_daily_loss_pct: 0.03 max_drawdown_pct: 0.1 max_trades_per_day: 20 allowed_symbols: - AAPL - MSFT - BTC-USD banned_symbols: [] execution: slippage_bps: 5 commission_pct: 0.001 min_order_value: 100 allow_short: false参数解释会在后面章节展开。这里先把配置集中写明后续所有模块都从该文件读取。2.5 获取市场数据DataFeed 模块负责给 agent 提供最新行情快照。下面是一个基于 yfinance 的最小实现import yfinance as yf class DataFeed: def __init__(self, symbols, lookback_days5): self.symbols symbols self.lookback_days lookback_days def fetch(self, symbol: str) - dict: ticker yf.Ticker(symbol) hist ticker.history( periodf{self.lookback_days}d, interval1d ) if hist.empty: raise ValueError(fno data for {symbol}) last hist.iloc[-1] return { symbol: symbol, price: float(last[Close]), volume: int(last[Volume]), timestamp: str(last.name), history: hist.tail(5).to_dict(orientrecords), }关键点是无论数据源是什么fetch 方法都返回一个统一结构的 dict。后面 agent 不关心数据来自 yfinance 还是自建行情服务这样替换成本低。如果想要离线运行可以增加一个 MockDataFeedclass MockDataFeed: def __init__(self, initial_prices): self.prices initial_prices def fetch(self, symbol: str) - dict: price self.prices[symbol] return { symbol: symbol, price: price, volume: 1000, timestamp: 2025-01-01, history: [], }2.6 让 LLM 生成交易意图先定义核心数据模型。这里使用 pydantic 的 BaseModel让字段校验更严格from enum import Enum from pydantic import BaseModel from typing import Optional class Action(str, Enum): BUY buy SELL sell HOLD hold class TradeIntent(BaseModel): action: Action symbol: str quantity: int reason: str confidence: float 0.5agent.py 的核心功能是调用 LLM并把模型输出解析为 TradeIntent。示例import json from openai import OpenAI SYSTEM_PROMPT 你是一个交易助手。根据用户提供的行情和账户数据输出JSON格式的交易意图。 JSON必须包含四个字段 - action: buy / sell / hold - symbol: 股票代码 - quantity: 正整数必须大于0 - reason: 简洁的决策理由 - confidence: 0到1之间的小数 只输出JSON不要输出任何解释。 class Agent: def __init__(self): self.client OpenAI() def decide(self, market_data: dict, portfolio: dict) - TradeIntent: user_content json.dumps( {market: market_data, portfolio: portfolio}, ensure_asciiFalse, ) resp self.client.chat.completions.create( modelgpt-4o-mini, response_format{type: json_object}, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_content}, ], ) content resp.choices[0].message.content raw json.loads(content) return TradeIntent(**raw)这里有几个细节需要注意response_format 强制模型返回 JSON但并不能保证结构完全正确仍需 pydantic 校验。quantity 必须设置为正整数否则后续风控和撮合会出现脏数据。不要把 API Key 写死在代码里。通过环境变量注入例如OPENAI_API_KEY。模型名要根据当前可用的接口调整不同服务的模型名可能不同。2.7 使用模拟撮合器执行订单paper_broker.py 负责模拟成交。它维护现金、持仓、日志并根据配置计算手续费和滑点。class PaperBroker: def __init__(self, config): account config[account] execution config[execution] self.cash float(account[initial_cash]) self.positions {} self.commission_pct execution[commission_pct] self.slippage_bps execution[slippage_bps] self.min_order_value execution[min_order_value] self.allow_short execution[allow_short] self.trades [] def execute(self, intent: TradeIntent, market_price: float) - dict: if intent.action hold: return {status: hold, reason: agent decided to hold} slip market_price * self.slippage_bps / 10000 if intent.action buy: exec_price market_price slip else: exec_price market_price - slip value intent.quantity * exec_price commission max(value * self.commission_pct, 0.01) if intent.action buy: if value commission self.cash: raise ValueError(insufficient cash) self.cash - value commission self.positions[intent.symbol] ( self.positions.get(intent.symbol, 0) intent.quantity ) else: held self.positions.get(intent.symbol, 0) if held intent.quantity: raise ValueError(insufficient position to sell) if not self.allow_short and intent.quantity held: raise ValueError(short selling is not allowed) self.cash value - commission self.positions[intent.symbol] held - intent.quantity trade { symbol: intent.symbol, action: intent.action.value, quantity: intent.quantity, exec_price: exec_price, commission: commission, value: value, } self.trades.append(trade) return {status: filled, trade: trade}模拟撮合器的关键不是追求精确而是保持一致。手续费和滑点不能一次算一次不算否则资金曲线会失真。3. 限制机制的四个关键层级限制层必须独立于 agent不能由 agent 自己判断“我该不该买”。下面按账户层、单笔交易层、日亏损层和拒绝策略四个层级展开。3.1 账户层限制账户层限制关注的是整体组合状态。常检查项目包括总持仓数量是否超过 max_positions。单一标的市值占总资产的比例是否超过 max_single_position_pct。目标标的是否在 allowed_symbols 之外。是否被 banned_symbols 禁止。如果是 sell是否持有足够数量。一个最小的账户层检查实现class RiskManager: def __init__(self, config): self.account config[account] self.max_positions self.account[max_positions] self.max_single_position_pct self.account[max_single_position_pct] self.allowed_symbols set(self.account[allowed_symbols]) self.banned_symbols set(self.account[banned_symbols]) def check_account_limits(self, intent, portfolio) - list[str]: errors [] if intent.symbol in self.banned_symbols: errors.append(symbol is banned) if intent.symbol not in self.allowed_symbols: errors.append(symbol is not in allowed list) if intent.action buy: if ( intent.symbol not in portfolio.positions and len(portfolio.positions) self.max_positions ): errors.append(max_positions exceeded) return errors为什么允许 already-held symbol 在持仓数满时买入因为加仓不增加新标的不会进一步分散风险可以在后续单笔限制中控制。这是工程取舍取决于策略偏好。3.2 单笔交易限制单笔交易限制约束单个订单的大小。核心参数是 max_single_order_pct表示单笔订单价值不能超过当前总资产的比例。示例def check_order_limits(self, intent, portfolio, market_price) - list[str]: errors [] order_value intent.quantity * market_price total_equity portfolio.total_equity(self.cash, self.market_prices) if order_value self.min_order_value: errors.append(order value below minimum) if order_value total_equity * self.max_single_order_pct: errors.append(single order size exceeds max_single_order_pct) if intent.quantity 0: errors.append(quantity must be positive) return errors单笔限制的作用是防止模型一次下太大的单。即使模型非常坚信某只股票会涨单笔订单也只能占很小一部分资金。参数速查配置项含义常见默认调大影响调小影响max_single_order_pct单笔订单价值上限0.110%单笔冲击成本高风险集中交易碎片化手续费占比高min_order_value最小订单价值100过滤小额测试单小额订单增多max_single_position_pct单一标的持仓上限0.2组合集中度升高分散度增加潜在收益变低3.3 日亏损限制和风控熔断单笔限制解决单次风险日亏损限制解决连续风险。当日累计亏损超过 max_daily_loss_pct 后系统应进入“只卖不买”的熔断状态。实现方式class RiskManager: def __init__(self, config): self.max_daily_loss_pct self.account[max_daily_loss_pct] self.circuit_open False def update_daily_pnl(self, daily_pnl: float, total_equity: float): loss_ratio -daily_pnl / total_equity if loss_ratio self.max_daily_loss_pct: self.circuit_open True def check_circuit(self, intent) - list[str]: if self.circuit_open and intent.action buy: return [circuit breaker open: buy blocked] return []为什么熔断后只拦截 buy不拦截 sell因为在亏损场景下卖出通常用于降低风险而买入会继续增加风险暴露。这个策略不一定适合所有市场但适合作为默认选项。3.4 拒绝 AI 意图的策略风险管理器不直接下单它只负责返回错误列表。主流程根据错误列表决定是否执行订单def evaluate(self, intent, portfolio, market_price) - tuple[bool, list[str]]: errors [] errors.extend(self.check_account_limits(intent, portfolio)) errors.extend(self.check_order_limits(intent, portfolio, market_price)) errors.extend(self.check_circuit(intent)) if errors: return False, errors return True, []被拒绝的意图必须写入日志不能静默丢弃。原因也要完整记录否则后续无法判断模型是否在频繁越界。日志字段至少包括timestampintent actionsymbolquantitymarket pricereject reasonsportfolio equity4. 运行、验证与绩效评估4.1 启动模拟交易main.py 负责把数据、agent、风险管理器、paper broker 串起来。一个最小循环import yaml def load_config(path): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def run_once(agent, risk_manager, broker, data_feed, symbol): market data_feed.fetch(symbol) portfolio broker.get_portfolio(market) intent agent.decide(market, portfolio) ok, errors risk_manager.evaluate(intent, portfolio, market[price]) if not ok: log_rejection(intent, errors) return {status: rejected, errors: errors} result broker.execute(intent, market[price]) risk_manager.update_daily_pnl(portfolio.calc_daily_pnl(), portfolio.total_equity()) return result if __name__ __main__: config load_config(config.yaml) agent Agent() risk_manager RiskManager(config) broker PaperBroker(config) data_feed DataFeed(config[account][allowed_symbols]) print(run_once(agent, risk_manager, broker, data_feed, AAPL))run_once 每次只处理一个标的和一个决策。真实系统里会循环多个标的并加入时间控制避免过密调用行情接口。4.2 观察限制是否生效启动后不要只看“有没有成交”。更重要的是验证限制层真的会拦截越界意图。推荐构造三个测试场景把 max_trades_per_day 设为 1运行两次第二次应向风险管理器返回拒绝原因。把 max_single_order_pct 设为 0.01让 agent 生成一个较大数量买入单检查订单是否被拦截。手动把 daily_pnl 设置为超过 max_daily_loss_pct再让 agent 生成 buy 意图检查熔断是否触发。测试时可以在风险管理器里临时加入日志输出print(f[risk] {intent.symbol} {intent.action.value} - {errors})如果拒绝日志出现说明限制链路生效。如果任何越界订单都通过了优先检查是否真的调用了 risk_manager.evaluate而不是只调用了 agent.decide。4.3 简单绩效统计模拟交易运行一段时间后统计收益和回撤。用 pandas 可以快速实现import pandas as pd def compute_metrics(equity_series): returns equity_series.pct_change().dropna() total_return equity_series.iloc[-1] / equity_series.iloc[0] - 1 max_drawdown (equity_series / equity_series.cummax() - 1).min() return { total_return: round(total_return, 4), max_drawdown: round(max_drawdown, 4), volatility: round(returns.std() * (252 ** 0.5), 4), }绩效指标不能只看收益率。一个回撤 50% 后涨回 100% 的账户风险承受难度远高于低回撤策略。所以 Total Return、Max Drawdown、Sharpe 三类指标要一起看。4.4 限制配置速查表配置项默认值检查层级拦截表现建议max_positions5账户层拒绝买入新标的结合策略集中度设置max_single_position_pct0.2账户层拒绝加仓超限标的过高会集中风险max_single_order_pct0.1单笔层拒绝订单价值超限单笔不宜过大max_daily_loss_pct0.03熔断层拒绝所有 buy超过后只允许降仓max_trades_per_day20频率层拒绝当天多余交易防止高频磨损allow_shortfalse账户层拒绝卖出开空多数学习环境建议关闭slippage_bps5执行层影响模拟成交价实盘前应参考真实滑点commission_pct0.001执行层影响现金流按券商实际费率调整5. 常见问题排查5.1 市场数据源连接失败现象程序运行时出现HTTPError、ConnectionError或 yfinance 的Exception: No data found。可能原因网络不可达或外部数据接口临时不可用。标的代码不在平台支持范围内。请求频率过高被限流。本地时区或日期处理导致查询区间为空。检查方式curl -I https://query1.finance.yahoo.com再运行一段最小脚本import yfinance as yf print(yf.Ticker(AAPL).history(period5d))处理建议增加重试逻辑指数退避。离线开发时切换到 MockDataFeed。给 fetch 方法增加超时参数。5.2 明明设置了限制订单还是超限现象日志里没有拒绝记录但订单明显超过max_single_order_pct。可能原因主流程没有调用 risk_manager.evaluate。调用顺序错误先执行后校验。修改 config.yaml 后没有重新加载。限制检查的是“订单名义价值”但实际成交价包含滑点和手续费导致超限发生。检查方式在 execute 前打印 intent 和 current equity。确认 risk_manager 初始化时使用的是最新配置。检查是否有多个 RiskManager 实例部分实例使用旧配置。处理建议在主流程统一入口处调用 evaluate。将配置对象的加载做成单例避免重复加载。如果担心滑点导致超限设置一个 buffer例如只允许用到 90% 的限制额度。5.3 AI 策略频繁被限制拦截现象日志中 rejection 数量远大于 filled 数量。可能原因限制过严模型没有足够空间表达策略。模型没有接收到账户余量信息盲目生成大额买单。prompt 中的输出格式约束不够严格。模型反复尝试同一只标的但该标的已被 banned_symbols 排除。处理建议把当前现金、持仓占比、剩余可买数量注入 prompt。在 prompt 中明确说明“只能买入你实际能支付的数量”。如果模型频繁给出同一越界建议要关注是不是配置本身不合理。不要为了让模型“少被拒绝”而放宽真实风险限制。5.4 模拟撮合和真实行情差异现象运行多次后组合资金变化不符合预期。常见原因有手续费重复计算。滑点方向写反买入时用了低价卖出时用了高价。卖出数量超过持仓时没有拦截。现金更新时忽略了 commission。回测中多次使用同一市场快照导致每次成交价都一样。检查方式打印每一笔 trade 的 exec_price、commission、cash。写一个最小测试购买 1 股再卖出 1 股检查现金是否回到初始值减去两次手续费。处理建议模拟撮合器增加单元测试。所有现金变化都通过cash - value commission或cash value - commission不要散落多处计算。日报表中增加“手续费总额”和“滑点总额”两个字段。6. 从 paper trading 到实盘生产环境补齐什么6.1 模拟盘与实盘的关键差异维度Paper Trading实盘资金虚拟资金真实资金订单路由本地模拟成交经券商 API 发往交易所成交价格按收盘价或延迟数据模拟受盘口深度、流动性和滑点影响风控内存中检查需要独立风控服务和人工干预通道审计本地日志即可需要不可篡改的审计日志合规无监管要求需要遵守券商和市场规则故障影响不影响真实资产可能造成实际损失模拟盘跑通并不代表实盘可用。最大的区别在于延迟和不确定性。模拟撮合假设价格是已知的实盘中订单可能部分成交、延迟成交或者被拒绝。6.2 生产环境必须有审计和告警生产环境建议增加以下机制所有 agent 意图和 reject reason 写入 JSONL 日志保留至少 180 天。所有 risk_manager 配置变更都要版本化最好通过 Git 管理。当日亏损接近限制值时通过邮件或 Webhook 告警。连续多次拒绝或连续执行失败时自动暂停策略。风险管理器必须独立于 agent 进程避免 agent 崩溃导致风控失效。推荐把配置做成外部化配置中心而不是写死在代码包中。这样生产环境调整限制时可以回滚也方便审计。6.3 发布前检查清单可以把这个清单打印出来实盘前逐项确认[ ] 数据源稳定性行情接口是否有重试、降级、超时机制。[ ] LLM 输出校验是否强制 JSON 输出是否校验必填字段。[ ] 风险限制所有限制参数是否与实盘账户规模匹配。[ ] 订单金额限制是否校验最小订单、最大订单和下单频率。[ ] 熔断机制日亏损超过阈值后是否能自动停止买入。[ ] 错误处理API 调用失败、数据缺失、模型解析失败是否有兜底。[ ] 日志审计每笔订单是否有完整记录。[ ] 权限控制实盘 API Key 是否经过加密存储是否最小权限。[ ] 回滚方案系统异常时能否一键停止所有订单。[ ] 监控告警是否有关键指标和异常告警。7. 扩展方向当前示例只完成了最小闭环。下一步可以从几个方向扩展第一把 LLM 决策和传统指标结合。模型不只读行情快照还可以读取均线、RSI、MACD 等指标把技术面信号和语义信号一起输入 prompt。第二引入多智能体协同。一个 agent 负责分析情绪一个 agent 负责生成交易计划一个 agent 负责复核风险限制仍然在最后统一执行。第三加入回测框架。让交易代理在历史数据上分批运行统计胜率、盈亏比、最大回撤再根据结果调整 prompt 和限制参数。第四动态限制。根据市场波动率自动调整单笔仓位限制例如波动率上升时自动缩小 max_single_order_pct。动态限制比固定值更接近实盘需求但实现时要注意逻辑可解释性。第五异常检测。监控 LLM 输出置信度、拒绝率、连续亏损次数当这些指标偏离正常范围时自动降级策略。在实际项目里最值得投入的不是让模型“更聪明”而是让限制层“更可靠”。一个能稳定拦截越界意图的 AI 交易代理即使策略平庸也至少不会因为系统缺陷导致灾难性损失。新手可以先从 paper trading 环境反复实验等日志、监控、回滚和审计都完善后再考虑对接真实券商接口。