Python+Tushare Pro：构建你的A股数据自动化采集系统

在量化投资和金融分析的领域，数据就像石油一样珍贵。但不同于石油的是，金融数据的获取往往更加复杂——要么需要支付高昂的费用，要么需要在各个财经网站间手动复制粘贴，既耗时又容易出错。这正是Tushare Pro这样的工具存在的意义：它提供了一个稳定、免费且可通过代码自动化访问的A股数据接口，让数据采集变得简单高效。

对于刚接触量化分析的Python开发者或个人投资者来说，建立一个本地化的金融数据库是迈向自动化交易和系统化分析的第一步。本文将带你从零开始，通过Tushare Pro构建一个完整的A股数据采集系统，涵盖行情、财务和市场参考数据，并教你如何将这些数据持久化存储，形成自己的"数据金矿"。

1. Tushare Pro入门：从注册到第一个API调用

1.1 为什么选择Tushare Pro

在众多金融数据接口中，Tushare Pro脱颖而出有几个关键原因：

• 免费额度充足 ：基础版提供每日500次调用，对个人用户足够
• 数据全面 ：覆盖股票、基金、期货、债券、外汇等多个市场
• 接口稳定 ：专业团队维护，API响应速度快且稳定
• Python友好 ：原生支持Python，与Pandas无缝集成

与其他数据源相比，Tushare Pro的优势显而易见：

数据源	免费额度	数据延迟	接口稳定性	Python支持
Tushare Pro	500次/日	实时	高	原生支持
新浪财经	无限制	15分钟	一般	需爬虫
东方财富	无限制	5分钟	一般	需爬虫
Wind终端	付费	实时	高	需插件

1.2 快速开始：安装与配置

安装Tushare Pro非常简单，只需一行pip命令：

pip install tushare --upgrade

安装完成后，你需要前往 Tushare Pro官网 注册账号并获取API Token。注册过程完全免费，只需提供基本的邮箱信息。

获取Token后，可以在Python中这样设置：

import tushare as ts

# 设置你的Token（替换为实际获取的值）
ts.set_token('你的Token')

# 创建API接口对象
pro = ts.pro_api()

提示：将Token保存在安全的地方，避免在代码中直接硬编码。可以考虑使用环境变量或配置文件管理敏感信息。

2. 核心数据接口详解与实战

2.1 行情数据：获取历史K线

行情数据是量化分析的基础，Tushare Pro提供了多种时间维度的行情接口：

# 获取日线行情
df_daily = pro.daily(ts_code='600519.SH', start_date='20200101', end_date='20201231')

# 获取周线行情
df_weekly = pro.weekly(ts_code='600519.SH', start_date='20200101', end_date='20201231')

# 获取月线行情
df_monthly = pro.monthly(ts_code='600519.SH', start_date='20200101', end_date='20201231')

返回的数据包含标准的OHLCV（开盘价、最高价、最低价、收盘价、成交量）格式，非常适合技术分析。

对于更复杂的行情需求，可以使用 pro_bar 接口获取复权数据：

# 获取前复权日线
df_adj = ts.pro_bar(ts_code='600519.SH', adj='qfq', start_date='20200101', end_date='20201231')

2.2 财务数据：深入公司基本面

财务数据是价值投资的核心，Tushare Pro提供了完整的财务报告接口：

# 获取利润表数据
df_income = pro.income(ts_code='600519.SH', start_date='20200101', end_date='20201231')

# 获取资产负债表
df_balancesheet = pro.balancesheet(ts_code='600519.SH', start_date='20200101', end_date='20201231')

# 获取现金流量表
df_cashflow = pro.cashflow(ts_code='600519.SH', start_date='20200101', end_date='20201231')

此外，Tushare Pro还提供了常用的财务指标接口，可以直接获取PE、PB等估值指标：

# 获取财务指标数据
df_fina = pro.fina_indicator(ts_code='600519.SH', start_date='20200101', end_date='20201231')

2.3 市场参考数据：把握整体市场脉搏

除了个股数据，Tushare Pro还提供了丰富的市场参考数据：

# 获取融资融券数据
df_margin = pro.margin(start_date='20200101', end_date='20201231')

# 获取沪深港通资金流向
df_moneyflow = pro.moneyflow_hsgt(start_date='20200101', end_date='20201231')

# 获取龙虎榜数据
df_toplist = pro.top_list(start_date='20200101', end_date='20201231')

这些数据对于判断市场整体情绪和资金流向非常有价值。

3. 构建本地数据仓库

3.1 数据存储方案比较

将Tushare Pro获取的数据本地化存储有多种选择：

存储方式	优点	缺点	适用场景
CSV	简单易用，无需数据库	查询效率低，无索引	小规模数据，临时分析
SQLite	轻量级，单文件	并发性能一般	个人使用，中等规模
MySQL	性能好，功能完善	需要单独安装维护	团队协作，大规模数据
MongoDB	灵活，适合非结构化	资源占用较大	复杂数据结构

3.2 使用SQLite构建本地数据库

对于个人用户，SQLite是一个理想的轻量级选择。以下是如何将Tushare数据存储到SQLite的完整示例：

import sqlite3
from datetime import datetime
import tushare as ts

# 初始化Tushare Pro
ts.set_token('你的Token')
pro = ts.pro_api()

# 创建SQLite数据库连接
conn = sqlite3.connect('stock_data.db')
cursor = conn.cursor()

# 创建日线行情表
cursor.execute('''
CREATE TABLE IF NOT EXISTS daily (
    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)
)
''')

# 获取并存储贵州茅台日线数据
df = pro.daily(ts_code='600519.SH', start_date='20200101', end_date='20201231')
df.to_sql('daily', conn, if_exists='append', index=False)

# 关闭连接
conn.close()

3.3 自动化数据更新策略

为了保持数据的时效性，可以设置定时任务自动更新数据。以下是一个简单的自动化脚本框架：

import schedule
import time
from datetime import datetime, timedelta

def update_daily_data():
    # 获取昨天的日期
    yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y%m%d')
    
    # 连接数据库
    conn = sqlite3.connect('stock_data.db')
    
    # 获取已存储的最新日期
    latest_date = pd.read_sql('SELECT MAX(trade_date) FROM daily', conn).iloc[0,0]
    
    # 如果数据库为空或需要更新
    if latest_date is None or latest_date < yesterday:
        # 获取数据并存储
        df = pro.daily(ts_code='600519.SH', start_date=latest_date, end_date=yesterday)
        df.to_sql('daily', conn, if_exists='append', index=False)
    
    conn.close()

# 每天下午6点执行更新
schedule.every().day.at("18:00").do(update_daily_data)

while True:
    schedule.run_pending()
    time.sleep(60)

注意：实际应用中需要考虑错误处理、API调用限制等因素，建议使用专业的任务调度工具如Airflow或Celery。

4. 高级应用与性能优化

4.1 批量获取多只股票数据

Tushare Pro的免费版对API调用有限制，合理设计请求可以最大化利用有限的调用次数：

# 获取股票列表
stocks = pro.stock_basic(exchange='', list_status='L')['ts_code'].tolist()

# 分批获取数据
batch_size = 10  # 每次获取10只股票
for i in range(0, len(stocks), batch_size):
    batch = stocks[i:i+batch_size]
    df = pro.daily(ts_code=','.join(batch), start_date='20200101', end_date='20201231')
    # 存储数据...

4.2 使用多线程加速数据采集

对于大量数据采集，可以使用多线程提高效率：

from concurrent.futures import ThreadPoolExecutor

def fetch_stock_data(ts_code):
    try:
        df = pro.daily(ts_code=ts_code, start_date='20200101', end_date='20201231')
        return df
    except Exception as e:
        print(f"Error fetching {ts_code}: {e}")
        return None

# 获取股票列表
stocks = pro.stock_basic(exchange='', list_status='L')['ts_code'].tolist()[:50]  # 限制前50只

# 使用线程池
with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(fetch_stock_data, stocks))

# 合并结果
all_data = pd.concat([df for df in results if df is not None])

4.3 数据质量检查与清洗

从API获取的数据可能存在缺失或异常值，需要进行质量检查：

def check_data_quality(df):
    # 检查缺失值
    missing = df.isnull().sum()
    
    # 检查异常值
    anomalies = {
        'negative_price': (df['close'] <= 0).sum(),
        'volume_zero': (df['vol'] == 0).sum(),
        'pct_chg_extreme': (df['pct_chg'].abs() > 20).sum()
    }
    
    return {'missing': missing, 'anomalies': anomalies}

# 示例使用
quality_report = check_data_quality(all_data)
print(quality_report)

对于发现的问题，可以使用Pandas进行清洗：

# 填充缺失值（前向填充）
df.fillna(method='ffill', inplace=True)

# 处理异常值（设置为NaN）
df.loc[df['close'] <= 0, 'close'] = None
df.loc[df['vol'] == 0, 'vol'] = None

5. 从数据到洞察：基础分析示例

有了本地数据仓库后，可以进行各种分析。以下是几个简单的示例：

5.1 计算移动平均线

# 从数据库读取数据
conn = sqlite3.connect('stock_data.db')
df = pd.read_sql('SELECT * FROM daily WHERE ts_code="600519.SH" ORDER BY trade_date', conn)
conn.close()

# 计算5日、20日、60日均线
df['ma5'] = df['close'].rolling(5).mean()
df['ma20'] = df['close'].rolling(20).mean()
df['ma60'] = df['close'].rolling(60).mean()

# 可视化
import matplotlib.pyplot as plt
plt.figure(figsize=(12,6))
plt.plot(df['trade_date'], df['close'], label='Close')
plt.plot(df['trade_date'], df['ma5'], label='MA5')
plt.plot(df['trade_date'], df['ma20'], label='MA20')
plt.plot(df['trade_date'], df['ma60'], label='MA60')
plt.legend()
plt.show()

5.2 财务指标分析

# 获取财务指标数据
df_fina = pro.fina_indicator(ts_code='600519.SH', start_date='20100101', end_date='20201231')

# 分析ROE变化
plt.figure(figsize=(12,6))
plt.plot(pd.to_datetime(df_fina['end_date']), df_fina['roe'], 'o-')
plt.title('ROE Trend')
plt.xlabel('Date')
plt.ylabel('ROE (%)')
plt.grid()
plt.show()

5.3 简单的选股策略

# 获取所有股票的每日指标
df_daily_basic = pro.daily_basic(trade_date='20201231')

# 筛选PE<15, PB<2, 市值>100亿的股票
condition = (df_daily_basic['pe'] < 15) & (df_daily_basic['pb'] < 2) & (df_daily_basic['total_mv'] > 1000000)
selected = df_daily_basic[condition]

# 获取这些股票的基本信息
stock_list = selected['ts_code'].tolist()
df_stocks = pro.stock_basic(ts_code=','.join(stock_list))

# 合并结果
result = pd.merge(selected, df_stocks, on='ts_code')
print(result[['ts_code', 'name', 'industry', 'pe', 'pb', 'total_mv']])

在实际项目中，我发现将Tushare Pro与本地数据库结合使用时，合理设计数据表结构和索引能显著提高查询效率。例如，为常用的查询字段（如ts_code和trade_date）创建复合索引，可以使数据检索速度提升数倍。此外，定期对数据库进行维护（如VACUUM和ANALYZE）也能保持其最佳性能。