简介这是一份面向量化交易初学者与期货数据分析人员的Python自动化工具解决手动下载同花顺期货历史行情费时易错的问题。脚本通过模拟操作或接口调用在启动同花顺并关闭广告后自动完成品种选择、周期设定与数据导出最终生成标准CSV文件便于后续导入pandas、Excel或回测框架使用。压缩包仅含1个核心Python源文件.py大小仅3KB轻量简洁无依赖臃肿库适合快速部署与二次开发。目前已有872人学习下载资源虽小但功能聚焦提供完整可运行的自动化流程逻辑、清晰的数据清洗与CSV写入实现、以及适配同花顺客户端实际交互的关键参数配置。对需要高频获取主力合约日线/分钟线数据的个人投资者、策略验证者及教学演示场景尤为实用。1. 项目概述与核心价值最近在整理期货策略回测数据时发现一个挺普遍的需求如何高效、稳定地获取同花顺PC版上的期货历史行情数据并转换成CSV这种通用格式。手动一个个页面点开、导出费时费力不说还容易出错。网上流传的“同花顺自动下载xls数据.rar_dirtyqj2”这类资源往往代码老旧、依赖混乱甚至包含一些不安全的“脏”代码直接运行风险不小。这个项目的核心就是利用Python构建一个健壮、透明、可维护的自动化工具替代那些来路不明的“一键脚本”真正实现同花顺期货历史数据的“一键获取与清洗”。这个工具能做什么简单说就是模拟用户在同花顺PC客户端上的操作自动查询指定期货合约下载其历史行情数据通常是Excel格式然后进行数据清洗、格式转换最终输出为规整的CSV文件。它解决的不仅仅是“下载”问题更是数据“可用性”问题。生成的CSV文件列名清晰如日期、开盘、最高、最低、收盘、成交量、持仓量格式统一缺失值处理得当可以直接被Pandas、NumPy、甚至Excel导入用于后续的数据分析、策略回测或可视化。适合谁来参考如果你是需要定期获取期货数据进行量化研究的学生、个人交易者或是需要为团队搭建基础数据源的开发者这个项目会非常有用。它不要求你是Python高手但需要你有一点编程基础和动手的意愿。整个过程我会把原理讲透把坑先踩一遍你跟着做就能得到一个属于自己的、可靠的数据获取管道。2. 逆向工程同花顺数据接口的探查与模拟2.1 客户端行为分析与接口定位同花顺PC客户端的数据下载本质上是一个“请求-响应”过程。我们的目标不是破解其核心协议而是找到它对外提供数据的那层“窗户”。最直接的方法就是利用抓包工具监控客户端在点击“导出”或“下载数据”时究竟向哪个服务器地址发送了什么请求。这里我推荐使用Fiddler Classic或Charles这类HTTP/HTTPS抓包工具。以Fiddler为例启动后需要配置其代理并信任根证书以便解密HTTPS流量操作时务必确保同花顺客户端走的是Fiddler代理。然后在同花顺客户端中手动操作一遍数据导出流程比如进入某个期货合约的K线图页面选择“数据导出”设置好日期范围、周期如日线点击导出为Excel。此时Fiddler的会话列表里会捕获到大量请求。我们需要从中筛选出关键的那个。通常这个请求会有以下特征URL可能包含明显的关键词如quote.xxx.com、datacenter.xxx.com、download、export、kline等。请求方法多为GET或POST参数中会包含合约代码、开始日期、结束日期、周期等。响应内容Response在Inspector的TextView或WebView标签页中直接显示为乱码因为是Excel二进制流或者是一串Base64编码的数据。找到这个请求后重点观察其Headers和Query String/Request Body。Headers里需要关注Cookie和User-Agent这是模拟浏览器或客户端身份的关键。Query String或Body里则包含了我们构造请求所需的全部参数。注意同花顺的接口可能会有反爬机制比如验证Referer请求来源、Cookie的有效期或对参数进行某种编码。直接复制抓到的URL可能下次就失效了。更稳健的做法是分析出参数的构成规则。2.2 请求参数的解构与构造假设我们抓到的请求URL简化后如下http://datacenter.xxx.com/servi ce/download?codeRB8888start20230101end20231231periodday我们需要理解每个参数code: 期货合约代码。这里有个关键点同花顺内部使用的代码可能与交易所官方代码不同。例如螺纹钢主力连续可能是RB8888而具体合约如RB2410可能是RB2410.SHF.SHF代表上期所。这需要通过多次尝试或查找同花顺内部的代码映射表来确定。start/end: 起止日期格式通常是YYYYMMDD。period: 数据周期。day代表日线week代表周线5可能代表5分钟线。这个也需要验证。此外可能还有隐藏参数如field指定下载的字段如开盘价、收盘价等、exporttype导出类型如xls等。构造请求的Python代码核心片段import requests def construct_download_url(contract_code, start_date, end_date, periodday): 构造数据下载请求URL。 注意这里的域名和参数路径是示例实际需要根据抓包结果替换。 base_url http://datacenter.xxx.com/service/download params { code: contract_code, start: start_date.strftime(%Y%m%d), # 假设start_date是datetime对象 end: end_date.strftime(%Y%m%d), period: period, exporttype: xls, # 指定导出Excel格式 # 可能还有其他必要参数如 field需根据实际情况添加 } # 注意真实的接口可能需要额外的认证信息如token可能放在headers或params里 return base_url, params2.3 会话维持与反反爬策略直接使用requests.get()可能很快就会被拒绝因为缺少了模拟真实浏览器会话的标识。User-Agent设置一个常见的浏览器UA。headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36 }Cookie这是维持登录状态的关键。你可以从抓包工具中复制完整的Cookie字符串但更自动化的方式是使用requests.Session()对象。先模拟登录如果接口需要或者直接使用从客户端抓取的固定Cookie注意有效期。session requests.Session() session.headers.update(headers) # 如果抓到了固定的Cookie可以这样设置需替换为实际值 session.cookies.set(some_cookie_name, cookie_value, domain.xxx.com) # 然后使用session进行请求 response session.get(base_url, paramsparams)Referer有些接口会检查请求来源。通常可以设置为数据页面的URL。headers[Referer] http://quote.xxx.com/xxxpage请求频率控制务必在请求间添加延时避免高频请求导致IP被封。使用time.sleep()。import time time.sleep(1) # 每次请求后暂停1秒这是一个比较保守且友好的间隔实操心得同花顺的接口并非官方公开API稳定性无法保证。我们的代码应该具备良好的错误处理和重试机制。一个健壮的下载函数应该包含超时设置、状态码检查、以及有限次数的重试逻辑。3. 核心工具链搭建从请求到清洗的完整流程3.1 环境准备与依赖库选择工欲善其事必先利其器。我们不需要一个庞大的环境只需要几个核心库。# 建议使用虚拟环境如 venv 或 conda # pip install 安装以下库 requests2.28.0 # 用于发送HTTP请求 pandas1.5.0 # 数据处理和分析的核心读写CSV/Excel openpyxl3.0.0 # Pandas读写.xlsx文件需要如果同花顺导出的新版本是.xlsx xlrd2.0.0 # Pandas读写旧版.xls文件需要注意xlrd 2.0 已不支持.xls需用旧版或改用其他引擎 # 对于.xls文件一个更好的选择是使用 pip install xlrd1.2.0旧版支持.xls或 pip install pyxlsb用于.xlsb # 更通用的方案是让pandas自动选择引擎但确保有备用的。 lxml4.9.0 # 可选用于解析HTML如果数据以网页表格形式返回库选型理由requestsPython事实上的HTTP库标准简单易用。pandas数据处理的瑞士军刀。read_excel和to_csv方法能轻松处理格式转换其DataFrame结构也是量化分析的基础。openpyxl/xlrd是Pandas读写Excel文件的底层引擎。由于同花顺导出的可能是.xls或.xlsx建议两者都安装让Pandas自动探测。3.2 数据下载模块实现我们将下载逻辑封装成一个函数提高代码的复用性和可读性。import requests import pandas as pd from io import BytesIO import time class THSHistoryDownloader: def __init__(self, cookie_strNone): self.session requests.Session() self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64)..., Referer: http://quote.10jqka.com.cn/ # 示例Referer } self.session.headers.update(self.headers) if cookie_str: # 简化处理将Cookie字符串转换为字典并添加到会话中 # 更严谨的做法是使用 http.cookies.SimpleCookie 解析 for cookie in cookie_str.split(; ): if in cookie: name, value cookie.split(, 1) self.session.cookies.set(name, value, domain.10jqka.com.cn) def download_history_excel(self, contract_code, start_date, end_date, periodday, retry3): 下载历史数据Excel文件返回pandas DataFrame。 参数: contract_code: 同花顺合约代码 (如 RB8888, AU2412.SHF) start_date: 开始日期字符串 YYYYMMDD 或 datetime对象 end_date: 结束日期字符串 YYYYMMDD 或 datetime对象 period: 周期day, week, month, 5 等 retry: 失败重试次数 返回: pandas.DataFrame 或 None (如果失败) # 1. 构造请求参数 (此处为示例需替换为真实接口参数) params { code: contract_code, start: pd.to_datetime(start_date).strftime(%Y%m%d), end: pd.to_datetime(end_date).strftime(%Y%m%d), period: period, export: excel, # 或 xls } url http://datacenter.10jqka.com.cn/ajax/download/xxx # 示例URL for attempt in range(retry): try: print(f尝试下载 {contract_code} {period}线数据第{attempt1}次...) resp self.session.get(url, paramsparams, timeout15) resp.raise_for_status() # 检查HTTP状态码是否为200 # 2. 检查响应内容是否为Excel content_type resp.headers.get(Content-Type, ) if application/vnd.ms-excel in content_type or application/octet-stream in content_type: # 3. 使用pandas直接读取二进制流 # 注意需要根据实际文件格式指定引擎。如果是.xls用xlrd旧版.xlsx用openpyxl # 这里让pandas自动推断 df pd.read_excel(BytesIO(resp.content)) print(f下载成功数据形状: {df.shape}) return df else: # 可能返回的是错误页面或JSON print(f响应类型非Excel: {content_type}) print(resp.text[:500]) # 打印前500字符以便调试 return None except requests.exceptions.RequestException as e: print(f网络请求失败: {e}) if attempt retry - 1: wait_time 2 ** attempt # 指数退避 print(f等待{wait_time}秒后重试...) time.sleep(wait_time) else: print(重试次数用尽下载失败。) return None except Exception as e: print(f处理数据时发生未知错误: {e}) return None重要提示上面的url和params字典是示例占位符。你必须使用抓包工具获取真实的URL和参数名及格式。直接使用示例代码将无法工作。3.3 数据清洗与CSV转换的精细化处理下载到的Excel数据往往不是“开箱即用”的。常见的“脏数据”问题包括多余的表头/尾行Excel里可能有标题行、说明行、空行、合计行。不规范的列名可能是中文可能包含空格或特殊字符。数据类型混乱日期列可能是字符串格式如“2023-01-01”也可能是Excel的序列值。数值列可能混入“-”或空字符串。缺失值可能用“-”、“NaN”、“NULL”或直接空单元格表示。清洗流程需要标准化。以下是一个清洗函数示例def clean_history_data(df, contract_code): 清洗从同花顺下载的原始DataFrame。 参数: df: 原始的pandas DataFrame contract_code: 合约代码用于日志或列名重命名参考 返回: 清洗后的pandas DataFrame if df is None or df.empty: print(输入数据为空无需清洗。) return df df_clean df.copy() # 1. 删除全为空值的行和列 df_clean.dropna(howall, inplaceTrue) df_clean.dropna(axis1, howall, inplaceTrue) # 2. 规范化列名假设原始列名是中文 # 常见的同花顺导出列名映射 column_mapping { 日期: date, 时间: time, # 对于分钟线 开盘: open, 最高: high, 最低: low, 收盘: close, 成交量: volume, 成交额: amount, # 股票常用期货可能没有或为‘成交金额’ 持仓量: open_interest, # 期货特有 涨跌: change, 涨跌幅: change_pct, } df_clean.rename(columnscolumn_mapping, inplaceTrue) # 对于未映射的列进行通用清洗去除空格转小写替换特殊字符 df_clean.columns [col.strip().replace( , _).replace((, ).replace(), ).lower() for col in df_clean.columns] # 3. 处理日期列假设清洗后列名为date if date in df_clean.columns: # 尝试多种日期解析格式 try: df_clean[date] pd.to_datetime(df_clean[date]) except Exception as e: print(f日期解析失败尝试其他方式: {e}) # 有时日期可能是数字格式Excel序列值 try: df_clean[date] pd.to_datetime(df_clean[date], unitD, origin1899-12-30) # Excel的日期系统 except: print(无法解析日期列请手动检查数据格式。) # 4. 处理数值列开盘、最高、最低、收盘、成交量、持仓量 numeric_columns [open, high, low, close, volume, open_interest, amount] for col in numeric_columns: if col in df_clean.columns: # 将非数字字符如‘-’‘--’替换为NaN然后转换类型 df_clean[col] pd.to_numeric(df_clean[col].replace([-, --, , NaN], pd.NA), errorscoerce) # 5. 按日期排序如果日期列已成功解析 if date in df_clean.columns and df_clean[date].dtype datetime64[ns]: df_clean.sort_values(date, inplaceTrue) df_clean.reset_index(dropTrue, inplaceTrue) # 6. 去除可能的重复行基于日期 if date in df_clean.columns: df_clean.drop_duplicates(subset[date], keepfirst, inplaceTrue) print(f数据清洗完成。原始形状: {df.shape}, 清洗后形状: {df_clean.shape}) return df_clean清洗完成后保存为CSV就非常简单了def save_to_csv(df, contract_code, period, output_dir./data): 将清洗后的DataFrame保存为CSV文件。 参数: df: 清洗后的DataFrame contract_code: 合约代码用于生成文件名 period: 数据周期 output_dir: 输出目录 import os os.makedirs(output_dir, exist_okTrue) # 生成文件名例如RB8888_day_20240101_20241231.csv filename f{contract_code.replace(., _)}_{period}_{pd.Timestamp.now().strftime(%Y%m%d)}.csv filepath os.path.join(output_dir, filename) # 保存为CSV索引不保存日期格式标准化 df.to_csv(filepath, indexFalse, encodingutf-8-sig) # utf-8-sig支持Excel直接打开不乱码 print(f数据已保存至: {filepath}) return filepath4. 实战构建自动化下载脚本与任务调度4.1 脚本集成与参数化将上述模块组合起来形成一个主脚本download_ths_futures.py。# download_ths_futures.py import argparse from datetime import datetime, timedelta import sys from pathlib import Path # 假设之前的类THSHistoryDownloader和函数clean_history_data, save_to_csv定义在同一个文件或已导入 def main(): parser argparse.ArgumentParser(description同花顺期货历史行情下载器) parser.add_argument(--code, requiredTrue, help同花顺合约代码如 RB8888 或 AU2412.SHF) parser.add_argument(--start, requiredTrue, help开始日期格式 YYYYMMDD) parser.add_argument(--end, requiredTrue, help结束日期格式 YYYYMMDD) parser.add_argument(--period, defaultday, choices[day, week, month, 5, 15, 30, 60], help数据周期默认日线) parser.add_argument(--output, default./data, helpCSV文件输出目录) parser.add_argument(--cookie, help手动设置Cookie字符串可选从抓包工具获取) args parser.parse_args() # 初始化下载器 downloader THSHistoryDownloader(cookie_strargs.cookie) # 下载数据 print(f开始下载合约 {args.code} 从 {args.start} 到 {args.end} 的 {args.period} 线数据...) raw_df downloader.download_history_excel( contract_codeargs.code, start_dateargs.start, end_dateargs.end, periodargs.period ) if raw_df is not None: # 清洗数据 cleaned_df clean_history_data(raw_df, args.code) # 保存CSV if not cleaned_df.empty: save_to_csv(cleaned_df, args.code, args.period, args.output) print(任务完成) else: print(清洗后数据为空请检查原始数据。) sys.exit(1) else: print(数据下载失败请检查网络、参数或Cookie。) sys.exit(1) if __name__ __main__: main()使用方式python download_ths_futures.py --code RB8888 --start 20230101 --end 20231231 --period day --output ./futures_data4.2 批量化下载与错误恢复如果需要下载多个合约或长时间序列的数据可以编写一个批处理脚本。# batch_download.py import subprocess import time from datetime import datetime contract_list [RB8888, AU2412.SHF, CU2411.SHF] # 合约列表 periods [day] # 可以扩展为 [day, 5] start_date 20230101 end_date 20231231 for contract in contract_list: for period in periods: print(f\n 处理合约: {contract}, 周期: {period} ) # 使用subprocess调用主脚本可以更好地隔离每次运行 cmd [ python, download_ths_futures.py, --code, contract, --start, start_date, --end, end_date, --period, period, --output, ./batch_output ] try: result subprocess.run(cmd, checkTrue, capture_outputTrue, textTrue, timeout60) print(result.stdout) if result.stderr: print(f警告: {result.stderr}) except subprocess.CalledProcessError as e: print(f命令执行失败返回码: {e.returncode}) print(f标准错误: {e.stderr}) # 可以在这里记录失败的合约和周期以便后续重试 with open(failed_tasks.log, a) as f: f.write(f{datetime.now()}: {contract} {period} failed.\n) except subprocess.TimeoutExpired: print(f任务超时: {contract} {period}) # 请求间隔避免触发反爬 time.sleep(3)4.3 简易定时任务与日志记录对于需要每日更新的数据可以结合系统定时任务。Linux/Mac (使用cron):# 编辑crontab: crontab -e # 每天下午6点运行 0 18 * * * cd /path/to/your/script /usr/bin/python3 /path/to/your/script/download_ths_futures.py --code RB8888 --start $(date -d -30 days \%Y\%m\%d) --end $(date \%Y\%m\%d) --period day --output /path/to/data /path/to/log/download.log 21Windows (使用任务计划程序): 创建一个批处理文件run_download.bat:echo off cd C:\path\to\your\script python download_ths_futures.py --code RB8888 --start %DATE:~0,4%%DATE:~5,2%%DATE:~8,2% --end %DATE:~0,4%%DATE:~5,2%%DATE:~8,2% --period day --output C:\path\to\data然后在Windows任务计划程序中创建任务定时执行此批处理文件。在Python脚本内增加日志记录import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(ths_downloader.log), logging.StreamHandler() # 同时输出到控制台 ] ) logger logging.getLogger(__name__) # 在代码中用logger代替print logger.info(f开始下载合约 {contract_code}...) try: # ... 下载逻辑 except Exception as e: logger.error(f下载失败: {e}, exc_infoTrue)5. 常见问题排查与进阶优化5.1 典型错误与解决方案速查表问题现象可能原因排查步骤与解决方案请求返回状态码403/4041. 接口URL或参数已变更。2. Cookie失效或未正确设置。3. IP或请求频率被限制。1.重新抓包确认最新接口地址和参数格式。2.更新Cookie从新抓取的请求中复制或检查Cookie有效期。3.增加延时更换网络环境如使用手机热点或尝试添加更多请求头如X-Requested-With: XMLHttpRequest。能下载文件但Pandas读取报错如Unsupported format1. 文件不是标准Excel格式可能是HTML或加密。2. 缺少对应的引擎如.xls文件但未安装xlrd1.2.0。3. 文件损坏。1.检查响应内容将resp.content前几百字节写入文件用文本编辑器或十六进制查看器检查文件头。2.指定引擎pd.read_excel(..., engineopenpyxl)或enginexlrd。3. 如果返回的是HTML尝试用pd.read_html(BytesIO(resp.content))。数据列名是乱码或非预期中文原始Excel编码问题或包含多余行。1.先查看原始DataFrameprint(df.head())和print(df.columns)。2. 在read_excel中尝试encoding参数如encodinggbk或utf-8。3. 使用skiprows参数跳过文件开头的非数据行。日期列无法正确解析日期格式多样字符串、Excel序列数、时间戳。1.先查看原始值print(df[日期列名].head())。2.手动转换如果看起来像数字如45205使用pd.to_datetime(df[日期列], unitD, origin1899-12-30)。3. 如果格式复杂使用pd.to_datetime(df[日期列], format%Y/%m/%d)指定格式。数值列中有“-”或空字符串导致类型为object数据中包含非数字占位符。在清洗函数中使用pd.to_numeric(..., errorscoerce)进行转换它会将无法转换的值设为NaN。下载速度慢或中途断开网络问题或服务器限流。1.增加超时时间requests.get(..., timeout(10, 30))连接10秒读取30秒。2.实现分片下载对于超长周期数据按年或按月分批请求。3.使用重试机制如前文代码所示配合指数退避。5.2 进阶优化方向异步并发下载如果需要下载大量合约可以使用aiohttp和asyncio进行异步请求大幅提升效率。但务必注意控制并发数避免对目标服务器造成过大压力。# 简略示例 import aiohttp import asyncio async def download_one(session, code, start, end): url ... async with session.get(url, params{...}) as resp: content await resp.read() # ... 处理内容 return content async def main(codes): async with aiohttp.ClientSession(headersheaders) as session: tasks [download_one(session, code, ...) for code in codes] results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果数据验证与完整性检查下载后自动检查数据行数是否在合理范围、是否有连续日期缺失、最新数据是否更新。def validate_data(df, contract_code): if df is None or df.empty: return False, 数据为空 if date not in df.columns: return False, 缺少日期列 # 检查日期是否连续针对日线 date_diff df[date].diff().dt.days.dropna() if not all(date_diff 1): missing date_diff[date_diff 1] return False, f日期不连续间隔有: {missing.unique()} # 检查必要数值列是否存在NaN过多 required_cols [open, high, low, close] for col in required_cols: if col in df.columns and df[col].isna().sum() len(df) * 0.1: # 缺失超过10% return False, f{col}列缺失值过多 return True, 数据验证通过配置文件管理将合约列表、时间范围、请求头、Cookie等配置信息写入config.yaml或config.ini文件使脚本更易于管理和维护。数据存储升级对于海量数据可以考虑存入SQLite或MySQL数据库而非单个CSV文件便于查询和管理。使用Pandas的to_sql方法可以轻松实现。5.3 最后的叮嘱逆向工程获取的数据其可用性和稳定性始终取决于数据源方。同花顺客户端的更新可能导致接口失效。因此这个脚本更适合作为个人或小范围使用的辅助工具而非生产环境的核心依赖。建议定期如每月检查脚本是否仍能正常工作。最重要的始终尊重数据源的版权和服务条款将获取的数据用于个人学习和研究避免商业用途和过量请求干扰服务器。整个流程走下来从抓包分析到脚本成型再到错误处理和优化其实就是一个典型的数据获取ETL抽取、转换、加载微流程。把这个流程吃透不仅解决了同花顺期货数据的问题其方法论也完全可以迁移到其他需要从客户端或网站抓取数据的场景中。工具是死的思路是活的掌握了如何分析请求、模拟会话、处理异常数据你就拥有了获取网络数据的主动权。本文还有配套的精品资源点击获取