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 免费 中低 很高 被动推送 ⭐⭐⭐⭐⭐
AutoGen 免费 高 很高 主动查询 ⭐⭐⭐⭐
自研 Agent 免费 高 最高 可控 ⭐⭐⭐

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

OpenCLAW 简介

什么是 OpenCLAW？

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

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

为什么需要 OpenCLAW？

传统量化系统的痛点：

  1. 需要主动打开软件查询
  2. 数据推送不及时
  3. 多平台切换麻烦
  4. 无法在日常聊天中获取数据

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：在飞书中添加机器人

  1. 登录飞书开放平台
  2. 创建企业自建应用
  3. 配置机器人
  4. 填写 Webhook URL：ws://your-server:18789/feishu
  5. 添加到群组

步骤 3：开始使用

在飞书群中直接发送：

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

优势总结

相比传统方案

维度 传统量化软件 Tushare + OpenCLAW
交互方式 打开软件→查询 聊天中直接问
推送及时性 手动查看 实时推送
学习成本 需要学习软件 会聊天就会用
集成度 单一平台 23+ 平台统一
扩展性 受限于软件 插件无限扩展

下一步

  1. 注册 Tushare 获取 token
  2. 部署 OpenCLAW（Docker 5 分钟搞定）
  3. 配置消息通道（飞书/钉钉/企业微信）
  4. 在聊天中开始查询

参考：

  • 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 年新兴的开源项目，更新较快，具体配置请以官方文档为准。

# 人工智能 # python # AIGC # 自然语言处理
Logo
龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

cover

QClaw体验：微信里的本地AI助手，让智能触手可及

avatar 龙虾开发者社区
cover

WorkBuddy使用心得：腾讯版“免部署小龙虾“的办公新体验

avatar 龙虾开发者社区

VibeVoice Pro流式TTS效果展示：300ms低延迟真实音频生成作品集

本文介绍了如何在星图GPU平台自动化部署VibeVoice Pro：零延迟流式音频引擎镜像，实现300ms低延迟的实时语音生成。该技术特别适用于智能助手对话场景，能够提供自然流畅的语音交互体验，显著提升用户满意度。

avatar 龙虾开发者社区