PythonStock实战：从tushare到akShare的平滑迁移，利用ak.stock_zh_a_hist重构历史数据模块

1. 为什么要从tushare迁移到akShare？

最近在维护一个Python股票分析系统时，遇到了一个棘手的问题：原先使用的tushare接口突然开始收费了。作为一个长期依赖免费数据源的项目，这无疑是个重大打击。经过一番调研，我发现akShare这个开源库不仅完全免费，而且数据源更丰富，更新也更及时。

tushare曾经是Python金融数据分析的首选工具，但自从它转向商业化运营后，免费接口的功能越来越受限。相比之下，akShare的优势很明显：

• 数据覆盖更全面，包含A股、港股、美股等多个市场
• 接口更新频率高，能及时跟上交易所规则变化
• 完全开源免费，没有调用次数限制
• 社区活跃，问题响应速度快

不过迁移过程并非一帆风顺。我在实际项目中遇到了几个典型问题：

1. 接口参数命名不一致（比如tushare用ts_code，akShare用symbol）
2. 返回数据字段有差异
3. 日期格式要求不同
4. 复权处理方式有别

最让我头疼的是历史数据接口的差异。原先使用tushare的daily接口获取日线数据，现在需要改用ak.stock_zh_a_hist。这个接口虽然功能强大，但参数设置和返回格式都需要仔细调整。

2. 认识ak.stock_zh_a_hist接口

2.1 接口基本用法

ak.stock_zh_a_hist是akShare中获取A股历史行情的主力接口。与tushare的pro.daily相比，它有以下几个特点：

import akshare as ak # 基本调用方式 df = ak.stock_zh_a_hist( symbol="600519", # 股票代码 start_date="20230101", # 开始日期 end_date="20231231", # 结束日期 adjust="hfq" # 复权方式 )

关键参数说明：

• symbol：支持带市场前缀（sh/sz）和不带前缀两种格式
• adjust：支持"qfq"(前复权)、"hfq"(后复权)、""(不复权)
• 日期格式统一为YYYYMMDD，比tushare更严格

2.2 返回数据结构解析

接口返回的是一个包含11列的DataFrame：

1. date：交易日期
2. open：开盘价
3. close：收盘价
4. high：最高价
5. low：最低价
6. volume：成交量（股）
7. amount：成交额（元）
8. amplitude：振幅
9. quote_change：涨跌幅
10. ups_downs：涨跌额
11. turnover：换手率

与tushare返回的数据相比，主要差异在：

• 字段命名更直观（如涨跌幅从pct_chg变为quote_change）
• 增加了振幅和换手率指标
• 成交量的单位是股而非手

2.3 版本兼容性注意

akShare更新频繁，不同版本接口可能有变化。比如在v1.1.0之前，还有个period参数控制周期，但在v1.1.1中已经移除了。建议通过help命令随时查看最新文档：

help(ak.stock_zh_a_hist)

如果遇到参数错误，很可能是版本不匹配导致的。可以通过pip install --upgrade akshare保持最新版本。

3. 迁移实战：代码改造指南

3.1 基础接口替换

原先使用tushare的代码可能是这样的：

import tushare as ts pro = ts.pro_api() df = pro.daily(ts_code='600519.SH', start_date='20230101', end_date='20231231')

改造后的akShare版本：

import akshare as ak df = ak.stock_zh_a_hist(symbol='600519', start_date='20230101', end_date='20231231')

需要注意的改造点：

1. 股票代码要去掉.SH/.SZ后缀
2. 日期格式保持YYYYMMDD
3. 返回字段名称需要统一修改

3.2 字段映射与转换

由于返回字段不同，需要建立映射关系：

特别要注意的是成交量单位转换。tushare返回的是"手"，而akShare返回的是"股"，需要做相应处理：

# 如果下游代码依赖tushare的成交量单位 df['volume'] = df['volume'] / 100 # 将股转换为手

3.3 复权处理差异

tushare和akShare的复权处理方式略有不同：

# tushare复权方式 df = pro.daily(ts_code='600519.SH', adj='hfq') # akShare等效写法 df = ak.stock_zh_a_hist(symbol='600519', adjust='hfq')

实测发现两种复权算法结果可能有微小差异，但对大多数分析场景影响不大。如果对精度要求极高，建议用不复权数据自行计算。

4. 性能优化与缓存策略

4.1 本地缓存实现

频繁请求网络接口不仅速度慢，还可能触发反爬限制。我在项目中实现了一个本地缓存方案：

import os import pandas as pd from datetime import datetime def get_hist_with_cache(code, start_date, end_date, adjust=""): cache_dir = f"./cache/{datetime.now().strftime('%Y%m')}/" os.makedirs(cache_dir, exist_ok=True) cache_file = f"{cache_dir}{code}_{start_date}_{end_date}_{adjust}.pkl" if os.path.exists(cache_file): return pd.read_pickle(cache_file) df = ak.stock_zh_a_hist( symbol=code, start_date=start_date, end_date=end_date, adjust=adjust ) df.to_pickle(cache_file) return df

这个方案有几个优化点：

1. 按月份分目录存储，便于管理
2. 使用pickle格式，读写速度快
3. 文件名包含所有查询参数，避免冲突

4.2 批量请求优化

获取多只股票历史数据时，可以结合多线程加速：

from concurrent.futures import ThreadPoolExecutor def batch_get_hist(codes, start_date, end_date): with ThreadPoolExecutor(max_workers=5) as executor: futures = [ executor.submit( get_hist_with_cache, code, start_date, end_date ) for code in codes ] return [f.result() for f in futures]

实测下来，5个线程并发可以将获取100只股票3年历史数据的时间从15分钟缩短到3分钟左右。

4.3 缓存更新策略

对于长期运行的系统，还需要考虑缓存更新问题。我的做法是：

1. 每日收盘后自动更新当天数据
2. 每周检查一次历史数据的完整性
3. 每月归档旧缓存，避免单个文件过大

def update_daily_cache(): today = datetime.now().strftime("%Y%m%d") for code in watch_list: df = get_hist_with_cache(code, today, today) # 后续处理...

5. 常见问题与解决方案

5.1 接口限流与重试机制

虽然akShare没有官方限流，但数据源网站可能有访问限制。建议添加重试逻辑：

import time from requests.exceptions import RequestException def safe_get_hist(code, start_date, end_date, retry=3): for i in range(retry): try: return ak.stock_zh_a_hist( symbol=code, start_date=start_date, end_date=end_date ) except RequestException as e: if i == retry - 1: raise time.sleep(2 ** i) # 指数退避

5.2 数据校验与清洗

从不同渠道获取的数据需要验证质量。我通常会检查：

1. 是否有缺失交易日
2. 价格异常值（如收盘价为0）
3. 成交量突变（超过3倍标准差）

def validate_data(df): # 检查日期连续性 dates = pd.to_datetime(df['date']) if (dates.diff().dropna() != pd.Timedelta(days=1)).any(): print("警告：存在缺失交易日") # 检查价格有效性 if (df['close'] <= 0).any(): print("警告：存在无效���格") # 检查成交量异常 mean_vol = df['volume'].mean() std_vol = df['volume'].std() if ((df['volume'] - mean_vol).abs() > 3 * std_vol).any(): print("警告：存在异常成交量")

5.3 时区与交易日处理

如果系统需要处理多市场数据，要注意时区问题。我的解决方案是：

import pytz def localize_time(df, market): tz_map = { 'SH': 'Asia/Shanghai', 'SZ': 'Asia/Shanghai', 'US': 'America/New_York' } df['date'] = pd.to_datetime(df['date']).dt.tz_localize(tz_map[market]) return df

对于交易日历，可以使用ak.tool_trade_date_hist_sina()接口获取。