利用AKShare批量获取上海期货交易所多品种历史行情数据

1. 为什么需要批量获取期货行情数据

做期货分析的朋友都知道，单日行情数据就像一张照片，只能反映某个瞬间的市场状态。而真正有价值的分析需要连续的时间序列数据，就像一段视频能记录市场变化的完整过程。我刚开始做期货分析时，每天手动下载数据再拼接，不仅效率低下，还经常漏掉关键交易日的数据。

AKShare这个开源库完美解决了这个问题。它封装了上海期货交易所的官方数据接口，我们可以用几行代码就获取任意时间段的多品种行情。比如最近我需要分析铜、铝、黄金三个品种过去半年的走势，传统方法可能要花半天时间收集整理数据，现在用AKShare只需要5分钟。

批量获取数据的核心价值在于：

• 趋势分析：计算20日均线、MACD等指标至少需要一个月的数据
• 策略回测：验证交易策略有效性通常需要1-3年的历史数据
• 相关性研究：不同品种间的价格联动分析需要同步的时间序列
• 风险监控：波动率计算需要足够长的历史数据作为基准

2. 环境准备与AKShare安装

在开始写代码前，我们需要准备好Python环境。推荐使用Anaconda创建独立的虚拟环境，避免与其他项目的依赖冲突。这是我常用的环境配置命令：

conda create -n futures python=3.8 conda activate futures pip install akshare pandas numpy

AKShare的安装可能会遇到几个常见问题：

1. 如果出现SSL证书错误，可以尝试：

   import ssl ssl._create_default_https_context = ssl._create_unverified_context

2. 网络不稳定时建议使用国内镜像源：

   pip install -i https://pypi.tuna.tsinghua.edu.cn/simple akshare

验证安装是否成功：

import akshare as ak print(ak.__version__) # 应该输出类似1.3.0的版本号

我建议用Jupyter Notebook进行开发，它的交互式特性特别适合数据探索。安装方法：

pip install jupyterlab jupyter lab

3. 单日行情获取基础

我们先从基础的单日数据获取开始，这是批量操作的基础。AKShare提供了get_shfe_daily()接口，接收一个格式为YYYYMMDD的日期字符串：

import akshare as ak from datetime import datetime # 获取当天行情 today = datetime.now().strftime("%Y%m%d") data = ak.get_shfe_daily(today) # 优化显示设置 import pandas as pd pd.set_option('display.max_columns', None) pd.set_option('display.width', 1000) print(data.head())

这个接口返回的DataFrame包含这些关键字段：

• symbol：合约代码，如CU2404表示2024年4月到期的铜期货
• open/high/low/close：开盘价、最高价、最低价、收盘价
• volume：成交量（手）
• open_interest：持仓量（手）
• turnover：成交金额（万元）
• settle：结算价
• pre_settle：前结算价

注意几个常见问题：

1. 节假日会返回空数据，建议先检查日期是否为交易日
2. 部分远期合约可能没有成交数据
3. 夜盘数据会包含在次日的数据中

4. 实现多日数据批量获取

批量获取的核心是构造日期序列，然后循环调用单日接口。这里分享我优化过的代码：

def get_shfe_multi_days(start_date, end_date): """获取指定日期范围的SHFE数据""" date_range = pd.date_range(start=start_date, end=end_date) all_data = [] for date in date_range: date_str = date.strftime("%Y%m%d") try: daily_data = ak.get_shfe_daily(date_str) if not daily_data.empty: daily_data['trade_date'] = date_str # 添加日期列 all_data.append(daily_data) print(f"已获取 {date_str} 数据") except Exception as e: print(f"获取 {date_str} 数据失败: {str(e)}") return pd.concat(all_data) if all_data else pd.DataFrame() # 示例：获取2024年1月数据 jan_data = get_shfe_multi_days("20240101", "20240131")

这段代码有几个优化点：

1. 自动跳过没有数据的日期（如节假日）
2. 为每条数据添加交易日期标记
3. 加入异常捕获避免中途失败
4. 显示进度反馈

对于大量数据获取，建议：

• 分批次获取（如按月获取）
• 添加随机延时避免被封禁
• 本地保存中间结果

5. 多品种数据处理技巧

获取到的数据包含所有品种，我们通常需要按品种拆分处理。这是我的标准处理流程：

# 按品种分组 grouped = jan_data.groupby('variety') # 示例：提取铜期货主力合约 def get_main_contract(group): return group[group['volume'] == group['volume'].max()].iloc[0] cu_main = grouped.get_group('CU').groupby('trade_date').apply(get_main_contract) # 保存到CSV cu_main.to_csv('cu_main_contract_jan2024.csv', index=False)

对于多品种分析，可以构建品种字典：

products = { 'CU': '铜', 'AL': '铝', 'AU': '黄金', 'SC': '原油' } for code, name in products.items(): try: df = grouped.get_group(code) df.to_csv(f'{name}_{code}_jan2024.csv', index=False) except KeyError: print(f"品种{code}在指定日期无数据")

高级处理技巧：

1. 主力合约连续拼接：选取每日成交量最大的合约
2. 远月价差计算：比较不同到期月的价格差
3. 品种间价比：如铜金比、油金比等

6. 实战案例：构建本地行情数据库

长期做期货分析，建议建立本地数据库。以下是SQLite方案：

import sqlite3 from tqdm import tqdm def init_db(db_path='shfe.db'): conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS daily_data ( symbol TEXT, trade_date TEXT, open REAL, high REAL, low REAL, close REAL, volume INTEGER, open_interest INTEGER, turnover REAL, settle REAL, pre_settle REAL, variety TEXT, PRIMARY KEY (symbol, trade_date) ) ''') conn.commit() return conn def update_db(conn, start_date, end_date): df = get_shfe_multi_days(start_date, end_date) if not df.empty: df.to_sql('daily_data', conn, if_exists='append', index=False) print(f"成功更新{len(df)}条数据") else: print("无新数据更新") # 初始化并更新数据库 conn = init_db() update_db(conn, "20240101", "20240331") conn.close()

这个数据库可以支持各种查询：

# 查询铜期货数据 query = ''' SELECT * FROM daily_data WHERE variety='CU' AND trade_date BETWEEN '20240101' AND '20240331' ORDER BY trade_date ''' cu_data = pd.read_sql(query, conn)

进阶功能建议：

1. 添加自动更新脚本（如每日收盘后运行）
2. 增加技术指标计算字段
3. 构建REST API提供数据服务

7. 常见问题与性能优化

在实际使用中，我遇到过这些问题和解决方案：

问题1：数据获取速度慢

• 解决方案：使用多线程获取

from concurrent.futures import ThreadPoolExecutor def fetch_day(date): date_str = date.strftime("%Y%m%d") try: return ak.get_shfe_daily(date_str) except: return pd.DataFrame() with ThreadPoolExecutor(max_workers=5) as executor: results = list(executor.map(fetch_day, date_range))

问题2：数据缺失或不完整

• 检查方法：

# 检查各品种数据完整性 coverage = jan_data.groupby('variety')['trade_date'].nunique() print(coverage.sort_values())

问题3：数据清洗需求

• 典型清洗操作：

# 处理异常值 def clean_data(df): df = df[df['volume'] > 0] # 剔除无成交数据 df = df[df['high'] >= df['low']] # 剔除价格异常 return df.drop_duplicates(['symbol', 'trade_date'])

性能优化建议：

1. 使用Polars替代Pandas处理大数据
2. 采用Parquet格式存储替代CSV
3. 对历史数据建立本地缓存

8. 数据可视化与分析示例

有了数据后，我们可以进行各种分析。以下是常用的可视化方法：

import matplotlib.pyplot as plt # 铜期货价格走势 cu_main = cu_main.sort_values('trade_date') plt.figure(figsize=(12, 6)) plt.plot(pd.to_datetime(cu_main['trade_date']), cu_main['close'], label='铜主力') plt.title('2024年1月铜期货价格走势') plt.xlabel('日期') plt.ylabel('价格(元/吨)') plt.grid() plt.legend() plt.show()

多品种对比分析：

# 准备数据 products = ['CU', 'AL', 'AU'] dfs = [grouped.get_group(p).groupby('trade_date').apply(get_main_contract) for p in products] # 标准化后绘图 plt.figure(figsize=(12, 6)) for df, name in zip(dfs, ['铜', '铝', '黄金']): norm_price = df['close'] / df['close'].iloc[0] # 首日标准化 plt.plot(pd.to_datetime(df['trade_date']), norm_price, label=name) plt.title('多品种价格走势对比(标准化)') plt.legend() plt.show()

进阶分析方向：

1. 计算各品种波动率
2. 构建相关性矩阵
3. 回测简单的均线策略
4. 分析期限结构变化

9. 扩展应用：与其他数据源结合

AKShare还提供其他相关数据，可以丰富我们的分析：

获取现货价格：

# 获取长江有色现货价格 spot_cu = ak.macro_china_lme_spot("铜")

获取库存数据：

# 获取LME库存 lme_stock = ak.futures_lme_stock("铜")

宏观经济数据：

# 获取PMI数据 pmi = ak.macro_china_pmi()

数据融合示例：

# 合并期货和现货数据 merged = pd.merge( cu_main[['trade_date', 'close']].rename(columns={'close': 'futures'}), spot_cu[['日期', '现货价格']].rename(columns={'日期': 'trade_date', '现货价格': 'spot'}), on='trade_date' ) # 计算基差 merged['basis'] = merged['spot'] - merged['futures']

10. 注意事项与最佳实践

经过多次实战，我总结出这些经验：

1. 数据更新频率：

   • 日线数据通常在收盘后1-2小时更新
   • 重大节假日前后注意数据更新时间变化

2. 数据质量检查：

   # 典型检查项 assert not df.duplicated(['symbol', 'trade_date']).any() assert (df['high'] >= df['low']).all() assert (df['volume'] >= 0).all()

3. 合规使用：

   • 避免高频请求（建议间隔至少1秒）
   • 商业用途需获得交易所授权
   • 注明数据来源

4. 备份策略：

   • 每日增量备份
   • 每月全量备份
   • 版本控制数据清洗脚本

5. 性能监控：

   # 记录执行时间 import time start = time.time() # 数据获取操作 print(f"耗时：{time.time()-start:.2f}秒")

最后建议将整套流程封装成类，方便复用：

class SHFEData: def __init__(self, db_path='shfe.db'): self.conn = sqlite3.connect(db_path) def update_data(self, start_date, end_date): # 实现数据更新逻辑 pass def query_data(self, product, start_date, end_date): # 实现数据查询逻辑 pass def __del__(self): self.conn.close()