OpenClaw接入A股数据：从零搭建AI盯盘监控系统

说实话，在看到“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 apscheduler

OpenClaw 的模型接入一般通过环境变量配置。在项目根目录建一个.env文件：

OPENAI_API_KEY=sk-xxx OPENCLAW_MODEL=gpt-4o-mini

不同版本的 OpenClaw 在配置项上可能有细微差别，以你手里那份仓库的文档为准。装完之后先跑一个最简单的验证：让 OpenClaw 调用一个你定义的 echo 函数，能正常返回就说明工具调用链路通了。这一步不要跳过，因为后面所有功能都依赖“智能体能调函数”这条能力，尽早验证能帮你把问题和数据源、调度模块隔离开。

2.2 A 股数据源怎么选

说实话，A 股数据源这块的水比很多人想象中深。我用了小一周的时间把主流方案都试了一遍，列个表给你直接对比：

我在最终方案里选择了“主用 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_ascii=False) df = df[df["代码"].isin(stock_codes)] cols = ["代码", "名称", "最新价", "涨跌幅", "涨跌额", "成交量", "成交额", "振幅", "最高", "最低", "今开", "昨收"] cols = [c for c in cols if c in df.columns] # 防止字段缺失直接崩 return df[cols].to_json(orient="records", force_ascii=False)

然后把它注册进 OpenClaw：

from openclaw import Agent # 具体导入路径以你仓库版本为准 agent = Agent(model="gpt-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_ascii=False)}" ) 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(f"records/{today}.csv", index=False) 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(dt=None): 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(dt=None): 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(timezone="Asia/Shanghai") @sched.scheduled_job("cron", day_of_week="mon-fri", hour=9, minute=25) def pre_market_snapshot(): if is_trading_day(): run_pre_market_snapshot() # 开盘前拉一次数据，确认系统正常 @sched.scheduled_job("interval", minutes=10) 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_week="mon-fri", hour=14, minute=50) 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, json=payload)

这里有个容易踩的坑：如果你在钉钉机器人配置里勾了加签，请求 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_retries=3, base_delay=1.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 常见问题速查表

5.5 几点避坑心得

最后把老经验里的精华浓缩成几条，每一条都是真金白银换来的。

第一，AkShare 的字段名是字符串不是变量名，每次升级后先打印一行列名再跑主逻辑，能省掉大量默默出错的调试时间。第二，密钥管理一定要规范：API key、webhook 地址不要硬编码在代码文件里，统一放.env，并且确保它不提交到 git 仓库。第三，系统启动时要推送一条“监控系统已启动”的消息，否则你可能某天突然发现监控已经挂了两天，却一直没察觉。第四，保留原始数据的 CSV 留档，这些数据是以后调整阈值、回测策略、诊断问题的依据，不可再生，删了就没了。

从我实际运行这段时间的体会来说，这套系统的价值不在于“让 AI 替你看盘”这个听起来很酷的概念，而在于把“死规则的判断”和“活语义的解读”拆开了。涨跌幅超阈值、量比异常这种逻辑，交给 Python 判断又快又准，不可能算错；而“今天市场的情绪怎么样、资金在往哪些方向走”这种需要经验才能回答的问题，交给大模型总结，刚好用上它语义组织能力强的长处。两者分工明确，系统才稳定可靠。

最后再分享一个建议，同时也是我自己踩过坑之后才想明白的：不要一上来就追求全自动、多通道告警的完整形态。先跑通最小闭环——每天收盘拉数据、生成一份日报、推送到自己的手机；稳定运行一周之后，再加盘中监控和异动告警；再之后才考虑多标的股票池、历史回测、更复杂的规则引擎。数据源限流、智能体输出不稳定这些坑，都会在最小闭环阶段集中暴露，那时候代码规模小，处理成本最低。一步到位只会让你同时面对十个问题，连哪里出的错都找不到。

另外要提醒一句：这套技术方案仅用于自动化数据获取与信息整理，不构成任何投资建议。股市有风险，实际操作前请自行评估，量力而行。