当前位置：首页 > news >正文

【期货量化进阶】期货市场数据质量评估方法（实战技巧）

news 2026/3/26 18:10:31

一、前言

数据质量直接影响策略表现。准确评估数据质量可以帮助我们发现数据问题、提高策略可靠性。本文将介绍各种数据质量评估方法。

数据质量指标
完整性检查
准确性验证
一致性检查
数据清洗

二、为什么选择天勤量化（TqSdk）

功能	说明
历史数据	支持获取完整历史数据
数据验证	支持数据验证功能
灵活分析	pandas/numpy支持分析
数据修复	支持数据修复

安装方法：

pipinstalltqsdk pandas numpy

三、数据质量指标

3.1 质量维度

维度	说明	重要性
完整性	数据是否完整	⭐⭐⭐⭐⭐
准确性	数据是否准确	⭐⭐⭐⭐⭐
一致性	数据是否一致	⭐⭐⭐⭐
及时性	数据是否及时	⭐⭐⭐

3.2 常见问题

问题	说明	影响
缺失数据	数据缺失	策略失效
异常值	数据异常	信号错误
重复数据	数据重复	计算错误
时间错乱	时间顺序错误	回测错误

四、完整性检查

4.1 缺失数据检查

#!/usr/bin/env python# -*- coding: utf-8 -*-""" 功能：数据质量评估 说明：本代码仅供学习参考 """fromtqsdkimportTqApi,TqAuthimportpandasaspdimportnumpyasnpdefcheck_missing_data(klines):"""检查缺失数据"""missing_stats={}forcolin['open','high','low','close','volume']:missing_count=klines[col].isna().sum()missing_ratio=missing_count/len(klines)missing_stats[col]={'missing_count':missing_count,'missing_ratio':missing_ratio}returnmissing_stats# 使用示例api=TqApi(auth=TqAuth("快期账户","快期密码"))klines=api.get_kline_serial("SHFE.rb2510",3600,500)api.wait_update()missing_stats=check_missing_data(klines)forcol,statsinmissing_stats.items():print(f"{col}: 缺失{stats['missing_count']}条 ({stats['missing_ratio']:.2%})")api.close()

4.2 时间连续性检查

defcheck_time_continuity(klines,expected_interval=3600):"""检查时间连续性"""time_diffs=klines.index.to_series().diff().dt.total_seconds()# 找出异常间隔abnormal_intervals=time_diffs[time_diffs!=expected_interval]return{'total_bars':len(klines),'abnormal_count':len(abnormal_intervals),'abnormal_intervals':abnormal_intervals.tolist()}# 使用示例continuity=check_time_continuity(klines)print(f"异常间隔数:{continuity['abnormal_count']}")

五、准确性验证

5.1 价格合理性检查

defcheck_price_validity(klines):"""检查价格合理性"""issues=[]# 检查OHLC关系invalid_ohlc=((klines['high']<klines['low'])|(klines['high']<klines['open'])|(klines['high']<klines['close'])|(klines['low']>klines['open'])|(klines['low']>klines['close']))ifinvalid_ohlc.any():issues.append(f"OHLC关系异常:{invalid_ohlc.sum()}条")# 检查价格变化幅度price_change=klines['close'].pct_change()extreme_changes=abs(price_change)>0.1# 单日变化超过10%ifextreme_changes.any():issues.append(f"价格变化异常:{extreme_changes.sum()}条")returnissues# 使用示例issues=check_price_validity(klines)ifissues:forissueinissues:print(f"问题:{issue}")

5.2 成交量检查

defcheck_volume_validity(klines):"""检查成交量合理性"""issues=[]# 检查负成交量negative_volume=klines['volume']<0ifnegative_volume.any():issues.append(f"负成交量:{negative_volume.sum()}条")# 检查异常大成交量volume_mean=klines['volume'].mean()volume_std=klines['volume'].std()extreme_volume=klines['volume']>volume_mean+5*volume_stdifextreme_volume.any():issues.append(f"异常大成交量:{extreme_volume.sum()}条")returnissues# 使用示例volume_issues=check_volume_validity(klines)ifvolume_issues:forissueinvolume_issues:print(f"问题:{issue}")

六、一致性检查

6.1 跨周期一致性

defcheck_cross_timeframe_consistency(api,symbol):"""检查跨周期一致性"""# 获取不同周期数据klines_1h=api.get_kline_serial(symbol,3600,100)klines_4h=api.get_kline_serial(symbol,14400,100)api.wait_update()# 检查收盘价一致性issues=[]# 4小时K线的收盘价应该等于对应1小时K线的最后一个收盘价foriinrange(len(klines_4h)):end_time=klines_4h.index[i]start_time=end_time-pd.Timedelta(hours=4)klines_1h_in_range=klines_1h[(klines_1h.index>=start_time)&(klines_1h.index<=end_time)]iflen(klines_1h_in_range)>0:expected_close=klines_1h_in_range['close'].iloc[-1]actual_close=klines_4h['close'].iloc[i]ifabs(expected_close-actual_close)>0.01:issues.append(f"时间{end_time}: 不一致")returnissues# 使用示例consistency_issues=check_cross_timeframe_consistency(api,"SHFE.rb2510")ifconsistency_issues:print(f"一致性问题:{len(consistency_issues)}条")

七、数据清洗

7.1 异常值处理

defclean_abnormal_values(klines,method='median'):"""清洗异常值"""cleaned=klines.copy()forcolin['open','high','low','close']:# 使用IQR方法识别异常值Q1=cleaned[col].quantile(0.25)Q3=cleaned[col].quantile(0.75)IQR=Q3-Q1 lower_bound=Q1-1.5*IQR upper_bound=Q3+1.5*IQR outliers=(cleaned[col]<lower_bound)|(cleaned[col]>upper_bound)ifmethod=='median':cleaned.loc[outliers,col]=cleaned[col].median()elifmethod=='forward_fill':cleaned.loc[outliers,col]=cleaned[col].fillna(method='ffill')returncleaned# 使用示例cleaned_klines=clean_abnormal_values(klines)

7.2 缺失值填充

deffill_missing_data(klines,method='forward_fill'):"""填充缺失值"""filled=klines.copy()ifmethod=='forward_fill':filled=filled.fillna(method='ffill')elifmethod=='backward_fill':filled=filled.fillna(method='bfill')elifmethod=='interpolate':filled=filled.interpolate()returnfilled# 使用示例filled_klines=fill_missing_data(klines,method='interpolate')

八、数据质量报告

8.1 生成报告

defgenerate_quality_report(klines):"""生成质量报告"""report={'total_bars':len(klines),'missing_data':check_missing_data(klines),'price_issues':check_price_validity(klines),'volume_issues':check_volume_validity(klines),'time_continuity':check_time_continuity(klines)}# 计算质量分数quality_score=100# 缺失数据扣分forcol,statsinreport['missing_data'].items():quality_score-=stats['missing_ratio']*20# 问题扣分quality_score-=len(report['price_issues'])*5quality_score-=len(report['volume_issues'])*5quality_score=max(0,quality_score)report['quality_score']=quality_scorereturnreport# 使用示例report=generate_quality_report(klines)print(f"数据质量分数:{report['quality_score']:.2f}")