36.黄金与中国国债收益率相关分析（美债作对照）

from pathlib import Path  # 导入路径工具
import pandas as pd  # 导入表格库
import numpy as np  # 导入有限值与方差检查工具
import hashlib  # 导入快照指纹工具
import h5py  # 读取原生HDF5复合数据集
GOLD_SECURITY_CODE='518880.SH'  # 配置命名中国黄金ETF稳定代码
GOLD_SECURITY_NAME='华安黄金ETF'  # 配置供人工核对的证券名称
root=Path('/home/ubuntu/r2_data_mount/data')  # 指定可用根目录
gold_path=root/'fund'/'funds.h5'  # 绑定课程规定基金文件
yield_path=root/'yield_curve'/'yield_curve.h5'  # 绑定课程规定收益率文件
if not gold_path.exists() or not yield_path.exists(): raise FileNotFoundError('未找到课程数据文件，请从课程数据下载入口获取并核对文件位置')  # 文件缺失即停止
fund_frame=pd.read_hdf(gold_path,key='/funds')  # 按真实Pandas HDF key读取
required_fund={'symbol','datetime','close'}  # 声明基金字段要求
if not required_fund.issubset(fund_frame.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 字段结构不符即停止
normalized_symbol=fund_frame['symbol'].astype(str).str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False)  # 统一交易所后缀
gold_frame=fund_frame.loc[normalized_symbol.eq(GOLD_SECURITY_CODE),['symbol','datetime','close']].copy()  # 精确选择华安黄金ETF
gold_frame['datetime']=pd.to_datetime(gold_frame['datetime'].astype(str),format='%Y%m%d%H%M%S',errors='coerce')  # 解析真实时间字段
gold_frame=gold_frame.loc[gold_frame['datetime'].between('2019-01-01','2025-12-31')]  # 固定课堂期间
security_column='symbol'  # 记录真实证券字段
audit_rows=[{'kind':'fund','file':gold_path.name,'key':'/funds','selector':GOLD_SECURITY_CODE,'matches':len(gold_frame),'schema_ok':True}]  # 保存绑定检查
if gold_frame.empty: raise ValueError(f'输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 证券缺失即停止

with h5py.File(yield_path,mode='r') as handle:  # 以原生HDF5只读打开
    if '/data' not in handle: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 数据集缺失即停止
    raw_curve=handle['/data'][()]  # 读取复合数组快照
    if not {'date','10Y'}.issubset(raw_curve.dtype.names): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 检查宽表字段
curve_frame=pd.DataFrame({'date':pd.to_datetime(raw_curve['date'].astype(str),format='%Y%m%d',errors='coerce'),'yield_10y':raw_curve['10Y']})  # 标准化日期与10Y列
curve_frame=curve_frame.loc[curve_frame['date'].between('2019-01-01','2025-12-31')].assign(tenor='10Y')  # 固定期间并显式保留期限
curve_date,tenor_column,yield_column,tenor_label='date','tenor','yield_10y','10Y'  # 绑定后续统一接口
yield_key='/data'  # 记录真实原生HDF5数据集路径
ten_year_curve=curve_frame.copy()  # 10Y已从宽表精确选择
audit_rows.append({'kind':'yield','file':yield_path.name,'key':yield_key,'selector':'10Y','matches':len(ten_year_curve),'schema_ok':True})  # 保存曲线检查
audit_table=pd.DataFrame(audit_rows)  # 形成文件/key/证券/期限检查表
print(audit_table)  # 挂载存在时发布完整候选检查，不只发布成功项

gold_date=next((column for column in ['datetime','date','trade_date'] if column in gold_frame.columns),None)  # 识别黄金日期字段
gold_value=next((column for column in ['close','nav','unit_nav'] if column in gold_frame.columns),None)  # 识别黄金价格字段
if None in [gold_date,gold_value,security_column,curve_date,tenor_column,yield_column] or gold_frame.empty or curve_frame.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 字段或数据无法对应时停止
def collapse_observations(frame,date_column,value_column,label,require_positive=False):  # 定义观测键冲突请先检查数据后再继续接口
    work=frame.assign(observation_date=pd.to_datetime(frame[date_column],errors='coerce'),observation_value=pd.to_numeric(frame[value_column],errors='coerce'))[['observation_date','observation_value']].copy()  # 统一日期与数值类型
    invalid=work['observation_date'].isna() | ~np.isfinite(work['observation_value']) | (work['observation_value'].le(0) if require_positive else False)  # 识别无效日期、非有限值或非正价格
    if invalid.any(): raise ValueError(f'输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 观测值不可用时停止
    duplicate_mask=work.duplicated('observation_date',keep=False)  # 识别同日重复观测
    conflicting_dates=int(work.loc[duplicate_mask].groupby('observation_date')['observation_value'].nunique(dropna=False).gt(1).sum()) if duplicate_mask.any() else 0  # 统计同日数值冲突键
    if conflicting_dates: raise ValueError(f'输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 同日数值冲突时停止
    collapsed=work.drop_duplicates(['observation_date','observation_value']).drop_duplicates('observation_date').sort_values('observation_date')  # 仅合并数值完全一致的重复观测
    audit={'label':label,'duplicate_rows':int(duplicate_mask.sum()),'conflicting_dates':conflicting_dates,'unique_dates':len(collapsed)}  # 形成观测键去重检查
    return collapsed.set_index('observation_date')['observation_value'],audit  # 返回唯一日期序列与检查
gold_daily,gold_observation_audit=collapse_observations(gold_frame,gold_date,gold_value,'gold',require_positive=True)  # 检查并合并完全一致的黄金同日观测
yield_daily,yield_observation_audit=collapse_observations(ten_year_curve,curve_date,yield_column,'ten-year yield')  # 检查并合并完全一致的十年期同日观测
gold_weekly_observed=gold_daily.resample('W-FRI').last()  # 将唯一日频黄金价格转为观测周末值
yield_weekly_observed=yield_daily.resample('W-FRI').last()  # 将唯一日频十年收益率转为观测周末值
calendar_start=max(gold_weekly_observed.index.min(),yield_weekly_observed.index.min())  # 固定两资产共同周频起点
calendar_end=min(gold_weekly_observed.index.max(),yield_weekly_observed.index.max())  # 固定两资产共同周频终点
if calendar_end<calendar_start: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 无共同周频窗口时停止

weekly_calendar=pd.date_range(calendar_start,calendar_end,freq='W-FRI')  # 建立不允许静默缺周的完整周五日历
gold_series=gold_weekly_observed.reindex(weekly_calendar)  # 按完整日历显式保留黄金缺周
ten_year_series=yield_weekly_observed.reindex(weekly_calendar)  # 按完整日历显式保留收益率缺周
def longest_missing_run(series):  # 定义连续缺周检查函数
    missing=series.isna()  # 标记缺失周
    groups=missing.ne(missing.shift()).cumsum()  # 为连续缺失段编号
    return int(missing.groupby(groups).sum().max()) if missing.any() else 0  # 返回最长连续缺周数
weekly_coverage_audit={'policy':'COMPLETE_WEEKLY_CALENDAR_NO_FILL_THEN_COMPLETE_CASE','calendar_weeks':len(weekly_calendar),'gold_observed_weeks':int(gold_series.notna().sum()),'gold_missing_weeks':int(gold_series.isna().sum()),'gold_longest_gap':longest_missing_run(gold_series),'yield_observed_weeks':int(ten_year_series.notna().sum()),'yield_missing_weeks':int(ten_year_series.isna().sum()),'yield_longest_gap':longest_missing_run(ten_year_series)}  # 发布双资产覆盖与缺口检查
minimum_weekly_coverage=0.80  # 事先设定课堂周频水平值覆盖最低要求
if gold_series.notna().mean()<minimum_weekly_coverage or ten_year_series.notna().mean()<minimum_weekly_coverage: raise ValueError(f'输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 任一资产覆盖不足时停止
if weekly_coverage_audit['gold_longest_gap']>4 or weekly_coverage_audit['yield_longest_gap']>4: raise ValueError(f'输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 连续缺口超过四周时停止
weekly_transforms=pd.concat({'gold_return':gold_series.pct_change(fill_method=None),'yield_change':ten_year_series.diff()},axis=1)  # 不填充地计算黄金收益与利率变动
weekly_coverage_audit['joint_transform_weeks']=int(weekly_transforms.notna().all(axis=1).sum())  # 记录同一时间段可用变动周数
aligned_frame=weekly_transforms.dropna()  # 仅在无填充变换后采用同一时间段完整观测
if len(aligned_frame)<52: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 周频窗口不足时停止
if aligned_frame[['gold_return','yield_change']].std().eq(0).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 常数序列禁止相关输出
full_correlation=aligned_frame.corr().loc['gold_return','yield_change']  # 计算全样本相关
rolling_correlation=aligned_frame['gold_return'].rolling(52).corr(aligned_frame['yield_change'])  # 计算52周滚动相关
lagged_correlation=pd.Series({lag:aligned_frame['gold_return'].corr(aligned_frame['yield_change'].shift(lag)) for lag in range(-4,5)},name='correlation')  # 计算正负4周滞后相关
stress_mask=aligned_frame['yield_change'].abs()>=aligned_frame['yield_change'].abs().quantile(0.90)  # 定义利率变动绝对值前10%的压力期

stress_frame=aligned_frame.loc[stress_mask]  # 固定压力期样本
if len(stress_frame)<5 or stress_frame.std().eq(0).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 压力窗检查要求
stress_correlation=stress_frame.corr().loc['gold_return','yield_change']  # 计算压力期相关
reported_rolling=rolling_correlation.dropna(); evidence=np.r_[full_correlation,stress_correlation,reported_rolling.to_numpy(),lagged_correlation.to_numpy()]  # 汇总拟发布依据
if reported_rolling.empty or not np.isfinite(evidence).all(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # NaN/inf不得发布
fingerprint=hashlib.sha256(pd.util.hash_pandas_object(aligned_frame,index=True).values.tobytes()).hexdigest()[:16]  # 生成同一时间段快照指纹
binding_contract={'fund_file':gold_path.name,'yield_file':yield_path.name,'hdf_key':yield_key,'fund_security':GOLD_SECURITY_CODE,'tenor':str(tenor_label),'schema':{'gold_date':gold_date,'gold_value':gold_value,'curve_date':curve_date,'yield_value':yield_column},'observation_audit':[gold_observation_audit,yield_observation_audit],'weekly_coverage_audit':weekly_coverage_audit,'period':[str(aligned_frame.index.min()),str(aligned_frame.index.max())],'fingerprint':fingerprint}  # 返回完整绑定、观测键与周频缺口检查
print(binding_contract,{'n':len(aligned_frame),'full':full_correlation,'stress':stress_correlation},reported_rolling.tail(),lagged_correlation)  # 输出绑定与四类相关依据

def classify_correlation(value):  # 定义与课文完全一致的相关区间分类器
    if not np.isfinite(value): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 非有限相关值请先检查数据后再继续
    if value < -0.5: return '较强负相关'  # 严格低于−0.5归入较强负相关
    if value < -0.3: return '中度负相关'  # −0.5至−0.3归入中度负相关
    if value < 0.3: return '弱相关或方向不明'  # −0.3至0.3归入弱或方向不明
    return '正相关背离'  # 0.3及以上归入正相关背离
correlation_class=classify_correlation(full_correlation)  # 使用统一函数分类全样本相关
classification={'measure':'full_correlation','value':full_correlation,'class':correlation_class,'cut_points':[-0.5,-0.3,0.3],'threshold_status':'课堂语言规则，非显著性判断或稳定投资信号'}  # 形成含切点与边界的分类依据
expected_classes={-0.5:['较强负相关','中度负相关','中度负相关'],-0.3:['中度负相关','弱相关或方向不明','弱相关或方向不明'],0.3:['弱相关或方向不明','正相关背离','正相关背离']}  # 事先设定三个切点的下/等/上预期类别
boundary_rows=[]  # 初始化可见边界测试行
for cut_point in [-0.5,-0.3,0.3]:  # 逐切点构造三个精确位置
    test_values=[np.nextafter(cut_point,-np.inf),cut_point,np.nextafter(cut_point,np.inf)]  # 构造下方、等于与上方浮点值
    for position,value,expected_class in zip(['below','at','above'],test_values,expected_classes[cut_point]):  # 将位置、精确值与预期类别对齐
        actual_class=classify_correlation(value)  # 调用权威分类器
        boundary_rows.append({'cut_point':cut_point,'position':position,'signed_delta':f'{value-cut_point:+.3e}','exact_value':value.hex(),'expected_class':expected_class,'actual_class':actual_class})  # 保存可见差值、精确表示与类别
boundary_sensitivity=pd.DataFrame(boundary_rows)  # 形成学生可逐行核对的边界表
if not boundary_sensitivity['actual_class'].eq(boundary_sensitivity['expected_class']).all(): raise AssertionError('correlation boundary classification mismatch')  # 权威分类与预期序列不一致时停止
try:  # 显式测试非有限值检查要求
    classify_correlation(np.nan)  # 将NaN送入分类器检查请先检查数据后再继续
    nonfinite_guard='FAILED'  # 若未异常则标记检查要求失败
except ValueError as error:  # 捕获预期的请检查数据
    nonfinite_guard=str(error)  # 保存非有限值检查要求依据
print(classification)  # 输出统一分类依据
print(boundary_sensitivity.to_string(index=False))  # 无省略地输出九行切点位置、精确值、差值与预期/实际类别
print({'nonfinite_guard':nonfinite_guard})  # 输出非有限值检查要求