量化数据工程实战：基于AmazingData的PIT原则、频率对齐与数据质量管控

一、量化数据的三层分类体系1.1 基本面数据Fundamental Data基本面数据来源于上市公司定期报告包括资产负债表、利润表和现金流量表。AmazingData银河证券星耀数智数据库提供了完整的财务数据接口。基本面数据的关键特征如下。更新频率为季度季报、半年度和年度。发布时间为季度结束后1个月内。数据量级相对较小A股约5000多家公司乘以100多个字段。滞后性较强财务数据反映的是过去经营状况。AmazingData财务数据接口的核心字段极为丰富。以资产负债表为例包含货币资金、应收账款、存货、固定资产、无形资产、资产总计、短期借款、应付账款、负债合计、股本、资本公积、未分配利润、股东权益合计等核心科目以及递延所得税资产/负债、合同资产/负债等新会计准则科目。现金流量表包含经营活动、投资活动和筹资活动的完整现金流明细。利润表包含营业收入、营业成本、营业利润、利润总额、净利润、每股收益等核心指标。常用量化因子包括三类。盈利能力因子包括ROE净资产收益率、ROA总资产收益率和毛利率。估值指标包括PE市盈率、PB市净率和PS市销率。财务质量因子包括产权比率和流动比率。以下是一个基于AmazingData的基本面因子计算示例import AmazingData as adimport pandas as pdimport numpy as np登录AmazingDataad.login(username‘your_username’, password‘your_password’,host‘your_host’, portyour_port)base_data ad.BaseData()info_data ad.InfoData()获取沪深A股代码列表和历史代码表calendar base_data.get_calendar()today calendar[-1]all_codes base_data.get_hist_code_list(security_type‘EXTRA_STOCK_A_SH_SZ’,start_date20130101,end_datetoday,local_path‘D://AmazingData_local_data//’)获取财务数据使用本地缓存加速balance_sheet info_data.get_balance_sheet(code_listall_codes,local_path‘D://AmazingData_local_data//’,is_localFalse # 首次设为False从服务器获取后续设为True从本地读取)cash_flow info_data.get_cash_flow(code_listall_codes,local_path‘D://AmazingData_local_data//’,is_localFalse)income info_data.get_income(code_listall_codes,local_path‘D://AmazingData_local_data//’,is_localFalse)def calc_fundamental_factors(balance_dict, income_dict, cash_flow_dict):“”计算基本面因子注意实际使用需严格遵循PIT原则“”factors {}for code in balance_dict.keys(): try: bs balance_dict[code] inc income_dict.get(code) cf cash_flow_dict.get(code) if bs is None or inc is None or len(bs) 0 or len(inc) 0: continue # 获取最新一期数据 latest_bs bs.iloc[-1] latest_inc inc.iloc[-1] factor {} # 盈利能力 equity latest_bs[TOT_SHARE_EQUITY_INCL_MININT] total_assets latest_bs[TOTALASSETS] net_profit latest_inc[NETPROINCLMIN_INT_INC] revenue latest_inc[TOTOPERAREV] if equity 0: factor[ROE] net_profit / equity if total_assets 0: factor[ROA] net_profit / total_assets if revenue 0: factor[gross_margin] (revenue - latest_inc[LESS_OPERA_COST]) / revenue # 估值指标需配合市值数据 # PE 市值 / 净利润 # PB 市值 / 净资产 # 市值从行情数据获取 # 财务质量 current_assets latest_bs[TOTALCURASSETS] current_liab latest_bs[TOTALCURLIAB] if current_liab 0: factor[current_ratio] current_assets / current_liab total_debt latest_bs[TOTAL LIAB] if equity 0: factor[debt_to_equity] total_debt / equity factors[code] factor except Exception as e: continue return pd.DataFrame(factors).T计算基本面因子fundamental_factors calc_fundamental_factors(balance_sheet, income, cash_flow)print(fundamental_factors.head())1.2 量价数据Price Volume Data量价数据按时间精度可分为四个层级。日频数据包含OHLCV适用于中低频策略和因子计算。分钟频数据同样包含OHLCV适用于日内策略和波动率估计。Tick快照包含最新价和买卖盘适用于高频策略和订单流分析。Level-2逐笔委托/成交包含委托队列、大单明细适用于超高频交易和市场微观结构研究。AmazingData提供了完整的K线数据查询接口支持从1分钟到年线的多种周期获取历史K线数据多周期支持market_data ad.MarketData(calendar)日频K线kline_day market_data.query_kline(code_listall_codes[:10], # 示例取前10只begin_date20200101,end_date20241231,periodad.constant.Period.day.value)分钟频K线kline_min5 market_data.query_kline(code_listall_codes[:10],begin_date20240101,end_date20241231,periodad.constant.Period.min5.value)周线kline_week market_data.query_kline(code_listall_codes[:10],begin_date20200101,end_date20241231,periodad.constant.Period.week.value)使用AmazingData时序函数计算技术指标from AmazingData.operator.time_series_function import TimeSeriesFunctiondef calc_technical_factors(kline_dict):“”基于K线数据计算技术指标因子“”tech_factors {}for code, kline in kline_dict.items(): if len(kline) 60: continue factors {} # 移动平均线 factors[MA5] TimeSeriesFunction.MA(kline[close], 5).iloc[-1] factors[MA20] TimeSeriesFunction.MA(kline[close], 20).iloc[-1] factors[MA60] TimeSeriesFunction.MA(kline[close], 60).iloc[-1] # 指数移动平均 factors[EMA12] TimeSeriesFunction.EMA(kline[close], 12).iloc[-1] factors[EMA26] TimeSeriesFunction.EMA(kline[close], 26).iloc[-1] # MACD dif factors[EMA12] - factors[EMA26] factors[DIF] dif # 波动率 returns kline[close].pct_change().dropna() factors[volatility_20] TimeSeriesFunction.STD(returns, 20).iloc[-1] # 动量 factors[momentum_20] ( kline[close].iloc[-1] / kline[close].iloc[-21] - 1 ) # 振幅 factors[amplitude] ( kline[high].iloc[-1] - kline[low].iloc[-1] ) / kline[pre_close].iloc[-1] tech_factors[code] factorsreturn pd.DataFrame(tech_factors).T1.3 另类数据Alternative DataAmazingData提供了多种另类数据源。融资融券数据包括成交汇总和交易明细可构建杠杆情绪指标。龙虎榜和大宗交易数据可追踪主力资金动向。ETF申赎数据和份额变动可反映机构资金流向。行业指数数据和成分股权重可用于行业轮动策略。可转债数据包括发行、转股、赎回、回售等完整生命周期数据。国债收益率数据支持多种期限3个月到30年可用于利率风险建模。另类数据的处理挑战在于频率不固定、信息差大、处理难度高。以下是一个基于AmazingData的融资融券因子示例def calc_margin_factor(info_data, codes):“”计算融资融券情绪因子“”margin_detail info_data.get_margin_detail(code_listcodes,local_path‘D://AmazingData_local_data//’,is_localFalse)margin_factors {} for code, df in margin_detail.items(): if df is None or len(df) 20: continue latest df.iloc[-1] prev df.iloc[-20] factor {} # 融资余额变化率 if prev[BORROWMONEYBALANCE] 0: factor[margin_balance_change] ( latest[BORROWMONEYBALANCE] / prev[BORROWMONEYBALANCE] - 1 ) # 融资买入额占成交额比例 factor[margin_buy_ratio] latest[PURCHWITHBORROWMONEY] / ( latest[PURCHWITHBORROWMONEY] latest[REPAYMENT_OF_BORROW_MONEY] ) # 融券余量变化 if prev[SEC_LENDING_BALANCE_VOL] 0: factor[short_balance_change] ( latest[SEC_LENDING_BALANCE_VOL] / prev[SEC_LENDING_BALANCE_VOL] - 1 ) margin_factors[code] factorreturn pd.DataFrame(margin_factors).T二、PIT原则防止未来函数的核心防线2.1 什么是未来函数Look-ahead Bias未来函数是指在模型构建或回测中使用了在当时时点尚未可获取的数据导致回测结果虚高。一个典型的陷阱示例如下假设今天是3月31日但年报4月30日才发布如果数据已经包含全年净利润就构成了未来函数。2.2 基于AmazingData的PIT原则实现AmazingData提供了历史代码表接口get_hist_code_list和历史证券信息接口get_history_stock_status可用于PIT验证。核心规则是任何数据的使用时点不得早于该数据实际可被获取的时点。以下是一个PIT数据处理器的实现class PITDataHandler:“”“PIT数据处理器”“”def __init__(self, info_data, base_data): self.info_data info_data self.base_data base_data def get_pit_financial_data(self, code_list, query_date): 获取query_date时点实际可获取的最新财务数据 使用AmazingData的财务数据和公告日期字段 # 获取财务数据包含公告日期 income self.info_data.get_income( code_listcode_list, local_pathD://AmazingData_local_data//, is_localTrue ) pit_data {} for code in code_list: if code not in income or income[code] is None: continue df income[code] # 筛选在query_date之前已公告的数据 # ACTUAL_ANN_DATE为实际公告日期 df[ACTUAL_ANN_DATE] pd.to_datetime(df[ACTUAL_ANN_DATE]) query_dt pd.to_datetime(str(query_date)) available df[df[ACTUAL_ANN_DATE] query_dt] if len(available) 0: continue # 取最新一期已公告的数据 latest available.iloc[-1] pit_data[code] latest return pd.DataFrame(pit_data).T def create_pit_panel(self, code_list, date_range): 生成完整的PIT面板数据 pit_panel [] for date in date_range: daily_data self.get_pit_financial_data(code_list, date) if len(daily_data) 0: daily_data[trade_date] date pit_panel.append(daily_data) return pd.concat(pit_panel, ignore_indexTrue) if pit_panel else None使用历史证券信息进行PIT验证def validate_pit_with_history_status(base_data, info_data, codes):“”使用历史证券信息验证PIT包含历史的涨跌停、ST、除权除息等信息“”history_status info_data.get_history_stock_status(code_listcodes,local_path‘D://AmazingData_local_data//’,is_localFalse)# 验证某日期某股票是否停牌、ST等 # IS_ST_SEC: 是否ST # IS_SUSP_SEC: 是否停牌 # IS_XD_SEC: 是否除息 # IS_XR_SEC: 是否除权return history_statusPIT验证清单包括四项。第一财报数据使用ACTUAL_ANN_DATE而非REPORTINGPERIOD。第二指数成分股使用调整生效日而非公告日。第三宏观数据使用发布日而非统计截止日。第四分析师预期使用研报发布日而非报告期。三、频率对齐多源数据的统一处理3.1 对齐策略研究通常统一至日频不同频率数据需进行转换。低频数据转日频采用Forward Fill方法。财报数据向后填充但只能填充到下一个有效值之前需确保不会用未来数据填充。高频数据转日频采用降频聚合。分钟数据降频为日频时开盘价取第一个值最高价取最大值最低价取最小值收盘价取最后一个值成交量取总和。VWAP计算为价格乘以成交量的总和除以成交量的总和。AmazingData的K线数据已经天然支持频率对齐可直接使用日频K线进行多因子合成。3.2 数据对齐的注意事项数据对齐需注意四个问题。第一是交易日与非交易日的统一需使用交易日历剔除节假日AmazingData的get_calendar接口提供完整交易日历。第二是不同市场开盘时间的统一需统一到主要市场收盘时点。第三是停牌处理需标记停牌日避免用前收盘价填充导致偏差可使用get_history_stock_status接口。第四是复权处理需统一使用前复权或后复权避免价格断裂AmazingData提供get_backward_factor和get_adj_factor接口。四、数据质量管控体系4.1 常见数据质量问题以下是一个基于AmazingData的数据质量检查器class DataQualityChecker:“”“数据质量检查器”“”def __init__(self, data_dict): self.data data_dict def check_missing(self, threshold0.05): 缺失值检查 missing_report {} for code, df in self.data.items(): if df is None: missing_report[code] 1.0 continue missing_ratio df.isnull().mean().mean() missing_report[code] missing_ratio problematic {k: v for k, v in missing_report.items() if v threshold} return problematic def check_price_limits(self, kline_dict): 涨跌停价格异常检查 base_data ad.BaseData() code_info base_data.get_code_info(security_typeEXTRA_STOCK_A) anomalies [] for code, kline in kline_dict.items(): if code not in code_info.index: continue high_limit code_info.loc[code, high_limited] low_limit code_info.loc[code, lowlimited] # 检查是否有价格超出涨跌停限制 invalid_high kline[kline[high] high_limit * 1.01] invalid_low kline[kline[low] low_limit * 0.99] if len(invalid_high) 0 or len(invalid_low) 0: anomalies.append({ code: code, invalid_high_count: len(invalid_high), invalid_low_count: len(invalid_low) }) return anomalies def check_survivorship_bias(self, current_codes, base_data): 幸存者偏差检查 # 获取历史所有出现过的代码 hist_codes base_data.get_hist_code_list( security_typeEXTRA_STOCK_A_SH_SZ, start_date20130101, end_datebase_data.get_calendar()[-1], local_pathD://AmazingData_local_data// ) historical_set set(hist_codes) current_set set(current_codes) missing historical_set - current_set return len(missing), list(missing)[:10] def check_look_ahead(self, factor_dates, announcement_dates): 未来函数检查 violations (announcement_dates factor_dates).sum() return violations4.2 数据质量评分卡数据质量评分卡包含五个维度。完整性占25%检查缺失值比例和字段覆盖率。准确性占25%检查异常值比例和逻辑一致性。及时性占20%检查数据延迟和更新频率。一致性占15%检查跨源比对和口径统一。可追溯占15%检查数据来源和变更记录。五、总结数据是量化研究的基石Garbage In, Garbage Out是永恒真理。第一PIT原则是防止未来函数的第一道防线。第二频率对齐确保多源数据的可比性。第三质量管控需要系统化的检查框架。第四文档记录每个数据源的获取方式和处理逻辑。AmazingData的本地缓存机制HDF5格式和增量更新功能为数据质量管控提供了基础设施支持。好的量化研究员首先是好的数据工程师。