35.上市公司营业利润增长率的计算

def select_latest_disclosure(frame,value_fields):  # 统一财报披露版本选择约定
    required={'ts_code','end_date',*value_fields}; versions=[c for c in ['ann_date','f_ann_date'] if c in frame.columns]  # 固定键与可接受版本字段
    if not required<=set(frame.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 缺列时停止
    work=frame.dropna(subset=list(required)).copy(); work['end_date']=pd.to_datetime(work['end_date'],errors='coerce')  # 清理必需字段
    for column in versions: work[column]=pd.to_datetime(work[column],errors='coerce')  # 解析披露日期
    keys=['ts_code','end_date']; work=work.dropna(subset=['end_date']); duplicate_mask=work.duplicated(keys,keep=False); before=int(duplicate_mask.sum())  # 统计选择前重复
    if before and (not versions or work.loc[duplicate_mask,versions].isna().all(axis=1).any()): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 无版本依据时停止
    work['_version']=work[versions].max(axis=1) if versions else pd.NaT  # 合并双版本字段为排序键
    latest_version=work.groupby(keys)['_version'].transform('max') if versions else pd.Series(pd.NaT,index=work.index)  # 为每个公司报告期定位最晚披露时点
    latest_rows=work.loc[~duplicate_mask | work['_version'].eq(latest_version)].copy()  # 仅保留单例或并列最晚版本
    tied_mask=latest_rows.duplicated(keys,keep=False); tied_latest_rows=int(tied_mask.sum())  # 统计并列最晚版本行
    conflicting_latest_keys=int(latest_rows.loc[tied_mask].groupby(keys,dropna=False)[value_fields].nunique(dropna=False).gt(1).any(axis=1).sum()) if tied_latest_rows else 0  # 统计并列且数值冲突的披露键
    if conflicting_latest_keys: raise ValueError(f'输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 并列最新值冲突时请先检查数据后再继续
    selected=latest_rows.drop_duplicates(keys+value_fields).drop_duplicates(keys).drop(columns='_version',errors='ignore')  # 仅合并数值完全一致的并列行
    after=int(selected.duplicated(keys).sum()); audit={'duplicate_rows_before':before,'tied_latest_rows':tied_latest_rows,'conflicting_latest_keys':conflicting_latest_keys,'duplicate_rows_after':after,'version_fields':versions}  # 返回并列与冲突检查
    if after: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 选择后仍重复则停止
    return selected,audit  # 返回唯一版本与检查依据

from pathlib import Path  # 导入路径工具
import pandas as pd  # 导入表格库
import numpy as np  # 导入有限值与安全除法工具
path=Path('/home/ubuntu/r2_data_mount/data/stock/financial_statement.h5')  # 指定财务报表
if not path.exists(): raise FileNotFoundError('未找到课程数据文件，请从课程数据下载入口获取并核对文件位置')  # 缺失即停止
financials=pd.read_hdf(path,key='financial_data').rename(columns={'order_book_id':'ts_code','quarter':'end_date','info_date':'ann_date','profit_from_operation':'operate_profit'})  # 按已核验key读取并标准化营业利润字段
financials['ts_code']=financials['ts_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False)  # 统一证券代码后缀
financials['end_date']=pd.PeriodIndex(financials['end_date'].str.upper(),freq='Q').end_time.normalize()  # 将季度键转换为报告期末日期
company,version_audit=select_latest_disclosure(financials.loc[financials['ts_code'].eq('600104.SH')],['operate_profit'])  # 筛选上海汽车最晚披露版本
if company.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 空结果时停止
company=company.sort_values('end_date').copy()  # 固定报告期时间顺序
if not np.isfinite(company['operate_profit']).all(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 非有限营业利润时停止
company['fiscal_year']=company['end_date'].dt.year  # 构造财政年键
company['fiscal_quarter']=company['end_date'].dt.quarter  # 构造财政季度键
if company.duplicated(['fiscal_year','fiscal_quarter']).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 期间键不唯一时停止
prior_profit=company[['fiscal_year','fiscal_quarter','operate_profit']].rename(columns={'operate_profit':'prior_year_profit'}).copy()  # 建立上年同季营业利润查找表
prior_profit['fiscal_year']=prior_profit['fiscal_year']+1  # 将历史年键前移一年以匹配当期
company=company.merge(prior_profit,on=['fiscal_year','fiscal_quarter'],how='left',validate='one_to_one')  # 按年与季度精确连接同比分母
denominator_scale=max(float(company['operate_profit'].abs().max()),1.0)  # 用样本量级定义接近零最低要求
near_zero_cutoff=denominator_scale*1e-6  # 设定课堂数值稳定性阈值

company['profit_change']=company['operate_profit']-company['prior_year_profit']  # 对所有可比期发布营业利润绝对变化
company['comparison_status']='CONVENTIONAL_GROWTH'  # 默认仅两期均为稳定正值时使用常规增长
company.loc[company['prior_year_profit'].isna(),'comparison_status']='MISSING_PRIOR_QUARTER'  # 标记缺少上年同季
company.loc[company['prior_year_profit'].notna() & company['prior_year_profit'].abs().le(near_zero_cutoff),'comparison_status']='ZERO_OR_NEAR_ZERO_BASE'  # 标记零或接近零基期
company.loc[company['prior_year_profit'].lt(-near_zero_cutoff) & company['operate_profit'].lt(0) & company['profit_change'].gt(0),'comparison_status']='LOSS_NARROWING'  # 负基期且亏损绝对额下降时标记亏损收窄
company.loc[company['prior_year_profit'].lt(-near_zero_cutoff) & company['operate_profit'].lt(0) & company['profit_change'].lt(0),'comparison_status']='LOSS_WIDENING'  # 负基期且亏损绝对额上升时标记亏损扩大
company.loc[company['prior_year_profit'].lt(-near_zero_cutoff) & company['operate_profit'].lt(0) & company['profit_change'].eq(0),'comparison_status']='LOSS_UNCHANGED'  # 标记亏损持平
company.loc[company['prior_year_profit'].lt(-near_zero_cutoff) & company['operate_profit'].ge(0),'comparison_status']='SIGN_CHANGE_LOSS_TO_NONLOSS'  # 标记亏损转非亏损
company.loc[company['prior_year_profit'].gt(near_zero_cutoff) & company['operate_profit'].le(0),'comparison_status']='SIGN_CHANGE_PROFIT_TO_NONPROFIT'  # 标记盈利转非盈利
company['denominator_issue']=company['comparison_status'].where(company['comparison_status'].ne('CONVENTIONAL_GROWTH'),'')  # 将非常规比较状态同步到异常清单
valid_yoy=company['comparison_status'].eq('CONVENTIONAL_GROWTH')  # 仅将两期均为稳定正值的精确同季对比纳入增长率
company['quarter_yoy']=np.where(valid_yoy,company['operate_profit']/company['prior_year_profit']-1,np.nan)  # 计算可解释的精确同季同比
abnormal_denominators=company.loc[~valid_yoy,['end_date','operate_profit','prior_year_profit','profit_change','comparison_status','denominator_issue']].copy()  # 生成含绝对变化与亏损/符号状态的异常清单
annual=company.loc[company['end_date'].dt.month.eq(12)].set_index(company.loc[company['end_date'].dt.month.eq(12),'end_date'].dt.year)['operate_profit']  # 提取年末累计营业利润
annual=annual.sort_index()  # 按年度固定CAGR端点顺序
annual_scale=max(float(annual.abs().max()),1.0)  # 用年度样本量级定义端点稳定性最低要求
annual_endpoint_cutoff=annual_scale*1e-6  # 设定接近零端点的课堂数值阈值
interpretable_annual=annual.loc[annual.gt(annual_endpoint_cutoff)]  # 排除非正或接近零的CAGR端点
if len(interpretable_annual)<2: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 可解释端点不足时停止
start_year,end_year=int(interpretable_annual.index[0]),int(interpretable_annual.index[-1])  # 固定首尾稳定正值年度

start_profit,end_profit=interpretable_annual.iloc[0],interpretable_annual.iloc[-1]  # 取得CAGR首尾营业利润
cagr=(end_profit/start_profit)**(1/(end_year-start_year))-1  # 按完整年数计算复合增长率
manual_check=(1+cagr)**(end_year-start_year)*start_profit  # 反推末值进行另行核对
assert abs(manual_check-end_profit)<max(1e-6,abs(end_profit)*1e-10)  # 验证CAGR可还原末值
endpoint_cases=[('基准',start_year,end_year),('起点后移一年',start_year+1,end_year),('终点前移一年',start_year,end_year-1)]  # 定义基准与相邻端点情景
endpoint_rows=[]  # 初始化端点敏感性输出
for label,case_start,case_end in endpoint_cases:  # 逐个检查相邻年端点的可解释性
    endpoints_available=case_start in annual.index and case_end in annual.index and case_end>case_start  # 检查两个指定年度均存在且间隔为正
    endpoints_stable=endpoints_available and annual.loc[case_start]>annual_endpoint_cutoff and annual.loc[case_end]>annual_endpoint_cutoff  # 检查两端均为稳定正值
    case_cagr=(annual.loc[case_end]/annual.loc[case_start])**(1/(case_end-case_start))-1 if endpoints_stable else np.nan  # 端点稳定时重算CAGR
    endpoint_rows.append({'scenario':label,'start_year':case_start,'end_year':case_end,'status':'OK' if endpoints_stable else 'UNINTERPRETABLE','cagr':case_cagr})  # 保存情景结果与停止状态
endpoint_sensitivity=pd.DataFrame(endpoint_rows)  # 形成可执行的相邻端点敏感性表
print(version_audit,company[['end_date','operate_profit','prior_year_profit','profit_change','comparison_status','quarter_yoy']].tail(),{'start_year':start_year,'end_year':end_year,'cagr':cagr,'manual_check':manual_check},abnormal_denominators,endpoint_sensitivity)  # 输出版本、绝对变化、状态、同比、CAGR与敏感性依据