33.数据分布情况分析

from pathlib import Path  # 导入路径工具
import matplotlib.pyplot as plt  # 导入绘图库
import pandas as pd  # 导入表格库
import numpy as np  # 导入有限值数据质量检查工具
from scipy import stats  # 导入Q-Q图与稳健分布工具
path=Path('/home/ubuntu/r2_data_mount/data/stock/stock_price_post_adjusted.parquet')  # 指定复权价格
if not path.exists(): raise FileNotFoundError('未找到课程数据文件，请从课程数据下载入口获取并核对文件位置')  # 缺失即停止
price_rows=pd.read_parquet(path,columns=['order_book_id','date','close'],filters=[('order_book_id','==','600104.XSHG')])  # 按实际字段结构与目标证券读取必需字段
price_rows=price_rows.rename(columns={'order_book_id':'ts_code','date':'trade_date'}).copy()  # 标准化证券与交易日字段供后续核对
if price_rows.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 目标证券缺失时停止
price_rows['trade_date']=pd.to_datetime(price_rows['trade_date'],errors='coerce')  # 解析交易日期键
price_rows['close']=pd.to_numeric(price_rows['close'],errors='coerce')  # 解析复权收盘价
if price_rows['trade_date'].isna().any() or not np.isfinite(price_rows['close']).all() or price_rows['close'].le(0).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 日期或价格不可用时停止
if price_rows.duplicated('trade_date').any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 目标日期键不唯一时停止
price_rows=price_rows.sort_values('trade_date')  # 按唯一日期键固定时间顺序
returns=price_rows['close'].pct_change(fill_method=None).dropna()  # 计算单只股票日收益率
if not np.isfinite(returns).all(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 派生收益非有限时停止
if len(returns)<30: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 最小收益样本不足时禁止发布分布依据
distribution_summary=returns.agg(['count','mean','median','std','skew'])  # 汇总中心、离散与偏度
fisher_kurtosis=returns.kurt()  # 按Pandas Fisher约定计算超额峰度且正态基准为0

figure,axes=plt.subplots(1,2)  # 创建直方图与箱线图画布
axes[0].hist(returns,bins=30)  # 绘制收益率直方图
axes[1].boxplot(returns,vert=True)  # 绘制收益率箱线图
qq_theoretical,qq_ordered=stats.probplot(returns,dist='norm',fit=False)  # 计算Q-Q图理论与样本分位点
lower_bound,upper_bound=returns.quantile([0.01,0.99])  # 定义1%缩尾边界
winsorized_returns=returns.clip(lower_bound,upper_bound)  # 生成稳健性比较序列
sensitivity=pd.DataFrame({'raw':[returns.skew(),fisher_kurtosis],'winsorized':[winsorized_returns.skew(),winsorized_returns.kurt()]},index=['skew','fisher_kurtosis'])  # 比较原始与缩尾统计
print(distribution_summary,{'fisher_kurtosis':fisher_kurtosis,'qq_points':len(qq_ordered)},sensitivity)  # 输出分布统计、Q-Q点数与敏感性结果