说实话在看到“OpenClaw 接入 A 股数据”这个需求时我第一反应是终于有人把智能体和真实金融数据流串起来了。过去我们聊 AI Agent总停留在问答、写文案、跑个 demo但一旦要接真实世界的动态数据尤其是 A 股这种高频、强时效、规则复杂的市场很多框架就露怯了。OpenClaw 作为一套本地部署的智能体运行框架刚好补上了“能规划任务、能自动调工具”的短板。这篇文章就把我从零部署到 24 小时自动监控的完整实战记录放出来。包括环境怎么搭、数据源怎么选、工具函数怎么注册、盘中异动监控和收盘复盘怎么实现、定时调度和告警推送怎么配置最后还有我在实际运行中踩过的坑。适合有 Python 基础、想自己搭一套“AI 盯盘助手”的开发者也适合对智能体工程落地感兴趣的读者。我按实际操作顺序写尽量不废话你可以直接照着做。1. 为什么要把 OpenClaw 接进 A 股数据1.1 先搞清楚 OpenClaw 到底是什么OpenClaw 这个名字有点意思“open claw”可以理解成“伸出去的开放之爪”——它解决的问题正是让 AI 不再只是“嘴上说说”而是能真正把手伸到外部世界里干活。它本质上是一个本地部署的智能体运行框架核心能力是给大语言模型提供三样东西工具调用、任务编排、记忆管理。你给 OpenClaw 一个目标它会自动拆解成步骤判断什么时候该查数据、什么时候该算指标、什么时候该发消息而不是像普通聊天机器人那样只能坐在对话框里给你打一段字。举个直观的例子。你直接问大模型“今天白酒板块表现怎么样”它只能凭训练数据里的记忆给你一个模糊回答甚至可能是几个月前的旧信息。但如果你把 OpenClaw 接上行情数据接口它会先调用get_realtime_quotes这个工具函数拉取实时行情再基于真实数据做总结。这个“自动决策该调什么工具、拿着工具结果继续推理”的能力就是它能接 A 股数据的根本前提。1.2 人工盯盘到底痛在哪里我不否认盯盘有它的乐趣但从效率角度看人工盯盘有三个很难解决的问题。第一是时效性。A 股盘中就四个小时9:30 到 11:30、13:00 到 15:00真正的异动往往就发生在一两分钟之内。你手动刷新行情页面永远慢半拍就算设置了涨跌幅提醒盯着一堆自选股也容易看漏。第二是精力。盯盘是反人性的活长时间保持注意力高度集中人一定会疲劳。你可能上午还能认真看每笔异动下午就开始犯困漏掉了真正重要的信号。而且盘中不只是看价格涨跌还要看成交量放大、板块联动、大单进出这些维度同时跟踪人工几乎做不到。第三是沉淀。收盘之后的复盘需要把当天的高低点、成交量变化、板块轮动、个股异动这些信息串起来形成一份结构化记录。纯手工整理既慢又容易遗漏。等到月底回看你连上周三市场发生了什么都要翻半天。这套系统解决的恰好就是这三件事定时拉数解决时效性规则过滤解决精力问题自动生成报告解决沉淀问题。它不会替你做投资决策但能让你把注意力只放在“看报告”这一步而不是淹没在原始数据里。1.3 整套方案的长什么样我用最简单的方式描述一下这条流水线数据层免费数据源如 AkShare→ 工具层Python 函数封装数据获取和计算→ 智能体层OpenClaw 大模型做语义分析→ 输出层Markdown 日报 钉钉/企业微信机器人告警选型逻辑其实很朴素。个人项目不烧钱所以数据源选开源免费的智能体编排要可控所以选本地部署的 OpenClaw 而不是某平台的云服务告警推送要稳定所以选免费的企业 IM 机器人 webhook。整套方案里除了大模型 API 调用要一点点 token 费用其他全免费跑一台普通 Linux 服务器甚至树莓派都够。2. 部署落地OpenClaw 环境搭建与数据源选型2.1 从零初始化运行环境先把环境捋顺。我建议 Python 版本不低于 3.10因为 OpenClaw 和 AkShare 新版都依赖 3.10 的新语法特性用老版本会在依赖安装阶段就卡住。# 创建虚拟环境避免把系统 Python 搞乱 python3 -m venv openclaw-env source openclaw-env/bin/activate # 克隆 OpenClaw 仓库并安装依赖 git clone 你的OpenClaw仓库地址 openclaw cd openclaw pip install -r requirements.txt # 安装本项目额外需要的包 pip install akshare pandas requests apschedulerOpenClaw 的模型接入一般通过环境变量配置。在项目根目录建一个.env文件OPENAI_API_KEYsk-xxx OPENCLAW_MODELgpt-4o-mini不同版本的 OpenClaw 在配置项上可能有细微差别以你手里那份仓库的文档为准。装完之后先跑一个最简单的验证让 OpenClaw 调用一个你定义的 echo 函数能正常返回就说明工具调用链路通了。这一步不要跳过因为后面所有功能都依赖“智能体能调函数”这条能力尽早验证能帮你把问题和数据源、调度模块隔离开。2.2 A 股数据源怎么选说实话A 股数据源这块的水比很多人想象中深。我用了小一周的时间把主流方案都试了一遍列个表给你直接对比数据源费用模式数据覆盖面限流情况上手难度最适合场景AkShare免费开源沪深A股、指数、板块、期货、基金、宏观数据全市场快照接口有明显频率限制极低pip 安装即可个人实时监控、盘中异动Tushare注册获取 token积分制沪深股票、财务数据、指数分钟级数据有积分门槛积分越高单位时间调用次数越多中等需要财务数据做深度筛选baostock免费历史行情稳定覆盖完整实时数据更新一般低历史回测、策略研究行情网站的非官方接口免费实时性好延迟低完全无保障字段说变就变中追求极致实时性能接受不稳定我在最终方案里选择了“主用 AkShare 辅助 baostock”的组合。实时看盘和盘中监控用 AkShare因为它在免费源里覆盖最全、接口更新最活跃社区维护力度大遇到问题也好找到解决方案。历史数据在做回测或分析时用 baostock 补因为它的历史数据更稳定。为什么不选商业数据终端不是不好是个人项目没必要。你的监控频率是分钟级不需要毫秒级行情免费接口的延迟和精度完全够用。而且商业终端动辄一年几千上万的费用自己搭个小系统花那个冤枉钱干嘛。2.3 数据源初始化与首次拉取验证AkShare 的安装很简单但首次拉数据一定要做一次验证确认接口能通、字段和预期一致。因为 AkShare 底层爬的是公开行情页面接口字段偶尔会变先看一眼真实返回能省掉后面大量调试时间。import akshare as ak # 拉取沪市A股全市场实时快照 df_spot ak.stock_zh_a_spot_em() print(df_spot.columns.tolist())正常输出会包含这些字段序号、代码、名称、最新价、涨跌幅、涨跌额、成交量、成交额、振幅、最高、最低、今开、昨收、量比等。注意这些都是字符串列名不是变量名后面在代码里用df[代码]这种形式筛选时不要手滑写错。我还建议你顺手测一下指数接口df_index ak.stock_zh_index_spot_em(symbol沪深重要指数) print(df_index.head())这个接口后面做收盘复盘时要用提前验证能确认它没被限流、字段完整。第一次拉全市场快照如果遇到网络波动多试两次就好不是你的代码问题。3. 核心实现从拉取行情到智能体监控脚本3.1 把“拉数据”变成 OpenClaw 的工具这是整个接入过程中最关键的一步。OpenClaw 之所以能干活是因为它有一个工具注册机制你可以把 Python 函数注册成智能体可调用的工具大模型在推理时看到“用户想查行情”会自动匹配到对应工具调用它拿结果再基于结果继续回答。我的第一个工具函数是获取指定股票列表的实时行情import akshare as ak import json def get_realtime_quotes(stock_codes: list) - str: 获取A股指定股票代码的实时行情快照返回JSON字符串。 一次拉全市场快照在内存中过滤目标股票避免多次请求。 df ak.stock_zh_a_spot_em() if df is None or df.empty: return json.dumps({error: empty data}, ensure_asciiFalse) df df[df[代码].isin(stock_codes)] cols [代码, 名称, 最新价, 涨跌幅, 涨跌额, 成交量, 成交额, 振幅, 最高, 最低, 今开, 昨收] cols [c for c in cols if c in df.columns] # 防止字段缺失直接崩 return df[cols].to_json(orientrecords, force_asciiFalse)然后把它注册进 OpenClawfrom openclaw import Agent # 具体导入路径以你仓库版本为准 agent Agent(modelgpt-4o-mini) agent.register_tool(get_realtime_quotes, get_realtime_quotes)这里有两个设计细节值得说。第一为什么是全市场快照 内存过滤而不是循环单个股票去请求因为 AkShare 的全市场快照是一次接口调用性能损耗最小如果你循环 20 只股票各请求一次等于 20 次 http 往返既慢又容易被限流。数据量方面全市场五千多只股票的 DataFrame 也就几十兆内存个人脚本完全扛得住。第二为什么返回 JSON 字符串而不是 DataFrame 对象因为 OpenClaw 和大模型交互的媒介是文本JSON 是结构化文本里最通用、解析成本最低的格式。大模型看到 JSON 能直接提取字段做总结不需要额外写代码解析。3.2 实现一次完整的盘中监控流程工具函数能跑通之后接下来就是监控逻辑。我不建议让大模型来算“哪些股票异动”原因后面细说更靠谱的做法是用 Python 写死规则把筛选出来的异常股票列表交给大模型做语义解读。先写规则判断函数def monitor_stocks(stock_codes: list, threshold: float 2.0, volume_ratio: float 2.0) - list: 监控指定股票池 - 涨跌幅绝对值超过 threshold默认2%的标的 - 成交量超过上一轮 volume_ratio 倍默认2倍的标的 返回异动股票列表。 quotes get_realtime_quotes(stock_codes) data json.loads(quotes) pre_volumes load_last_volumes(stock_codes) # 从缓存里读取上一周期的成交量 events [] for item in data: pct float(item.get(涨跌幅, 0)) vol float(item.get(成交量, 0)) code item.get(代码, ) pre_vol pre_volumes.get(code, 0) if abs(pct) threshold: item[异动原因] 涨跌幅超阈值 events.append(item) elif pre_vol 0 and vol / pre_vol volume_ratio: item[异动原因] 成交量突增 events.append(item) return events这个函数里我特意没写太复杂量比的计算在真实场景里会更精细比如要用过去 5 日均量做分母、去掉集合竞价的前几分钟噪声。但逻辑骨架就是对的一一用规则把“异常”号筛选出来再交给下一步。筛选出异动股票之后构造 prompt 让 OpenClaw 生成速报prompt ( 你是A股盘中监控助手。以下是刚过滤出的异动股票列表请生成一份简明速报 按异动幅度从高到低排序说明每只股票的名称、代码、最新价、涨跌幅、成交量和异动原因。 不要自行计算或补充数据中没有的信息。\n\n f{json.dumps(events, ensure_asciiFalse)} ) report agent.chat(prompt) print(report)你可能会问既然 Python 都已经把异动股票筛出来了还要大模型做什么这里面的分工逻辑是Python 擅长确定性计算大模型擅长语义组织。规则筛选只能告诉你“这 5 只股票涨跌幅超过了 3%”但“这几只股票同时异动意味着什么、有没有板块效应、市场情绪偏向什么方向”就得靠大模型结合经验做定性解读。把计算交给代码、把表达交给模型系统才稳定。3.3 生成每日收盘复盘报告盘中监控是过程收盘复盘才是沉淀。我每天在 14:50 跑一次收盘数据拉取把全市场快照存成 CSV 留档同时生成一份 Markdown 复盘报告。import datetime def save_daily_snapshot(): 收盘前拉取全市场快照存储为CSV留档。 df ak.stock_zh_a_spot_em() today datetime.date.today().strftime(%Y%m%d) df.to_csv(frecords/{today}.csv, indexFalse) return df df save_daily_snapshot() # 统计关键指标 up_count (df[涨跌幅] 0).sum() down_count (df[涨跌幅] 0).sum() amount_total df[成交额].sum() / 1e8 # 亿元 top5 df.nlargest(5, 涨跌幅)[[代码, 名称, 涨跌幅]].to_dict(records) bottom5 df.nsmallest(5, 涨跌幅)[[代码, 名称, 涨跌幅]].to_dict(records)这些统计全部在 Python 里算完然后传给大模型让它基于这些确定的数字写复盘。Prompt 模板我固定成下面这样summary_prompt f 今天是{datetime.date.today()}以下是收盘后的市场概况 - 上涨家数{up_count}下跌家数{down_count} - 两市成交额{amount_total:.0f}亿 - 涨幅前5{top5} - 跌幅前5{bottom5} 请生成一份300字左右的收盘复盘包含大盘表现、多空信号、值得关注的方向。 请严格基于以上提供的数据进行总结不要自行计算或补充未提供的信息。 把大模型输出的 Markdown 写入reports/20250630.md一天的工作就闭环了。这样坚持跑一个月你会积累一份完整的每日复盘档案数据留档也在后续想量化分析什么都方便。4. 24 小时接管定时调度与告警通路4.1 交易日历判断比 cron 更关键的一步很多人一上来就写 cron 定时任务然后发现周六周日、节假日它也在跑拉一堆空数据回来还发告警烦得要死。A 股的特殊性在于交易日不是简单地“周一至周五”还有法定节假日和调休判断逻辑必须体现在代码里。我维护了一个简单的节假日集合import datetime # 按年初官方安排维护每年更新一次 HOLIDAYS { 2025-01-01, 2025-01-28, 2025-01-29, 2025-01-30, 2025-01-31, 2025-02-03, 2025-02-04, 2025-04-04, # ... 按当年具体安排补齐 } def is_trading_day(dtNone): dt dt or datetime.datetime.now() if dt.weekday() 5: return False if dt.strftime(%Y-%m-%d) in HOLIDAYS: return False return True另外还要判断是否在盘中的可操作时间窗口。A 股一天有两个交易时段中间有午休收盘后数据接口还有一段稳定期def in_market_window(dtNone): dt dt or datetime.datetime.now() t dt.time() return (datetime.time(9, 30) t datetime.time(11, 30)) or \ (datetime.time(13, 0) t datetime.time(15, 0))不要小看这两个函数它们是整个调度系统不出错的基石。没有它们定时任务早晚会把你搞崩溃。4.2 用 APScheduler 把任务挂起来调度框架我选了 APScheduler而不是系统自带的 crontab。原因很简单交易日判断、时间窗口判断都要写在 Python 里用 APScheduler 可以把调度逻辑和业务逻辑放在同一个进程统一管理、统一日志改规则时也不用去碰系统配置。直接看代码骨架from apscheduler.schedulers.blocking import BlockingScheduler sched BlockingScheduler(timezoneAsia/Shanghai) sched.scheduled_job(cron, day_of_weekmon-fri, hour9, minute25) def pre_market_snapshot(): if is_trading_day(): run_pre_market_snapshot() # 开盘前拉一次数据确认系统正常 sched.scheduled_job(interval, minutes10) def intraday_monitor(): if is_trading_day() and in_market_window(): events monitor_stocks(CODE_POOL) if events: send_alert(events) cache_volumes() # 缓存本轮成交量供下一轮量比判断 sched.scheduled_job(cron, day_of_weekmon-fri, hour14, minute50) def daily_review(): if is_trading_day(): generate_review() # 生成收盘复盘报告 sched.start()几个参数细节说一下。盘中监控的interval我设的是 10 分钟。这个频率对于个人监控已经足够因为一分钟级别的异动即使捕捉到人工也来不及反应10 分钟一次的采样反而能过滤掉大量盘中噪声。9:25 开盘前快照的作用是“系统自检”确认数据源通、程序活着有问题早发现早处理。复盘定在 14:50 而不是 15:00是因为行情接口在收盘后几分钟内数据才完全稳定14:50 的盘面已接近收盘拉到的数据差异很小还能避开收盘瞬间的接口拥堵。4.3 告警推送的多通道配置监控和调度都跑起来了最后拼图是告警推送。我优先接的是钉钉自定义机器人因为免费、配置简单、手机端推送及时。企业微信机器人和 Server酱原理一样都是 webhook POST你按需加就行。钉钉机器人的关键代码import requests import hashlib import hmac import base64 import time DINGTALK_WEBHOOK https://oapi.dingtalk.com/robot/send?access_token你的token def send_dingtalk(title: str, text: str, secret: str ): url DINGTALK_WEBHOOK if secret: # 钉钉加签模式时间戳 密钥做 HMAC-SHA256 签名 ts str(round(time.time() * 1000)) string_to_sign f{ts}\n{secret} hmac_code hmac.new( secret.encode(), string_to_sign.encode(), hashlib.sha256 ).digest() sign base64.b64encode(hmac_code).decode() url f{DINGTALK_WEBHOOK}timestamp{ts}sign{sign} payload { msgtype: markdown, markdown: {title: title, text: text}, } requests.post(url, jsonpayload)这里有个容易踩的坑如果你在钉钉机器人配置里勾了加签请求 URL 必须带上timestamp和sign两个参数否则返回sign not match。我第一次就漏了 timestamp排查了半天才发现是加签模式的问题。告警推送还有个体验问题就是“重复轰炸”。盘中一个标的 3 分钟内反复触发涨跌幅阈值如果每次都推你的手机基本就废了。加一个简单的节流逻辑last_alert_time {} def throttle_alert(code: str, seconds: int 600) - bool: 同一标的在 seconds 秒内只告警一次避免重复轰炸。 now time.time() if now - last_alert_time.get(code, 0) seconds: return False last_alert_time[code] now return True每次发告警前先过这道节流阀同一个标的 10 分钟内最多推一次信息量够用又不打扰人。5. 常见问题与排查技巧实录5.1 AkShare 频繁拉取被限流这是个人监控项目里遇到最多的问题。现象很典型连续拉几次全市场快照之后接口返回空 DataFrame或者直接抛超时、连接中断的异常。原因是免费接口有访问频率限制尤其是全市场快照这种“重量级”接口短时间内高频请求很容易被服务器限流。我的解决方案是三管齐下。第一请求之间加随机延迟避免固定间隔被识别成机器行为time.sleep(random.uniform(1, 3))。第二写一个重试装饰器失败后指数退避重试避免一次网络抖动直接导致任务挂掉import time import random from functools import wraps def retry(max_retries3, base_delay1.0): def decorator(func): wraps(func) def wrapper(*args, **kwargs): for i in range(max_retries): try: return func(*args, **kwargs) except Exception as e: if i max_retries - 1: raise delay base_delay * (2 ** i) random.uniform(0, 1) time.sleep(delay) return None return wrapper return decorator第三也是最关键的优化盘中监控不要每次实时拉全市场而是把行情数据缓存下来10 分钟一个周期。这样数据源请求量直接从“每分钟几十次”降到“每 10 分钟一次”限流问题基本消失。缓存方案用 Python 的字典就能实现不需要上 Redis。5.2 大模型“自作聪明”算错数我最早做收盘复盘时把“上涨家数多少、下跌家数多少”的计算也交给了大模型结果它对数字的把握非常不靠谱输出“大约 3000 家上涨”实际是 2100 家。原因很简单大模型本质是概率预测它擅长的是语义理解而不是精确算术。这个坑让我彻底确立了前面提到的分工原则所有精确计算一律在 Python 里完成大模型只能基于给定的数字做解读。同时我在 prompt 模板里固定加了一句“请严格基于以上提供的数据进行总结不要自行计算或补充未提供的信息”。这个教训也适用于盘中监控。比如你想判断“今天医药板块是不是整体走强”让大模型从原始数据里自己算板块平均值远不如在 Python 里先算好板块涨跌幅再让大模型解读这个数值。5.3 长时间运行后进程假死系统跑了几天后我发现定时任务偶尔不执行了日志停在某一行没有报错。排查下来的主要原因是长时间运行带来的资源泄漏requests 连接未释放、内存逐渐增长、某个接口调用卡住导致线程池耗尽。针对这个问题我从三个方面做了加固。第一把每轮任务体包裹在try/except里并加超时熔断任何单轮任务的异常都不能拖垮整个调度进程。第二日志用RotatingFileHandler做滚动避免日志文件无限膨胀把磁盘写满。第三用 systemd 或 supervisor 托管进程崩溃后自动拉起基本做到无人值守。如果你不想引入 systemd还有一个土办法写一个看门狗脚本每天凌晨 3 点重启一次监控进程。A 股凌晨没有交易重启完全无感但能把内存泄漏和连接数问题一次性清零。我现在的生产配置就是“APScheduler 调度 每天自动重启 启动后推送一条‘监控系统已上线’的消息”。5.4 常见问题速查表现象常见原因解决方案AkShare 返回空 DataFrame免费接口被限流加随机延迟、重试退避、降频拉取接口返回字段缺失AkShare 版本升级、字段名变化打印df.columns核对代码里对字段做存在性判断大模型算错数字LLM 不擅长精确算术统计计算放 Pythonprompt 禁止自行推导钉钉机器人推送报 sign not match加签模式下请求 URL 缺参数拼接timestamp和sign参数定时任务周六日也在跑缺少交易日判断用is_trading_day()做过滤进程跑几天后假死内存/连接数泄漏加超时熔断、日志滚动、每日重启守护告警轰炸同一标的反复触发规则加时间戳节流10 分钟内不重复推送5.5 几点避坑心得最后把老经验里的精华浓缩成几条每一条都是真金白银换来的。第一AkShare 的字段名是字符串不是变量名每次升级后先打印一行列名再跑主逻辑能省掉大量默默出错的调试时间。第二密钥管理一定要规范API key、webhook 地址不要硬编码在代码文件里统一放.env并且确保它不提交到 git 仓库。第三系统启动时要推送一条“监控系统已启动”的消息否则你可能某天突然发现监控已经挂了两天却一直没察觉。第四保留原始数据的 CSV 留档这些数据是以后调整阈值、回测策略、诊断问题的依据不可再生删了就没了。从我实际运行这段时间的体会来说这套系统的价值不在于“让 AI 替你看盘”这个听起来很酷的概念而在于把“死规则的判断”和“活语义的解读”拆开了。涨跌幅超阈值、量比异常这种逻辑交给 Python 判断又快又准不可能算错而“今天市场的情绪怎么样、资金在往哪些方向走”这种需要经验才能回答的问题交给大模型总结刚好用上它语义组织能力强的长处。两者分工明确系统才稳定可靠。最后再分享一个建议同时也是我自己踩过坑之后才想明白的不要一上来就追求全自动、多通道告警的完整形态。先跑通最小闭环——每天收盘拉数据、生成一份日报、推送到自己的手机稳定运行一周之后再加盘中监控和异动告警再之后才考虑多标的股票池、历史回测、更复杂的规则引擎。数据源限流、智能体输出不稳定这些坑都会在最小闭环阶段集中暴露那时候代码规模小处理成本最低。一步到位只会让你同时面对十个问题连哪里出的错都找不到。另外要提醒一句这套技术方案仅用于自动化数据获取与信息整理不构成任何投资建议。股市有风险实际操作前请自行评估量力而行。