从Tushare迁移到AKShare v1.1.1：手把手教你用Python缓存股票历史数据，提速20分钟

从Tushare迁移到AKShare v1.1.1：Python股票数据缓存实战指南

在金融数据分析领域，数据源的稳定性和获取效率直接影响着量化策略的执行效果。最近不少开发者发现，曾经广泛使用的Tushare库由于接口变动，导致原有股票分析系统出现数据获取失败的情况。与此同时，AKShare作为新兴的金融数据接口库，凭借更丰富的接口和活跃的维护，正成为替代方案的热门选择。

本文将带您完成从Tushare到AKShare v1.1.1的完整迁移过程，重点解决三个核心问题：如何正确使用AKShare的新接口、如何设计高效的数据缓存机制，以及如何通过优化将数据获取时间缩短20分钟以上。无论您是在维护现有股票分析系统，还是构建新的量化交易框架，这些实战经验都能为您节省大量试错时间。

1. 迁移准备与环境配置

1.1 为什么选择AKShare？

AKShare相较于Tushare有几个显著优势：

• 接口丰富度：覆盖A股、港股、美股、期货、期权等多市场数据
• 维护活跃度：GitHub仓库保持高频更新，问题响应迅速
• 数据质量：直接对接交易所和主流财经平台的一手数据源
• 社区支持：中文文档完善，案例丰富

迁移前需要确认环境配置：

pip install akshare==1.1.1 pandas>=1.0.0

提示：建议使用虚拟环境隔离项目依赖，避免版本冲突

1.2 接口差异对比分析

Tushare和AKShare在历史数据接口上存在几个关键差异：

2. 核心接口迁移实战

2.1 基础数据获取

AKShare的stock_zh_a_hist接口是获取A股历史行情的主要入口。典型调用方式：

import akshare as ak def get_hist_data(code, start_date, end_date): """ 获取股票历史数据 :param code: 股票代码，如'000001' :param start_date: 开始日期，格式'20200101' :param end_date: 结束日期，格式'20201231' :return: DataFrame格式的历史行情 """ try: df = ak.stock_zh_a_hist( symbol=code, start_date=start_date, end_date=end_date, adjust="hfq" # 可选qfq(前复权), hfq(后复权), ""(不复权) ) # 统一列名格式 df.columns = ['date', 'open', 'close', 'high', 'low', 'volume', 'amount', 'amplitude', 'quote_change', 'ups_downs', 'turnover'] return df.sort_values('date') except Exception as e: print(f"获取{code}历史数据失败: {str(e)}") return None

2.2 常见问题排查

迁移过程中可能会遇到以下典型问题：

1. 参数不匹配：AKShare的adjust参数替代了Tushare的adj
2. 字段差异：AKShare默认返回更多技术指标数据
3. 版本差异：v1.1.1移除了period参数，统一使用日期范围查询
4. 数据精度：AKShare的成交量单位是股，而Tushare可能是手

注意：AKShare接口更新频繁，建议定期检查help(ak.stock_zh_a_hist)查看最新参数说明

3. 性能优化与缓存设计

3.1 缓存架构设计

直接频繁请求网络接口不仅速度慢，还可能触发反爬机制。我们设计了一个基于本地文件的二级缓存系统：

1. 内存缓存：使用Python字典暂存最近访问的数据
2. 磁盘缓存：将数据序列化保存到本地文件
3. 压缩存储：使用gzip压缩减少磁盘占用

缓存目录结构示例：

data/ ├── 2023-08/ # 按月归档 │ ├── 20230801_000001.pkl.gz │ └── 20230801_600000.pkl.gz └── 2023-09/ └── 20230901_000001.pkl.gz

3.2 缓存实现代码

import os import gzip import pickle from datetime import datetime import pandas as pd class StockDataCache: def __init__(self, base_dir="data"): self.base_dir = base_dir self.memory_cache = {} # 简单的内存缓存 self.cache_hits = 0 def _get_cache_path(self, code, end_date): """生成缓存文件路径""" month = end_date[:6] # 取年月部分 dir_path = os.path.join(self.base_dir, month) os.makedirs(dir_path, exist_ok=True) return os.path.join(dir_path, f"{end_date}_{code}.pkl.gz") def get_data(self, code, start_date, end_date): """优先从缓存获取数据""" cache_key = f"{code}_{start_date}_{end_date}" # 内存缓存检查 if cache_key in self.memory_cache: self.cache_hits += 1 return self.memory_cache[cache_key] # 磁盘缓存检查 cache_file = self._get_cache_path(code, end_date) if os.path.exists(cache_file): try: with gzip.open(cache_file, 'rb') as f: df = pickle.load(f) self.memory_cache[cache_key] = df self.cache_hits += 1 return df except: pass # 网络请求 df = ak.stock_zh_a_hist( symbol=code, start_date=start_date, end_date=end_date, adjust="" ) if df is not None: # 标准化列名 df.columns = ['date', 'open', 'close', 'high', 'low', 'volume', 'amount', 'amplitude', 'quote_change', 'ups_downs', 'turnover'] df = df.sort_values('date') # 更新缓存 self.memory_cache[cache_key] = df with gzip.open(cache_file, 'wb') as f: pickle.dump(df, f) return df

3.3 性能对比测试

我们对三种数据获取方式进行了基准测试（获取100只股票3年历史数据）：

4. 高级应用与技巧

4.1 批量获取与并行处理

当需要获取多只股票数据时，可以使用并发编程加速：

from concurrent.futures import ThreadPoolExecutor def batch_fetch(stock_list, start_date, end_date, max_workers=5): """ 批量获取股票历史数据 :param stock_list: 股票代码列表 :param start_date: 开始日期 :param end_date: 结束日期 :param max_workers: 最大并发数 :return: 字典{code: DataFrame} """ cache = StockDataCache() results = {} with ThreadPoolExecutor(max_workers=max_workers) as executor: future_map = { executor.submit(cache.get_data, code, start_date, end_date): code for code in stock_list } for future in concurrent.futures.as_completed(future_map): code = future_map[future] try: results[code] = future.result() except Exception as e: print(f"股票{code}获取失败: {str(e)}") return results

4.2 缓存维护策略

长期运行的系统中，缓存管理同样重要：

1. 定期清理：删除过期的月度缓存文件夹
2. 缓存验证：对比本地数据与最新网络数据
3. 容量监控：设置最大缓存大小限制

def clean_cache(months_to_keep=3): """保留最近N个月的缓存数据""" now = datetime.now() cutoff = (now.year * 100 + now.month) - months_to_keep cutoff = max(cutoff, 202300) # 保留2023年之后的数据 for dir_name in os.listdir(cache.base_dir): dir_path = os.path.join(cache.base_dir, dir_name) if os.path.isdir(dir_path): dir_date = int(dir_name.replace('-', '')) if dir_date < cutoff: shutil.rmtree(dir_path) print(f"已删除过期缓存: {dir_path}")

在实际项目中，这套缓存系统将数据获取时间从原来的45分钟缩短到了8分钟左右，同时减少了90%以上的网络请求。对于需要频繁回测或实时监控的场景，这种优化带来的效率提升尤为明显。