企业官网建设流程全解析

当你第一次看到“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 交易 Agent

AI 交易 Agent 是一个具备“感知、决策、执行”能力的自动化系统。

• 感知：读取市场数据、持仓数据、账户资金。
• 决策：通过大模型或强化学习模型，基于当前状态生成交易指令。
• 执行：将交易指令发送到模拟撮合引擎或真实交易所 API。
• 记忆：保存历史决策、持仓变化、市场摘要，供后续决策参考。

它与传统量化策略最大的区别在于，决策部分可以由自然语言驱动。你可以直接在 Prompt 里写“当 RSI 低于 30 时买入，单笔仓位不超过总资金的 10%”，模型会尝试理解并执行。这种灵活性降低了策略开发门槛，但也对“约束能力”提出了更高要求。

2.2 用户设定的限制：limits 是什么

“limits”并不是一个简单的最大仓位字段，而是一组完整的风险边界。常见维度包括：

这些限制要求在 Agent 的 Prompt 中出现，也要求在交易执行层被强制校验。不要把安全边界完全交给大模型自觉遵守，因为模型输出具有概率性，一旦偶尔越界，必须有代码把它拦下来。

2.3 Paper Trading（模拟交易）为什么重要

模拟交易的含义，是用实时或模拟行情进行撮合，但资金是虚拟的。它和回测有本质区别：

• 回测是过去的数据，有未来函数和过拟合风险。
• 模拟交易是当下行情，Agent 在真实环境里做决策，但账户是模拟账户。

对于 Agent 类系统，模拟交易还有一个额外价值：它验证的是完整链路，而不是单个策略。模型 Prompt 是否正确解析、仓位校验是否生效、Agent 记忆是否被正确维护、异常情况下是否会触发熔断，这些只有放进模拟环境才能暴露问题。

2.4 适用场景与人群

这个项目最适合三类人：

1. 正在学习 Agent 开发的人。交易场景是 Agent 工程化的绝佳练习场，因为它同时涉及工具调用、记忆、安全边界、外部 API。
2. 有一定量化基础但想用大模型升级策略的个人开发者。可以先从模拟盘看效果，再逐步小资金实盘。
3. 想要搭建内部金融 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 requests

• openai：用于调用大模型接口。
• 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.yaml

account: 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 (f"cash={self.cash:.2f}, total_value={total_value:.2f}, " f"positions={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", encoding="utf-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']}, " f"rsi14={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": f"Portfolio: {portfolio_ctx}\nMarket:\n{market_ctx}\nDecide a single order."}, ] response = self.client.chat.completions.create( model=self.config["model"]["model_name"], messages=messages, temperature=self.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( action=decision["action"], symbol=symbol, quantity=quantity, ) 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", encoding="utf-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_KEY="sk-xxxx" python main.py

Windows 用户可以使用：

set OPENAI_API_KEY=sk-xxxx python main.py

如果使用 OpenAI 兼容接口的第三方服务，可以在agent.py中补充：

self.client = OpenAI( base_url="https://your-endpoint.example.com/v1", api_key="sk-xxxx" )

6. 运行结果与效果验证

运行python main.py后，预期输出类似：

--- cycle 1 --- [order] {'status': 'filled', 'symbol': 'AAPL', 'action': 'buy', 'qty': 100, 'price': 152.34} cash=84766.00, total_value=... , positions={'AAPL': (100, 152.34)}, drawdown=0.0000 --- cycle 2 --- [hold] agent chose to hold cash=84766.00, total_value=...

因为行情是随机生成的，数字每次不同。但有几个行为特征可以用来验证系统是否正确：

1. 订单状态只有 rejected 或 filled，不能出现自定义状态。
2. 所有订单数量都在 100 到 500 之间，如果模型输出 50 或 1000，会被拒绝。
3. 只允许交易 AAPL、TSLA、MSFT，其它品种一律 rejected。
4. 当回撤超过 5% 时，系统会打印风险熔断提示，尽管随机行情下这个概率较低。

如果你发现订单状态出现了rejected，这是正常的，说明限制校验生效了。真正需要担心的是：如果模型连续输出quantity=999999还成交了，那才是执行层校验失效。

这里有一个很可能的坑：模型输出 JSON 字段时，有时会把quantity写成字符串，例如"quantity": "one hundred"。所以我在run_cycle中先用int(decision.get("quantity", 0))做了强转。如果转换失败，程序会直接报错。你可以在转换外层加 try-except，给模型一次重新生成的机会。

7. 常见问题与排查思路

在实际开发中，你大概率会遇到下面这些问题。我把它们整理成表格，方便排查。

这里特别想强调一次“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 模拟盘跑通后，不要急着上实盘

模拟盘通过，只代表“链路通畅”，不代表“策略有效”。因为模拟盘往往没有真实的滑点、手续费、流动性冲击。即使模拟盘收益曲线很漂亮，也要保持警惕。

如果你真的要走向实盘，建议按这个顺序推进：

1. 在模拟盘上连续运行至少 4 周，观察极端行情下的表现。
2. 接入历史数据做回测，确认策略逻辑不是巧合。
3. 使用券商提供的 paper trading API，而不只是本地模拟撮合。
4. 用最小资金实盘，并且第一周只观察不干预，验证真实行情链路。
5. 实时监控 Agent 的行为，设置人工熔断开关，确保可以随时手动接管。

8.5 注意合规与风险边界

这篇文章的技术示例只是学习用途，不构成任何投资建议。在真实交易中，你还需要遵守所在国家或地区的法律法规，确认使用自动交易软件的合规性，并对任何资金损失承担全部责任。如果你是帮机构开发，务必让风控团队参与评审，并设置多层审批机制。

9. 总结与后续学习方向

现在再回头看这个项目标题，你会发现它的高明之处：它没有强调“AI 能赚钱”，而是强调“AI 在限制内交易”和“从模拟盘开始”。这两件事，恰好是 Agent 交易系统最容易被忽略、却又最决定成败的环节。

这篇文章里，我们完成了以下内容：

• 理解了 AI 交易 Agent 的核心环节：感知、决策、执行、记忆。
• 明白了用户设定限制（limits）的价值，以及如何用代码实现硬校验。
• 通过一个最小化 Demo，跑通了模拟交易 Agent 的完整链路。
• 总结了从模拟盘走向实盘前必须做的工程准备。

下一步，你可以从这几个方向继续深入：

1. 接入真实市场数据，把PaperExchange.get_snapshot替换为真实 API，并在模拟撮合中增加手续费和滑点。
2. 完善 Agent 记忆，引入向量数据库存储历史决策和市场事件，让模型拥有更长期的市场上下文。
3. 引入模型评测机制，定期用历史区间数据评测 Agent 决策质量，防止模型策略漂移。
4. 设计多 Agent 架构，让一个 Agent 负责行情分析，另一个 Agent 负责风险审核，再有一个 Agent 负责生成订单，通过“双人复核”机制降低单点错误率。
5. 关注 Agent 安全方向，包括 Prompt 注入防护、外发请求白名单、模型输出合规检查。这些在金融场景下同样重要。

如果你正在学习 Agent 开发，交易场景是一个值得多花时间练习的方向。它逼着你把问题从“模型会不会回答”升级到“系统能不能在约束下稳定运行”，而后者才是 Agent 工程化的真正门槛。