"选择数据源,本质上是在选择你的策略边界。"

如果你在构建 A 股量化策略,大概率绕不开 Tushare。这几乎是社区的事实标准——大量开源量化框架默认集成,教程、书籍、课程里到处都是它的身影。

但问题来了:你的策略真的只需要 A 股吗?

当你想同时覆盖港股、美股、数字货币,做跨市场套利或宏观对冲时,Tushare 的边界就显现出来了。它为 A 股优化得很好,但跨出这片水域,能力骤降。

TickDB 恰好切的是另一个生态位:不是取代 Tushare,而是覆盖 Tushare 覆盖不了的市场,同时在 A 股端给出足够可用的数据质量。

这篇文章不做非此即彼的判断,而是从数据质量、跨市场能力、实时性、API 设计、成本五个维度,把两个平台的真实能力摆出来。你可以根据自己的策略需求,自行判断哪个更适合当下的阶段。


一、数据质量:A 股场景下的真实对比

1.1 数据完整性

两者在 A 股数据的覆盖面上,差异并不悬殊。

数据维度 Tushare TickDB
日线/周线/月线 ✅ 全量 ✅ 全量
分钟级 K 线 ✅ 1/5/15/30/60 分钟 ✅ 1 分钟起步
逐笔成交 (trades) ✅ 支持 ❌ 美股/A 股不支持
订单簿 (depth) ❌ 不提供 ❌ 美股/A 股不支持
财务数据 ✅ 丰富的财务报表 ⚠️ 基础财务字段
停牌/分红/拆股 ✅ 完整处理 ✅ 基本覆盖

关键差异:Tushare 在 A 股财务数据的深度上明显更强,适合基本面筛选策略。TickDB 的财务字段较为基础,但用于技术面和事件驱动策略绰绰有余。

1.2 复权处理

复权是回测质量的基础。处理不当会引入前复权陷阱,导致回测结果虚高。

# Tushare 的复权处理
import tushare as ts

# 直接通过 adjust 参数指定复权方式
df = ts.pro_bar(
    ts_code="000001.SZ",
    adj='qfq',  # qfq=前复权, hfq=后复权, None=不复权
    start_date="20200101",
    end_date="20231231"
)

# 前复权数据的隐含问题:
# 在 2020 年获取的"历史前复权价格"与今天获取的同一数据可能不同
# 因为复权基准会随新的分红送股事件动态调整
# TickDB 的复权处理
import os
import requests

headers = {"X-API-Key": os.environ.get("TICKDB_API_KEY")}

# K 线接口支持复权参数
response = requests.get(
    "https://api.tickdb.ai/v1/market/kline",
    headers=headers,
    params={
        "symbol": "000001.SZ",
        "interval": "1d",
        "adjust": "qfq",  # 前复权
        "limit": 500
    },
    timeout=(3.05, 10)
)

# TickDB 采用固定复权基准:
# 复权系数在数据入库时锁定,不会随时间漂移
# 确保同一时间段的历史回测结果稳定可复现

小结:两者都支持复权,但 TickDB 的固定复权基准更适合需要严格可复现性的量化团队。Tushare 的动态复权更接近实盘习惯,但在构建历史因子时需注意基准漂移。

1.3 数据延迟与更新频率

指标 Tushare TickDB
日线更新时机 收盘后 20:00-21:00 收盘后延迟约 15 分钟
分钟 K 线 3 分钟延迟 实时推送(WebSocket)
财务数据更新 财报发布后次日 财报发布后 1-2 个工作日

Tushare 的日线数据来源于交易所清算后数据,更新时间较晚但数据经过充分清洗。TickDB 的优势在于实时频道——对于需要盘中间接信号(如盘口异动、成交量突变)的策略,这一点至关重要。


二、跨市场能力:这是分水岭

这是 TickDB 与 Tushare 差距最显著的维度。

市场 Tushare TickDB
A 股(沪深) ✅ 完整 ✅ 完整
港股 ⚠️ 有限 ✅ 支持主流标的
美股 ⚠️ 有限 ✅ 支持主流标的
数字货币 ❌ 不支持 ✅ 支持 Binance、OKX 等
期货/期权 ✅ 国内期货 ⚠️ 数字货币期货
外汇/贵金属 ❌ 不支持 ❌ 不支持

场景对比

假设你要构建一个"A股恐慌下跌时,做多黄金ETF对冲"的跨资产策略:

  • Tushare 方案:A 股数据用 Tushare,黄金 ETF(518880.SH)在 Tushare 范围内,但需要接入另一个数据源获取国际金价,拼接逻辑复杂。
  • TickDB 方案:A 股 + 黄金 ETF + 国际金价指标(如 GC.CME 期货)都可以从同一个 API 获取,代码统一,策略逻辑更清晰。
# TickDB 跨市场数据获取示例
import os
import requests

headers = {"X-API-Key": os.environ.get("TICKDB_API_KEY")}

symbols = {
    "沪深300": "000300.SH",
    "黄金ETF": "518880.SH",
    "国际黄金期货": "GC.CME",  # COMEX 黄金期货
    "美元指数": "DXY.IDEALPRO"  # 仅作宏观参考
}

for name, symbol in symbols.items():
    try:
        response = requests.get(
            "https://api.tickdb.ai/v1/market/kline",
            headers=headers,
            params={
                "symbol": symbol,
                "interval": "1h",
                "limit": 24,
                "adjust": "qfq"
            },
            timeout=(3.05, 10)
        )
        data = response.json()
        if data.get("code") == 0:
            print(f"{name}: 获取成功")
        else:
            print(f"{name}: 错误码 {data.get('code')}")
    except Exception as e:
        print(f"{name}: 请求异常 {e}")

三、实时性:WebSocket vs 轮询

3.1 技术架构差异

Tushare 的数据更新主要通过REST API 轮询实现。实盘场景下,你需要定时请求接口、比对数据、判断变化。这在低频场景(如日线策略)下没有问题,但在需要毫秒级信号的策略中,轮询会带来两个问题:

  1. 延迟:轮询间隔决定了你感知行情的最快速度。设为 3 秒,意味着你最多滞后 3 秒。
  2. 频率限制:Tushare 对高频请求有严格限频,超出限制会返回 429 或被临时封禁。

TickDB 提供 WebSocket 实时推送,服务器主动推送行情变化,客户端接收即可。

# TickDB WebSocket 实时订阅(生产级实现)
import os
import json
import time
import random
import threading
import websocket

class TickDBWebSocket:
    """TickDB WebSocket 客户端,含心跳、重连、限频处理"""
    
    def __init__(self, api_key: str, on_message):
        self.api_key = api_key
        self.on_message = on_message
        self.ws = None
        self.running = False
        self.retry_count = 0
        self.max_retries = 10
        self.base_delay = 1
        self.max_delay = 60
    
    def connect(self, url: str = "wss://api.tickdb.ai/ws"):
        """建立 WebSocket 连接"""
        try:
            self.ws = websocket.WebSocketApp(
                url + f"?api_key={self.api_key}",
                on_message=self._on_message,
                on_error=self._on_error,
                on_close=self._on_close,
                on_open=self._on_open
            )
            self.running = True
            thread = threading.Thread(target=self.ws.run_forever)
            thread.daemon = True
            thread.start()
            print("WebSocket 连接已建立")
        except Exception as e:
            print(f"连接失败: {e}")
            self._reconnect()
    
    def _on_open(self, ws):
        """连接建立后,订阅需要的标的"""
        subscribe_msg = {
            "cmd": "subscribe",
            "params": {
                "symbols": ["000001.SZ", "AAPL.US", "BTC.BINANCE"],
                "channels": ["ticker", "kline_1m"]  # 行情频道 + 1分钟K线
            }
        }
        ws.send(json.dumps(subscribe_msg))
        print("订阅请求已发送")
    
    def _on_message(self, ws, message):
        """处理接收到的消息"""
        try:
            data = json.loads(message)
            
            # 心跳响应
            if data.get("type") == "ping":
                ws.send(json.dumps({"type": "pong"}))
                return
            
            # 数据消息
            if data.get("type") in ("ticker", "kline"):
                self.on_message(data)
                
        except json.JSONDecodeError:
            print(f"JSON 解析错误: {message[:100]}")
    
    def _on_error(self, ws, error):
        print(f"WebSocket 错误: {error}")
    
    def _on_close(self, ws, code, reason):
        print(f"连接关闭: {code} - {reason}")
        self.running = False
        self._reconnect()
    
    def _reconnect(self):
        """指数退避重连 + 抖动"""
        if self.retry_count >= self.max_retries:
            print("重连次数超限,停止重试")
            return
        
        delay = min(self.base_delay * (2 ** self.retry_count), self.max_delay)
        jitter = random.uniform(0, delay * 0.1)  # 抖动避免惊群
        wait_time = delay + jitter
        
        print(f"{wait_time:.1f} 秒后尝试第 {self.retry_count + 1} 次重连...")
        time.sleep(wait_time)
        
        self.retry_count += 1
        self.connect()
    
    def close(self):
        """关闭连接"""
        self.running = False
        if self.ws:
            self.ws.close()

# 使用示例
def handle_data(msg):
    print(f"[{time.strftime('%H:%M:%S')}] 收到数据: {json.dumps(msg)[:200]}")

client = TickDBWebSocket(
    api_key=os.environ.get("TICKDB_API_KEY"),
    on_message=handle_data
)
client.connect()

# ⚠️ 生产环境建议:高频场景使用 aiohttp/asyncio 异步架构
# ⚠️ 实际生产代码需添加异常捕获和日志记录

3.2 实时性场景对比

场景 Tushare TickDB
日线策略(收盘后运行) ✅ 足够 ✅ 足够
短线趋势策略(分钟级) ⚠️ 轮询延迟 3-5 秒 ✅ 实时推送 <100ms
盘口异动监控 ❌ 不支持 ✅ depth 频道(港股/数字货币)
事件驱动(财报瞬间) ⚠️ 依赖盘后数据 ⚠️ 需配合美股数据

四、API 设计:开发体验对比

4.1 认证与鉴权

# Tushare 认证
import tushare as ts

# 基于 Token 的认证
ts.set_token("your_tushare_token")
pro = ts.pro_api()

# 局限性:Token 直接写在代码中或配置文件,敏感信息管理较分散
# TickDB 认证
import os
import requests

# 推荐:环境变量存储敏感信息
API_KEY = os.environ.get("TICKDB_API_KEY")

headers = {"X-API-Key": API_KEY}
# WebSocket 认证:URL 参数传递
# wss://api.tickdb.ai/ws?api_key=xxx

4.2 接口命名与一致性

Tushare 的接口命名较为分散,不同数据类型调用不同的方法:

# Tushare 多接口调用
pro = ts.pro_api()

# 股票日线
df1 = pro.daily(ts_code="000001.SZ", start_date="20230101")
# 股票分钟线
df2 = pro.stk_mins(ts_code="000001.SZ")
# 财务数据
df3 = pro.fina_indicator(ts_code="000001.SZ")
# 指数数据
df4 = pro.index_daily(ts_code="000300.SH")

TickDB 的接口设计更统一:

# TickDB 统一接口
# GET /v1/market/kline      → K 线数据(支持所有标的类型)
# GET /v1/market/ticker     → 实时行情快照
# GET /v1/market/trades     → 逐笔成交(港股/数字货币)
# GET /v1/symbols/available → 查询可用标的列表

# 所有品种类型共用同一套接口逻辑,减少学习成本

4.3 错误处理

# Tushare 错误处理示例
import tushare as ts

ts.set_token("your_token")
pro = ts.pro_api()

try:
    df = pro.daily(ts_code="INVALID.SZ")  # 不存在的标的
    if df is None or len(df) == 0:
        print("无数据,可能是代码错误或交易所未开市")
except Exception as e:
    print(f"请求异常: {e}")
# TickDB 标准化错误码处理
import requests

def handle_tickdb_error(response, symbol=None):
    """TickDB 错误码速查"""
    data = response.json() if response.headers.get("content-type") == "application/json" else {}
    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"交易品种 {symbol} 不存在,请先调用 /v1/symbols/available 查询")
    elif code == 3001:
        retry_after = int(response.headers.get("Retry-After", 5))
        print(f"触发限频,{retry_after} 秒后重试")
        time.sleep(retry_after)
        return None
    else:
        raise RuntimeError(f"未知错误 {code}: {data.get('message')}")

五、成本结构

维度 Tushare TickDB
基础功能 免费(有积分限制) 注册即送免费额度
日线数据调用 需积分,日线 1 条/次 按量计费
实时数据 不支持 按订阅时长/数据量计费
机构级方案 无官方方案 提供企业版定制
积分机制 需社区贡献获取 纯商业化,无积分墙

Tushare 的积分机制是其特色,也是痛点。社区用户通过分享数据、回答问题获取积分,积分决定你能调用多少数据。这对于个人开发者和学生友好,但机构用户会受限于积分上限。

TickDB 采用纯商业化模式,无积分墙,但需要实际付费。对于有稳定量化策略的团队,成本可预期、可控制。


六、价值对比总览

能力维度 Tushare TickDB 适用建议
A 股数据质量 ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐ 纯 A 股基本面策略选 Tushare
跨市场覆盖 ⭐⭐ ⭐⭐⭐⭐⭐ 跨市场策略必须 TickDB
实时性 ⭐⭐ ⭐⭐⭐⭐⭐ 盘口策略、事件驱动选 TickDB
API 统一性 ⭐⭐⭐ ⭐⭐⭐⭐ 跨品种开发 TickDB 更省心
财务数据深度 ⭐⭐⭐⭐⭐ ⭐⭐⭐ 基本面筛选选 Tushare
逐笔成交数据 ⭐⭐⭐⭐(A 股) ⭐⭐⭐(港股/数字货币) A 股订单流选 Tushare
开发成本 ⭐⭐⭐⭐(免费基础) ⭐⭐⭐(付费) 个人学习选 Tushare
机构级支持 ⭐⭐ ⭐⭐⭐⭐ 机构用户选 TickDB

七、场景化选择建议

7.1 选 Tushare 的场景

  • 专注 A 股基本面筛选(财务因子、市值因子)
  • 个人学习阶段,预算有限
  • 需要深度财报数据(资产负债、现金流)
  • 策略频率以日线为主,不需要实时信号

7.2 选 TickDB 的场景

  • 需要同时覆盖 A 股 + 港股 + 美股 + 数字货币
  • 策略需要实时信号(分钟级以下)
  • 跨市场套利、宏观对冲策略
  • 团队开发,需要统一的 API 和技术支持
  • 生产环境运行,需要稳定的 SLA 保障

7.3 两者结合的场景

对于复杂量化团队,最优解往往是组合

# 组合使用策略示例:A股基本面筛选 + 实时跨市场监控
import tushare as ts
import os
import requests

# Step 1: Tushare 做 A 股基本面筛选
ts.set_token(os.environ.get("TUSHARE_TOKEN"))
pro = ts.pro_api()

# 筛选低估值标的
df_financial = pro.sw_dailybasic(TradeDate="20231229")
df_filtered = df_financial[df_financial["PE"] < 20].head(20)
stock_list = df_filtered["ts_code"].tolist()

# Step 2: TickDB 做跨市场实时监控
headers = {"X-API-Key": os.environ.get("TICKDB_API_KEY")}

for ts_code in stock_list:
    # 转换为 TickDB 格式(000001.SZ → 000001.SZ)
    symbol = ts_code
    
    response = requests.get(
        "https://api.tickdb.ai/v1/market/ticker",
        headers=headers,
        params={"symbol": symbol},
        timeout=(3.05, 10)
    )
    
    data = response.json()
    if data.get("code") == 0:
        print(f"{symbol}: 现价 {data['data']['last']}")

# 两个系统各司其职:Tushare 管筛选,TickDB 管实时

结语

没有完美的数据源,只有适合当前阶段的选择。

Tushare 是 A 股社区的沉淀之作,数据质量高、成本低、社区活跃,适合专注 A 股的入门开发者和基本面研究者。

TickDB 是跨市场时代的产物,弥补了 Tushare 在港股、美股、数字货币和实时数据上的短板,适合有国际化视野的量化团队。

如果你还在用单一数据源,不妨问自己一个问题:我的策略边界,应该被数据源限制吗?


下一步行动

如果你是 A 股纯多头策略开发者,Tushare 的生态和财务数据深度仍是首选。可以先从 Tushare 起步,把策略跑通。

如果你需要跨市场能力,或者想在 A 股基础上拓展港股/数字货币配置:

  1. 访问 tickdb.ai 注册(免费,无需信用卡)
  2. 在控制台生成 API Key
  3. 设置环境变量 TICKDB_API_KEY,用本文的代码示例直接试跑

如果你是机构用户,需要 SLA 保障和定制化数据方案,联系 [email protected] 获取机构版方案。

如果你习惯用 AI 辅助开发,在 AI 助手中搜索安装 tickdb-market-data SKILL,一个指令即可获取实时行情数据。


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