1. 为什么需要批量获取期货行情数据做期货分析的朋友都知道单日行情数据就像一张照片只能反映某个瞬间的市场状态。而真正有价值的分析需要连续的时间序列数据就像一段视频能记录市场变化的完整过程。我刚开始做期货分析时每天手动下载数据再拼接不仅效率低下还经常漏掉关键交易日的数据。AKShare这个开源库完美解决了这个问题。它封装了上海期货交易所的官方数据接口我们可以用几行代码就获取任意时间段的多品种行情。比如最近我需要分析铜、铝、黄金三个品种过去半年的走势传统方法可能要花半天时间收集整理数据现在用AKShare只需要5分钟。批量获取数据的核心价值在于趋势分析计算20日均线、MACD等指标至少需要一个月的数据策略回测验证交易策略有效性通常需要1-3年的历史数据相关性研究不同品种间的价格联动分析需要同步的时间序列风险监控波动率计算需要足够长的历史数据作为基准2. 环境准备与AKShare安装在开始写代码前我们需要准备好Python环境。推荐使用Anaconda创建独立的虚拟环境避免与其他项目的依赖冲突。这是我常用的环境配置命令conda create -n futures python3.8 conda activate futures pip install akshare pandas numpyAKShare的安装可能会遇到几个常见问题如果出现SSL证书错误可以尝试import ssl ssl._create_default_https_context ssl._create_unverified_context网络不稳定时建议使用国内镜像源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 lab3. 单日行情获取基础我们先从基础的单日数据获取开始这是批量操作的基础。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前结算价注意几个常见问题节假日会返回空数据建议先检查日期是否为交易日部分远期合约可能没有成交数据夜盘数据会包含在次日的数据中4. 实现多日数据批量获取批量获取的核心是构造日期序列然后循环调用单日接口。这里分享我优化过的代码def get_shfe_multi_days(start_date, end_date): 获取指定日期范围的SHFE数据 date_range pd.date_range(startstart_date, endend_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)这段代码有几个优化点自动跳过没有数据的日期如节假日为每条数据添加交易日期标记加入异常捕获避免中途失败显示进度反馈对于大量数据获取建议分批次获取如按月获取添加随机延时避免被封禁本地保存中间结果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, indexFalse)对于多品种分析可以构建品种字典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, indexFalse) except KeyError: print(f品种{code}在指定日期无数据)高级处理技巧主力合约连续拼接选取每日成交量最大的合约远月价差计算比较不同到期月的价格差品种间价比如铜金比、油金比等6. 实战案例构建本地行情数据库长期做期货分析建议建立本地数据库。以下是SQLite方案import sqlite3 from tqdm import tqdm def init_db(db_pathshfe.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_existsappend, indexFalse) print(f成功更新{len(df)}条数据) else: print(无新数据更新) # 初始化并更新数据库 conn init_db() update_db(conn, 20240101, 20240331) conn.close()这个数据库可以支持各种查询# 查询铜期货数据 query SELECT * FROM daily_data WHERE varietyCU AND trade_date BETWEEN 20240101 AND 20240331 ORDER BY trade_date cu_data pd.read_sql(query, conn)进阶功能建议添加自动更新脚本如每日收盘后运行增加技术指标计算字段构建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_workers5) 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])性能优化建议使用Polars替代Pandas处理大数据采用Parquet格式存储替代CSV对历史数据建立本地缓存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, labelname) plt.title(多品种价格走势对比(标准化)) plt.legend() plt.show()进阶分析方向计算各品种波动率构建相关性矩阵回测简单的均线策略分析期限结构变化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}), ontrade_date ) # 计算基差 merged[basis] merged[spot] - merged[futures]10. 注意事项与最佳实践经过多次实战我总结出这些经验数据更新频率日线数据通常在收盘后1-2小时更新重大节假日前后注意数据更新时间变化数据质量检查# 典型检查项 assert not df.duplicated([symbol, trade_date]).any() assert (df[high] df[low]).all() assert (df[volume] 0).all()合规使用避免高频请求建议间隔至少1秒商业用途需获得交易所授权注明数据来源备份策略每日增量备份每月全量备份版本控制数据清洗脚本性能监控# 记录执行时间 import time start time.time() # 数据获取操作 print(f耗时{time.time()-start:.2f}秒)最后建议将整套流程封装成类方便复用class SHFEData: def __init__(self, db_pathshfe.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()
利用AKShare批量获取上海期货交易所多品种历史行情数据
1. 为什么需要批量获取期货行情数据做期货分析的朋友都知道单日行情数据就像一张照片只能反映某个瞬间的市场状态。而真正有价值的分析需要连续的时间序列数据就像一段视频能记录市场变化的完整过程。我刚开始做期货分析时每天手动下载数据再拼接不仅效率低下还经常漏掉关键交易日的数据。AKShare这个开源库完美解决了这个问题。它封装了上海期货交易所的官方数据接口我们可以用几行代码就获取任意时间段的多品种行情。比如最近我需要分析铜、铝、黄金三个品种过去半年的走势传统方法可能要花半天时间收集整理数据现在用AKShare只需要5分钟。批量获取数据的核心价值在于趋势分析计算20日均线、MACD等指标至少需要一个月的数据策略回测验证交易策略有效性通常需要1-3年的历史数据相关性研究不同品种间的价格联动分析需要同步的时间序列风险监控波动率计算需要足够长的历史数据作为基准2. 环境准备与AKShare安装在开始写代码前我们需要准备好Python环境。推荐使用Anaconda创建独立的虚拟环境避免与其他项目的依赖冲突。这是我常用的环境配置命令conda create -n futures python3.8 conda activate futures pip install akshare pandas numpyAKShare的安装可能会遇到几个常见问题如果出现SSL证书错误可以尝试import ssl ssl._create_default_https_context ssl._create_unverified_context网络不稳定时建议使用国内镜像源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 lab3. 单日行情获取基础我们先从基础的单日数据获取开始这是批量操作的基础。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前结算价注意几个常见问题节假日会返回空数据建议先检查日期是否为交易日部分远期合约可能没有成交数据夜盘数据会包含在次日的数据中4. 实现多日数据批量获取批量获取的核心是构造日期序列然后循环调用单日接口。这里分享我优化过的代码def get_shfe_multi_days(start_date, end_date): 获取指定日期范围的SHFE数据 date_range pd.date_range(startstart_date, endend_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)这段代码有几个优化点自动跳过没有数据的日期如节假日为每条数据添加交易日期标记加入异常捕获避免中途失败显示进度反馈对于大量数据获取建议分批次获取如按月获取添加随机延时避免被封禁本地保存中间结果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, indexFalse)对于多品种分析可以构建品种字典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, indexFalse) except KeyError: print(f品种{code}在指定日期无数据)高级处理技巧主力合约连续拼接选取每日成交量最大的合约远月价差计算比较不同到期月的价格差品种间价比如铜金比、油金比等6. 实战案例构建本地行情数据库长期做期货分析建议建立本地数据库。以下是SQLite方案import sqlite3 from tqdm import tqdm def init_db(db_pathshfe.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_existsappend, indexFalse) print(f成功更新{len(df)}条数据) else: print(无新数据更新) # 初始化并更新数据库 conn init_db() update_db(conn, 20240101, 20240331) conn.close()这个数据库可以支持各种查询# 查询铜期货数据 query SELECT * FROM daily_data WHERE varietyCU AND trade_date BETWEEN 20240101 AND 20240331 ORDER BY trade_date cu_data pd.read_sql(query, conn)进阶功能建议添加自动更新脚本如每日收盘后运行增加技术指标计算字段构建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_workers5) 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])性能优化建议使用Polars替代Pandas处理大数据采用Parquet格式存储替代CSV对历史数据建立本地缓存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, labelname) plt.title(多品种价格走势对比(标准化)) plt.legend() plt.show()进阶分析方向计算各品种波动率构建相关性矩阵回测简单的均线策略分析期限结构变化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}), ontrade_date ) # 计算基差 merged[basis] merged[spot] - merged[futures]10. 注意事项与最佳实践经过多次实战我总结出这些经验数据更新频率日线数据通常在收盘后1-2小时更新重大节假日前后注意数据更新时间变化数据质量检查# 典型检查项 assert not df.duplicated([symbol, trade_date]).any() assert (df[high] df[low]).all() assert (df[volume] 0).all()合规使用避免高频请求建议间隔至少1秒商业用途需获得交易所授权注明数据来源备份策略每日增量备份每月全量备份版本控制数据清洗脚本性能监控# 记录执行时间 import time start time.time() # 数据获取操作 print(f耗时{time.time()-start:.2f}秒)最后建议将整套流程封装成类方便复用class SHFEData: def __init__(self, db_pathshfe.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()
