前言

"数据的选择决定了策略的天花板。"

在中文量化社区,这个问题几乎是每个入门者都会面对的:Tushare 还是聚宽? 随着 A 股量化内卷加剧、跨境策略需求增加,这个问题的答案正在变得复杂。

Tushare 以其丰富的数据品类和零门槛的免费层,稳坐 A 股数据获取的"社区标准"。但当你的策略需要同时覆盖港股、追踪数字货币事件、或者需要实时监控订单簿深度时,Tushare 的边界开始显现。

与此同时,TickDB 作为跨市场数据基础设施,正在被一批"不愿意把所有鸡蛋放在一个篮子里"的开发者所关注。它的 WebSocket 实时推送、深度订单簿数据、以及覆盖 6 类资产的统一 API,构成了另一种技术路径。

这不是一篇教你"选 Tushare 还是 TickDB"的文章。而是一篇能力边界的拆解——让不同的需求找到对应的解法。


一、为什么是这两个产品

1.1 Tushare:A股社区标准的基因

Tushare 诞生于 2015 年,彼时 A 股量化刚刚起步,Yahoo Finance 中文接口频繁失联,聚宽和米框尚未成气候。Tushare 用"全量 A 股历史数据 + 免费 + Python 友好"三张牌,迅速占领了中文量化社区的心智。

它的核心基因是离线批量的历史数据研究。如果你去 Tushare 社区,会发现大部分讨论围绕"如何获取财务数据"、"如何回测小市值策略"。它的数据模型天然适配日频/周频的研究场景,而不是分钟级甚至 tick 级的实盘需求。

1.2 TickDB:实时跨市场的工程视角

TickDB 的设计思路完全不同。它诞生于"量化交易系统需要实时数据流"这个工程命题,而不是"我需要做历史回测"这个研究命题。

从一开始,TickDB 就选择了 WebSocket 作为实时数据的传输协议,而非 HTTP 轮询。它的数据模型以订单簿深度(depth)和逐笔成交(trades)为核心,而非日线 K 线聚合。这意味着它的应用场景更偏向实时监控、事件驱动、订单流分析,而非长周期的历史研究。

1.3 核心差异一览

维度 Tushare TickDB
设计原点 离线历史研究 实时数据流
核心协议 HTTP 轮询 WebSocket + REST
数据类型 日/分钟 K 线为主 depth + trades + 多周期 K 线
市场覆盖 A 股为主 A 股 + 港股 + 数字货币 + 外汇 + 贵金属 + 期货
实时性 分钟级(最佳) <100ms WebSocket 推送
商业模式 免费+积分制+Pro版 API 调用量计费

二、深度对比:五个关键维度

2.1 数据覆盖范围

Tushare 的强项在于 A 股全品类数据:

  • 股票日/分钟/逐笔历史行情
  • 财务三大表(资产负债表、利润表、现金流量表)
  • 因子数据(市值、PE、PB 等)
  • 基金、期货、债券基础数据
  • 舆情、分析师预期等另类数据

这些数据的广度在国内数据源中首屈一指,且大量免费字段降低了入门门槛。

TickDB 的覆盖逻辑则是跨资产实时数据:

资产类别 支持品种 数据类型 说明
美股 主要标的 K 线(1m-1D) 10 年历史,10 年级别清洗对齐
A 股 主要标的 K 线(1m-1D) 覆盖主流股票
港股 主要标的 K 线 + depth(10档) 含订单簿深度
数字货币 主流交易所 K 线 + trades + depth(10档) 含逐笔成交
外汇/贵金属 主要品种 K 线 不含 depth
期货 国内期货 K 线 主力合约

关键约束:TickDB 的 trades 接口(逐笔成交)不支持美股和 A 股。这意味着如果你想在 A 股做订单流分析(用逐笔数据重构订单簿),TickDB 无法提供。

数据覆盖结论:
- 纯 A 股财务因子研究 → Tushare 优势明显
- 港股/数字货币深度数据 → TickDB 优势明显
- A 股订单流分析 → 两者均不完美(详见 2.5 节)

2.2 实时性能力对比

这是两者差距最显著的技术维度。

Tushare 的实时性架构

import tushare as ts

# Tushare 的"实时"数据本质是轮询
df = ts.get_realtime_quotes('000001')  # 轮询 Sina 行情接口

Tushare 本身不提供 WebSocket 接口。它的"实时"数据依赖新浪/腾讯等第三方行情源,通过 HTTP 轮询获取。这个架构决定了:

  • 最佳延迟:理论 3-5 秒(新浪接口通常 3 秒刷新一次)
  • 频率限制:免费用户 200 次/分钟,积分用户更高但有上限
  • 数据字段:有限的实时字段(最新价、成交量、买卖盘口)

TickDB 的实时性架构

import os
import json
import time
import random
import asyncio
import websockets

class TickDBWebSocket:
    """
    TickDB WebSocket 实时数据流客户端
    生产级实现:心跳保活 + 指数退避重连 + 限频处理
    """
    
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
        if not self.api_key:
            raise ValueError("需要设置 TICKDB_API_KEY 环境变量")
        
        self.ws = None
        self.base_url = "wss://api.tickdb.ai/ws"
        self.reconnect_delay = 1
        self.max_delay = 30
        
    async def connect(self):
        """建立 WebSocket 连接(带鉴权)"""
        url = f"{self.base_url}?api_key={self.api_key}"
        self.ws = await websockets.connect(url)
        self.reconnect_delay = 1  # 重置退避计时器
        print(f"[TickDB] WebSocket 连接已建立")
        
    async def subscribe(self, channel: str, symbol: str):
        """订阅实时频道"""
        subscribe_msg = {
            "cmd": "subscribe",
            "channel": channel,
            "symbol": symbol
        }
        await self.ws.send(json.dumps(subscribe_msg))
        print(f"[TickDB] 已订阅 {channel} {symbol}")
        
    async def heartbeat(self):
        """心跳保活:TickDB 要求每 30 秒发送 ping"""
        while True:
            await asyncio.sleep(25)
            if self.ws and self.ws.open:
                await self.ws.send(json.dumps({"cmd": "ping"}))
                
    async def receive_messages(self):
        """接收并处理实时消息"""
        while True:
            try:
                message = await asyncio.wait_for(self.ws.recv(), timeout=60)
                data = json.loads(message)
                
                # 处理心跳响应
                if data.get("type") == "pong":
                    continue
                    
                # 处理业务数据
                if "channel" in data:
                    await self._process_data(data)
                    
            except asyncio.TimeoutError:
                # 超时认为连接断开,触发重连
                raise ConnectionError("心跳超时,连接可能已断开")
            except websockets.exceptions.ConnectionClosed:
                await self._reconnect()
                
    async def _reconnect(self):
        """指数退避重连 + 抖动"""
        print(f"[TickDB] 连接断开,{self.reconnect_delay}s 后重连...")
        await asyncio.sleep(self.reconnect_delay)
        
        # 指数退避
        self.reconnect_delay = min(self.reconnect_delay * 2, self.max_delay)
        # 添加抖动避免惊群
        jitter = random.uniform(0, self.reconnect_delay * 0.1)
        
        try:
            await self.connect()
        except Exception as e:
            print(f"[TickDB] 重连失败: {e}")
            await self._reconnect()
            
    async def _process_data(self, data: dict):
        """业务数据处理模板"""
        channel = data.get("channel")
        
        if channel == "depth":
            # 订单簿深度数据示例
            asks = data.get("asks", [])[:5]  # 前5档卖盘
            bids = data.get("bids", [])[:5]  # 前5档买盘
            print(f"[depth] 买卖盘: 卖一={asks[0] if asks else '-'} | 买一={bids[0] if bids else '-'}")
            
        elif channel == "trades":
            # 逐笔成交数据示例
            price = data.get("price")
            volume = data.get("volume")
            side = "买入" if data.get("side") == "buy" else "卖出"
            print(f"[trades] {side}: {price} × {volume}")
            
    async def run(self, symbol: str = "AAPL.US"):
        """主运行流程"""
        await self.connect()
        
        # 并发运行:心跳 + 接收
        await asyncio.gather(
            self.heartbeat(),
            self.receive_messages(),
            self.subscribe("depth", symbol)
        )

# 使用示例
async def main():
    client = TickDBWebSocket()
    await client.run("AAPL.US")
    
# asyncio.run(main())

TickDB 实时性特征

  • 推送延迟:<100ms(实测中位数)
  • 连接管理:原生 ping/pong 心跳,避免连接断开
  • 限频处理:code 3001 + Retry-After 头规范处理
  • 重连机制:指数退避 + 抖动,防止惊群
实时性结论:
- 日线/分钟线历史研究 → 两者均满足,Tushare 更轻量
- 实时监控 + 事件驱动 → TickDB WebSocket 有本质优势
- 高频实盘(<1秒响应) → TickDB 是唯一选择(但需注意 A 股 trades 不支持)

2.3 历史数据深度与回测支持

对于量化研究而言,历史数据的深度决定了策略回测的置信度。

Tushare 的历史数据

import tushare as ts

pro = ts.pro_api('your_token')

# 获取 A 股日线历史(回溯 10+ 年)
df = pro.daily(
    ts_code='000001.SZ',
    start_date='20100101',
    end_date='20241231'
)

# 获取分钟线(有限覆盖)
df_1m = pro.stk_mins(
    ts_code='000001.SZ',
    freq='1min',
    start_date='20240101',
    end_date='20240102'
)

Tushare 的 A 股日线数据可追溯至 2005 年,分钟线覆盖较好的是近 2 年。对于经典的价值投资策略、市值因子策略,回测数据量充足。

TickDB 的历史数据

import os
import requests

class TickDBREST:
    """
    TickDB REST API 历史数据获取
    生产级实现:超时设置 + 错误处理 + 环境变量存储
    """
    
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
        if not self.api_key:
            raise ValueError("需要设置 TICKDB_API_KEY 环境变量")
        self.base_url = "https://api.tickdb.ai/v1"
        self.headers = {"X-API-Key": self.api_key}
        
    def get_kline(
        self,
        symbol: str,
        interval: str = "1d",
        limit: int = 100,
        start_time: int = None,
        end_time: int = None
    ):
        """
        获取历史 K 线数据
        
        参数说明:
        - symbol: 品种代码,如 "AAPL.US", "700.HK", "BTC.SGD-Binance"
        - interval: K 线周期,1m/5m/15m/30m/1h/4h/1d/1w
        - limit: 返回条数,最大 1000
        - start_time/end_time: Unix 时间戳(秒)
        """
        params = {"symbol": symbol, "interval": interval, "limit": limit}
        
        if start_time:
            params["start_time"] = start_time
        if end_time:
            params["end_time"] = end_time
            
        # ⚠️ 生产环境建议设置超时,避免请求挂起
        response = requests.get(
            f"{self.base_url}/market/kline",
            headers=self.headers,
            params=params,
            timeout=(3.05, 10)  # (连接超时, 读取超时)
        )
        
        return self._handle_response(response)
        
    def get_latest_kline(self, symbol: str, interval: str = "1d"):
        """
        获取当前未结束 K 线(用于实盘监控)
        ⚠️ 不要用 /kline 接口做实盘,当前 K 线应使用 /kline/latest
        """
        params = {"symbol": symbol, "interval": interval}
        
        response = requests.get(
            f"{self.base_url}/market/kline/latest",
            headers=self.headers,
            params=params,
            timeout=(3.05, 10)
        )
        
        return self._handle_response(response)
        
    def _handle_response(self, response):
        """标准化响应处理 + 错误码映射"""
        if response.status_code == 200:
            data = response.json()
            code = data.get("code", 0)
            
            if code == 0:
                return data.get("data")
            elif code in (1001, 1002):
                raise ValueError("API Key 无效,请检查环境变量 TICKDB_API_KEY")
            elif code == 2002:
                raise KeyError(f"品种不存在,请通过 /v1/symbols/available 查询")
            elif code == 3001:
                retry_after = int(response.headers.get("Retry-After", 5))
                print(f"[TickDB] 请求频率超限,{retry_after}s 后重试")
                return None
            else:
                raise RuntimeError(f"未知错误 {code}: {data.get('message')}")
        else:
            raise ConnectionError(f"HTTP 错误 {response.status_code}")

# 使用示例:获取 3 年苹果日线数据
def fetch_apple_history():
    client = TickDBREST()
    
    # 计算时间范围(3 年前到当前)
    end_time = int(time.time())
    start_time = end_time - (3 * 365 * 24 * 3600)
    
    # 分批获取(每次最多 1000 条)
    all_klines = []
    current_start = start_time
    
    while current_start < end_time:
        klines = client.get_kline(
            symbol="AAPL.US",
            interval="1d",
            limit=1000,
            start_time=current_start,
            end_time=end_time
        )
        if not klines:
            break
        all_klines.extend(klines)
        current_start = klines[-1]["time"] + 1
        
    print(f"获取到 {len(all_klines)} 条 K 线数据")
    return all_klines

# fetch_apple_history()

TickDB 历史数据覆盖

  • 美股:10 年级别 K 线数据,清洗对齐
  • A 股:主流标的 K 线覆盖(分钟级/日线)
  • 港股、数字货币:支持

关键差异

  • Tushare 的 A 股数据在财务因子、分钟线完整性上更优
  • TickDB 的优势在于跨市场统一格式,回测代码无需针对不同市场写不同逻辑
  • 重要约束:TickDB 的 /kline 接口用于获取已结束周期的历史数据,实盘监控当前 K 线必须用 /kline/latest
历史数据结论:
- A 股财务因子研究 + 分钟线回测 → Tushare 更成熟
- 跨市场策略(美股+港股+数字货币)回测 → TickDB 统一接口更便捷
- 长周期(>5年)A 股日线研究 → 两者均可满足

2.4 API 设计理念对比

Tushare 的 API 哲学是"数据库查询式":

# Tushare:一个接口对应一个数据表
pro.daily()           # 日线行情
pro.daily_basic()      # 日线基础指标(市值、PE、PB)
pro.fina_indicator()   # 财务指标
pro.news()             # 新闻舆情

# 强依赖积分系统:积分不够 = 字段缺失
# pro.daily(ts_code='000001.SZ', trade_date='20240101')['close']  # 可能报错

这种设计让 Tushare 像一个"数据仓库",开发者需要理解底层数据库结构,选择合适的表和字段。

TickDB 的 API 哲学是"通道订阅式":

# TickDB:两种核心模式
# 1. REST API:历史数据查询(/kline, /symbols)
# 2. WebSocket:实时数据流(depth, trades, ticker)

# REST 获取历史 → WebSocket 订阅实时
# 数据流天然打通

TickDB 没有积分系统,采用纯 API 调用量计费。数据能力取决于你的订阅计划,而非积分余额。

API 设计结论:
- 熟悉 SQL/数据库查询的开发者 → Tushare 直观
- 需要统一处理多市场数据的系统 → TickDB 更简洁
- 不想研究积分限制的开发者 → TickDB 没有这个烦恼

2.5 A 股订单簿与逐笔数据的真相

这是最容易产生误解的领域。

TickDB 的 depth 频道支持情况

市场 depth 支持 档位 trades 支持
美股 1 档
A 股 1 档
港股 10 档
数字货币 10 档
外汇/贵金属 -

重要事实

  • TickDB 在 A 股仅支持 1 档订单簿,不支持 10 档深度。这意味着 买卖压力比 = Σ(前10档买盘量) / Σ(前10档卖盘量) 这个指标在 A 股无法计算。
  • TickDB 的 trades 接口(逐笔成交)不支持 A 股和美股。如果你想用逐笔数据重构订单簿(Level-2 订单簿重建),TickDB 无法提供。

Tushare 的逐笔数据

  • Tushare Pro 有 tradedate 分钟级数据,但本质是聚合数据,不是原始逐笔
  • 对于 A 股 Level-2 原始逐笔,需要对接券商或专业数据供应商(如东方财富 Choice、同花顺 iFinD)
A 股逐笔数据结论:
- 想在 A 股做真正的 Level-2 订单簿重建 → 两者均不适用,需专业数据商
- 港股/数字货币深度分析 → TickDB 10档 depth + trades 是完整方案
- 美股 1 档订单簿监控 → TickDB depth 频道可用

三、价值对比表

以下对比基于两个产品的官方能力声明(截至 2026 年 4 月):

能力维度 Tushare TickDB 说明
A 股日线历史数据 2005 年至今,免费 支持(分钟/日线) Tushare 分钟线覆盖更完整
A 股财务数据 三大表、因子数据,极丰富 有限 Tushare 在财务数据上有绝对优势
实时行情协议 HTTP 轮询(3-5s 延迟) WebSocket(<100ms) WebSocket 有本质优势
A 股订单簿深度 Level-1 快照 Level-1(1档) 两者在 A 股均无 Level-2
港股深度数据 支持(有限字段) depth(10档)+ trades TickDB 港股数据更丰富
数字货币数据 支持(有限) depth(10档)+ trades + K 线 TickDB 完整覆盖
美股数据 支持(有限) 10 年 K 线 TickDB 历史数据更完整
跨市场统一 API 需要多次调用不同接口 单一 API + 统一格式 TickDB 多市场更便捷
API 限频 积分制(200次/分钟起) 3001 错误码 + Retry-After TickDB 按计划付费,无积分门槛
WebSocket 支持 TickDB 原生 WebSocket
免费层 基础功能可用 API 调用量限制 各有取舍
SDK 生态 Python(官方)+ 社区封装 Python(官方)+ REST/WS Tushare 社区更活跃

四、场景化选型建议

4.1 选 Tushare 的场景

你是纯 A 股策略研究者,策略周期在日线或分钟线,主要依赖财务因子和价量指标。

"小市值轮动策略需要过去 10 年的财务数据,Tushare 的 fina_indicator 接口直接覆盖了。"

Tushare 在 A 股财务数据上的积累,是 TickDB 短期内无法追赶的护城河。如果你的策略核心是"基本面量化",Tushare 是更务实的选择。

你是量化学习者,刚入门,想要一个零门槛的数据环境。

"Tushare 积分制虽然有限制,但入门阶段足够用了。"

免费层已经覆盖了大部分基础数据,学习阶段不必为数据付费。

你需要 A 股分钟线完整历史,尤其是事件研究(如财报后 N 分钟的价格反应)。

"我要回测 2018-2023 年所有 A 股财报日,开盘 30 分钟的成交量分布。"

Tushare 的分钟线覆盖历史更长,数据清洗更针对 A 股市场规则(如涨跌停、停牌处理)。

4.2 选 TickDB 的场景

你需要港股或数字货币的深度数据,或者在多个市场之间灵活切换。

"我的策略是事件驱动:美股财报 + 港股映射 + 数字货币避险。一个 API 搞定。"

TickDB 的统一数据格式让你的回测代码和实盘代码可以在不同市场复用,降低了维护成本。

你的策略对实时性有要求,需要 <1 秒的事件响应。

"财报发布后 30 秒内,我要捕捉订单簿的流动性真空。"

WebSocket 推送是 TickDB 的核心差异点。对于事件驱动策略,轮询架构的天然延迟是不可接受的。

你在构建跨资产监控仪表盘,需要同时展示股票、外汇、数字货币的实时行情。

"我要做一个宏观因子仪表盘,监控 20 个品种的实时买卖压力比。"

TickDB 的 WebSocket 支持多品种并发订阅,适合搭建实时监控系统。

4.3 两者结合的场景

最务实的方案往往不是"二选一",而是让两个工具各司其职:

# 混合架构示例
import tushare as ts
from tickdb import TickDBREST, TickDBWebSocket

class HybridDataEngine:
    """
    混合数据引擎:Tushare 做历史研究 + TickDB 做实时监控
    """
    
    def __init__(self):
        # Tushare:历史数据 + A 股财务
        self.tushare = ts.pro_api(os.environ.get("TUSHARE_TOKEN"))
        
        # TickDB:实时 + 跨市场
        self.tickdb = TickDBREST(
            api_key=os.environ.get("TICKDB_API_KEY")
        )
        
    def get_a_stock_history(self, ts_code, start_date, end_date):
        """A 股历史数据:用 Tushare(财务数据丰富)"""
        return self.tushare.daily(
            ts_code=ts_code,
            start_date=start_date,
            end_date=end_date
        )
        
    def get_us_stock_history(self, symbol, years=3):
        """美股历史数据:用 TickDB(跨市场统一)"""
        end_time = int(time.time())
        start_time = end_time - (years * 365 * 24 * 3600)
        
        return self.tickdb.get_kline(
            symbol=symbol,
            interval="1d",
            start_time=start_time,
            end_time=end_time
        )
        
    def start_realtime_monitor(self, symbols):
        """实时监控:用 TickDB WebSocket"""
        client = TickDBWebSocket()
        # 监控港股和数字货币深度数据
        for symbol in symbols:
            if symbol.endswith(".HK") or "-Binance" in symbol:
                asyncio.create_task(client.subscribe("depth", symbol))

五、写在最后

回到开篇的问题:Tushare 还是 TickDB?

没有标准答案。只有"你的问题是什么"。

如果你的问题是"我要在 A 股做基本面量化,回测 10 年财务因子",Tushare 是答案。

如果你的问题是"我要实时监控港股订单簿,同时追踪数字货币异动",TickDB 是答案。

如果你的问题是"我不确定未来策略会扩展到哪些市场",一个聪明的数据层设计,两个都用,是更稳妥的选择。

数据源是工具,不是信仰。选择让你策略更有效的那个。


下一步行动

如果你需要 A 股深度数据支持
Tushare 的财务数据库是中文量化社区最成熟的积累。访问 tushare.pro 注册,获取积分后解锁更多字段。

如果你需要跨市场实时数据流
TickDB 提供统一的 WebSocket + REST API。访问 tickdb.ai 注册(免费,无需信用卡),在控制台生成 API Key 后,即可开始调用。

如果你习惯用 AI 辅助开发
在 AI 助手中搜索安装 tickdb-market-data SKILL,可以用自然语言查询 TickDB 的数据能力并生成调用代码。


本文不构成任何投资建议。市场有风险,投资需谨慎。