Tushare量化数据 + AI 智能体 + OpenCLAW：构建全渠道量化数据系统

Tushare + AI 智能体 + OpenCLAW：构建全渠道量化数据系统

在 AI Agent 火爆的 2026 年，量化数据获取方式正在发生变革。这篇文章展示如何将 Tushare、AI 智能体与最新的 OpenCLAW 框架结合，实现自然语言查询、多通道推送、自动化数据采集等功能。

技术演进

第一代：API 调用（2020-2023）

df = pro.daily(ts_code='000001.SZ', start_date='20230101', end_date='20231231')

第二代：AI Agent（2024-2025）

agent.chat("帮我查一下贵州茅台 2023 年的数据")

第三代：OpenCLAW 多通道（2026）

# 在飞书/钉钉/Telegram 中直接发送：
"贵州茅台最近走势如何？"
# 自动获取数据、分析、推送报告

方案对比

选择：LangChain + OpenCLAW 组合，前者负责智能分析，后者负责多通道推送。

OpenCLAW 简介

什么是 OpenCLAW？

OpenCLAW 是 2025-2026 年最火的开源 AI 智能体框架，GitHub 星标突破 18.8 万。它是一个自托管、多通道的 AI 网关，核心能力：

• 23+ 消息平台统一接入：飞书、钉钉、企业微信、WhatsApp、Telegram、Discord 等
• 7×24 小时后台运行：守护进程模式
• 长期记忆系统：向量搜索 + Markdown 持久化
• 任务执行能力：操作终端、浏览器、文件系统
• 插件扩展：丰富的技能生态

为什么需要 OpenCLAW？

传统量化系统的痛点：

OpenCLAW 的解决方案：

• 在常用聊天工具中直接查询
• 定时自动推送日报/周报
• 异常数据实时告警
• 自然语言交互，零学习成本

技术架构

┌─────────────────────────────────────────┐
│ 用户（自然语言输入） │
│ 飞书/钉钉/Telegram/企业微信 │
└────────────────┬────────────────────────┘
│ WebSocket
↓
┌─────────────────────────────────────────┐
│ OpenCLAW Gateway │
│ – 消息路由 │
│ – 插件管理 │
│ – 记忆系统 │
└────────────────┬────────────────────────┘
│
┌────────┴────────┐
↓ ↓
┌──────────────┐ ┌──────────────┐
│ Tushare │ │ LangChain │
│ 数据获取 │ │ 智能分析 │
└──────────────┘ └──────────────┘
↓ ↓
┌──────────────┐ ┌──────────────┐
│ 原始数据 │ │ 分析报告 │
└──────────────┘ └──────────────┘
│
↓
┌─────────────────────────────────────────┐
│ OpenCLAW 推送 │
│ 格式化报告 → 用户聊天界面 │
└─────────────────────────────────────────┘

环境准备

# 基础依赖
pip install tushare pandas numpy sqlalchemy

# LangChain 相关
pip install langchain langchain-openai chromadb

# OpenCLAW 安装（2026 年 3 月最新版）
# 方式 1：Docker（推荐）
docker pull openclaw/openclaw:latest
docker run -d -p 18789:18789 openclaw/openclaw

# 方式 2：源码安装
git clone https://github.com/openclaw/openclaw.git
cd openclaw
npm install
npm start

# 国产大模型（可选）
pip install zhipuai # 智谱 AI
pip install dashscope # 通义千问

核心实现

Tushare 客户端封装

import tushare as ts
import pandas as pd
from typing import Optional, Dict, Any
from dataclasses import dataclass

@dataclass
class StockData:
"""股票数据结构"""
ts_code: str
name: str
data: pd.DataFrame
metadata: Dict[str, Any]

class TushareClient:
"""支持 OpenCLAW 调用的 Tushare 客户端"""

def __init__(self, token: str):
ts.set_token(token)
self.pro = ts.pro_api()
self._cache = {}

def get_stock_list(self) –> pd.DataFrame:
"""获取股票列表"""
return self.pro.stock_basic(
exchange='',
list_status='L',
fields='ts_code,symbol,name,industry,list_date'
)

def get_daily_data(
self,
ts_code: str,
start_date: str,
end_date: str
) –> Optional[pd.DataFrame]:
"""获取日线数据"""
cache_key = f"{ts_code}_{start_date}_{end_date}"

if cache_key in self._cache:
return self._cache[cache_key]

try:
df = self.pro.daily(ts_code=ts_code, start_date=start_date, end_date=end_date)
df['trade_date'] = pd.to_datetime(df['trade_date'])
df = df.sort_values('trade_date').reset_index(drop=True)
self._cache[cache_key] = df
return df
except Exception as e:
print(f"Failed to fetch {ts_code}: {e}")
return None

def query_by_name(self, name: str) –> Optional[str]:
"""根据股票名称查找代码"""
stocks = self.get_stock_list()
match = stocks[stocks['name'].str.contains(name, case=False)]
if not match.empty:
return match.iloc[0]['ts_code']
return None

def calculate_statistics(self, df: pd.DataFrame) –> Dict[str, float]:
"""计算统计数据"""
if df is None or df.empty:
return {}

return {
'count': len(df),
'avg_close': float(df['close'].mean()),
'max_close': float(df['close'].max()),
'min_close': float(df['close'].min()),
'price_change': float((df['close'].iloc[–1] / df['close'].iloc[0] – 1) * 100),
'volatility': float(df['pct_chg'].std() * (252 ** 0.5)) if 'pct_chg' in df.columns else None
}

def generate_daily_report(self, ts_code: str, days: int = 5) –> str:
"""
生成日报摘要

OpenCLAW 专用：返回格式化的文本报告
"""
from datetime import datetime, timedelta

end_date = datetime.now()
start_date = end_date – timedelta(days=days*7) # 考虑周末

df = self.get_daily_data(
ts_code,
start_date.strftime('%Y%m%d'),
end_date.strftime('%Y%m%d')
)

if df is None or df.empty:
return "数据获取失败"

stats = self.calculate_statistics(df)
latest = df.iloc[–1]

report = f"""
📊 **股票日报**
股票代码：{ts_code}
报告日期：{end_date.strftime('%Y-%m-%d')}

📈 **最新数据**（{latest['trade_date'].strftime('%Y-%m-%d')}）
– 收盘价：¥{latest['close']:.2f}
– 涨跌幅：{latest['pct_chg']:.2f}%
– 成交量：{latest['vol']/10000:.1f}万手
– 成交额：{latest['amount']/10000:.1f}万元

📊 **统计指标**（近{days}日）
– 平均收盘价：¥{stats['avg_close']:.2f}
– 最高价：¥{stats['max_close']:.2f}
– 最低价：¥{stats['min_close']:.2f}
– 区间涨跌幅：{stats['price_change']:.2f}%
– 波动率：{stats['volatility']:.2f}%

💡 **简评**
{self._generate_comment(latest, stats)}
"""

return report

def _generate_comment(self, latest: pd.Series, stats: Dict) –> str:
"""生成简评"""
pct = latest['pct_chg']
if pct > 5:
return "近期表现强势，涨幅较大，注意回调风险"
elif pct > 0:
return "走势稳健，保持关注"
elif pct > –5:
return "小幅调整，属正常波动"
else:
return "近期表现较弱，注意风险控制"

LangChain Agent 集成

from langchain.agents import initialize_agent, Tool
from langchain.llms import OpenAI
from langchain.memory import ConversationBufferMemory
import os

class TushareAgent:
"""Tushare 数据 Agent"""

def __init__(self, token: str, openai_api_key: str = None):
self.client = TushareClient(token)

# 初始化 LLM
if openai_api_key:
os.environ['OPENAI_API_KEY'] = openai_api_key
self.llm = OpenAI(temperature=0, model='gpt-3.5-turbo-instruct')
else:
from langchain.llms import ZhipuAI
self.llm = ZhipuAI(temperature=0)

self.tools = self._create_tools()
self.agent = self._create_agent()

def _create_tools(self):
"""创建工具函数"""
return [
Tool(
name="查询股票数据",
func=self._query_stock_data,
description="查询某支股票的日线数据，参数格式：ts_code,start_date,end_date"
),
Tool(
name="根据名称查代码",
func=lambda name: str(self.client.query_by_name(name)),
description="根据股票名称查找股票代码"
),
Tool(
name="生成日报",
func=lambda ts_code: self.client.generate_daily_report(ts_code, days=5),
description="生成股票近 5 日的分析报告"
)
]

def _query_stock_data(self, args: str) –> str:
"""查询股票数据"""
try:
ts_code, start_date, end_date = args.split(',')
df = self.client.get_daily_data(ts_code.strip(), start_date.strip(), end_date.strip())

if df is None:
return "数据获取失败"

stats = self.client.calculate_statistics(df)
return f"""股票代码：{ts_code}
数据点数：{stats['count']}
平均收盘价：{stats['avg_close']:.2f}
最高价：{stats['max_close']:.2f}
最低价：{stats['min_close']:.2f}
区间涨跌幅：{stats['price_change']:.2f}%
波动率：{stats['volatility']:.2f}%

最近 5 天：
{df.tail().to_string()}
"""
except Exception as e:
return f"查询失败：{e}"

def _create_agent(self):
"""创建 Agent"""
return initialize_agent(
tools=self.tools,
llm=self.llm,
agent="chat-conversational-react-description",
memory=ConversationBufferMemory(memory_key="chat_history", return_messages=True),
verbose=False # OpenCLAW 中关闭详细输出
)

def chat(self, query: str) –> str:
"""与 Agent 对话"""
return self.agent.run(query)

OpenCLAW 插件配置

创建 OpenCLAW 插件文件 tushare_plugin.py：

# plugins/tushare_plugin.py
from openclaw.plugin import BasePlugin
from typing import Dict, Any

class TusharePlugin(BasePlugin):
"""Tushare 数据查询插件"""

def __init__(self):
super().__init__()
self.client = TushareClient(token='your_tushare_token')
self.agent = TushareAgent(token='your_tushare_token')

async def handle_message(self, message: Dict[str, Any]) –> Dict[str, Any]:
"""
处理消息

支持的命令：
– 查询 XXX 股票
– XXX 走势如何
– 生成 XXX 日报
– 推荐一些股票
"""
text = message.get('text', '').strip()

# 关键词匹配
if any(kw in text for kw in ['查询', '走势', '日报', '数据']):
# 提取股票代码（简化版，实际需要用 NLP）
stock_name = self._extract_stock_name(text)
if stock_name:
ts_code = self.client.query_by_name(stock_name)
if ts_code:
report = self.client.generate_daily_report(ts_code)
return {
'status': 'success',
'response': report
}

# 使用 Agent 处理复杂查询
response = self.agent.chat(text)
return {
'status': 'success',
'response': response
}

def _extract_stock_name(self, text: str) –> str:
"""提取股票名称（简化版）"""
# 实际项目中应该用 NLP 模型
keywords = ['贵州茅台', '平安银行', '五粮液', '宁德时代']
for kw in keywords:
if kw in text:
return kw
return None

# 注册插件
plugin = TusharePlugin()

OpenCLAW 配置文件

创建 config.yaml：

# OpenCLAW 配置文件
gateway:
port: 18789
host: 0.0.0.0

channels:
# 飞书配置
feishu:
enabled: true
app_id: "your_app_id"
app_secret: "your_app_secret"

# 钉钉配置
dingtalk:
enabled: true
app_key: "your_app_key"
app_secret: "your_app_secret"

# 企业微信配置
wechat_work:
enabled: true
corp_id: "your_corp_id"
agent_id: "your_agent_id"
secret: "your_secret"

plugins:
– name: tushare
path: ./plugins/tushare_plugin.py
enabled: true

ai:
provider: zhipu # 或 openai
api_key: "your_api_key"
model: "glm-4" # 或 gpt-3.5-turbo

memory:
enabled: true
type: chromadb
persist_directory: ./data/memory

使用场景

场景 1：飞书/钉钉查询

你（在飞书中）：贵州茅台最近走势如何？

OpenCLAW + Tushare Agent：
📊 **股票日报**
股票代码：600519.SH
报告日期：2026-03-09

📈 **最新数据**（2026-03-08）
– 收盘价：¥1750.00
– 涨跌幅：+1.25%
– 成交量：15.2 万手
– 成交额：26.5 亿元

📊 **统计指标**（近 5 日）
– 平均收盘价：¥1735.50
– 最高价：¥1760.00
– 最低价：¥1720.00
– 区间涨跌幅：+2.15%
– 波动率：25.30%

💡 **简评**
走势稳健，保持关注

场景 2：定时推送日报

配置 OpenCLAW 定时任务：

# schedulers/daily_report.py
from openclaw.scheduler import Schedule
from datetime import datetime

class DailyReportSchedule(Schedule):
"""每日收盘后推送日报"""

def __init__(self):
super().__init__(
name="daily_report",
cron="0 16 * * 1-5" # 工作日 16:00
)
self.client = TushareClient('your_token')

async def execute(self):
"""执行定时任务"""
# 关注的股票列表
watchlist = ['600519.SH', '000001.SZ', '000858.SZ']

for ts_code in watchlist:
report = self.client.generate_daily_report(ts_code)
# 推送到指定群组
await self.send_to_group("量化交流群", report)

schedule = DailyReportSchedule()

场景 3：异常告警

# monitors/price_alert.py
from openclaw.monitor import BaseMonitor

class PriceAlertMonitor(BaseMonitor):
"""价格异动监控"""

def __init__(self):
super().__init__(
name="price_alert",
interval=300 # 5 分钟检查一次
)
self.client = TushareClient('your_token')
self.threshold = 5.0 # 涨跌幅超过 5% 告警

async def check(self):
"""检查价格异动"""
watchlist = ['600519.SH', '000001.SZ']

for ts_code in watchlist:
# 获取实时数据（需要分钟线接口）
data = self.client.get_realtime_quote(ts_code)

if abs(data['pct_chg']) > self.threshold:
alert_msg = f"""
🚨 **价格异动告警**
股票：{ts_code}
当前涨跌幅：{data['pct_chg']:.2f}%
时间：{datetime.now().strftime('%H:%M:%S')}

请密切关注！
"""
await self.send_alert(alert_msg)

monitor = PriceAlertMonitor()

场景 4：多平台同步

# config.yaml – 多平台配置
channels:
feishu:
enabled: true
groups:
– "量化交流群"
– "投研小组"

dingtalk:
enabled: true
groups:
– "投资部"

telegram:
enabled: true
channels:
– "@quant_channel"

# 同一消息同步到所有平台
broadcast:
enabled: true
channels:
– feishu
– dingtalk
– telegram

完整工作流示例

步骤 1：启动 OpenCLAW

# 使用 Docker
docker run -d \\
-p 18789:18789 \\
-v $(pwd)/config.yaml:/app/config.yaml \\
-v $(pwd)/plugins:/app/plugins \\
-v $(pwd)/data:/app/data \\
openclaw/openclaw:latest

# 查看日志
docker logs -f openclaw

步骤 2：在飞书中添加机器人

步骤 3：开始使用

在飞书群中直接发送：

"贵州茅台今天怎么样？"
"生成一份五粮液的日报"
"推荐一些低 PE 的银行股"

优势总结

相比传统方案

下一步

参考：

• Tushare 文档：https://tushare.pro/document/2
• OpenCLAW GitHub：https://github.com/openclaw/openclaw
• OpenCLAW 文档：https://docs.openclaw.com
• LangChain 文档：https://python.langchain.com

说明：OpenCLAW 是 2025-2026 年新兴的开源项目，更新较快，具体配置请以官方文档为准。