★ FEATURED ARTICLE 做量化交易的人通常都有这种体会模拟盘跑得再顺心里也始终悬着一块石头毕竟模拟和实盘永远隔着一堵看不见的墙。我这个项目最开始就是想让一个 AI Agent 自己去看行情、自己决策、自己在模拟盘里下单同时把每一次买卖和理由同步到飞书像带了一个能实时汇报的交易助理。整套系统基于 OpenClaw 这个开源 Agent 执行框架加上飞书开放平台的自定义机器人应用我把从零到模拟盘跑通、再到为实盘做准备的全过程完整走了一遍踩了不少坑也总结了一套可以复用的搭建方法。这篇文章就是把这段经历原原本本记录下来适合对自动化交易系统、Agent 工具调用、飞书机器人开发感兴趣的开发者参考。先说明白这套系统本身不构成任何投资建议代码和参数完全用于技术验证。尤其是“实盘”部分我只讲对接思路和风控准备不会劝任何人把真钱交给 AI 去管。1. 项目设计思路这套架构到底在解决什么问题我先把你可能最关心的问题放最前面为什么偏偏用 OpenClaw 加飞书 Agent 这个组合而不是直接写一个 Python 定时脚本原因在于传统定时脚本解决的是“到点执行”而 AI 自主系统要解决的是“根据当前状态自主判断下一步做什么”。OpenClaw 的核心价值是把大语言模型变成一个可以调用外部工具的执行体你要做的不是把每一步都写死而是给它注册好工具、设定好目标让它自己决定在什么条件下调用买入、卖出、查询、通知这些动作。飞书 Agent 则解决人机协作的问题——Agent 在后台再怎么自主运行你也不可能一直盯着终端看把它接进飞书之后每个决策、每笔模拟成交、每日持仓汇总都会以消息形式推送到你的手机你只需要在群里做最终确认和干预。这套系统的整体架构可以拆成四层每一层都有明确边界交互层飞书机器人应用负责接收用户指令、推送交易报告和预警消息。决策层OpenClaw 运行时中的 Agent 核心负责调用策略模块、读取行情、判断买卖信号。撮合层自建的模拟交易引擎用虚拟资金模拟真实交易中的手续费、印花税、持仓成本和成交逻辑。数据层免费或低成本的行情数据接口提供 K 线、实时报价和账户快照。1.1 OpenClaw 在系统里扮演的角色OpenClaw 在这套系统里相当于“管事的那个大脑”但它本身不内置炒股知识它更像一个调度中枢。你要做的事情是把数据源、策略判断、模拟下单、消息推送这些能力做成一个个具名的工具函数注册进 OpenClaw 的工具列表里然后给 Agent 一段总目标 Prompt告诉它“你是一个模拟交易助手负责监控这几只标的遵循策略模块给出的信号执行操作并把关键动作汇报到飞书”。这种设计最直接的好处是逻辑边界非常清楚。行情获取、策略计算、下单撮合这些偏重逻辑的事情用传统代码实现而什么时候该看盘、遇到异常怎么处理、如何组织汇报语言这些偏重判断的事情交给大模型去完成。OpenClaw 跑起来之后每一次周期循环就是一轮标准的 Agent 工作流读取工具状态 → 调用工具获取数据 → 根据模型输出决定动作 → 执行动作 → 写日志 → 进入下一轮。我自己没有把 OpenClaw 魔改成什么很复杂的东西因为它的收益恰恰来自克制。工具只注册了四个核心函数查询行情、执行模拟买入、执行模拟卖出、推送飞书消息。Agent 能做的事再大最终也要落到这四个函数上。1.2 飞书 Agent 解决的人机协作问题把系统接入飞书是我觉得整个项目里影响体验最关键的一步。早期我用最简单的方式调试也就是把日志写到文件里等想起来再去看一眼。后来发现系统自主运行的时候决策理由、执行结果、账户权益这些信息如果不在第一时间推送到手边你根本没法判断它是不是在正常工作。飞书这边我有两种接入方式可以用两者的定位差别很明显自定义机器人应用走飞书开放平台注册一个企业自建应用拿到 App ID 和 App Secret通过接口主动发消息给群聊。能力更完整可以接收消息回调、支持复杂交互适合做正式的 Agent 交互窗口。群自定义机器人 Webhook在飞书群里添加一个自定义机器人提供一个 Webhook 地址用 HTTP POST 就能把消息推到群里。实现成本最低适合快速验证消息链路。第二个方案虽然简单但它只能发消息不能收消息对于“人工干预指令”这种需求支持有限。我在项目正式跑起来之后换成了第一种方案因为我还需要能在飞书群里直接发一句“暂停交易”或“清仓”让 Agent 响应这些指令。1.3 模拟盘与实盘的边界标题里写了“从零到实盘”但我必须把模拟和实盘这两个阶段说清楚因为它们对应的系统要求完全不同。模拟盘阶段你的账户资金是假的行情是真实的目标是验证策略逻辑、Agent 决策质量、消息链路稳定性和故障恢复能力。这个阶段可以放开手脚让 Agent 每天自动跑跑得越久暴露的问题越充分。实盘阶段资金变成真的这时候系统要求会发生质变接口稳定性、鉴权安全、交易额度限制、异常中断恢复、人工熔断机制每一个都是硬指标。当时我把这套系统连接到真实券商账户之前实际上在模拟盘上连续跑了几周确认了两件事一是 Agent 的每次操作都有完整可追溯的日志二是所有操作在极端行情下也不会出现明显超出风控参数的交易。即便这样实盘真正上线时我也只放了极小的一笔资金而且保留了随时人工接管的权利。把这套系统完整做完之后我最大的感受是架构本身不难难的是把每一个细节都摆到它该在的位置上。下面我先讲环境搭建和基础接入过程。2. 前期准备飞书机器人、OpenClaw 运行环境和行情源第一步不是写核心代码而是把地基打好。这一阶段我花了大约两天时间主要精力在飞书应用配置和行情接口的稳定性测试上。2.1 OpenClaw 运行环境搭建OpenClaw 是一个以 Python 为主的 Agent 执行框架。我是在一台云服务器上部署的配置不需要太高2 核 4G 内存足够跑起来因为真正消耗资源的大模型推理走的是 API 调用本地只负责逻辑编排和工具执行。环境准备工作包括安装 Python 3.10建议用虚拟环境隔离依赖。通过 Git Clone 拉取项目之后把环境变量文件拷贝一份在里面配置大模型 API 地址和模型参数。确认网络能正常访问行情接口和飞书开放平台接口这两者都是标准 HTTPS 请求没有特殊要求。用官方自带的 demo 跑一遍工具调用流程确认 Agent 能成功执行一次简单的函数调用。这一步其实不用搞得很复杂。OpenClaw 本身提供了工具注册的接口你只需要把自定义函数通过装饰器或者配置表挂上去就能被 Agent 看到。初次跑 demo 时我犯过一个低级错误把模型 API 的地址填成了局域网地址导致 Agent 一直处于等待响应的状态日志看起来像卡死实际是网络不通。2.2 创建飞书自定义机器人应用飞书这边的完整步骤我一步步记录一下因为配置过程中容易漏掉回调地址的校验。登录飞书开放平台进入开发者后台创建一个“企业自建应用”。在应用能力里开启“机器人”能力这样应用在群里表现得像一个普通机器人。在“凭证与基础信息”页面拿到 App ID 和 App Secret这两个值后面代码里要用。在“事件订阅”里配置请求地址选择监听消息事件。注意飞书会先发送一个 URL 验证请求你需要让服务端正确响应 challenge 字段否则配置保存不了。发布应用版本等待管理员审核通过。对于仅推送消息的场景其实拿到 App ID 和 App Secret 之后就能调接口了。核心逻辑是先用凭证换取一个tenant_access_token再用这个 token 调消息发送接口把消息以机器人身份发送到指定群聊。这里有个细节飞书的 token 有有效期一般是两小时所以代码里最好做一层缓存不要每次发消息都重新请求。我自己的实现里写了一个简单的飞书客户端把获取 token 和发送消息封装成两个函数token 快过期时才自动刷新。这样 Agent 在调用推送工具时不需要关心底层凭证的逻辑。2.3 行情数据接入免费源就能跑起来行情数据是整个系统决策的依据这一步我建议优先考虑免费开源数据源因为模拟盘阶段对行情实时性的要求并没有想象中那么高。我用的方案是 Python 的开源数据接口库通过它把 A 股的历史日线数据和实时行情都拿到本地。它支持沪深京三地股票也支持指数和基金基本覆盖了我们常见的需求。对于更专业的分钟级数据也可以申请一些数据平台的测试 token但免费额度通常有限做模拟盘够用了。行情模块只需要封装三个底层能力获取单只标的的最新价格获取单只标的最近一段时间的日 K 数据获取整个监控列表的行情快照。为了让 Agent 调用更顺手我把这三个能力封装成一个统一函数入参是股票代码返回一个包含最新价、涨跌幅、近五日收盘价的字典。Agent 只关心“这个标的价格是多少、趋势怎么样”这类语义化的问题数据获取的技术细节被完全屏蔽在函数内部。这里要注意免费数据接口在盘中请求频率过高容易被临时限制访问。我自己踩过这个坑Agent 循环间隔设得太短每分钟去拉一次全列表行情结果跑了一个多小时数据源就返回空数据了。后面把轮询间隔拉长到五分钟并为接口加了一组备用源切换逻辑是主源返回空数据或者连续超时的时候自动切换。3. 核心模块一个一个写模拟撮合、决策循环和消息推送地基搭好之后就可以开始把系统的五脏六腑填充起来了。这一部分是整个项目工作量最大的地方也是让我觉得最有收获的地方。3.1 模拟撮合引擎虚拟资金先跑起来模拟撮合引擎的核心职责就是让买卖操作在虚拟资金池里产生真实的成本和盈亏。它不需要跟真实券商系统对接但一定要尽量贴近实际交易的费用规则。我用一个 Python 类实现了最简版本的模拟券商from dataclasses import dataclass, field from typing import Dict dataclass class Position: symbol: str shares: int avg_cost: float dataclass class MockAccount: cash: float positions: Dict[str, Position] field(default_factorydict) trade_log: list field(default_factorylist) class MockBroker: def __init__(self, initial_cash: float 100000.0): # 初始资金默认10万元纯虚拟 self.account MockAccount(cashinitial_cash) def market_buy(self, symbol: str, shares: int, price: float): cost price * shares # 佣金按万2.5计算不足5元按5元收取 commission max(cost * 0.00025, 5) total_cost cost commission if self.account.cash total_cost: raise RuntimeError(f现金不足无法买入 {symbol}) self.account.cash - total_cost if symbol in self.account.positions: pos self.account.positions[symbol] new_total_shares pos.shares shares # 移动加权平均成本 pos.avg_cost (pos.avg_cost * pos.shares cost) / new_total_shares pos.shares new_total_shares else: self.account.positions[symbol] Position(symbol, shares, price) self.account.trade_log.append( (买入, symbol, shares, price, commission, 0) ) def market_sell(self, symbol: str, shares: int, price: float): pos self.account.positions.get(symbol) if not pos or pos.shares shares: raise RuntimeError(f持仓不足无法卖出 {symbol}) proceeds price * shares commission max(proceeds * 0.00025, 5) # 印花税按卖出金额的千分之0.5计算 stamp_tax proceeds * 0.0005 net_proceeds proceeds - commission - stamp_tax self.account.cash net_proceeds pos.shares - shares if pos.shares 0: del self.account.positions[symbol] self.account.trade_log.append( (卖出, symbol, shares, price, commission, stamp_tax) ) def total_assets(self, price_map: Dict[str, float]) - float: stock_value sum( pos.shares * price_map.get(symbol, pos.avg_cost) for symbol, pos in self.account.positions.items() ) return self.account.cash stock_value这个引擎把几个关键规则都覆盖了买入时计算佣金卖出时额外计算印花税持仓成本用移动加权平均法计算账户权益等于现金加持仓市值。模拟盘的成本结构和真实交易已经非常接近得出的收益曲线也更有参考价值。在真实代码里我会把交易日志同时写入一个 CSV 文件保证 Agent 崩溃重启后还能通过日志恢复账户状态。这是模拟盘阶段容易被忽略、实盘阶段却非常致命的问题。3.2 决策模块规则策略怎么接入 Agent决策模块是整个系统里最符合“策略”二字的部分。我没有在一开始就设计很复杂的模型而是先用一个基于均线关系的规则策略跑通全链路。原因很简单复杂模型出错了你很难判断是策略问题还是系统问题简单策略则可以让所有异常都暴露在可解释的逻辑里。这个策略的逻辑非常朴素对每个监控标的计算近 5 日均线和近 20 日均线当 5 日均线上穿 20 日均线时产生买入信号当 5 日均线下穿 20 日均线时产生卖出信号其他情况保持当前状态不变。def make_decision(kline_list, position_shares): kline_list: 按时间升序排列的收盘价列表 position_shares: 当前持仓股数 if len(kline_list) 20: return hold, 数据不足 closes [k[close] for k in kline_list] short_ma sum(closes[-5:]) / 5 long_ma sum(closes[-20:]) / 20 if short_ma long_ma and position_shares 0: return buy, f5日均线{short_ma:.2f}上穿20日均线{long_ma:.2f} if short_ma long_ma and position_shares 0: return sell, f5日均线{short_ma:.2f}下穿20日均线{long_ma:.2f} return hold, f均线状态无变化short{short_ma:.2f}, long{long_ma:.2f}为什么要把决策逻辑放到一个普通函数里而不是直接让大模型自己算均线因为大模型不擅长精确数值计算让它算均线既慢又容易出错。正确的分工是数值计算交给代码判断和调度交给模型。Agent 的职责是调用 make_decision 获取信号然后把买入或卖出信号翻译成一次 mock_buy 或 mock_sell 调用。这个经验是项目中的一个重要转折点。最初我试着让 Agent 自己做技术指标计算和信号判断结果十次里有三次方向是反的而且给出错误信号后它还会一本正经地生成一段解释显得自己特别有道理。后来我把所有需要精确计算的部分全部下沉为工具函数Agent 只做动作选择整个系统的可靠性一下子提升了一个台阶。3.3 OpenClaw 任务循环让它自己跑起来OpenClaw 的任务循环是这套系统的发动机。为了让 Agent 真正“自主”我们得给它注册好工具再设定一个大目标让它在这个目标框架内循环决策。工具注册的关键代码大致长这样# 在 OpenClaw 中注册自定义工具 def register_stock_tools(claw): claw.register_tool( nameget_quote, description获取指定股票的实时价格和近期K线数据, parameters{ symbol: {type: string, description: 股票代码例如 600000} }, funcget_quote, ) claw.register_tool( namemock_buy, description在模拟账户中买入指定股票传入股票代码和股数, parameters{ symbol: {type: string}, shares: {type: integer, description: 买入股数必须是100的整数倍}, }, funcmock_buy, ) claw.register_tool( namemock_sell, description在模拟账户中卖出指定股票传入股票代码和股数, parameters{ symbol: {type: string}, shares: {type: integer, description: 卖出股数不能超过当前持仓}, }, funcmock_sell, ) claw.register_tool( namepush_feishu, description推送文本消息到飞书群聊, parameters{ text: {type: string, description: 要推送的文本内容}, }, funcpush_feishu, )注册好之后给 Agent 设置一个总目标 Prompt。我用的是这个方向的描述“你是一个自主模拟炒股助手。你拥有查询行情、模拟买入、模拟卖出、推送飞书消息四个工具。你的任务是每隔一段时间检查监控列表中的标的根据工具返回的数据调用决策函数判断买卖信号。每次执行买卖后必须调用推送飞书消息工具把操作内容和原因发送到群里。如果账户现金不足或者持仓不足不要重试超过两次直接把异常信息推送至飞书。”这个 Prompt 有三个关键设计明确告诉 Agent 有哪些工具可用避免它自己臆造不存在的函数。明确要求它“每次执行买卖后必须推送消息”用机制保证可观测性。明确异常处理方式避免 Agent 陷入无意义的死循环。OpenClaw 的调度频率我是通过外部定时任务控制的比如每隔 10 分钟触发一次 Agent 运行。这是因为大模型 API 调用本身有延迟和费用你不可能让 Agent 像高频交易系统一样每秒跑一次也没有必要。3.4 飞书消息推送的实现飞书消息推送这块我前面提到有两个方案正式版我采用了自定义机器人应用。核心代码其实就是两步换 token 和发消息import requests import time class FeishuClient: def __init__(self, app_id: str, app_secret: str): self.app_id app_id self.app_secret app_secret self.token None self.token_expire_at 0 def _get_token(self) - str: # token 两小时有效在有效期内直接复用 if self.token and time.time() self.token_expire_at: return self.token resp requests.post( https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal, json{app_id: self.app_id, app_secret: self.app_secret}, timeout10, ) data resp.json() self.token data[tenant_access_token] self.token_expire_at time.time() data.get(expire, 7200) - 60 return self.token def send_text(self, chat_id: str, text: str): token self._get_token() resp requests.post( fhttps://open.feishu.cn/open-apis/im/v1/messages?receive_id_typechat_id, headers{Authorization: fBearer {token}}, json{ receive_id: chat_id, msg_type: text, content: json.dumps({text: text}), }, timeout10, ) return resp.json()这里有一个飞书 API 的细节content 字段必须是一个字符串化的 JSON直接传字典会被接口拒绝。我第一次接入时在这里卡了快二十分钟一直报参数错误后来仔细查文档才发现这个坑。把推送功能接好后我让 Agent 在消息里用固定格式输出比如【模拟交易】买入 600000 浦发银行 200股 成交价格10.25 触发原因5日均线上穿20日均线 账户剩余现金98620.50固定格式好处非常多它让历史记录更容易追溯而且你扫一眼飞书消息就能判断 Agent 行为是否正常不需要逐行翻日志。4. 从模拟盘到实盘我踩过的坑和过渡清单模拟盘和实盘之间的距离比大多数人想象中要远得多。模拟盘里跑出的漂亮收益曲线换成实盘后可能会因为各种微小差异而面目全非。我把自己一路从模拟切到实盘的过程整理成下面几个要点。4.1 模拟盘验证的重点指标模拟盘阶段最重要的事情不是看收益有多高而是验证系统在长时间运行下能不能保持稳定。我给自己定了四个核心验证指标连续运行时长系统是否能连续运行一两周不崩溃、不卡死。Agent 框架、数据源接口、消息推送任何一个环节掉链子都会中断自主交易。操作可追溯性每一笔模拟交易的触发原因、成交价格、费用明细是否完整记录。不可追溯的操作等于没有发生。信号执行一致性Agent 是否严格按决策模块的信号执行有没有出现信号是买入、结果却变成卖出的情况。异常自恢复能力数据源超时、接口限流、飞书 token 过期这类常见故障发生后系统能否自动恢复并在下个周期继续正常工作。模拟盘阶段如果把这四个指标练扎实了切换实盘时心里会稍微有底一些。4.2 实盘接入的注意事项实盘接入并不是简单地换一个真实账户接口就完事了。它涉及券商选择、接口权限申请、资金划转、风控参数配置等一系列环节。首先任何实盘操作都必须通过券商官方提供的量化交易接口或者合法合规的交易软件进行。个人开发者对接实盘时优先找支持这种接入方式的券商开户的时候直接咨询量化接口的开通条件。我这边最终用的是一个支持 Python 接口的量化柜台申请通过后券商提供一个独立的资金账号这个账号和你平时用的交易账号隔离可以单独设置交易权限和资金上限。其次安全方面有一个容易忽视的细节API 密钥和账号信息绝对不能写在代码里或者提交到代码仓库。我自己专门在服务器上配置了一个环境变量文件权限设置为只有运行用户可读程序启动时从环境变量加载账号信息。实盘下单前还有一个必须做的动作在券商的客户端里手动设置风控参数包括单笔最大委托金额、单日最大委托次数、最大持仓数量等。这些限制是资金安全的最后一道防线万一 Agent 或代码出现逻辑失控券商的系统会自动拦截异常委托。4.3 风控清单我把实盘阶段必须做到的风控项做成了一份清单每次调整系统后我都会对着清单过一遍资金额度实盘账户只放总资金的一小部分比例设定到即使全部亏完也不会影响生活的水平。人工熔断开关在飞书群里预留一个指令比如“紧急暂停”Agent 收到后立即停止所有后续操作。这个功能对 Agent 来说是一种安全绳。操作限额在模拟撮合引擎之外再加一层实盘专用的参数校验任何单笔金额超过设定上限的委托都会直接拒绝。盯盘阈值设置一个最大单日亏损百分比触发后自动停止当日后续交易。账号隔离实盘账户和测试账户的凭据严格分开避免误操作把测试代码发到实盘。日志备份每一笔实盘委托和回报都同步写入数据库且每 5 分钟做一次增量备份。这个清单不是万能的但它能把最坏情况造成的损失控制在可接受范围内。5. 常见问题与排查思路实录整个项目开发和运行期间我记录了一批典型问题。我把它们整理成一份问题速查表形式方便你直接对照排查。现象可能原因排查步骤与解决方案Agent 长时间无响应大模型 API 网络不通或配置错误先手动调用一次模型接口确认连通性再查看 Agent 日志中最后一次请求的状态码飞书机器人发不出消息token 过期或 chat_id 错误检查 token 刷新逻辑用飞书 Open API 调试工具手工发一条消息定位问题行情数据突然为空免费数据源被限流或网络异常检查接口返回的字段切换备用数据源并适当降低请求频率Agent 重复执行同一买入操作状态记录缺失导致重复决策检查账户状态是否持久化确保 Agent 每次读到的都是最新持仓模拟账户现金变成负数买入前未检查可用资金在 mock_buy 里增加资金校验并在异常时通过推送通知人工处理实盘委托被券商拒绝超出券商设置的风控阈值查看券商客户端内的委托拒单原因调整 Agent 参数或券商风控配置还有几个排查经验比较琐碎但对效率影响很大。第一日志必须带时间戳和工具名。Agent 执行链路长没有时间戳的日志在排障时几乎是废的。我自己后来在 OpenClaw 里给每个工具调用都包了一层装饰器自动记录调用时间、入参、出参和耗时。第二异常不要静默吞掉。最初我在一些接口调用上用了 try except 加 pass结果系统看起来一切正常实际上已经好几天没有执行任何交易了。正确的做法是任何异常都要推送飞书即使不影响主流程也要让你知道系统发生了异常。第三模拟盘数据输出格式要稳定。行情接口偶尔会返回缺失字段如果程序直接抛异常或输出格式乱七八糟Agent 的决策质量会急剧劣化。解决方案是写一个统一的数据清洗函数所有数据源返回后先过清洗逻辑再交给 Agent。第四升级代码前一定要备份当前状态。模拟盘阶段还好实盘阶段任何一个代码变更都要先在测试环境完整跑一遍确认没有副作用后再部署到实盘。我在实盘系统上线前的最后一次演练是把所有异常情况都主动触发了一遍断掉行情接口、伪造飞书 token 过期、给模拟引擎造资金不足的假数据然后记录了系统每个环节的反应。这套故障演练方法比修好任何一个具体 bug 都更有价值因为它能验证整个系统的韧性。最后再分享一点个人体会。做 AI 自主交易系统最容易踩的坑是过分崇拜 Agent 的能力。大模型确实擅长理解和生成但它完全不擅长精确计算和纪律执行而交易系统偏偏对精确和纪律要求极高。把这两者的优势结合起来让模型负责判断和交互让代码负责计算和执行这套系统才能走得远。我现在每天还是会翻一遍飞书群里的交易记录看看 Agent 的行为是否合乎预期。这不是不信任它而是作为系统的负责人你永远需要保留对系统行为的最终解释权和掌控权。
阅读完成 · 觉得有帮助？
