
手动交易最大的敌人往往不是行情而是自己的情绪和执行力。明明设好了止损线价格剧烈波动时却忍不住手动干预计划好单笔最大亏损连续几次亏损后却开始上头加仓。复盘时会发现真正拉低收益的往往不是策略本身而是交易纪律的缺失。这也是为什么越来越多的开发者开始尝试用 AI agent 来执行交易——把规则写死、把限额管住、把操作交给自动化的代理程序并且先在纸面交易Paper Trading环境里跑通再考虑后续迭代。本文将围绕“一个在用户设定限额内进行交易的 AI agent”这一主题完整拆解它的核心概念、系统设计、代码实现和工程落地要点。正文会从最基础的 agent 工作原理讲起逐步实现一个包含限额控制、模拟行情、自动决策和交易执行的纸面交易示例。无论你是刚接触 AI agent 的新手还是已经在写量化策略的开发者都可以从这篇文章里找到可以上手的代码和值得借鉴的设计思路。1. 背景与核心概念1.1 什么是“在限额内交易”的 AI agent先拆两个关键词AI agent 和 limit限额。AI agent 可以粗浅地理解为“一个能自己看、自己想、自己动手的智能程序”。它不再是简单的 if-else 规则引擎而是可以接收环境信息比如当前价格、持仓、账户余额经过模型推理或策略逻辑最终做出动作比如买入、卖出、观望并把动作反馈到环境中去。而“trades inside limits you set”的意思是这个 agent 并不是无约束的自动交易机器人用户会在运行前设定一系列边界条件。典型的限额包括单笔交易最大金额。最大持仓数量。单日最大亏损额度。单笔最大亏损比例。允许交易的标的范围。禁止交易的时段。这些限制的本质是把“自动交易”的自主权关进笼子里。尤其是早期版本agent 的智能程度可能并不高很容易因为策略缺陷或数据异常做出极端操作。限额体系相当于安全网能让错误的影响范围可控这也是这类项目普遍选择先纸面交易的原因。1.2 为什么先做纸面交易Paper Trading纸面交易也叫模拟交易是指 agent 在一个模拟环境里进行买卖操作它看到的是模拟行情下单后走模拟撮合逻辑影响的是虚拟账户资金不涉及真实金钱。先纸面交易有非常实际的好处验证策略逻辑是否行得通。验证 agent 是否严格遵循用户设定限额。验证 API 调用、数据流、状态管理是否稳定。暴露风控漏洞但不会造成真实亏损。为后续小资金实盘提供参考依据。很多从“Show HN”或开源社区走出来的交易 agent 项目都会把 Paper Trading 作为默认运行模式。真实金额交易意味着额外的合规、授权、安全责任而纸面交易让项目可以先用技术能力取信于人。1.3 和传统量化交易程序的区别传统量化交易程序通常采用“策略 执行引擎”的方式策略由人编写成规则或数学模型程序负责批量下单。而 AI agent 的差异点在于更强调“感知 - 决策 - 执行 - 反馈”的完整循环。可能接入大模型LLM作为决策模块用自然语言描述策略而非纯代码策略。允许策略动态调整agent 可以根据市场状态自行判断是加大仓位还是暂停交易。更关注“工具调用Tool Calling”比如通过调用函数来查询余额、下单、读取行情。所以它不仅仅是把指标公式跑一遍而是一个更接近“自主交易员”的软件形态。2. 环境准备与版本说明本文的实战示例使用 Python 实现核心逻辑不依赖重量级框架便于读者在本地直接运行和改造。你需要准备以下基础环境操作系统Windows 10/11、macOS 或 Linux 均可。Python 版本建议 3.9 及以上。依赖库标准库 random、dataclasses、enum、datetime 即可完成核心演示如果希望输出更美观的表格可以安装 pandas但不是必须。IDE 或编辑器VS Code、PyCharm 或任意你习惯的工具都可以。如需接入真实行情或真实券商 API需要额外申请相应权限和密钥但本文不涉及真实行情源仅使用模拟数据生成器方便你完整跑通流程。版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。3. 核心原理与系统设计3.1 Agent 交易的基本循环一个交易 agent 的核心运行循环可以概括为四步观察Observe获取当前行情、持仓、账户权益、历史交易记录。观察频率可以是每根K线、每分钟或每个 tick。决策Decide基于策略逻辑或大模型推理给出目标操作。例如“当前价格低于近 5 日均价 2%可以买入 10 股”。执行Execute在决策指导下执行交易操作同时必须在此阶段校验所有限额。反馈Feedback将交易结果、持仓变化、账户权益变化记录回状态中供下一步观察使用。整个循环是一个闭环系统任何一步出错都可能导致后续决策失真。因此在实际工程项目里每一步都应该有日志、有状态快照、有异常捕获。3.2 限额管理Limits模块设计限额管理是整个 agent 的灵魂。一个合理的做法是把限额抽象成独立模块每次执行交易前都调用一个统一的检查函数。一个最小可用的限额模型可以包含max_trade_amount单笔交易最大金额。max_position单标的持仓最大数量。max_loss_per_trade单笔交易最大亏损比例。max_daily_loss单日最大亏损额度。allowed_symbols允许交易的标的列表。每次判断买入时需同时检查当前持仓 计划买入数量 最大持仓。计划买入金额 单笔最大交易金额。标的在允许列表里。每次判断卖出时需检查持仓数量是否足够以及卖出后是否使账户权益跌破单日最大亏损线。当任一限额不满足时agent 应当放弃该操作并记录原因而不是强行执行。3.3 纸面交易撮合引擎纸面交易引擎需要在不接真实券商的情况下模拟真实交易环境。核心职责包括维护虚拟账户余额和持仓。接收 agent 的交易指令。按模拟价格撮合成交通常可以添加一个滑点比例。记录交易明细。更新账户权益。价格数据可以使用简单随机游走生成也可以用历史数据回放。本文使用随机游走方式生成模拟行情方便快速演示。3.4 引入大模型作为决策模块可选对于希望体验“AI”部分的读者可以预留一个LLMStrategy类内部调用 OpenAI、DeepSeek、通义千问等模型的 Function Calling 能力让模型根据输入信息决定是否交易。不过需要注意的是大模型存在幻觉和数字理解不稳定的问题。它可能把“买入 10 股”理解成“买入 10 手”也可能在压力行情下违背既定限额。因此无论 agent 的决策层是简单规则还是大模型限额校验层都必须是硬代码、不可被模型绕过。也就是说模型可以“提议”交易但最终“批准”权在限额检查器手中。这一点也是工程上最重要的一条原则AI 只拥有建议权限额系统才拥有最终决定权。4. 完整实战从零实现一个限额内纸面交易 agent下面我们来实现一个可以直接运行的示例项目。这个项目包含模拟行情源、虚拟账户、限额检查器、规则策略、纸面交易引擎和主循环。4.1 创建项目结构建议创建如下目录结构paper-trading-agent/ ├── agent.py # agent 主逻辑 ├── strategy.py # 决策策略规则策略 LLM预留接口 ├── risk.py # 限额检查器 ├── models.py # 数据模型 ├── market.py # 模拟行情源 ├── broker.py # 纸面交易引擎 └── main.py # 程序入口为了减少文件切换的复杂度本文演示将以多个代码片段讲解最后在main.py中串联。4.2 定义数据模型models.py首先定义交易方向、订单、持仓等基础数据结构。文件路径models.py。# 文件路径models.py from dataclasses import dataclass from enum import Enum from typing import Optional class OrderSide(str, Enum): BUY BUY SELL SELL class OrderStatus(str, Enum): PENDING PENDING FILLED FILLED REJECTED REJECTED dataclass class MarketSnapshot: 一条行情快照 symbol: str price: float timestamp: str dataclass class Order: 一条交易订单 symbol: str side: OrderSide quantity: int price: float status: OrderStatus OrderStatus.PENDING reject_reason: Optional[str] None这里使用dataclass来减少样板代码。MarketSnapshot代表某个时间点某个标的的价格信息Order代表一次交易订单包含方向、数量、价格和最终状态。4.3 模拟行情源market.py随机游走生成模拟行情。为了让结果可复现可以接受随机种子。文件路径market.py。# 文件路径market.py import random from datetime import datetime, timedelta from models import MarketSnapshot class RandomWalkMarket: 基于随机游走的模拟行情源 def __init__(self, symbols: list, seed: int 42): self.symbols symbols self.prices {symbol: 100.0 for symbol in symbols} random.seed(seed) def next_price(self, symbol: str) - float: current self.prices[symbol] # 每次价格变动在 -2% ~ 2% 之间浮动 change random.uniform(-0.02, 0.02) next_price max(1.0, current * (1 change)) self.prices[symbol] next_price return round(next_price, 2) def tick(self) - dict: 生成一轮所有标的的行情快照 timestamp (datetime.now() timedelta(secondsrandom.randint(0, 1000))).strftime(%Y-%m-%d %H:%M:%S) snapshots {} for symbol in self.symbols: price self.next_price(symbol) snapshots[symbol] MarketSnapshot( symbolsymbol, priceprice, timestamptimestamp, ) return snapshotsRandomWalkMarket的tick()方法每调用一次就生成所有标的的最新价格快照。随机游走虽然不能真实反映市场规律但足够用来测试 agent 的买卖逻辑和限额检查。4.4 虚拟账户broker.py虚拟账户需要维护现金余额、持仓数量和交易记录。文件路径broker.py。# 文件路径broker.py from models import Order, OrderSide, OrderStatus class PaperBroker: 纸面交易引擎模拟撮合与持仓管理 def __init__(self, initial_cash: float 10000.0): self.initial_cash initial_cash self.cash initial_cash self.positions {} # symbol - quantity self.trades [] # 交易记录 def get_position(self, symbol: str) - int: return self.positions.get(symbol, 0) def market_value(self, prices: dict) - float: 按最新价格计算持仓市值 total 0.0 for symbol, qty in self.positions.items(): price prices.get(symbol) if price: total qty * price return round(total, 2) def equity(self, prices: dict) - float: 总权益 现金 持仓市值 return round(self.cash self.market_value(prices), 2) def fill_order(self, order: Order, market_price: float) - Order: 执行订单并处理买卖 这里不考虑滑点但如果需要可以扩展 slippage 参数 if order.side OrderSide.BUY: cost order.quantity * market_price if cost self.cash: order.status OrderStatus.REJECTED order.reject_reason 现金不足 return order self.cash - cost self.positions[order.symbol] self.positions.get(order.symbol, 0) order.quantity order.price market_price order.status OrderStatus.FILLED else: current_qty self.positions.get(order.symbol, 0) if order.quantity current_qty: order.status OrderStatus.REJECTED order.reject_reason 持仓不足 return order self.cash order.quantity * market_price self.positions[order.symbol] current_qty - order.quantity if self.positions[order.symbol] 0: del self.positions[order.symbol] order.price market_price order.status OrderStatus.FILLED self.trades.append(order) return orderPaperBroker实现了最核心的功能现金管理、持仓管理、买卖撮合和交易记录。4.5 限额检查器risk.py限额检查器负责在订单真正执行前进行拦截。文件路径risk.py。# 文件路径risk.py from dataclasses import dataclass from models import Order, OrderSide dataclass class TradingLimits: 用户可以设定的限额 max_trade_amount: float 1000.0 # 单笔最大交易金额 max_position: int 20 # 单标的最大持仓数量 max_daily_loss: float 500.0 # 单日最大亏损 allowed_symbols: list None # 允许交易的标的列表 def __post_init__(self): if self.allowed_symbols is None: self.allowed_symbols [AAPL, MSFT, GOOG] class RiskManager: 限额检查器统一校验每一笔订单 def __init__(self, limits: TradingLimits): self.limits limits def check_order(self, order: Order, current_position: int, current_equity: float, initial_equity: float) - tuple: 返回 (是否通过, 拒绝原因) if order.symbol not in self.limits.allowed_symbols: return False, f标的不允许交易: {order.symbol} if order.side OrderSide.BUY: # 检查单笔金额 order_amount order.quantity * order.price if order_amount self.limits.max_trade_amount: return False, f单笔交易金额 {order_amount:.2f} 超过限额 {self.limits.max_trade_amount:.2f} # 检查持仓数量 if current_position order.quantity self.limits.max_position: return False, f持仓数量将超过限额 {self.limits.max_position} # 检查单日亏损 if current_equity initial_equity - self.limits.max_daily_loss: return False, f当前亏损超过单日最大亏损限额 {self.limits.max_daily_loss:.2f} return True, 这个检查器目前覆盖了三个最核心的限制条件单笔金额、持仓数量、单日亏损。在实际项目中还需要扩展更多规则比如交易时段限制、涨跌停限制、禁止高价股等。4.6 决策策略strategy.py决策策略模块负责根据行情快照和账户状态生成目标操作。本文先实现一个简单的规则策略当价格比上一轮低时买入比上一轮高时卖出。同时预留了LLMStrategy的接口。文件路径strategy.py。# 文件路径strategy.py from models import Order, OrderSide from broker import PaperBroker class SimpleRuleStrategy: 一个简单的均值回归风格规则策略 - 如果当前价格低于上一轮价格则小幅买入。 - 如果当前价格高于上一轮价格则小幅卖出。 仅作演示不构成投资建议。 def __init__(self, trade_quantity: int 5): self.trade_quantity trade_quantity self.last_price {} def generate_orders(self, snapshots: dict, broker: PaperBroker) - list: orders [] for symbol, snapshot in snapshots.items(): price snapshot.price last self.last_price.get(symbol) if last is None: self.last_price[symbol] price continue if price last: # 价格下跌买入 orders.append(Order( symbolsymbol, sideOrderSide.BUY, quantityself.trade_quantity, priceprice, )) elif price last: # 价格上涨卖出 current_qty broker.get_position(symbol) sell_qty min(self.trade_quantity, current_qty) if sell_qty 0: orders.append(Order( symbolsymbol, sideOrderSide.SELL, quantitysell_qty, priceprice, )) self.last_price[symbol] price return orders class LLMStrategy: 预留接口后续可接入大模型 API将行情快照与账户状态交给 LLM 决策。 实际接入时建议使用 Function Calling将 check_order 和 broker 操作方法 封装成工具函数由模型调用。同时必须保留 RiskManager 作为最终拦截层。 def __init__(self, api_key: str None, model: str gpt-4o-mini): self.api_key api_key self.model model def generate_orders(self, snapshots: dict, broker: PaperBroker) - list: # 这里仅作示例真实调用需要安装 openai 等 SDK # 并补全函数调用协议。为避免误导不提供具体 SDK 代码。 return []SimpleRuleStrategy的逻辑很简单但足以演示“决策 - 检查 - 执行”链路。LLMStrategy作为一个接口占位告诉你未来可以往哪个方向扩展。4.7 Agent 主循环agent.pyAgent 主循环把行情源、策略、风控、交易引擎串起来。文件路径agent.py。# 文件路径agent.py from market import RandomWalkMarket from broker import PaperBroker from risk import RiskManager, TradingLimits from strategy import SimpleRuleStrategy class TradingAgent: 限额内交易的 agent 主程序 def __init__(self, symbols, initial_cash10000.0): self.market RandomWalkMarket(symbolssymbols, seed42) self.broker PaperBroker(initial_cashinitial_cash) self.strategy SimpleRuleStrategy(trade_quantity5) self.limits TradingLimits() self.risk RiskManager(self.limits) self.initial_equity initial_cash self.prices {} def run(self, rounds: int 20): 运行多轮模拟交易。 每一轮 1. 获取最新行情 2. 策略生成订单 3. 风控检查 4. 风控通过则撮合执行 5. 输出账户状态 print(f初始资金: {self.initial_equity:.2f}) print(- * 70) for i in range(1, rounds 1): snapshots self.market.tick() self.prices {s: snap.symbol for s, snap in snapshots.items()} price_map {s: snap.price for s, snap in snapshots.items()} orders self.strategy.generate_orders(snapshots, self.broker) for order in orders: allowed, reason self.risk.check_order( orderorder, current_positionself.broker.get_position(order.symbol), current_equityself.broker.equity(price_map), initial_equityself.initial_equity, ) if not allowed: print(f[第{i}轮] 订单被拒绝: {reason}) continue market_price price_map[order.symbol] filled_order self.broker.fill_order(order, market_price) if filled_order.status.name FILLED: print( f[第{i}轮] 成交: {order.side.value} {order.symbol} f{order.quantity}股 {market_price:.2f} ) else: print(f[第{i}轮] 订单失败: {filled_order.reject_reason}) equity self.broker.equity(price_map) print( f[第{i}轮] 权益: {equity:.2f} | 现金: {self.broker.cash:.2f} | f持仓: {self.broker.positions} ) print(- * 70) print(f最终权益: {self.broker.equity(price_map):.2f})4.8 运行入口main.py最后写一个入口文件直接创建 agent 并运行。文件路径main.py。# 文件路径main.py from agent import TradingAgent if __name__ __main__: agent TradingAgent(symbols[AAPL, MSFT, GOOG], initial_cash10000.0) agent.run(rounds30)在项目根目录执行python main.py预期输出大概是这样的具体数值因随机种子而定初始资金: 10000.00 ---------------------------------------------------------------------- [第1轮] 权益: 10045.00 | 现金: 10000.00 | 持仓: {} [第2轮] 成交: BUY AAPL 5股 98.73 [第2轮] 成交: BUY MSFT 5股 102.01 ... [第30轮] 最终权益: 10230.554.9 结果说明这个示例展示了一个完整的最小闭环模拟行情源驱动策略策略产生订单风控检查器对订单进行限额校验校验通过的订单进入纸面交易引擎撮合成交账户权益与持仓状态随之更新。关键点在于即使你没有写任何复杂的量化指标也没有接入任何真实行情源这样一个不到 200 行的系统已经具备了一个“限额内交易 agent”的全部骨架。接下来你可以把SimpleRuleStrategy替换成真实策略把RandomWalkMarket替换成历史数据回放或真实行情 API把PaperBroker替换成券商接口整个系统就能继续向生产环境演进。5. 常见问题与排查思路5.1 常见问题排查表问题现象常见原因解决思路agent 不产生任何交易订单策略逻辑可能认为当前没有满足条件的信号或行情源未正确返回数据打印策略输入信息检查last_price是否更新订单重复执行主循环在产生订单后未清除last_price或状态记录确保策略使用上一轮状态避免同一条信号重复触发风控拦截导致无法成交单笔金额、持仓数量或亏损限额设置过小调整TradingLimits参数并确认账户初始资金是否充足模拟行情价格全部相同随机种子固定或价格序列未正确更新检查RandomWalkMarket的next_price是否被调用大模型决策不稳定时买时卖LLM 对数字理解不稳定或提示词约束不足将大模型改为“只推荐方向”实际数量由代码计算并让风控统一校验单日亏损判断失效初始权益变化后没有重置记录每日开盘权益而不是始终使用首次初始权益卖出时提示持仓不足持仓数量计算错误或同一标的多次买入记录不完整在PaperBroker中打印每次成交细节确认持仓 dict 是否正确更新5.2 调试建议在实际调试过程中建议在 agent 主循环里加入更详细的日志print(f[DEBUG] symbol{order.symbol} side{order.side} qty{order.quantity} price{order.price:.2f} current_position{current_position})日志是交易 agent 最重要的调试工具。没有日志你很难判断一个订单到底是被策略拒绝、被风控拒绝、还是因为资金不足被交易引擎拒绝。5.3 关于 LLM 决策的重要提醒如果你的 agent 引入了大模型请务必注意以下几点大模型可能会编造不存在的持仓数量。大模型可能忽略小数位或数量单位。大模型可能受历史对话影响比如前一轮亏损后变得“激进”。大模型的 API 调用可能超时或返回异常格式。因此最稳妥的做法是让大模型只负责“判断方向”即输出“buy”、“sell”或“hold”再由代码层根据当前持仓、余额和限额计算出具体的买卖数量。这样既保留了 AI 的灵活性也限制了它的破坏边界。6. 最佳实践与工程建议6.1 不要把限额当成摆设限额管理不是加一个 if 判断那么简单。工程上建议限额配置与代码分离可以通过配置文件、环境变量或配置中心动态调整。每次检查限额时记录日志并且保留原始订单快照。如果 agent 多次触发限额拒绝应发出告警而不是静默忽略。上线前用历史数据回放验证限额在各种极端行情下都会生效。6.2 状态持久化与恢复真实交易环境中agent 可能随时崩溃或重启。如果状态只存在内存里重启后账户持仓、成交记录、当天累计亏损都会丢失。建议使用 SQLite、PostgreSQL 或 Redis 持久化账户状态。每次交易成交后立即写入交易流水。记录初始化权益和每日权益快照。6.3 从纸面交易走向真实环境的安全边界当你想把纸面交易升级为真实交易时有几个原则绝对不能妥协保持最小权限只在需要交易时连接券商 API避免 agent 拥有全部账户权限。设置资金上限给 agent 分配专属的小额资金不要把主资金放在同一账户。保留人工熔断机制当单日亏损达到一定比例时应自动停止 agent并通知管理员。灰度上线先以最小下单数量运行数周确认行为符合预期再逐步放开。合规意识参与金融市场的自动交易需要遵守当地法律法规及券商使用条款务必提前确认授权和合规要求。6.4 策略和模型的可观测性AI agent 的运行过程往往是黑盒尤其是使用大模型作为决策层时你需要建立完整的可观测性体系记录每一次模型输入和输出。记录每一次工具调用及其参数。记录每一次限额拒绝的原因和上下文。定期回测策略的长期收益和回撤。没有可观测性的 agent本质上是一个不受控的黑盒子。你很难回答“为什么今天手欠买入了那只股票”这类问题。6.5 避免过度优化在开发纸面交易 agent 时很容易陷入在模拟盘上把收益“做得很好看”的误区。如果你的策略只在某个随机种子或某段历史数据上表现好很可能就是过拟合。好的做法是使用不同的随机种子或不同的历史时间段进行交叉验证。关注回撤、夏普比率等风险调整后指标而不只是看总收益。永远假设未来行情可能与历史不同策略需要具备鲁棒性。7. 总结与学习路线本文从一个“在限额内交易的 AI agent首先运行在纸面交易环境”的项目出发完整讲解了交易 agent 的基本概念、系统组成、限额管理的重要性以及纸面交易引擎的实现方式。代码示例覆盖了行情生成、订单管理、风控检查、策略生成和主循环调度你可以直接复制下来运行再逐步替换成自己的策略和数据源。下一步你可以按以下路线继续深入学习更真实的回测框架比如 backtrader、vectorbt 或自研回测模块。深入研究风控体系包括止损、止盈、仓位管理和组合风险。接触券商 API了解真实下单接口和权限控制方式。尝试接入大模型用小样本验证 LLM 在交易决策中的可行性。关注开源社区里类似的 agent 项目重点研究它们的限额设计和异常处理方式。如果你准备做一个自己的交易 agent建议给自己设定一个纪律先跑通纸面交易再谈真实资金先做好限额熔断再考虑策略收益。技术能力决定了 agent 能飞多高而限额设计决定了它摔下来时有多痛。希望这篇文章对你有帮助欢迎动手实践也欢迎在留言区分享你的问题和改进思路。