Python获取大商所期货历史行情数据：基于AkShare的完整实现方案

1. 项目概述与核心价值做量化研究、策略回测或者仅仅是跟踪某个大宗商品的价格走势获取准确、结构化的历史行情数据是第一步也是最关键的一步。很多朋友可能习惯用一些商业数据终端但对于想深入定制、或者希望将数据获取流程自动化集成到自己系统中的开发者来说直接通过程序从源头获取数据是更灵活、更可控的选择。今天要聊的就是如何精准地获取大连商品交易所DCE在指定日期的期货行情数据。这个需求听起来简单但实际操作中会遇到不少“坑”。比如数据源选哪个是爬交易所官网还是用第三方接口获取的数据格式是否规整非交易日或节假日怎么处理历史数据的分时、Tick粒度如何选择这些问题如果不事先搞清楚很容易在代码写到一半时陷入僵局。我结合自己这些年对接国内期货数据的经验把从思路梳理、工具选型到代码实现、避坑指南的完整流程梳理一遍。无论你是刚入门的数据分析爱好者还是需要构建稳定数据管道的量化开发者这篇内容都能提供一个可直接参考、复现的解决方案。我们将聚焦于一个非常具体的场景输入一个日期例如2023年10月26日程序自动获取当天大连商品交易所所有上市期货合约的行情数据包括但不限于开盘价、最高价、最低价、收盘价、成交量、持仓量等关键字段并以结构化的格式如CSV、Pandas DataFrame返回或保存。我们会优先选择稳定、免费且合法的数据源并详细解释每一步背后的考量。2. 数据源评估与方案选型在动手写代码之前花点时间评估数据源是最高效的投资。错误的数据源会导致项目后期维护成本剧增甚至因政策变动而前功尽弃。2.1 主流数据源对比分析国内获取期货行情数据的途径大致分为以下几类各有优劣数据源类型代表渠道优点缺点适用场景交易所官方接口上期所、大商所、郑商所、中金所的信息商系统数据最权威、实时性最高、字段最全。接入门槛高通常需要成为会员或购买信息商服务费用昂贵个人开发者难以承受。高频交易、机构级实盘系统。第三方金融数据平台AkShare、Tushare、Baostock、JoinQuant API、RiceQuant API免费或低成本接口简单易用社区活跃通常已清洗好数据。数据可能有轻微延迟深度历史数据或Tick数据可能受限接口稳定性依赖平台运营。个人研究、量化策略原型开发、中低频回测。财经网站数据抓取新浪财经、东方财富、网易财经等网站的期货板块直接可见无需注册适合临时、小批量抓取。网页结构变更频繁爬虫维护成本高数据格式不统一有反爬风险法律风险需注意。临时性数据补全、非程序化的小规模数据获取。商业数据终端Wind、同花顺iFinD、Choice数据全面、准确、包含丰富的基本面数据工具链完善。价格极其昂贵通常按年订阅是机构级开支。专业金融机构的投资研究与决策。注意对于个人学习和小型项目强烈不建议直接爬取交易所官网或大型财经网站的核心数据这不仅有技术风险反爬机制更可能涉及法律风险。应优先使用提供了合法API的第三方平台。2.2 本项目方案确定AkShare综合考量易用性、稳定性、免费性和社区支持度我们选择AkShare作为本项目的核心数据获取工具。它是一个基于Python的开源金融数据接口库旨在提供优雅、简洁的API来获取各类金融数据。对于国内期货数据AkShare封装了相对可靠的数据源如新浪财经返回规整的Pandas DataFrame极大简化了数据处理流程。为什么是AkShare零成本完全免费这对于学习和研究阶段至关重要。接口友好函数命名清晰参数直观几行代码就能拿到数据。数据质量尚可对于日线级别的历史行情数据其准确性能满足大部分研究需求。活跃的社区项目在GitHub上持续更新遇到问题容易找到解决方案或同类使用者交流。当然它也有局限比如极端历史日期如十年前的数据可能不全获取实时行情或深度Tick数据能力较弱。但对于“获取指定日期历史行情”这个目标它是当前的最优解。3. 环境准备与核心工具链工欲善其事必先利其器。我们来搭建一个轻量、可复现的Python数据分析环境。3.1 Python环境与依赖安装建议使用 Python 3.8 及以上版本。使用pip安装必要的库# 核心数据获取与处理库 pip install akshare pandas # 可选用于更复杂的数据分析和可视化 pip install numpy matplotlibpandas数据处理和分析的基石AkShare返回的数据大多是DataFrame格式。akshare主角用于获取数据。numpy matplotlib后续进行数据分析和图表展示时会用到非必需但推荐。3.2 验证AkShare基础功能安装完成后可以在Python交互环境或脚本中简单测试确保能正常获取数据。import akshare as ak # 尝试获取大商所豆一主力合约的近期日线数据 futures_zh_daily_sina_df ak.futures_zh_daily_sina(symbolA0, start_date20231020, end_date20231026) print(futures_zh_daily_sina_df.head())如果运行后能打印出一个包含日期、开盘、收盘等信息的表格说明环境配置成功。这里symbolA0代表豆一主力合约后续我们会详细讲解合约代码的规则。4. 核心实现获取指定日期行情数据这是最核心的部分。我们将把任务分解为几个步骤并编写健壮的函数来完成。4.1 理解大商所期货合约代码规则在获取数据前必须清楚如何指代你想要的那个合约。国内期货合约代码有一套标准格式。 例如a2311a品种代码代表“豆一”。大商所a-豆一b-豆二m-豆粕y-豆油c-玉米i-铁矿石等。2311到期月份代表2023年11月。23是年份后两位11是月份。有时接口也支持使用如A0这样的代码表示“豆一主力合约”但为了精确获取指定日期的所有合约或特定合约我们必须使用完整的合约代码。AkShare的futures_zh_daily_sina接口通常需要完整的合约代码。我们需要一个方法根据指定日期找出当天所有上市交易的合约列表。4.2 步骤一获取指定日期所有上市合约列表遗憾的是AkShare没有直接提供一个日期来获取当日所有合约列表的函数。一个实用的策略是获取某个品种所有曾有交易的历史合约列表。根据指定日期过滤出在该日期已经上市即合约月份大于等于指定日期且尚未到期的合约。这里我们可以利用ak.futures_main_sina()获取主力合约映射但更全面的方法是先获取所有合约的基础信息。实际操作中一个更直接的方法是我们通常更关心主力合约或特定几个活跃合约。对于全品种全合约我们可以预先定义一个我们关注的核心品种列表。import akshare as ak import pandas as pd from datetime import datetime def get_dce_contracts_for_date(target_date_str): 获取大商所在指定日期可能存在的活跃合约列表基于常见品种。 这是一个近似方法因为无法动态获取历史上某一天的确切上市合约列表。 Args: target_date_str (str): 目标日期格式 YYYYMMDD 或 YYYY-MM-DD。 Returns: list: 合约代码列表例如 [a2311, m2311, y2401, ...] # 将字符串日期转换为datetime对象用于计算 try: target_date pd.to_datetime(target_date_str) except Exception as e: print(f日期格式错误: {e}) return [] # 大商所主要品种代码拼音首字母 dce_symbols [a, b, m, y, c, cs, jd, rr, l, v, pp, j, jm, i, fb, bb, p] contracts [] # 对于每个品种我们生成从目标日期往后推一段时间的可能合约 # 期货合约通常在到期前数月就开始交易。我们假设目标日期时当前月和后续几个月的合约都已上市。 for symbol in dce_symbols: year target_date.year % 100 # 取后两位 month target_date.month # 生成未来4个交割月的合约代码这是一个经验值可根据需要调整 for i in range(4): future_month month i future_year year if future_month 12: future_month - 12 future_year 1 contract_code f{symbol}{future_year:02d}{future_month:02d} contracts.append(contract_code) # 去重并返回 return list(set(contracts)) # 示例获取2023年10月26日可能存在的合约列表 target_date 2023-10-26 contract_list get_dce_contracts_for_date(target_date) print(f为日期 {target_date} 生成的合约列表示例: {contract_list[:10]}...) # 打印前10个实操心得实际上没有任何一个免费接口能完美回溯历史上任意一天所有上市合约的精确列表。上述方法是一个“足够好”的近似它基于合约交易的生命周期规律。对于回测更常见的做法是使用主力连续合约或提前准备好一份完整的合约上市日期表。如果追求绝对精确可能需要购买商业数据库。4.3 步骤二遍历合约获取日线数据并处理异常有了合约列表即使是近似的我们就可以遍历它调用AkShare接口获取每个合约在指定日期的数据。这里必须加入异常处理因为很多生成的合约代码在历史上可能不存在未上市或已退市。def get_dce_daily_data_by_date(target_date_str, contract_listNone, max_workers5): 获取指定日期大商所多个合约的日线行情数据。 Args: target_date_str (str): 目标日期格式 YYYYMMDD。 contract_list (list, optional): 合约代码列表。如果为None则使用内部函数生成。 max_workers (int, optional): 并发线程数用于加速。默认为5。 Returns: pandas.DataFrame: 合并后的行情数据每个合约一行。 import concurrent.futures from tqdm import tqdm # 可选用于显示进度条 if contract_list is None: contract_list get_dce_contracts_for_date(target_date_str) all_data [] failed_contracts [] # 定义一个内部函数用于单个合约的数据获取 def fetch_single_contract(contract): try: # 注意这里使用 futures_zh_daily_sina日期格式为 YYYYMMDD # 我们只获取目标日期当天的数据所以start和end日期相同 df ak.futures_zh_daily_sina(symbolcontract, start_datetarget_date_str.replace(-, ), end_datetarget_date_str.replace(-, )) if not df.empty: # 添加一列记录合约代码 df[contract] contract return df else: # 数据为空可能是该合约在该日期无交易未上市、已退市或停牌 return None except Exception as e: # 记录失败的合约 print(f获取合约 {contract} 数据失败: {e}) failed_contracts.append(contract) return None # 使用线程池并发获取提高效率 with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务 future_to_contract {executor.submit(fetch_single_contract, contract): contract for contract in contract_list} # 使用tqdm创建进度条需安装tqdm库 for future in tqdm(concurrent.futures.as_completed(future_to_contract), totallen(contract_list), desc获取数据): result future.result() if result is not None: all_data.append(result) print(f数据获取完成。成功: {len(all_data)} 个合约, 失败/无数据: {len(failed_contracts)} 个合约) if failed_contracts: print(f失败合约示例: {failed_contracts[:10]}) # 合并所有有效数据 if all_data: final_df pd.concat(all_data, ignore_indexTrue) # 按合约代码排序 final_df.sort_values(bycontract, inplaceTrue) # 重置索引 final_df.reset_index(dropTrue, inplaceTrue) return final_df else: return pd.DataFrame() # 示例获取2023年10月26日的数据 target_date_str 20231026 data_df get_dce_daily_data_by_date(target_date_str) if not data_df.empty: print(f获取到 {len(data_df)} 条数据记录) print(data_df[[contract, date, open, high, low, close, volume]].head()) else: print(未获取到任何数据。)关键点解析并发请求使用ThreadPoolExecutor并发请求数据可以显著缩短获取数十个合约数据的时间。但线程数不宜过高max_workers建议在5-10之间避免对数据源服务器造成过大压力或被封IP。异常处理将每个合约的请求用try...except包裹确保一个合约失败不影响其他合约的获取。失败的合约会被记录下来。空数据判断即使请求成功没有抛出异常返回的DataFrame也可能是空的。这通常意味着该合约在指定日期没有交易数据非交易日、合约未上市等。if not df.empty:这个判断非常重要。日期格式AkShare的futures_zh_daily_sina接口需要start_date和end_date参数为 “YYYYMMDD” 格式所以用replace(-, )去掉了分隔符。4.4 步骤三数据清洗与规整化从接口获取的原始数据可能包含一些我们需要处理的问题字段名可能是中文如“日期”、“开盘”、“收盘”等不利于程序化处理。数据格式数字可能是字符串需要转换为数值型。缺失值对于停牌等情况价格数据可能为NaN或特定值。重复数据理论上不应该有但检查一下更安全。我们来编写一个数据清洗函数def clean_futures_daily_data(raw_df): 清洗和规整从AkShare获取的期货日线数据。 Args: raw_df (pandas.DataFrame): 原始的行情DataFrame。 Returns: pandas.DataFrame: 清洗后的DataFrame。 if raw_df.empty: return raw_df df raw_df.copy() # 1. 重命名列如果接口返回的是中文列名可统一为英文 # 先查看原始列名 print(原始列名:, df.columns.tolist()) # 常见的列名映射根据实际接口返回调整 column_mapping { 日期: date, 开盘价: open, 最高价: high, 最低价: low, 收盘价: close, 成交量: volume, 持仓量: open_interest, 结算价: settlement, 前结算: pre_settlement, 涨跌: change, 涨跌幅: change_pct, contract: contract # 我们自己添加的列 } # 只重命名存在的列 df.rename(columns{col: column_mapping[col] for col in column_mapping if col in df.columns}, inplaceTrue) # 2. 确保日期列是datetime类型 if date in df.columns: df[date] pd.to_datetime(df[date]) # 3. 将价格、成交量等数值列转换为float类型处理千分位逗号等问题 numeric_columns [open, high, low, close, volume, open_interest, settlement, pre_settlement] for col in numeric_columns: if col in df.columns: # 先转换为字符串移除逗号再转换为float df[col] pd.to_numeric(df[col].astype(str).str.replace(,, ), errorscoerce) # 4. 计算涨跌幅如果接口未提供 if change_pct not in df.columns and close in df.columns and pre_settlement in df.columns: df[change_pct] (df[close] - df[pre_settlement]) / df[pre_settlement] * 100 # 5. 按日期和合约排序 sort_by [] if date in df.columns: sort_by.append(date) if contract in df.columns: sort_by.append(contract) if sort_by: df.sort_values(bysort_by, inplaceTrue) df.reset_index(dropTrue, inplaceTrue) # 6. 去除完全重复的行 df.drop_duplicates(inplaceTrue) return df # 应用清洗函数 cleaned_df clean_futures_daily_data(data_df) if not cleaned_df.empty: print(清洗后的数据列信息:) print(cleaned_df.dtypes) print(\n清洗后的数据样例:) print(cleaned_df.head())清洗后我们得到了一个列名规范、数据类型正确、排序整齐的DataFrame可以很方便地用于后续的分析、存储或可视化。5. 数据存储与后续处理建议获取并清洗数据后下一步通常是持久化存储以便后续分析。5.1 存储到本地文件最常用的格式是CSV和Parquet。def save_data_to_file(df, target_date_str, file_formatcsv): 将数据保存到本地文件。 Args: df (pandas.DataFrame): 要保存的数据。 target_date_str (str): 目标日期用于文件名。 file_format (str): 文件格式支持 csv 或 parquet。 import os # 创建数据目录如果不存在 data_dir ./dce_data os.makedirs(data_dir, exist_okTrue) filename os.path.join(data_dir, fdce_daily_{target_date_str}.{file_format}) try: if file_format.lower() csv: df.to_csv(filename, indexFalse, encodingutf-8-sig) # utf-8-sig支持Excel直接打开 print(f数据已保存为 CSV 文件: {filename}) elif file_format.lower() parquet: df.to_parquet(filename, indexFalse) print(f数据已保存为 Parquet 文件: {filename}) else: print(f不支持的格式: {file_format}) except Exception as e: print(f保存文件时出错: {e}) # 示例保存为CSV和Parquet if not cleaned_df.empty: save_data_to_file(cleaned_df, target_date_str, csv) save_data_to_file(cleaned_df, target_date_str, parquet)格式选择建议CSV通用性强任何文本编辑器都能打开便于快速查看和交换。但文件体积较大读写速度较慢。Parquet列式存储格式压缩率高读写速度快特别是用Pandas或PySpark非常适合存储大规模结构化数据。是生产环境的首选。5.2 存储到数据库对于长期、系统化的数据积累建议存入数据库如SQLite轻量、MySQL或PostgreSQL。import sqlite3 from sqlalchemy import create_engine def save_data_to_sqlite(df, db_path./dce_data/futures_data.db, table_namedaily_price): 将数据存储到SQLite数据库。 Args: df (pandas.DataFrame): 要存储的数据。 db_path (str): SQLite数据库文件路径。 table_name (str): 表名。 import os os.makedirs(os.path.dirname(db_path), exist_okTrue) # 使用SQLAlchemy创建引擎更通用支持多种数据库 engine create_engine(fsqlite:///{db_path}) try: # if_existsappend 表示如果表存在则追加数据 # 可以改为 replace 或 fail df.to_sql(table_name, conengine, if_existsappend, indexFalse) print(f数据已成功写入数据库表 {table_name}) except Exception as e: print(f写入数据库时出错: {e}) # 示例存入SQLite if not cleaned_df.empty: save_data_to_sqlite(cleaned_df)注意事项直接使用to_sql的if_existsappend时需要注意数据重复的问题。如果重复运行脚本可能会向数据库插入相同日期的重复数据。在生产环境中你应该在写入前先检查表中是否已存在该日期的数据或者设计表结构时使用(date, contract)作为复合主键来防止重复插入。6. 完整脚本封装与使用示例将以上所有步骤整合成一个完整的、可执行的脚本并增加命令行参数支持会让这个工具更加实用。# get_dce_daily.py import argparse import pandas as pd import akshare as ak from datetime import datetime import concurrent.futures import os import sys from tqdm import tqdm # 这里需要将之前定义的函数 get_dce_contracts_for_date, get_dce_daily_data_by_date, clean_futures_daily_data 复制过来 # 为了简洁此处省略函数具体定义假设它们已定义在同一个文件或模块中 def main(): parser argparse.ArgumentParser(description获取指定日期大连商品交易所期货日线行情数据) parser.add_argument(date, typestr, help目标日期格式YYYYMMDD 或 YYYY-MM-DD) parser.add_argument(--output, -o, typestr, defaultcsv, choices[csv, parquet, db, none], help输出格式: csv, parquet, db (SQLite), 或 none (仅返回DataFrame)) parser.add_argument(--symbols, -s, typestr, nargs, help指定合约代码列表例如 a2311 m2311。若不指定则自动生成常见合约列表。) args parser.parse_args() target_date_str args.date.replace(-, ) print(f开始获取 {target_date_str} 大连商品交易所期货日线数据...) # 确定合约列表 if args.symbols: contract_list args.symbols print(f使用用户指定的合约列表: {contract_list}) else: contract_list get_dce_contracts_for_date(args.date) print(f自动生成合约列表共 {len(contract_list)} 个合约。) # 获取数据 data_df get_dce_daily_data_by_date(target_date_str, contract_list) if data_df.empty: print(警告未获取到任何有效数据。可能的原因) print( 1. 输入日期是非交易日周末、节假日。) print( 2. 生成的合约列表在该日均未上市交易。) print( 3. 网络问题或数据源暂时不可用。) sys.exit(1) # 清洗数据 cleaned_df clean_futures_daily_data(data_df) print(f数据清洗完成共获取 {len(cleaned_df)} 条有效记录。) # 输出数据 if args.output csv: save_data_to_file(cleaned_df, target_date_str, csv) elif args.output parquet: save_data_to_file(cleaned_df, target_date_str, parquet) elif args.output db: save_data_to_sqlite(cleaned_df, f./dce_data/dce_{target_date_str}.db, daily_price) else: # none print(cleaned_df.head()) print(f\n数据已准备就绪DataFrame格式共 {len(cleaned_df)} 行。) print(任务完成) if __name__ __main__: main()使用方式将上述所有代码包括之前定义的函数保存为一个Python文件例如get_dce_daily.py。在命令行中运行# 获取2023年10月26日数据保存为CSV python get_dce_daily.py 20231026 --output csv # 获取2023-11-01数据保存为Parquet python get_dce_daily.py 2023-11-01 -o parquet # 仅获取指定合约的数据不保存文件直接打印预览 python get_dce_daily.py 20231026 -o none --symbols a2311 m2311 y23117. 常见问题、排查技巧与进阶优化在实际操作中你几乎一定会遇到下面这些问题。这里是我踩过坑后总结的排查思路和解决方案。7.1 数据获取失败或返回空数据这是最常见的问题。现象可能原因排查步骤与解决方案单个合约失败报错如超时、连接错误网络不稳定数据源接口临时故障合约代码错误。1.重试机制在fetch_single_contract函数中加入重试逻辑如retry库。2.检查合约代码确认合约代码格式正确如a2311且在该日期真实存在。可手动访问新浪财经对应合约页面验证。3.降低并发度减少max_workers数量避免请求过快被限制。单个合约成功但返回空DataFrame该合约在指定日期无交易未上市、已退市、停牌、非交易日。1.验证日期确认日期是交易日。可查日历或尝试获取已知活跃合约如A0主力合约的数据来验证。2.合约生命周期检查合约的上市和退市日期。大商所官网会公布合约表。所有合约都失败或返回空目标日期是非交易日周末、法定节假日数据源整体不可用网络完全中断。1.检查日期这是最可能的原因。先换一个已知的工作日如昨天测试。2.测试接口连通性运行ak.futures_zh_daily_sina(symbolA0, start_date20231026, end_date20231026)看是否能返回数据。3.查看AkShare版本pip show akshare。过旧版本可能接口失效尝试升级pip install --upgrade akshare。给fetch_single_contract函数增加重试机制的示例import time from functools import wraps def retry_on_failure(max_retries3, delay1): 简单的重试装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): last_exception None for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: last_exception e if attempt max_retries - 1: time.sleep(delay * (attempt 1)) # 延迟时间递增 print(f第 {attempt 1} 次重试...) else: print(f函数 {func.__name__} 在 {max_retries} 次重试后仍失败。) raise last_exception return wrapper return decorator # 然后装饰 fetch_single_contract 函数 retry_on_failure(max_retries2, delay2) def fetch_single_contract(contract): # ... 原有代码 ...7.2 数据字段缺失或格式异常问题清洗后发现open,high等价格字段全是NaN。排查打印原始数据df.head().to_dict()查看原始值。可能数据源返回了“-”或“NaN”字符串。解决在数据清洗的数值转换步骤中加强错误处理。# 在 clean_futures_daily_data 函数的数值转换循环内 for col in numeric_columns: if col in df.columns: # 尝试多种转换方式 df[col] pd.to_numeric(df[col].astype(str).str.replace(,, ).replace(-, ).replace(NaN, ), errorscoerce)7.3 性能优化建议缓存机制如果你需要频繁获取相同日期的数据比如在调试策略可以将获取的数据缓存到本地文件或数据库下次直接读取避免重复网络请求。增量更新构建数据管道时记录已获取的最新日期下次只获取新日期的数据。使用更高效的数据结构如果合约数量非常多数百个考虑使用concurrent.futures.ProcessPoolExecutor进行多进程并发但要注意进程间通信开销或者使用异步IO库如aiohttp进行请求。对于AkShare这种可能内部有全局锁的库多进程有时比多线程更有效。向量化操作在数据清洗和计算环节尽量使用Pandas的向量化操作避免低效的循环。7.4 法律与合规提醒重要本项目示例仅用于个人学习、研究和教育目的。在使用任何金融数据时请务必遵守以下原则尊重数据版权了解并遵守数据提供方如新浪财经、AkShare的服务条款。免费数据通常禁止用于商业用途。合规使用不得利用获取的数据从事市场操纵、内幕交易等违法违规活动。避免过度请求在程序中使用合理的请求间隔如添加time.sleep(0.5)不要对数据源服务器进行高频、密集的访问以免对其造成负担也避免自己的IP被封锁。数据准确性免责免费数据源可能存在错误或延迟不保证100%准确。对于关键交易决策建议通过权威渠道复核数据。获取数据只是量化研究的第一步但也是最容易让人卡壳的一步。希望这份详尽的指南能帮你把“获取指定日期大商所期货行情数据”这个任务从一个模糊的需求变成一套清晰、可执行、可扩展的代码方案。在实际操作中最考验人的往往不是代码本身而是对数据源特性的理解、对异常情况的处理以及构建一个健壮、可维护的数据流水线的思维。