当你第一次看到“AI Agent 自动交易”这个概念时脑子里想的很可能是让 AI 看行情、做判断、自动下单然后等着收益曲线往上走。但真正做过交易系统或者研究过 Agent 工程化的人会立刻警觉——一个没有边界的 Agent在金融市场里就是一个“无法控制的赌徒”。它不会恐惧不会犹豫也不会在亏损超过承受范围时主动停手。所以当我看到 “Show HN: An AI agent that trades inside limits you set, starting on paper” 这个开源项目时印象最深的不是“AI 会交易”而是它把两件事放在了最前面limits用户设定的限制和paper trading模拟盘。这个设计思路恰恰是 AI 交易 Agent 最容易被人忽略、却最核心的部分。这篇文章我想完整拆解一下这类系统的技术链路它到底解决了什么问题、Agent 如何在一个有边界的框架内做决策、模拟交易环境怎么搭建、代码怎么写、以及从模拟盘走向实盘之前必须想清楚的工程细节。读完你可以跑通一个最小可用的 AI 模拟交易 Agent并且理解为什么“限制”和“安全边界”是这类系统里最不可跳过的部分。1. 这篇文章真正要解决的问题AI 交易 Agent 的讨论很容易陷入“模型预测能力”的单点竞争。今天有人说大模型能预测次日涨跌明天有人说量化模型已经跑赢指数但实际上大部分个人开发者尝试 AI 交易时第一个遇到的问题根本不是模型不够聪明而是不知道如何让 Agent 安全地执行交易。具体来说有四个痛点第一Agent 容易产生“失控”行为。大模型存在幻觉可能因为一条新闻、一个异常指标就生成一个远超风险承受范围的订单。没有约束的 Agent本质上是一个带着资金的随机数生成器。第二策略与交易执行严重耦合。很多人直接把“大模型返回的结果”当成订单扔给交易所中间没有仓位审核、没有资金校验、没有交易时段限制。这在真实环境中非常危险。第三缺乏可复现的验证环境。实盘验证成本太高回测又容易被过拟合。对个人开发者来说模拟盘paper trading是验证“Agent 决策 执行链路”的最优环境。第四Agent 的记忆和审计被忽略。模型依赖上下文理解持仓但很多交易 Agent 不维护历史决策记录出了问题也复盘不了。这篇文章要做的就是把 AI 交易 Agent 拆成“有边界的决策系统”来讨论。核心不是教你怎么用模型预测涨跌而是教你如何设计一套约束机制让 Agent 在用户设定的风险边界内做决策并通过模拟盘验证它真的没有越界。2. AI 交易 Agent 的核心概念与适用场景2.1 什么是 AI 交易 AgentAI 交易 Agent 是一个具备“感知、决策、执行”能力的自动化系统。感知读取市场数据、持仓数据、账户资金。决策通过大模型或强化学习模型基于当前状态生成交易指令。执行将交易指令发送到模拟撮合引擎或真实交易所 API。记忆保存历史决策、持仓变化、市场摘要供后续决策参考。它与传统量化策略最大的区别在于决策部分可以由自然语言驱动。你可以直接在 Prompt 里写“当 RSI 低于 30 时买入单笔仓位不超过总资金的 10%”模型会尝试理解并执行。这种灵活性降低了策略开发门槛但也对“约束能力”提出了更高要求。2.2 用户设定的限制limits 是什么“limits”并不是一个简单的最大仓位字段而是一组完整的风险边界。常见维度包括限制维度示例作用品种白名单只允许交易 AAPL、TSLA、MSFT防止 Agent 去交易不熟悉的品种单笔仓位上限单次买入不超过总资金 20%防止单笔决策造成过大风险最大回撤触发权益回撤超过 5% 时停止交易防止亏损持续扩大交易时段只在 09:30-16:00 下发订单防止盘后或异常时段入场订单数量范围每次买入数量在 100-500 股之间控制交易粒度行为白名单只允许 buy、sell、hold禁止做空、加杠杆等复杂操作这些限制要求在 Agent 的 Prompt 中出现也要求在交易执行层被强制校验。不要把安全边界完全交给大模型自觉遵守因为模型输出具有概率性一旦偶尔越界必须有代码把它拦下来。2.3 Paper Trading模拟交易为什么重要模拟交易的含义是用实时或模拟行情进行撮合但资金是虚拟的。它和回测有本质区别回测是过去的数据有未来函数和过拟合风险。模拟交易是当下行情Agent 在真实环境里做决策但账户是模拟账户。对于 Agent 类系统模拟交易还有一个额外价值它验证的是完整链路而不是单个策略。模型 Prompt 是否正确解析、仓位校验是否生效、Agent 记忆是否被正确维护、异常情况下是否会触发熔断这些只有放进模拟环境才能暴露问题。2.4 适用场景与人群这个项目最适合三类人正在学习 Agent 开发的人。交易场景是 Agent 工程化的绝佳练习场因为它同时涉及工具调用、记忆、安全边界、外部 API。有一定量化基础但想用大模型升级策略的个人开发者。可以先从模拟盘看效果再逐步小资金实盘。想要搭建内部金融 Agent 平台的团队。可以用这个结构做产品原型验证风控流程。不适合一上来就直接接实盘账户也不适合把它当成“躺赢工具”。它本质上是一个工程系统不是理财建议。3. 环境准备与前置条件我们要实现的 Demo 基于 Python代码量不大但依赖需要提前装好。以下版本建议以你的实际环境为准重点演示通用思路。3.1 基础环境Python 3.9 或更高版本。一个可以调用大模型 API 的 Key本文以 OpenAI 兼容接口为例。如果没有可用的外部模型 API可以使用本地模型但需要额外工程适配。3.2 依赖清单在项目目录下创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate pip install openai pyyaml pandas requestsopenai用于调用大模型接口。pyyaml用于解析 limits 配置文件。pandas可用于后续扩展数据处理逻辑。requests用于扩展真实行情 API 调用。3.3 模型接口说明代码中会使用OpenAI()客户端对象但只需要模型 API 兼容 OpenAI 格式即可。你可以在配置里指定模型名例如gpt-4o-mini。如果使用国内大模型服务或本地模型通常只需要修改base_url和api_key后续代码逻辑不变。3.4 项目目录结构paper-trade-agent/ ├── limits.yaml # 用户设定的限制配置 ├── agent.py # Agent 决策核心 ├── paper_exchange.py # 模拟交易环境 ├── main.py # 运行入口 └── requirements.txt # 依赖清单这里的paper_exchange.py是整个演示的关键。它模拟了一个带风控校验的交易所撮合引擎专门用来验证 Agent 的下单是否越界。4. 核心流程拆解AI 交易 Agent 的核心流程可以拆成五个步骤。我建议你按这个顺序实现不要一上来就写模型调用。4.1 加载用户限制配置所有限制参数集中在一个 YAML 文件里方便调整和审计。Agent 在启动时读取该配置并把关键限制注入系统 Prompt。执行层也要读取同一份配置保证“提示词约束”和“代码约束”一致。如果配置分散在代码里很容易出现“模型以为只允许 10% 仓位代码却允许了 30%”的问题。4.2 获取市场数据与账户状态Agent 需要一个市场摘要包括当前价格、技术指标、涨跌幅等。同时要提供账户摘要让模型知道当前现金、持仓、回撤情况。这一步要注意不要直接喂给模型一大堆原始数据。模型上下文有限最好用结构化摘要。在交易场景里信息密度比信息数量更重要。4.3 生成决策Agent 调用大模型让模型基于当前市场状态和账户状态输出一个结构化的 JSON 指令。系统 Prompt 中要写清楚所有限制并要求模型严格输出 JSON。很多初学者在这一步犯的错误是 Prompt 过于开放例如“根据行情决定是否交易”。这会让模型输出各种奇怪的非结构化文本。正确做法是规定动作枚举、字段含义、输出格式。4.4 交易执行与强制校验模型输出 JSON 后不能直接发给“交易所”。必须在执行层重新校验品种是否在白名单内数量是否在最小值/最大值范围内买入金额是否超过单仓上限当前回撤是否已超过熔断线当前时间是否在允许交易时段内。任何一项不满足订单都应该被拒绝并记录原因。这才是“限制”真正落地的位置。4.5 记录与记忆维护决策记录、订单记录、账户权益曲线都需要持久化。Agent 的记忆窗口可以保留最近若干轮决策和成交结果作为下一次决策的上下文。没有记忆的交易 Agent就像一次只做一个动作的机器人无法处理需要连续观察的场景。但记忆也不能无限增长一般只保留最近 10 到 20 条避免浪费上下文。5. 完整示例代码实现下面直接给出一个可运行的最小实现。这段代码重点展示“限制校验”和“模拟撮合”行情数据使用随机模拟不依赖真实行情源。5.1 用户限制配置limits.yamlaccount: initial_cash: 100000 # 初始模拟资金 max_position_pct: 0.2 # 单仓最大仓位占总资金 20% max_drawdown_pct: 0.05 # 权益最大回撤 5%超过则停止交易 trading: symbols: [AAPL, TSLA, MSFT] # 只允许交易这三个标的 order_size_min: 100 # 单笔最小数量 order_size_max: 500 # 单笔最大数量 allowed_actions: [buy, sell, hold] trading_hours_start: 09:30 trading_hours_end: 16:00 model: model_name: gpt-4o-mini # 按实际可用模型调整 temperature: 0.2这份配置就是“用户设定的限制”。后续无论是 Prompt 生成还是下单校验都以它为准。5.2 模拟交易所paper_exchange.py# 文件路径paper_exchange.py from datetime import datetime import random class PaperExchange: def __init__(self, config): self.cash config[account][initial_cash] self.initial_equity self.cash self.peak_equity self.cash self.positions {} self.history [] self.symbols config[trading][symbols] self.max_position_pct config[account][max_position_pct] self.max_drawdown_pct config[account][max_drawdown_pct] self.order_size_min config[trading][order_size_min] self.order_size_max config[trading][order_size_max] self.allowed_actions config[trading][allowed_actions] def get_snapshot(self, symbols): 模拟行情快照实际项目应替换为真实数据源。 snapshot [] for symbol in symbols: price round(random.uniform(50, 500), 2) snapshot.append({ symbol: symbol, price: price, ema20: round(price * random.uniform(0.97, 1.03), 2), rsi14: round(random.uniform(30, 70), 1), change_pct: round(random.uniform(-3, 3), 2), }) return snapshot def get_portfolio_summary(self): total_value self._total_value() drawdown self._calculate_drawdown() return (fcash{self.cash:.2f}, total_value{total_value:.2f}, fpositions{self.positions}, drawdown{drawdown:.4f}) def submit_order(self, action, symbol, quantity): if action not in self.allowed_actions: return {status: rejected, reason: action not allowed} if action hold: return {status: rejected, reason: hold does not submit order} if symbol not in self.symbols: return {status: rejected, reason: symbol not allowed} if not (self.order_size_min quantity self.order_size_max): return {status: rejected, reason: quantity out of range} price self._latest_price(symbol) notional price * quantity if action buy: if notional self.cash: return {status: rejected, reason: insufficient cash} if self._position_pct(notional) self.max_position_pct: return {status: rejected, reason: position limit exceeded} self.cash - notional self.positions[symbol] self.positions.get(symbol, (0, 0)) old_qty, old_avg self.positions[symbol] new_qty old_qty quantity new_avg (old_qty * old_avg notional) / new_qty if new_qty 0 else 0 self.positions[symbol] (new_qty, new_avg) elif action sell: if symbol not in self.positions: return {status: rejected, reason: no position} qty, avg self.positions[symbol] if qty quantity: return {status: rejected, reason: insufficient position} self.cash notional remaining qty - quantity if remaining 0: del self.positions[symbol] else: self.positions[symbol] (remaining, avg) self.history.append({ time: datetime.utcnow().isoformat(), symbol: symbol, action: action, qty: quantity, price: price, }) return {status: filled, symbol: symbol, action: action, qty: quantity, price: price} def check_drawdown_limit(self): if self._calculate_drawdown() self.max_drawdown_pct: return False return True def _latest_price(self, symbol): return round(random.uniform(50, 500), 2) def _total_value(self): total self.cash for symbol, (qty, _) in self.positions.items(): total qty * self._latest_price(symbol) return total def _position_pct(self, notional): return notional / max(self._total_value(), 1) def _calculate_drawdown(self): equity self._total_value() self.peak_equity max(self.peak_equity, equity) if self.peak_equity 0: return 0.0 return (self.peak_equity - equity) / self.peak_equity这段代码的核心价值是在提交订单前做了多重校验。你可以看到即使模型输出了一个越界指令也会在这里被拦截。这是“用户设定限制”在工程层面的实现。5.3 Agent 决策核心agent.py# 文件路径agent.py import json import re import yaml from openai import OpenAI class TradeAgent: def __init__(self, config_path, exchange): with open(config_path, r, encodingutf-8) as f: self.config yaml.safe_load(f) self.exchange exchange self.client OpenAI() # 读取环境变量 OPENAI_API_KEY self.memory [] def _build_system_prompt(self): account self.config[account] trading self.config[trading] return f You are a paper trading agent. You must strictly obey these limits: - Allowed symbols: {trading[symbols]} - Allowed actions: {trading[allowed_actions]} - Order quantity must be between {trading[order_size_min]} and {trading[order_size_max]} - Max position percentage per symbol: {account[max_position_pct]} - Max drawdown allowed: {account[max_drawdown_pct]} Reply in JSON only, with format: {{action: buy|sell|hold, symbol: ..., quantity: integer, reason: short reason}} def _get_market_ctx(self): snapshot self.exchange.get_snapshot(self.config[trading][symbols]) lines [ f{item[symbol]}: price{item[price]}, ema20{item[ema20]}, frsi14{item[rsi14]}, change{item[change_pct]:.2f}% for item in snapshot ] return \n.join(lines) def decide(self): market_ctx self._get_market_ctx() portfolio_ctx self.exchange.get_portfolio_summary() messages [ {role: system, content: self._build_system_prompt()}, *self.memory[-10:], {role: user, content: fPortfolio: {portfolio_ctx}\nMarket:\n{market_ctx}\nDecide a single order.}, ] response self.client.chat.completions.create( modelself.config[model][model_name], messagesmessages, temperatureself.config[model][temperature], ) raw response.choices[0].message.content.strip() decision self._safe_parse_json(raw) self.memory.append({role: assistant, content: raw}) return decision def run_cycle(self): decision self.decide() if decision is None: print([error] agent failed to produce valid decision) return if decision.get(action) hold: print([hold] agent chose to hold) return symbol decision.get(symbol, ) quantity int(decision.get(quantity, 0)) result self.exchange.submit_order( actiondecision[action], symbolsymbol, quantityquantity, ) print(f[order] {result}) if not self.exchange.check_drawdown_limit(): print([risk] drawdown limit exceeded, halt trading) staticmethod def _safe_parse_json(text): try: return json.loads(text) except json.JSONDecodeError: match re.search(r\{.*\}, text, re.S) if match: try: return json.loads(match.group(0)) except json.JSONDecodeError: return None return None这里值得注意的设计细节是_safe_parse_json。大模型经常会在 JSON 外面加解释文字即使你在 Prompt 里要求“只输出 JSON”。所以需要做一层解析容错。5.4 运行入口main.py# 文件路径main.py import yaml from paper_exchange import PaperExchange from agent import TradeAgent def main(): with open(limits.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) exchange PaperExchange(config) agent TradeAgent(limits.yaml, exchange) total_cycles 5 for cycle in range(total_cycles): print(f--- cycle {cycle 1} ---) agent.run_cycle() print(exchange.get_portfolio_summary()) print() print( final history ) for order in exchange.history[-10:]: print(order) if __name__ __main__: main()5.5 运行前配置环境变量在终端中设置你的 API Key以 macOS / Linux 为例export OPENAI_API_KEYsk-xxxx python main.pyWindows 用户可以使用set OPENAI_API_KEYsk-xxxx python main.py如果使用 OpenAI 兼容接口的第三方服务可以在agent.py中补充self.client OpenAI( base_urlhttps://your-endpoint.example.com/v1, api_keysk-xxxx )6. 运行结果与效果验证运行python main.py后预期输出类似--- cycle 1 --- [order] {status: filled, symbol: AAPL, action: buy, qty: 100, price: 152.34} cash84766.00, total_value... , positions{AAPL: (100, 152.34)}, drawdown0.0000 --- cycle 2 --- [hold] agent chose to hold cash84766.00, total_value...因为行情是随机生成的数字每次不同。但有几个行为特征可以用来验证系统是否正确订单状态只有 rejected 或 filled不能出现自定义状态。所有订单数量都在 100 到 500 之间如果模型输出 50 或 1000会被拒绝。只允许交易 AAPL、TSLA、MSFT其它品种一律 rejected。当回撤超过 5% 时系统会打印风险熔断提示尽管随机行情下这个概率较低。如果你发现订单状态出现了rejected这是正常的说明限制校验生效了。真正需要担心的是如果模型连续输出quantity999999还成交了那才是执行层校验失效。这里有一个很可能的坑模型输出 JSON 字段时有时会把quantity写成字符串例如quantity: one hundred。所以我在run_cycle中先用int(decision.get(quantity, 0))做了强转。如果转换失败程序会直接报错。你可以在转换外层加 try-except给模型一次重新生成的机会。7. 常见问题与排查思路在实际开发中你大概率会遇到下面这些问题。我把它们整理成表格方便排查。问题现象可能原因排查方式解决方案agent输出无法解析为 JSON模型在 JSON 外添加了说明文字打印raw原文用正则提取首个 JSON 片段或要求模型只输出代码块agent terminated due to error或execution terminatedAPI 调用异常、上下文过长、模型返回不合法查看异常堆栈和最近一次 Prompt 长度缩短历史记忆窗口增加重试机制捕获openai.APIError类异常订单总是rejected: quantity out of range模型不理解数量范围检查系统 Prompt 中order_size_min/max是否清晰在 Prompt 里额外给出示例quantitymust be integer between 100 and 500模型频繁买入但不卖出没有持仓状态反馈查看get_portfolio_summary中positions是否传入模型在 user message 中明确展示现有持仓和均价回撤熔断不触发_calculate_drawdown逻辑有误打印peak_equity和total_value维护峰值权益曲线并在每次成交后更新实盘 API 与模拟盘不一致模拟环境缺少滑点、手续费、涨跌停对比真实券商 API 文档在PaperExchange中增加滑点模型和手续费字段API 请求超时网络环境问题或模型负载高查看日志中APITimeoutError设置timeout参数并增加指数退避重试这里特别想强调一次“Agent 执行层报错”的处理经验。当你看到类似于agent execution terminated due to error的信息时第一反应不该是修改模型 Prompt而是先看错误发生在哪一层如果错误发生在decide()之前说明是 API 调用或上下文管理问题。如果错误发生在submit_order()说明是执行层校验逻辑写得不完善。如果错误发生在_safe_parse_json()说明是模型输出格式不稳定。分层排查能帮你快速定位到底是大模型的问题还是工程代码的问题。8. 最佳实践与工程建议8.1 不要把风控责任全部交给模型这是全文最重要的一条建议。Prompt 里的限制是“软限制”代码里的校验是“硬限制”。即使模型今天听话也不能保证它下一条输出仍然听话。所有关键限制必须在执行层强制校验包括仓位、品种、数量、时段。8.2 为 Agent 设计完整的记忆管理策略在交易场景中Agent 需要记住这几类信息当前持仓与均价最近 N 次决策与结果最近一次下单被拒原因当前账户回撤状态。但记忆不是越多越好。建议只保留最近 10 到 20 条决策记录并且把账户状态用结构化文本拼接而不是把完整 JSON 丢给模型。上下文越短模型输出质量越稳定。8.3 建立可审计的日志体系交易系统最忌讳“黑箱”。每一步决策都应该记录触发决策时的市场快照模型返回的完整原始输出执行层的校验结果最终成交回报。可以用 JSON Lines 格式落盘方便后续复盘和回放。比如{time: 2025-01-01T10:00:00Z, step: infer, raw_model_output: ..., parsed_decision: {...}} {time: 2025-01-01T10:00:01Z, step: exchange_validation, order: {...}, result: rejected, reason: quantity out of range}有了这些日志Agent 出现预期外行为时你才能快速定位是模型问题还是规则问题。8.4 模拟盘跑通后不要急着上实盘模拟盘通过只代表“链路通畅”不代表“策略有效”。因为模拟盘往往没有真实的滑点、手续费、流动性冲击。即使模拟盘收益曲线很漂亮也要保持警惕。如果你真的要走向实盘建议按这个顺序推进在模拟盘上连续运行至少 4 周观察极端行情下的表现。接入历史数据做回测确认策略逻辑不是巧合。使用券商提供的 paper trading API而不只是本地模拟撮合。用最小资金实盘并且第一周只观察不干预验证真实行情链路。实时监控 Agent 的行为设置人工熔断开关确保可以随时手动接管。8.5 注意合规与风险边界这篇文章的技术示例只是学习用途不构成任何投资建议。在真实交易中你还需要遵守所在国家或地区的法律法规确认使用自动交易软件的合规性并对任何资金损失承担全部责任。如果你是帮机构开发务必让风控团队参与评审并设置多层审批机制。9. 总结与后续学习方向现在再回头看这个项目标题你会发现它的高明之处它没有强调“AI 能赚钱”而是强调“AI 在限制内交易”和“从模拟盘开始”。这两件事恰好是 Agent 交易系统最容易被忽略、却又最决定成败的环节。这篇文章里我们完成了以下内容理解了 AI 交易 Agent 的核心环节感知、决策、执行、记忆。明白了用户设定限制limits的价值以及如何用代码实现硬校验。通过一个最小化 Demo跑通了模拟交易 Agent 的完整链路。总结了从模拟盘走向实盘前必须做的工程准备。下一步你可以从这几个方向继续深入接入真实市场数据把PaperExchange.get_snapshot替换为真实 API并在模拟撮合中增加手续费和滑点。完善 Agent 记忆引入向量数据库存储历史决策和市场事件让模型拥有更长期的市场上下文。引入模型评测机制定期用历史区间数据评测 Agent 决策质量防止模型策略漂移。设计多 Agent 架构让一个 Agent 负责行情分析另一个 Agent 负责风险审核再有一个 Agent 负责生成订单通过“双人复核”机制降低单点错误率。关注 Agent 安全方向包括 Prompt 注入防护、外发请求白名单、模型输出合规检查。这些在金融场景下同样重要。如果你正在学习 Agent 开发交易场景是一个值得多花时间练习的方向。它逼着你把问题从“模型会不会回答”升级到“系统能不能在约束下稳定运行”而后者才是 Agent 工程化的真正门槛。