4. 金融数据获取与存储：从Tushare到SQLite的实战之路

做量化投资，数据就是你的弹药库。没有数据，再牛的策略也是纸上谈兵。

这一章，我带你走一遍完整的数据流水线——从获取到存储，再到维护。说白了，就是教会你如何搭建自己的数据仓库。

核心要点：数据获取与存储是量化系统的地基。地基不稳，上层建筑随时会塌。

4.1 数据获取：Tushare vs AkShare

国内做A股数据，绕不开两个库：Tushare和AkShare。我个人的习惯是，两个都装，哪个顺手用哪个。

Tushare 是老牌选手，数据质量高，但需要注册获取token。免费版有调用频率限制，一天最多200次。嗯，这个限制其实挺烦人的。

AkShare 是后起之秀，完全免费，接口丰富。它直接爬取各大财经网站的数据，实时性不错。但有个问题——数据源不稳定，有时候网站改版，接口就挂了。

我曾经在项目中同时用两个库做数据校验。Tushare拿日线，AkShare拿实时行情，交叉验证。结果发现，偶尔会有几毛钱的差异。你想想看，这要是做高频交易，几毛钱就是天壤之别。

4.1.1 安装与配置

# 安装Tushare
pip install tushare

# 安装AkShare
pip install akshare

# Tushare需要注册获取token
import tushare as ts
ts.set_token('你的token')
pro = ts.pro_api()

小技巧：Token不要硬编码在代码里。我习惯放在环境变量或配置文件中，避免泄露。

4.1.2 获取A股日线数据

# 使用Tushare获取平安银行日线数据
df = pro.daily(ts_code='000001.SZ', start_date='20230101', end_date='20231231')
print(df.head())

# 使用AkShare获取同样数据
import akshare as ak
df_ak = ak.stock_zh_a_hist(symbol="000001", period="daily", 
                           start_date="20230101", end_date="20231231", adjust="qfq")
print(df_ak.head())

这里有个坑要注意。Tushare返回的日期是字符串格式，AkShare返回的是datetime格式。我刚开始做的时候没注意，后面做时间序列分析时直接报错。嗯，数据类型一致性很重要。

4.2 数据存储：从CSV到Excel

数据拿到手了，存哪里？最简单的方案就是CSV或Excel。

CSV 轻量、通用、跨平台。我个人的习惯是，做快速验证时用CSV，方便用Excel打开看看。

Excel 适合小规模数据，可视化方便。但超过10万行就卡得不行。我曾经试过把一年的分钟线数据存到Excel里，打开花了5分钟，直接劝退。

4.2.1 存储到CSV

# 存储到CSV
df.to_csv('000001_daily.csv', index=False, encoding='utf-8-sig')

# 读取CSV
df_read = pd.read_csv('000001_daily.csv', parse_dates=['trade_date'])

注意：编码一定要用utf-8-sig，否则中文列名在Excel里会乱码。我踩过这个坑，折腾了半天才发现是编码问题。

4.2.2 存储到Excel

# 存储到Excel
with pd.ExcelWriter('stock_data.xlsx', engine='openpyxl') as writer:
    df.to_excel(writer, sheet_name='000001', index=False)
    df2.to_excel(writer, sheet_name='000002', index=False)

# 读取Excel
df_excel = pd.read_excel('stock_data.xlsx', sheet_name='000001')

Excel的好处是可以把多只股票放在不同sheet里。但说实话，数据量一大，还是得用数据库。

4.3 数据存储：SQLite数据库

SQLite是我个人最推荐的小型量化系统数据库。为什么？

  • 零配置：不需要安装数据库服务，一个文件搞定
  • 轻量级：整个库只有几百KB
  • 支持SQL：查询、聚合、连接操作都很方便
  • 跨平台：Windows、Linux、macOS都能用

我曾经在项目中用MySQL，后来发现单机跑量化策略，SQLite完全够用。而且部署简单，不用配数据库账号密码，省心不少。

4.3.1 创建数据库和表

import sqlite3
import pandas as pd

# 连接数据库（不存在则自动创建）
conn = sqlite3.connect('a_stock_data.db')
cursor = conn.cursor()

# 创建日线数据表
cursor.execute('''
CREATE TABLE IF NOT EXISTS daily_data (
    ts_code TEXT,
    trade_date TEXT,
    open REAL,
    high REAL,
    low REAL,
    close REAL,
    pre_close REAL,
    change REAL,
    pct_chg REAL,
    vol REAL,
    amount REAL,
    PRIMARY KEY (ts_code, trade_date)
)
''')

conn.commit()

设计思路：用(ts_code, trade_date)作为联合主键，确保同一只股票同一天的数据不会重复插入。这个设计我在项目中一直沿用，效果很好。

4.3.2 批量插入数据

# 批量插入数据（使用executemany）
data_list = df.values.tolist()
cursor.executemany('''
INSERT OR REPLACE INTO daily_data 
(ts_code, trade_date, open, high, low, close, pre_close, change, pct_chg, vol, amount)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', data_list)
conn.commit()

# 或者直接用pandas的to_sql
df.to_sql('daily_data', conn, if_exists='append', index=False)

这里我建议用INSERT OR REPLACE。为什么？因为如果某天数据有误，重新跑一遍就能覆盖更新，不会产生重复记录。你想想看，这比先删后插省事多了。

4.3.3 查询数据

# 查询某只股票某段时间的数据
query = '''
SELECT * FROM daily_data 
WHERE ts_code = '000001.SZ' 
AND trade_date BETWEEN '2023-01-01' AND '2023-12-31'
ORDER BY trade_date
'''
df_query = pd.read_sql(query, conn)

# 查询多只股票
query_multi = '''
SELECT ts_code, trade_date, close 
FROM daily_data 
WHERE ts_code IN ('000001.SZ', '000002.SZ')
AND trade_date >= '2023-06-01'
'''
df_multi = pd.read_sql(query_multi, conn)

4.4 数据更新与维护策略

数据不是一次性的活。股票市场每天都在产生新数据，你的数据库也得跟着更新。

我总结了一套「增量更新+全量校验」的策略，用了好几年，没出过问题。

4.4.1 增量更新策略

def update_daily_data(ts_code, start_date=None):
    """增量更新日线数据"""
    conn = sqlite3.connect('a_stock_data.db')
    
    # 获取数据库中已有数据的最大日期
    query = f"SELECT MAX(trade_date) FROM daily_data WHERE ts_code = '{ts_code}'"
    max_date = pd.read_sql(query, conn).iloc[0, 0]
    
    if max_date is None:
        # 没有数据，从指定日期开始获取
        start = start_date or '20100101'
    else:
        # 已有数据，从最大日期的下一天开始
        start = (pd.to_datetime(max_date) + pd.Timedelta(days=1)).strftime('%Y%m%d')
    
    # 获取新数据
    end = datetime.now().strftime('%Y%m%d')
    if start >= end:
        print(f"{ts_code} 数据已是最新")
        conn.close()
        return
    
    df_new = pro.daily(ts_code=ts_code, start_date=start, end_date=end)
    
    if df_new.empty:
        print(f"{ts_code} 没有新数据")
        conn.close()
        return
    
    # 插入新数据
    df_new.to_sql('daily_data', conn, if_exists='append', index=False)
    conn.commit()
    conn.close()
    print(f"{ts_code} 更新了 {len(df_new)} 条数据")

核心逻辑：先查数据库里最新日期，只拉取缺失的部分。这样既省时间，又省API调用次数。

4.4.2 全量校验策略

增量更新有个隐患——如果某天的数据后来被修正了，你的数据库里还是旧数据。所以，我建议定期做全量校验。

def verify_data(ts_code, year):
    """校验某年数据是否完整"""
    conn = sqlite3.connect('a_stock_data.db')
    
    # 从数据库查询该年数据
    db_query = f"""
    SELECT trade_date, close FROM daily_data 
    WHERE ts_code = '{ts_code}' 
    AND trade_date LIKE '{year}%'
    ORDER BY trade_date
    """
    df_db = pd.read_sql(db_query, conn)
    
    # 从Tushare重新获取该年数据
    df_api = pro.daily(ts_code=ts_code, 
                       start_date=f'{year}0101', 
                       end_date=f'{year}1231')
    
    # 对比数据条数
    if len(df_db) != len(df_api):
        print(f"{ts_code} {year}年数据不完整，数据库{len(df_db)}条，API{len(df_api)}条")
        # 重新拉取该年数据
        df_api.to_sql('daily_data', conn, if_exists='append', index=False)
        print("已重新拉取并覆盖")
    
    conn.close()

注意：全量校验不要频繁做，建议每月或每季度一次。毕竟Tushare有调用次数限制，你想想看，一天200次，校验100只股票就没了。

4.4.3 定时任务自动化

手动跑脚本太累了。我习惯用系统的定时任务来自动化。

Windows下用任务计划程序：

# 创建一个bat文件
python D:\quant\update_data.py

Linux/Mac下用crontab：

# 每天下午6点更新数据
30 18 * * * /usr/bin/python3 /home/quant/update_data.py

4.5 知识体系总览

下面这张图，是我对本章知识体系的总结。你看一眼，就能明白整个数据流水线的脉络。

金融数据获取与存储知识体系 数据源层 Tushare Pro AkShare 其他数据源 数据获取层 日线数据 分钟线数据 基本面数据 数据存储层 CSV文件 Excel文件 SQLite数据库 其他数据库 维护策略层 增量更新 全量校验 定时任务自动化

4.6 避坑指南

做数据获取和存储，我踩过的坑不少。分享几个给你，省得你重走弯路。

  • API调用频率限制：Tushare免费版一天200次，别一次性拉太多数据。我建议分批拉，每次拉完sleep一下。
  • 数据对齐问题：不同数据源的日期格式可能不同，一定要统一格式再存储。我习惯用YYYY-MM-DD格式。
  • 磁盘空间：SQLite文件会越来越大。我建议定期做VACUUM操作，回收空间。
  • 并发写入：SQLite不支持高并发写入。如果你的策略需要多进程同时写数据，考虑用PostgreSQL。

我的经验：刚开始做量化时，我每天手动跑脚本更新数据。后来数据量大了，手动跑太累，才做了自动化。建议你一开始就设计好自动化流程，省得后面重构。

好了，数据获取与存储就讲到这里。这套流程我用了好几年，从几千只股票到上亿条数据，都没出过问题。你照着做，也能搭建一个稳定的数据仓库。

专注资料整理

← 上一章 📖 返回目录 下一章 →