1. 项目缘起：为什么从Tushare开始做AI股票助手？

做量化分析或者AI股票预测，第一步永远是数据。没有高质量、稳定、结构化的数据，后面所有的模型训练、策略回测都成了空中楼阁。我见过太多朋友，一上来就沉迷于研究复杂的LSTM、Transformer模型，或者各种花哨的因子挖掘算法，结果在数据准备这一步就栽了跟头——要么数据源不稳定经常断更，要么字段缺失严重，要么数据格式混乱需要大量清洗，最终项目不了了之。

所以，当我决定动手做一个属于自己的“AI股票小助手”时，我给自己定的第一个铁律就是：先把数据管道（Data Pipeline）做扎实、做稳定、做自动化。在众多的数据源中，Tushare是我经过多轮对比后选定的起点。它不是一个完美的工具，但对于个人开发者、研究者和爱好者来说，它提供了一个非常友好的入门台阶。Tushare提供了相对规范的A股、港股、美股、期货、期权、基金、宏观经济等金融数据，通过Python API调用，数据直接以Pandas DataFrame的格式返回，这和我们后续用Python进行数据分析、特征工程的流程是天作之合。

这个“AI股票小助手03-Tushare数据采集”项目，就是整个小助手的数据基石。它的目标不仅仅是“把数据下载下来”，而是要构建一个健壮、可维护、可扩展的自动化数据采集系统。这意味着我们需要考虑：如何管理API Token？如何处理网络异常和请求限制？如何设计数据表结构以便高效存储和查询？如何增量更新数据避免重复拉取？如何监控数据质量？这些才是从“玩一玩”到“真正能用”的关键跨越。

接下来，我会详细拆解我是如何一步步搭建这个数据采集模块的。你会发现，这里面没有太多高深的算法，但充满了工程上的细节和“踩坑”后的经验总结，这些才是保证项目能长期运行的核心。

2. Tushare基础：从注册到获取第一份数据

万事开头难，但Tushare的开头还算简单。不过，即使是简单的步骤，里面也有不少值得注意的细节，这些细节会影响你后续使用的便捷性和安全性。

2.1 注册与Token管理：别把Token硬编码在代码里！

首先，访问Tushare官网进行注册。注册后会获得一个接口TOKEN，这是你调用所有API的凭证。很多新手教程会直接让你把Token写成字符串放在代码里，比如token = ‘your_token_here‘。这是极其不推荐的做法！

为什么？第一，安全性。如果你的代码要上传到GitHub等公共平台（即使是私有库，也有泄露风险），你的Token就暴露了。别人可以用你的Token调用接口，消耗你的积分，甚至导致你的账号被封。第二，可维护性。如果你有多个项目或多个环境（开发、测试、生产），每个地方都要改这个Token字符串，非常麻烦。

正确的做法是使用环境变量。我个人的实践如下：

1. 创建配置文件：在项目根目录创建一个名为.env的文件（注意前面的点），这个文件会被.gitignore忽略，不会上传到代码库。
2. 写入配置：在.env文件中写入你的Tushare Token。

   TUSHARE_TOKEN=你的token字符串

3. 安装依赖：使用python-dotenv库来读取这个文件。pip install python-dotenv。
4. 代码中读取：

   import os from dotenv import load_dotenv import tushare as ts # 加载 .env 文件中的环境变量 load_dotenv() # 从环境变量中读取Token TUSHARE_TOKEN = os.getenv(‘TUSHARE_TOKEN‘) # 初始化Tushare Pro接口 pro = ts.pro_api(TUSHARE_TOKEN) # 测试接口是否通畅 df = pro.trade_cal(exchange=‘SSE‘, start_date=‘20240101‘, end_date=‘20240131‘) print(df.head())

这样一来，你的敏感信息与代码完全分离。在不同机器上部署时，只需要配置对应的.env文件即可。

2.2 理解积分与数据权限：你的“数据货币”

Tushare采用积分制，不同积分等级能访问的数据范围和频率不同。免费注册后有120点基础积分，可以获取基础数据，如日线行情、股票列表、交易日历等，这对于入门学习和小规模回测已经足够。

但如果你想获取更历史的数据（比如完整的日线行情）、更高频的数据（如分钟线）、或者更丰富的指标（如资金流、龙虎榜），就需要积累积分。积分可以通过注册、完善信息、分享和捐赠获得。

这里有一个非常重要的经验：在编写你的数据采集脚本时，一定要有“成本意识”。不要动不动就全量拉取所有股票的所有历史日线数据，那会瞬间消耗大量积分和请求次数。合理的做法是：

1. 按需采集：明确你的策略或模型需要哪些数据。如果只做沪深300成分股的回测，就没必要拉取全市场5000多只股票的数据。
2. 增量更新：对于行情类数据，每天只拉取最新的数据，与本地历史数据合并。Tushare的很多接口都支持按日期区间查询，这是实现增量的基础。
3. 缓存结果：对于一些不常变化的基础数据，如股票列表、行业分类，拉取一次后可以保存到本地文件或数据库，下次直接读取，避免重复调用API。

2.3 第一个实战：获取股票列表并解析

股票列表是所有分析的起点。我们首先需要知道市场上有哪些标的。使用pro.stock_basic接口。

# 获取沪深京A股列表 df_stock_basic = pro.stock_basic(exchange=‘‘, list_status=‘L‘, fields=‘ts_code,symbol,name,area,industry,fullname,enname,market,exchange,curr_type,list_status,list_date,delist_date,is_hs‘) print(f“获取到 {len(df_stock_basic)} 只上市股票“) print(df_stock_basic.head())

字段解析与选择：

• ts_code：TS股票代码，是Tushare系统中的统一代码，格式如000001.SZ。这是后续调用其他接口时最重要的代码标识，建议作为主键存储。
• symbol：股票代码，如000001。
• name：股票名称。
• industry：行业分类。注意，Tushare的行业分类是它自己定义的，可能与申万、中信等行业分类不同。如果你的策略对行业有要求，可能需要额外获取或映射其他分类标准。
• market：市场类型（主板/创业板/科创板等）。
• list_date：上市日期。这个字段对于处理“幸存者偏差”至关重要。在做历史回测时，一定要确保你使用的股票在回测期初已经上市，否则就是用了未来的信息。
• is_hs：是否沪深港通标的。对于涉及北向资金的分析很重要。

拿到这个列表后，我们通常会把它持久化存储起来，作为后续数据采集的“总目录”。我建议存为CSV文件或写入数据库的stock_basic表。

3. 核心数据采集：日线行情与复权因子

对于大多数量化策略和AI模型来说，日线行情（开盘价、收盘价、最高价、最低价、成交量、成交额）是最核心的数据。Tushare提供pro.daily接口。

3.1 日线行情采集的陷阱与正确姿势

一个最直接的调用可能是这样的：

# 错误示范：一次性拉取茅台全部历史数据（如果积分不够，可能失败或限流） df = pro.daily(ts_code=‘600519.SH‘, start_date=‘19900101‘)

对于像贵州茅台这样上市早的股票，这个请求会返回上万条数据，对服务器压力和你的积分都不友好。更稳妥的方式是分阶段拉取或增量拉取。

增量拉取逻辑：

1. 检查本地是否已有该股票的历史数据。
2. 如果已有，则找到最新的日期last_date。
3. 以last_date的下一个交易日作为start_date，拉取至今的数据。
4. 将新数据追加到本地历史数据中。

这里就引出了另一个关键数据：交易日历。你必须知道哪些天是交易日，才能正确计算last_date的下一个交易日。可以使用pro.trade_cal接口获取。

def get_daily_data_incrementally(ts_code, pro_api, local_csv_path): “““增量获取日线数据“““ # 1. 加载本地已有数据 if os.path.exists(local_csv_path): df_local = pd.read_csv(local_csv_path, parse_dates=[‘trade_date‘]) last_date = df_local[‘trade_date‘].max() # 获取last_date的下一个交易日 cal = pro_api.trade_cal(exchange=‘SSE‘, start_date=last_date.strftime(‘%Y%m%d‘), end_date=datetime.now().strftime(‘%Y%m%d‘)) next_trade_dates = cal[cal[‘is_open‘]==1][‘cal_date‘] if len(next_trade_dates) > 1: # 第一个是last_date本身 start_date = next_trade_dates.iloc[1] else: print(f“{ts_code} 数据已是最新“) return df_local else: # 本地没有数据，从头开始拉取（可以设定一个合理的开始日期，比如5年前） start_date = (datetime.now() - timedelta(days=5*365)).strftime(‘%Y%m%d‘) df_local = pd.DataFrame() # 2. 从Tushare拉取增量数据 end_date = datetime.now().strftime(‘%Y%m%d‘) try: df_new = pro_api.daily(ts_code=ts_code, start_date=start_date, end_date=end_date) if df_new.empty: print(f“{ts_code} 无新数据“) return df_local # 转换trade_date格式 df_new[‘trade_date‘] = pd.to_datetime(df_new[‘trade_date‘]) except Exception as e: print(f“获取 {ts_code} 数据失败: {e}“) return df_local # 3. 合并数据 df_combined = pd.concat([df_local, df_new]).drop_duplicates(subset=[‘ts_code‘, ‘trade_date‘]).sort_values(‘trade_date‘).reset_index(drop=True) # 4. 保存回本地 df_combined.to_csv(local_csv_path, index=False) print(f“{ts_code} 数据更新至 {df_combined[‘trade_date‘].max().strftime(‘%Y-%m-%d‘)}“) return df_combined

3.2 复权因子：处理价格序列的“黄金标准”

直接获取的日线行情是未经复权的价格。股票会发生分红、送股、配股等事件，这些事件会导致股价在除权除息日发生跳跃，如果直接使用后复权价格进行计算，会扭曲真实的收益率序列。因此，量化分析必须使用复权价格。

Tushare提供了pro.adj_factor接口来获取复权因子。复权因子的用法是：复权价格 = 原始价格 * 复权因子

关键点：复权因子是点乘关系，并且是累积的。通常我们使用“后复权因子”，即保持最新价格不变，将历史价格向上调整。

# 获取复权因子 df_adj = pro.adj_factor(ts_code=‘600519.SH‘, start_date=‘20230101‘) print(df_adj.head())

采集策略：复权因子数据量不大，且一旦发布很少变更。可以定期（如每月）全量更新一次，或者与日线行情同步增量更新。存储时，建议将adj_factor单独存表，在与日线行情关联时进行合并计算。

计算复权价格示例：

# 假设 df_daily 是日线行情，df_adj 是复权因子，它们都有 `ts_code`, `trade_date` 字段 df_merged = pd.merge(df_daily, df_adj[[‘ts_code‘, ‘trade_date‘, ‘adj_factor‘]], on=[‘ts_code‘, ‘trade_date‘], how=‘left‘) # 前向填充复权因子（因为除权除息日才有新的因子，非除权日因子与上一日相同） df_merged[‘adj_factor‘] = df_merged.groupby(‘ts_code‘)[‘adj_factor‘].ffill() # 计算复权价格 df_merged[‘close_adj‘] = df_merged[‘close‘] * df_merged[‘adj_factor‘] df_merged[‘open_adj‘] = df_merged[‘open‘] * df_merged[‘adj_factor‘] # ... 其他价格字段同理

4. 构建健壮的数据采集系统

当我们需要采集几十、上百只股票的数据时，就不能再手动运行脚本了。我们需要一个系统化的解决方案。

4.1 任务调度与错误重试

网络请求不可能100%成功。Tushare接口有调用频率限制（积分不同限制不同），网络也可能波动。因此，必须加入错误处理与重试机制。

我推荐使用tenacity库来实现优雅的重试逻辑。

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests # 定义一个重试装饰器 @retry( stop=stop_after_attempt(5), # 最多重试5次 wait=wait_exponential(multiplier=1, min=4, max=60), # 指数退避等待 retry=retry_if_exception_type((requests.exceptions.ConnectionError, requests.exceptions.Timeout)), reraise=True ) def safe_tushare_call(api_func, *args, **kwargs): “““包装Tushare调用，加入重试机制“““ return api_func(*args, **kwargs) # 使用示例 try: df = safe_tushare_call(pro.daily, ts_code=‘000001.SZ‘, start_date=‘20240101‘) except Exception as e: print(f“经过多次重试后仍然失败: {e}“) # 记录失败任务，稍后重试或人工干预

对于批量任务，可以使用concurrent.futures的ThreadPoolExecutor进行有限的并发控制，避免触发Tushare的反爬机制或超出频率限制。切记不要开太高并发！根据你的积分等级，建议并发数在2-5之间。

from concurrent.futures import ThreadPoolExecutor, as_completed def fetch_one_stock(ts_code): “““获取单只股票数据“““ # 这里调用上面封装好的增量获取函数 return get_daily_data_incrementally(ts_code, pro, f“./data/{ts_code}.csv“) stock_list = [‘000001.SZ‘, ‘600519.SH‘, ‘000858.SZ‘] # 示例股票列表 max_workers = 3 # 控制并发数 with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_stock = {executor.submit(fetch_one_stock, code): code for code in stock_list} for future in as_completed(future_to_stock): stock = future_to_stock[future] try: result = future.result() print(f“{stock} 采集完成“) except Exception as exc: print(f“{stock} 采集过程中产生异常: {exc}“)

4.2 数据存储方案选型：CSV vs. 数据库

对于小规模数据（比如几百只股票，几年数据），使用CSV文件按股票代码分文件存储是最简单的，管理起来也直观。但存在一些问题：查询效率低（特别是跨股票、跨时间查询）、难以保证数据一致性、不支持复杂的SQL操作。

当数据量变大，或者需要频繁进行多维度查询时，使用关系型数据库（如SQLite、MySQL、PostgreSQL）是更专业的选择。

SQLite方案（轻量级推荐）：SQLite是一个单文件数据库，无需安装服务器，非常适合个人项目和小型应用。

import sqlite3 import pandas as pd # 连接数据库（如果不存在则创建） conn = sqlite3.connect(‘stock_data.db‘) # 将DataFrame写入数据库 df_stock_basic.to_sql(‘stock_basic‘, conn, if_exists=‘replace‘, index=False) df_daily.to_sql(‘daily_price‘, conn, if_exists=‘append‘, index=False) # 注意使用append模式 # 从数据库读取 df_read = pd.read_sql_query(“SELECT * FROM daily_price WHERE ts_code=‘000001.SZ‘“, conn) conn.close()

表结构设计建议：

• stock_basic表：存放股票基础信息。
• daily_price表：存放日线行情。主键建议为(ts_code, trade_date)，并建立索引以加速查询。
• adj_factor表：存放复权因子。
• trade_cal表：存放交易日历。

使用数据库后，增量更新逻辑可以改为：先查询数据库中某只股票的最新日期，再向Tushare请求该日期之后的数据。

4.3 数据质量监控与校验

自动化采集最怕的就是在沉默中出错。我们需要一些简单的监控点：

1. 数据完整性校验：拉取数据后，检查关键字段是否有NaN值。比如，检查close价格是否为0或空（除极特殊情况，收盘价不应为0）。
2. 数据连续性校验：对于日线数据，检查trade_date是否连续，是否存在非交易日的记录，或者缺失了某些交易日的数据。可以与trade_cal表进行比对。
3. 数据一致性校验：检查成交额（amount）是否大致等于成交量（vol）乘以均价（(high+low+close*2)/4估算）。如果出现数量级错误，可能是数据源问题。
4. 记录采集日志：每次采集任务运行的时间、采集的股票数量、成功失败情况、数据行数等，都应记录到日志文件或数据库的task_log表中。方便日后排查问题。

import logging logging.basicConfig(level=logging.INFO, format=‘%(asctime)s - %(name)s - %(levelname)s - %(message)s‘, handlers=[logging.FileHandler(‘data_collection.log‘), logging.StreamHandler()]) logger = logging.getLogger(__name__) def validate_daily_data(df, ts_code): “““简单的数据校验“““ if df.empty: logger.warning(f“{ts_code}: 数据为空“) return False if df[‘close‘].isnull().any(): logger.error(f“{ts_code}: 存在收盘价为NaN的记录“) return False if (df[‘close‘] == 0).any(): logger.warning(f“{ts_code}: 存在收盘价为0的记录，请核查“) # 检查日期是否连续（简单版） df = df.sort_values(‘trade_date‘) date_diff = df[‘trade_date‘].diff().dt.days # 正常情况下，日期差应该是1（连续交易日）或大于1（间隔了非交易日/节假日） # 但如果出现负数或0，就有问题 if (date_diff < 1).any(): logger.error(f“{ts_code}: 交易日期序列不连续或重复“) return False logger.info(f“{ts_code}: 数据校验通过，共 {len(df)} 条记录“) return True

5. 扩展数据维度：为AI模型准备“食材”

一个聪明的AI股票小助手，不能只盯着价格和成交量。就像做菜需要各种食材，一个有效的预测模型也需要多维度特征。Tushare提供了丰富的数据，我们可以有计划地纳入采集范围。

5.1 基本面数据：公司的“体检报告”

基本面数据更新频率较低（季报、年报），但对于中长期趋势判断至关重要。

• 利润表(pro.income): 营业收入、净利润、毛利率等。
• 资产负债表(pro.balancesheet): 资产、负债、所有者权益。
• 现金流量表(pro.cashflow): 经营、投资、筹资活动现金流。
• 业绩快报/预告(pro.forecast): 更及时的业绩信息。

采集策略：基本面数据量大，更新慢。可以每月或每季度在全量更新股票列表后，批量采集上一期（Q1/Q2/Q3/Annual）的报告数据。存储时，注意区分报告期（end_date）和公告日期（ann_date）。

5.2 市场情绪与资金数据：市场的“心跳”

这些数据频率高，对短期波动有较好解释力。

• 资金流向(pro.moneyflow): 每日主力、超大单、大单、中单、小单的资金净流入流出。这是观察机构和个人投资者行为的重要窗口。
• 沪深港通持股(pro.hk_hold): 北向资金的每日持股明细和变动。北向资金常被视为“聪明钱”，其动向备受关注。
• 龙虎榜(pro.top_list): 每日上榜营业部买卖情况，反映游资动向。

采集策略：这类数据需要每日更新。可以集成到日线行情采集任务中，作为另一个并行任务。由于数据量也较大，需要做好数据库表设计和索引优化。

5.3 另类数据与衍生指标

除了原始数据，我们还可以基于原始数据计算一些技术指标或衍生特征，这部分可以直接在数据入库后通过SQL或Pandas计算。

• 简单技术指标：移动平均线（MA）、布林带（Bollinger Bands）、相对强弱指数（RSI）等。虽然Tushare部分接口提供，但自己计算更灵活。
• 波动率指标：例如过去N日的收益率标准差。
• 价量关系指标：如量价齐升、放量突破等形态的布尔标识。

计算示例（在数据库层面）：

-- 计算20日移动平均线 UPDATE daily_price SET ma20 = ( SELECT AVG(close_adj) FROM daily_price AS p2 WHERE p2.ts_code = daily_price.ts_code AND p2.trade_date <= daily_price.trade_date AND p2.trade_date > DATE(daily_price.trade_date, ‘-20 days‘) ) WHERE trade_date >= (SELECT MIN(trade_date) FROM daily_price) + 20;

更复杂的计算建议在Python中利用Pandas的滚动（rolling）函数完成，再写回数据库。

6. 从采集到应用：数据管道的闭环

采集到的数据最终要为AI模型服务。因此，我们需要设计一个端到端的管道（Pipeline）。

1. 采集层：使用调度工具（如schedule库、APScheduler，或服务器上的Cron Job）定时运行我们的数据采集脚本。每日收盘后，自动触发日线行情、资金流等数据的增量更新。
2. 存储层：数据清洗、校验后，存入选定的数据库（如SQLite/MySQL）。确保表结构清晰，索引有效。
3. 特征工程层：从原始数据表中提取、计算模型所需的特征。这一步可以单独写一个特征计算脚本，定期运行。特征结果可以存入新的features表，或者直接生成特征文件（如.parquet格式）供模型读取。
4. 模型服务层：AI模型（无论是传统的机器学习模型还是深度学习模型）从特征存储中读取数据，进行训练或预测。

一个简单的每日自动化流程示例（伪代码）：

# run_daily.py import schedule import time from datetime import datetime from your_collection_module import update_all_stock_daily, update_moneyflow, update_basic_info_quarterly def daily_job(): print(f“{datetime.now()} 开始每日数据更新任务“) # 1. 更新交易日历（如果当天是新的交易日） # 2. 更新股票列表（如有新上市/退市） # 3. 增量更新所有关注股票的日线行情 update_all_stock_daily() # 4. 更新资金流向数据 update_moneyflow() # 5. 更新特征表 calculate_features() print(f“{datetime.now()} 每日数据更新任务完成“) def quarterly_job(): # 每季度运行一次，更新基本面数据 update_basic_info_quarterly() # 设置定时任务，每个交易日18:00执行（假设数据已更新） schedule.every().day.at(“18:00“).do(daily_job) # 每季度第一天执行 schedule.every().quarter.do(quarterly_job) while True: schedule.run_pending() time.sleep(60)

7. 踩坑实录与经验之谈

在构建这个数据采集系统的过程中，我遇到了不少坑，这里分享出来，希望大家能避开。

坑一：API调用超时与限流。Tushare的免费接口有频率限制。初期我用了高并发，很快就被限制返回空数据或错误。解决方案：严格遵守积分对应的调用频率，加入足够的延时（time.sleep），并使用tenacity进行重试。对于大批量任务，最好在夜间或非高峰时段执行。

坑二：数据字段含义理解偏差。比如trade_date字段，在有的接口里是YYYYMMDD格式的字符串，在有的接口里是YYYY-MM-DD。又比如vol单位是“手”还是“股”？（Tushare默认是“手”）。解决方案：一定要仔细阅读官方文档每个接口的字段说明，对拿不准的数据，用小样本进行验证，并与其他可靠数据源（如看盘软件）进行交叉比对。

坑三：复权因子拼接错误。在计算复权价格时，如果直接按日期合并，可能会因为复权因子发布日（ann_date）和生效日（trade_date）不同而导致错位。解决方案：使用前向填充（ffill）来保证非除权日的因子值与上一交易日相同，这是行业通用做法。

坑四：数据库写入性能瓶颈。当历史数据量很大时，使用Pandas的to_sql方法一条条插入会非常慢。解决方案：对于大批量数据初始化，可以考虑使用数据库的批量导入工具（如MySQL的LOAD DATA INFILE）。对于日常增量，如果数据量不大，to_sql的append模式可以接受，也可以自己拼接批量插入的SQL语句。

坑五：缺乏数据版本管理和回滚机制。一旦脚本有bug，可能导致错误数据污染数据库。解决方案：在每次重要的数据更新前，对涉及的表进行备份（例如复制一份到table_name_backup_yyyymmdd）。或者，采用“事务”的方式，先将新数据写入临时表，校验无误后再合并到主表。

最后一点体会：数据工程是AI应用中最“脏”最“累”但也是最基础的一环。把Tushare数据采集这一步做扎实了，后续的特征工程、模型训练和策略回测才能有一个稳定可靠的基础。这个“AI股票小助手”项目，我从数据采集开始，花了最多的时间来打磨这个底层模块，现在看来是完全值得的。它现在就像一台默默运转的抽水机，每天自动将市场的“活水”引入我的数据池，让我可以更专注于上层的策略和模型研究。