用Python+akshare搞定三大交易所期权数据，这个脚本我用了三年

用Pythonakshare构建稳定可靠的期权数据采集系统三年实战经验总结期权数据作为金融衍生品市场的核心要素对量化交易者和金融数据分析师而言至关重要。但获取稳定、准确的交易所期权数据并非易事——数据格式多变、接口不稳定、反爬机制复杂等问题常常让初学者望而却步。本文将分享我基于akshare库构建的三大交易所上交所、深交所、中金所期权数据采集系统这套代码经过三年生产环境检验日均处理数十万条数据记录已成为我量化策略的可靠数据基石。1. 系统架构设计与核心组件一个健壮的期权数据采集系统需要考虑数据获取、异常处理、日志记录和持久化存储等多个环节。经过多次迭代我的系统形成了以下核心架构class OptionDataCollector: def __init__(self): self.logger self._setup_logger() self.session requests.Session() self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) ..., Referer: http://www.sse.com.cn/ } def _setup_logger(self): logger logging.getLogger(__name__) logger.setLevel(logging.INFO) handler logging.FileHandler(option_data.log) formatter logging.Formatter(%(asctime)s - %(levelname)s - %(message)s) handler.setFormatter(formatter) logger.addHandler(handler) return logger关键组件说明请求会话管理使用requests.Session()保持连接池显著提升请求效率日志系统详细记录每次数据采集的详细信息便于问题排查异常处理机制针对各类网络异常设计重试策略数据缓存层避免重复请求已获取的数据提示在实际运行中建议为每个交易所实现单独的子类处理各自特有的数据格式和接口特性。2. 三大交易所数据获取实战2.1 上交所(SSE)期权数据采集上交所期权数据接口相对稳定但需要注意以下几点必须设置正确的Referer头部数据返回格式为CSV但编码为GBK接口有频率限制建议控制请求间隔def get_sse_option_data(self, trade_date): date_str trade_date.strftime(%Y%m%d) url fhttp://query.sse.com.cn/derivative/downloadRisk.do?trade_date{date_str}productType0 try: response self.session.get(url, headersself.headers, timeout10) response.encoding gbk data pd.read_csv(StringIO(response.text)) data[trade_date] trade_date return data except Exception as e: self.logger.error(fSSE数据获取失败 {trade_date}: {str(e)}) return None2.2 深交所(SZSE)期权数据处理深交所的数据获取有几个特别需要注意的坑接口返回的是xlsx格式但直接读取可能遇到编码问题需要处理历史数据格式变更2019年前后格式不同随机参数需要合理设置以避免被反爬def get_szse_option_data(self, trade_date): date_str trade_date.strftime(%Y-%m-%d) url fhttp://www.szse.cn/api/report/ShowReport?SHOWTYPExlsxCATALOGIDoption_hyfxzbTABKEYtab1txtSearchDate{date_str} try: response self.session.get(url, headersself.headers, timeout10) with tempfile.NamedTemporaryFile(suffix.xlsx) as tmp: tmp.write(response.content) tmp.seek(0) data pd.read_excel(tmp.name) data[trade_date] trade_date return data except Exception as e: self.logger.error(fSZSE数据获取失败 {trade_date}: {str(e)}) return None2.3 中金所(CFFEX)期权数据解析中金所数据以XML格式提供解析时需要注意字段可能缺失需要完善的异常处理某些日期的数据接口可能返回空值需要处理命名空间等XML特性def get_cffex_option_data(self, trade_date): date_str trade_date.strftime(%Y%m%d) url fhttp://www.cffex.com.cn/sj/hqsj/rtj/{date_str[:6]}/{date_str[6:]}/index.xml try: response self.session.get(url, headersself.headers, timeout10) root ET.fromstring(response.content) data [] for daily_data in root.findall(.//dailydata): record {child.tag: child.text for child in daily_data} record[trade_date] trade_date data.append(record) return pd.DataFrame(data) except Exception as e: self.logger.error(fCFFEX数据获取失败 {trade_date}: {str(e)}) return None3. 生产环境中的关键优化点经过三年实战我总结了以下几个对系统稳定性至关重要的优化点3.1 智能重试机制网络请求失败是常态而非例外。我设计了分级重试策略瞬时错误立即重试最多3次临时性错误等待30秒后重试持久性错误记录日志并跳过待下次定时任务处理def safe_request(self, url, max_retries3, backoff_factor1): for attempt in range(max_retries): try: response self.session.get(url, headersself.headers, timeout10) response.raise_for_status() return response except requests.exceptions.RequestException as e: if attempt max_retries - 1: raise sleep_time backoff_factor * (2 ** attempt) self.logger.warning(f请求失败{sleep_time}秒后重试: {url}) time.sleep(sleep_time)3.2 数据一致性校验采集到的数据需要经过严格校验检查必填字段是否存在验证数值范围是否合理对比前后交易日的数据变化是否在合理范围内def validate_data(self, df, exchange): # 基本校验 if df.empty: raise ValueError(空数据) # 交易所特定校验 if exchange SSE: required_cols [合约编码, 前结算价, 开盘价] elif exchange SZSE: required_cols [合约简称, 行权价, 成交量] else: required_cols [instrumentid, openprice, volume] missing_cols [col for col in required_cols if col not in df.columns] if missing_cols: raise ValueError(f缺失必要列: {missing_cols}) # 数值校验 if (df[volume] 0).any(): raise ValueError(成交量为负值)3.3 性能优化技巧处理大量历史数据时性能成为关键考量优化手段效果实现方式多线程采集提升吞吐量使用concurrent.futures.ThreadPoolExecutor内存缓存减少重复请求使用functools.lru_cache装饰器批量写入降低I/O开销使用pandas.DataFrame.to_sql的chunksize参数连接复用减少TCP握手保持requests.Session长连接4. 数据存储与后续处理采集到的原始数据需要合理存储以便后续分析。我推荐以下存储方案数据库选择开发环境SQLite简单易用生产环境PostgreSQL功能全面或ClickHouse分析性能优异数据分区策略按交易所分区按日期分区特别是处理历史数据时def save_to_database(self, df, table_name): engine create_engine(postgresql://user:passwordlocalhost:5432/option_data) # 仅追加新数据 existing_dates pd.read_sql( fSELECT DISTINCT trade_date FROM {table_name}, engine )[trade_date].tolist() new_data df[~df[trade_date].isin(existing_dates)] if not new_data.empty: new_data.to_sql( table_name, engine, if_existsappend, indexFalse, chunksize1000 ) self.logger.info(f新增{len(new_data)}条记录到{table_name})数据质量监控每日数据完整性检查异常值自动报警历史数据一致性验证这套系统经过三年不断迭代已经处理了超过200万条期权数据记录支持了我多个量化策略的研发和回测。最关键的体会是稳定性比功能丰富更重要。一个每天可靠运行的基础数据采集系统远比功能复杂但不可靠的系统有价值得多。