18  上市公司财务与行情综合分析

18.1 从投资问题到可审计证据

一家研究长三角制造业与科技企业的资产管理机构，希望回答三个彼此关联的问题：企业过去一年的盈利能力如何，资产负债表承担了多大杠杆，市场又为这些基本面支付了怎样的估值？这不是把几张表按季度名称拼接起来便能回答的问题。行情每天变化，利润表的季度字段通常是年初至报告期末的累计数，资产负债表是某一日的存量，而财务信息只有在披露后才进入投资者的信息集。若在 3 月 31 日直接使用同年一季报，即使公式完全正确，研究设计仍然含有未来信息。

本章把数据血缘、会计口径和信息时点视为分析的一部分。核心案例只使用本地真实数据，研究对象限定为上海、江苏、浙江和安徽的上市公司，并以恒瑞医药、上汽集团、海康威视和科大讯飞作为贯穿全章的审计锚点。完成本章后，读者应能够：

  • 目标 18.1：解释不复权、前复权和后复权价格各自回答的问题，并诊断停牌与缺失观测；

  • 目标 18.2：将利润表累计数还原为单季流量，再构造滚动十二个月指标；

  • 目标 18.3：以 info_date 为可得时点执行向后 merge_asof，并用断言证明没有前视偏差；

  • 目标 18.4：计算并解释 ROE、净利率、权益乘数、资产负债率、PE 与 PB，完成样本和行业层面的审计；

  • 目标 18.5：区分数据能够支持的结论与由于修订历史、行业异质性和样本选择而不能支持的结论。

目标—活动—核心练习—答案证据映射

表 18.1: 第十八章教学闭环映射
正式目标 学习活动或示例 可观察产出 核心练习 完整答案中的评分证据
目标 18.1 小节 18.2.2 与 小节 18.2.3 的三价格和共同交易日审计 输出复权收益差异、公司行动候选日与缺失率最高公司 小节 18.8.1 代数推导、真实价格断言及缺失/零成交审计表共同计分
目标 18.2 小节 18.3.2 与 小节 18.3.3 还原单季利润并构造 TTM 盈利和存量比率 小节 18.8.2 望远镜推导、四季完整条件和公司年度误差断言
目标 18.3 小节 18.4.2、小节 18.4.3 与 小节 18.4.4 生成安全面板并证明匹配时点不晚于决策日 小节 18.8.3 错误合并前视比例、安全合并未来记录数为零的断言
目标 18.4 小节 18.3.3、小节 18.5.1 与 小节 18.5.2 输出公司截面及行业中位数/市值加权对照 小节 18.8.4 ROE 两种汇总口径、差异排序和 PE/PB/杠杆解释框架
目标 18.5 小节 18.5.3 与 小节 18.6 将可描述结论与修订、异质性、选择偏差限制逐项配对 小节 18.8.4 答案末尾的三类“不能支持”陈述及所需补充证据逐项评分

表 表 18.1 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。

全章路线如下：小节 18.2 冻结样本并区分复权与缺失，小节 18.3 统一累计和单季口径，小节 18.4 建立信息集约束，小节 18.5 解释公司与行业结果，小节 18.6 界定证据边界，最后由 小节 18.8 完成分层检验。

18.2 数据合同与研究样本

本章使用六个 HDF5 文件。股票基本信息给出公司、地区和行业；三套价格分别保存前复权、后复权和不复权行情；财务报表提供会计指标及其 info_date；季度估值表提供 PE、PB 和市值。代码块 列表 18.1 首先建立跨平台路径，并把所有后续读取限定在本地数据根目录。

列表 18.1: 第十八章环境、路径与出版绘图设置 
import platform  # 识别操作系统，以便在同一教材代码中定位本地数据。
from pathlib import Path  # 使用路径对象避免手工拼接跨平台分隔符。
import matplotlib.pyplot as plt  # 生成基于真实计算结果的出版级图形。
import numpy as np  # 执行收益率、滚动窗口与数值断言。
import pandas as pd  # 读取 HDF5 并完成面板数据操作。

DATA_ROOT = Path('C:/qiufei/data' if platform.system() == 'Windows' else '/home/ubuntu/r2_data_mount/data')  # 按项目合同选择真实数据根目录。
STOCK_ROOT = DATA_ROOT / 'stock'  # 将本章所有输入限定在股票数据目录。
ANALYSIS_START = '2021-01-01'  # 留出四年以上行情以估计滚动市场特征。
ANALYSIS_END = '2025-12-31'  # 与本地行情和估值快照的覆盖终点一致。
plt.rcParams['font.family'] = ['Source Han Serif SC']  # 采用项目指定的中文出版字体。
plt.rcParams['axes.unicode_minus'] = False  # 保证负收益率刻度正确显示。

18.2.1 真实读取基本信息并冻结样本

样本不是从结局最好的公司中反向挑选。代码块 表 18.2 先限定地域、正常证券状态、2020 年以前上市以及四个具有可比经营含义的中信一级行业，再在每个行业按证券代码取前八家公司；四家审计锚点若未被抽中则补入。这一规则是确定性的，因而任何样本变化都能追溯到基本信息表。

basic_columns = ['order_book_id', 'symbol', 'province', 'status', 'special_type', 'listed_date', 'citics_2019_l1_name']  # 只读取筛样与解释所需字段。
stock_basic = pd.read_hdf(STOCK_ROOT / 'stock_basic_data.h5', columns=basic_columns)  # 从本地基础信息表读取真实公司属性。
stock_basic['listed_date'] = pd.to_datetime(stock_basic['listed_date'], errors='coerce')  # 将上市日标准化为可比较时间戳。
yrd_provinces = ['上海市', '江苏省', '浙江省', '安徽省']  # 按本书口径定义长三角四省市。
focus_industries = ['医药', '汽车', '电子', '计算机']  # 选择本地样本充足且经济含义不同的非金融行业。
is_eligible = stock_basic['province'].isin(yrd_provinces) & stock_basic['status'].eq('Active')  # 排除非长三角和已退市证券。
is_eligible &= stock_basic['special_type'].eq('Normal') & stock_basic['listed_date'].le('2020-01-01')  # 排除特别处理证券并保证足够历史。
is_eligible &= stock_basic['citics_2019_l1_name'].isin(focus_industries)  # 仅保留预先声明的行业。
eligible_companies = stock_basic.loc[is_eligible].sort_values(['citics_2019_l1_name', 'order_book_id'])  # 用稳定排序保证抽样可复现。
industry_sample = eligible_companies.groupby('citics_2019_l1_name', group_keys=False).head(8)  # 每个行业等额抽取八家公司。
anchor_codes = ['600276.XSHG', '600104.XSHG', '002415.XSHE', '002230.XSHE']  # 固定四省市的审计锚点。
anchor_companies = eligible_companies.loc[eligible_companies['order_book_id'].isin(anchor_codes)]  # 从合格总体中核验锚点身份。
sample_companies = pd.concat([industry_sample, anchor_companies]).drop_duplicates('order_book_id')  # 补入锚点且不重复公司。
sample_company_codes = sample_companies['order_book_id'].sort_values().tolist()  # 生成后续 HDF5 选择性读取的代码列表。
assert set(anchor_codes).issubset(sample_company_codes)  # 若地域、状态或行业口径变化则立即停止分析。
sample_companies.groupby(['province', 'citics_2019_l1_name']).size().rename('company_count').reset_index()  # 展示地域与行业覆盖。
表 18.2: 长三角真实公司样本的数据合同
province citics_2019_l1_name company_count
0 上海市 汽车 1
1 上海市 计算机 2
2 安徽省 医药 1
3 安徽省 汽车 2
4 安徽省 计算机 3
5 江苏省 医药 2
6 江苏省 汽车 2
7 江苏省 电子 5
8 江苏省 计算机 2
9 浙江省 医药 6
10 浙江省 汽车 4
11 浙江省 电子 4
12 浙江省 计算机 1

样本选择的含义

这是教学用的分层案例样本，不是可代表整个 A 股的随机样本。固定上市年限降低了新股历史不足造成的机械缺失，但也引入存续者选择。后文的行业中位数只能描述这组公司，不能直接解释为行业总体参数。

18.2.2 三种价格、复权与公司行动

设原始收盘价为 \(P_t\)，累计复权因子为 \(A_t\)。复权序列通常可写成比例等价的两种形式：

\[ P_t^{\mathrm{pre}} = P_t\frac{A_t}{A_T}, \qquad P_t^{\mathrm{post}} = P_t\frac{A_t}{A_0}. \tag{18.1}\]

只要两个因子的比例在同一份数据快照中是常数，前复权与后复权的理论区间收益相同，价格水平却不同。实际文件分别舍入价格小数位，计算收益后会留下约百万分之几的数值差。前复权把最新价格锚定在原始报价附近，适合展示近期价格；后复权把早期价格锚定，适合构造长期财富指数；不复权价格则保留投资者当时看到的报价，适合核对成交与除权除息缺口，但不能把除息日的机械下跌直接解释为经济损失。

代码块 列表 18.2 通过 HDF5 的 where 和 columns 只读取样本公司、分析区间和三个必要字段，而不是把三份一千多万行的表全部载入内存。

列表 18.2: 选择性读取三种真实日频行情 
def load_price_variant(file_name, variant_name):  # 封装一致的价格读取合同，避免三套行情处理口径漂移。
    '''读取样本证券的收盘价、成交量和成交额，并返回标准长表。'''  # 明确函数输出的业务字段与结构。
    price_path = STOCK_ROOT / file_name  # 将文件名解析到受控的本地股票目录。
    row_filters = [f'order_book_id in {sample_company_codes}', f'date >= {ANALYSIS_START!r}', f'date <= {ANALYSIS_END!r}']  # 在磁盘层过滤公司与日期。
    price_columns = ['close', 'volume', 'total_turnover']  # 只保留收益、停牌和流动性审计所需列。
    price_frame = pd.read_hdf(price_path, where=row_filters, columns=price_columns).reset_index()  # 从表格式 HDF5 选择性加载。
    price_frame['variant'] = variant_name  # 记录价格口径，防止后续透视时混淆来源。
    return price_frame  # 返回证券—交易日唯一的标准化长表。

pre_adjusted = load_price_variant('stock_price_pre_adjusted.h5', 'pre_adjusted')  # 读取前复权行情。
post_adjusted = load_price_variant('stock_price_post_adjusted.h5', 'post_adjusted')  # 读取后复权行情。
no_adjusted = load_price_variant('stock_price_no_adjust.h5', 'no_adjusted')  # 使用本地真实文件名读取不复权行情。
assert pre_adjusted[['order_book_id', 'date']].equals(post_adjusted[['order_book_id', 'date']])  # 核验三种口径共享同一证券日历。
assert pre_adjusted[['order_book_id', 'date']].equals(no_adjusted[['order_book_id', 'date']])  # 防止把覆盖差异误认为复权差异。

表 18.3 寻找恒瑞医药样本期内复权收益与原始报价收益分歧最大的交易日。复权与不复权收益的差异提示公司行动，而不是自动证明数据错误。

anchor_price_long = pd.concat([pre_adjusted, post_adjusted, no_adjusted], ignore_index=True)  # 合并三套真实行情以便同日比较。
anchor_price_long = anchor_price_long.loc[anchor_price_long['order_book_id'].eq('600276.XSHG')]  # 聚焦江苏恒瑞医药作逐日审计。
anchor_price_wide = anchor_price_long.pivot(index='date', columns='variant', values='close').sort_index()  # 将三种收盘价对齐到同一交易日。
anchor_returns = anchor_price_wide.pct_change(fill_method=None).add_suffix('_return')  # 禁止隐式填充，直接计算各口径收益。
adjustment_audit = anchor_price_wide.join(anchor_returns)  # 同表保存价格水平与收益率。
adjustment_audit['adjusted_return_gap'] = (adjustment_audit['pre_adjusted_return'] - adjustment_audit['post_adjusted_return']).abs()  # 核验两种复权收益的一致性。
adjustment_audit['raw_return_gap'] = (adjustment_audit['post_adjusted_return'] - adjustment_audit['no_adjusted_return']).abs()  # 定位原始报价受公司行动影响的日期。
assert adjustment_audit['adjusted_return_gap'].dropna().max() < 5e-6  # 允许两套价格各自舍入造成的百万分之几差异。
adjustment_audit.nlargest(5, 'raw_return_gap').reset_index()[['date', 'pre_adjusted_return', 'post_adjusted_return', 'no_adjusted_return', 'raw_return_gap']]  # 输出最值得回查的五日。
表 18.3: 恒瑞医药复权口径差异最大的五个交易日
variant date pre_adjusted_return post_adjusted_return no_adjusted_return raw_return_gap
0 2021-06-10 0.043477 0.043479 -0.132501 0.175980
1 2024-07-12 0.021525 0.021526 0.016502 0.005024
2 2022-07-05 -0.012194 -0.012194 -0.016031 0.003837
3 2025-05-23 -0.016249 -0.016250 -0.019784 0.003535
4 2023-06-08 0.005512 0.005511 0.001978 0.003533

易混淆概念辨析：复权不是补值

复权处理的是分红、送转等公司行动造成的价格不可比；停牌处理的是某个交易日没有成交记录。把停牌日价格向前填充会人为制造零收益，并改变波动率估计；调整历史价格也不会凭空创造停牌日观测。两者必须分别审计。

18.2.3 停牌、零成交与缺失观测

以样本中任一证券出现过交易的日期作为共同 A 股观测日历。表 18.4 把每家公司重新索引到这个日历，只做诊断而不填充价格。missing_quote_days 可能来自停牌、尚未上市或数据缺口；由于本章已限定 2020 年以前上市，样本期内的缺失更可能是停牌或数据问题，仍需结合公告才能最终定性。

market_trading_dates = pd.Index(post_adjusted['date'].drop_duplicates().sort_values(), name='date')  # 用样本联合交易日排除周末和法定休市日。
expected_panel_index = pd.MultiIndex.from_product([sample_company_codes, market_trading_dates], names=['order_book_id', 'date'])  # 构造公司与有效交易日的期望网格。
observed_post = post_adjusted.set_index(['order_book_id', 'date']).sort_index()  # 建立可与期望网格对齐的后复权面板。
price_grid = observed_post.reindex(expected_panel_index)  # 仅为审计显式暴露缺失，不进行价格插补。
missing_audit = price_grid.groupby(level='order_book_id').agg(observed_days=('close', 'count'), missing_quote_days=('close', lambda series: series.isna().sum()), zero_volume_days=('volume', lambda series: series.eq(0).sum()))  # 分证券统计三类覆盖指标。
missing_audit['missing_rate'] = missing_audit['missing_quote_days'] / len(market_trading_dates)  # 将缺失天数转成可跨公司比较的比例。
missing_audit = missing_audit.join(sample_companies.set_index('order_book_id')[['symbol', 'citics_2019_l1_name']])  # 补充公司简称与行业以支持人工复核。
missing_audit.sort_values(['missing_rate', 'zero_volume_days'], ascending=False).head(10).reset_index()  # 展示缺失最严重的十家公司。
表 18.4: 样本公司的停牌与行情缺失审计
order_book_id observed_days missing_quote_days zero_volume_days missing_rate symbol citics_2019_l1_name
0 000980.XSHE 1212 0 17 0.0 众泰汽车 汽车
1 000411.XSHE 1212 0 11 0.0 英特集团 医药
2 000153.XSHE 1212 0 10 0.0 丰原药业 医药
3 000559.XSHE 1212 0 10 0.0 万向钱潮 汽车
4 000727.XSHE 1212 0 1 0.0 冠捷科技 电子
5 000868.XSHE 1212 0 1 0.0 安凯客车 汽车
6 002077.XSHE 1212 0 1 0.0 大港股份 电子
7 000581.XSHE 1212 0 0 0.0 威孚高科 汽车
8 000700.XSHE 1212 0 0 0.0 模塑科技 汽车
9 000705.XSHE 1212 0 0 0.0 浙江震元 医药

后文计算持有期收益时使用相邻实际成交价，因此停牌期间的累计价格变化在复牌日一次体现；估计日波动率时则不把缺失日填成零收益。两个处理分别对应财富变化和价格发现强度，不能混用。

18.3 财务报表：从累计数到单季与滚动十二个月

18.3.1 选择性读取与修订记录

利润表中的 operating_revenue 与 net_profit_parent_company 是年初至季度末的累计流量；total_assets、total_liabilities 和 equity_parent_company 是报告期末存量。代码块 表 18.5 从表格式 HDF5 选择性读取这些字段，并保留 info_date 与 if_adjusted。若同一公司—报告期存在多条修订记录，只有 info_date 最新的一条被保留，但其较晚的披露时点不会被改写。

financial_columns = ['order_book_id', 'quarter', 'info_date', 'if_adjusted', 'operating_revenue', 'net_profit_parent_company', 'total_assets', 'total_liabilities', 'equity_parent_company']  # 限定会计指标与时点字段。
financial_filters = [f'order_book_id in {sample_company_codes}', '''quarter >= '2020q1' ''']  # 留出计算 TTM 和期初权益所需历史。
financial_raw = pd.read_hdf(STOCK_ROOT / 'financial_statement.h5', where=financial_filters, columns=financial_columns)  # 从本地表格式 HDF5 选择性读取。
financial_raw['report_period'] = pd.PeriodIndex(financial_raw['quarter'], freq='Q')  # 将报告期字符串转成有序季度。
financial_raw['info_date'] = pd.to_datetime(financial_raw['info_date'])  # 统一披露时间类型以比较修订版本与可得日。
duplicate_rows = financial_raw.duplicated(['order_book_id', 'report_period'], keep=False).sum()  # 统计潜在修订或重复记录。
financial = financial_raw.sort_values(['order_book_id', 'report_period', 'info_date'])  # 让每组最后一行对应最新可见修订。
financial = financial.drop_duplicates(['order_book_id', 'report_period'], keep='last').copy()  # 每个公司报告期只保留最新记录。
financial_contract_audit = pd.DataFrame({'raw_rows': [len(financial_raw)], 'duplicate_rows': [duplicate_rows], 'deduplicated_rows': [len(financial)], 'adjusted_rows': [financial['if_adjusted'].eq(1).sum()], 'min_info_date': [financial['info_date'].min()], 'max_info_date': [financial['info_date'].max()]})  # 汇总血缘与修订状态。
financial_contract_audit  # 输出财务数据合同审计表。
表 18.5: 财务报表读取、重复与修订状态审计
raw_rows duplicate_rows deduplicated_rows adjusted_rows min_info_date max_info_date
0 875 0 875 746 2021-04-17 2026-04-30

本地快照在案例样本中可能只保留每个报告期的最新调整值，而不保留每次历史版本。这意味着 info_date 很可能晚于法定首次披露日。严格使用这个日期会牺牲样本新鲜度，却不会把尚未在当前快照中可证明可得的数据提前使用；这是比猜测首次披露日更保守的选择。

18.3.2 单季流量的推导与核验

令 \(Y_{i,y,q}^{\mathrm{YTD}}\) 表示公司 \(i\) 在年份 \(y\)、第 \(q\) 季度披露的累计流量。单季流量为

\[ Y_{i,y,q}^{\mathrm{SQ}} = \begin{cases} Y_{i,y,1}^{\mathrm{YTD}}, & q=1, \\ Y_{i,y,q}^{\mathrm{YTD}}-Y_{i,y,q-1}^{\mathrm{YTD}}, & q\in\{2,3,4\}. \end{cases} \tag{18.2}\]

差分后的单季数只有在当前累计数和同一财年紧邻的上一季度累计数都可得时才可得。因此，它的可得日是两个来源 info_date 的较晚者，而不是机械使用当前报告期末。若 Q2、Q3 或 Q4 的前一季度缺失，代码必须保留缺口而不能跨季度作差；跨年时 Q1 重新以本期累计值起算。代码块 列表 18.3 用季度序号显式实现这一规则。

列表 18.3: 将累计利润表还原为时点安全的单季流量 
def add_single_quarter_flows(frame):  # 用严格相邻季度守卫还原累计流量。
    result = frame.sort_values(['order_book_id', 'report_period']).reset_index(drop=True).copy()  # 固定公司内报告期顺序。
    result['fiscal_year'] = result['report_period'].dt.year  # 提取当前财政年度。
    result['quarter_number'] = result['report_period'].dt.quarter  # 提取当前季度编号。
    result['quarter_serial'] = result['report_period'].astype('int64')  # 把季度转为可精确作差的连续序号。
    company_groups = result.groupby('order_book_id', sort=False)  # 所有滞后只允许在同一公司内发生。
    result['previous_report_period'] = company_groups['report_period'].shift(1)  # 记录物理上一行的报告期。
    result['previous_info_date'] = company_groups['info_date'].shift(1)  # 记录上一报告的可得日。
    result['quarter_gap'] = result['quarter_serial'].sub(company_groups['quarter_serial'].shift(1))  # 计算相邻观测相差几个季度。
    is_first_quarter = result['quarter_number'].eq(1)  # Q1 累计值本身就是单季值。
    same_year = result['previous_report_period'].dt.year.eq(result['fiscal_year'])  # 防止跨财年差分。
    previous_is_exact = result['previous_report_period'].dt.quarter.eq(result['quarter_number'].sub(1))  # 要求季度编号也紧邻。
    result['has_exact_previous_quarter'] = result['quarter_gap'].eq(1) & same_year & previous_is_exact  # 汇总三项连续性守卫。
    previous_revenue = company_groups['operating_revenue'].shift(1)  # 取得候选上一累计收入。
    previous_profit = company_groups['net_profit_parent_company'].shift(1)  # 取得候选上一累计利润。
    result['revenue_single_quarter'] = result['operating_revenue'].where(is_first_quarter)  # 先只填入无需差分的 Q1。
    result['profit_single_quarter'] = result['net_profit_parent_company'].where(is_first_quarter)  # 为 Q1 保留本期累计利润。
    valid_later = ~is_first_quarter & result['has_exact_previous_quarter']  # 仅允许同年紧邻季度作差。
    result.loc[valid_later, 'revenue_single_quarter'] = result.loc[valid_later, 'operating_revenue'].sub(previous_revenue.loc[valid_later])  # 还原有效后续季度收入。
    result.loc[valid_later, 'profit_single_quarter'] = result.loc[valid_later, 'net_profit_parent_company'].sub(previous_profit.loc[valid_later])  # 还原有效后续季度利润。
    result['single_info_date'] = pd.NaT  # 缺少精确前季时保持不可得。
    result.loc[is_first_quarter, 'single_info_date'] = result.loc[is_first_quarter, 'info_date']  # Q1 只依赖本期披露。
    result.loc[valid_later, 'single_info_date'] = result.loc[valid_later, ['info_date', 'previous_info_date']].max(axis=1)  # 后续季度等待两个累计数都可得。
    return result  # 返回带连续性证据的单季面板。

函数定义后立即对真实财务快照执行，并把被连续性守卫阻断的记录作为显式审计输出：

financial = add_single_quarter_flows(financial)  # 对真实财务快照执行严格单季还原。
later_quarters = financial['quarter_number'].ne(1)  # 标记理论上需要上一季度的观测。
quarter_gap_audit = financial.loc[later_quarters & ~financial['has_exact_previous_quarter'], ['order_book_id', 'report_period', 'previous_report_period', 'quarter_gap', 'info_date']].copy()  # 列出缺季而被阻断的记录。
quarter_gap_audit  # 输出缺口证据，不静默跨季计算。
order_book_id report_period previous_report_period quarter_gap info_date

一个必要的会计恒等式检查是：同一年度内，单季流量重新累加后应回到原累计数。表 18.6 以相对误差而不是绝对元数比较，避免大公司规模主导诊断。

year_groups = financial.groupby(['order_book_id', 'fiscal_year'], sort=False)  # 按公司和财年建立重构边界。
financial['reconstructed_revenue_ytd'] = year_groups['revenue_single_quarter'].transform(lambda series: series.cumsum(skipna=False))  # 缺季后保持缺失，禁止跳过断点继续累加。
financial['reconstructed_profit_ytd'] = year_groups['profit_single_quarter'].transform(lambda series: series.cumsum(skipna=False))  # 对利润使用相同的连续前缀规则。
revenue_scale = financial['operating_revenue'].abs().clip(lower=1)  # 用至少一元的尺度避免零分母。
profit_scale = financial['net_profit_parent_company'].abs().clip(lower=1)  # 对利润采用同样的稳定相对尺度。
financial['revenue_reconciliation_error'] = (financial['reconstructed_revenue_ytd'] - financial['operating_revenue']).abs() / revenue_scale  # 计算收入重构相对误差。
financial['profit_reconciliation_error'] = (financial['reconstructed_profit_ytd'] - financial['net_profit_parent_company']).abs() / profit_scale  # 计算利润重构相对误差。
reconciliation_audit = financial.groupby('order_book_id').agg(max_revenue_error=('revenue_reconciliation_error', 'max'), max_profit_error=('profit_reconciliation_error', 'max'), blocked_gap_rows=('has_exact_previous_quarter', lambda values: int((~values & later_quarters.loc[values.index]).sum()))).reset_index()  # 同时汇总可验证前缀误差和缺季阻断数。
assert reconciliation_audit[['max_revenue_error', 'max_profit_error']].max().max() < 1e-10  # 若累计口径理解错误则阻断后续指标。
reconciliation_audit.sort_values('max_profit_error', ascending=False).head()  # 输出数值误差最大的五家公司。
表 18.6: 单季流量重构累计数的数值核验
order_book_id max_revenue_error max_profit_error blocked_gap_rows
6 000727.XSHE 0.000000e+00 6.285000e-16 0
3 000581.XSHE 0.000000e+00 3.762291e-16 0
8 000868.XSHE 0.000000e+00 2.992729e-16 0
22 002230.XSHE 0.000000e+00 2.150783e-16 0
27 002331.XSHE 1.117787e-16 2.018224e-16 0

18.3.3 TTM 盈利能力与资产负债表比率

滚动十二个月归母利润只在最近四条记录恰好覆盖连续四个日历季度、且四个单季分量均可得时求和。为了避免用期末权益替代全年投入资本，本章以当前期末和恰好四个季度前的期末归母权益均值作为分母：

\[ \mathrm{ROE}_{i,q}^{\mathrm{TTM}} = \frac{\sum_{j=0}^{3} NP_{i,q-j}^{\mathrm{SQ}}} {\left(E_{i,q}+E_{i,q-4}\right)/2}. \tag{18.3}\]

同时定义净利率、权益乘数和资产负债率：

\[ \mathrm{Margin}_{i,q}^{\mathrm{TTM}}=\frac{NP_{i,q}^{\mathrm{TTM}}}{Revenue_{i,q}^{\mathrm{TTM}}},\qquad \mathrm{Leverage}_{i,q}=\frac{Assets_{i,q}}{Equity_{i,q}},\qquad \mathrm{DebtRatio}_{i,q}=\frac{Liabilities_{i,q}}{Assets_{i,q}}. \tag{18.4}\]

ROE 可以近似分解为净利率、资产周转率和权益乘数的乘积。高 ROE 可能源于强盈利能力，也可能源于高杠杆，因此只报告 ROE 而不报告杠杆会产生不完整的经济解释。

代码块 列表 18.4 将 式 18.3 与 式 18.4 落实为可得时点完整的真实指标。

列表 18.4: 构造带完整可得时点的 TTM 财务指标 
company_groups = financial.groupby('order_book_id', sort=False)  # 后续滚动只允许在同一公司内进行。
financial['quarter_lag_3_serial'] = company_groups['quarter_serial'].shift(3)  # 取得四条窗口中最早报告的季度序号。
financial['has_four_contiguous_quarters'] = financial['quarter_serial'].sub(financial['quarter_lag_3_serial']).eq(3)  # 四条唯一有序记录跨度为三才连续。
revenue_window_complete = company_groups['revenue_single_quarter'].transform(lambda series: series.notna().rolling(4, min_periods=4).sum().eq(4))  # 要求四个收入分量均可得。
profit_window_complete = company_groups['profit_single_quarter'].transform(lambda series: series.notna().rolling(4, min_periods=4).sum().eq(4))  # 要求四个利润分量均可得。
financial['has_valid_ttm_window'] = financial['has_four_contiguous_quarters'] & revenue_window_complete & profit_window_complete  # 合并时间和数值完整性守卫。
revenue_ttm_candidate = company_groups['revenue_single_quarter'].transform(lambda series: series.rolling(4, min_periods=4).sum())  # 先计算候选四季收入和。
profit_ttm_candidate = company_groups['profit_single_quarter'].transform(lambda series: series.rolling(4, min_periods=4).sum())  # 先计算候选四季利润和。
financial['revenue_ttm'] = revenue_ttm_candidate.where(financial['has_valid_ttm_window'])  # 只发布精确连续窗口的收入 TTM。
financial['profit_ttm'] = profit_ttm_candidate.where(financial['has_valid_ttm_window'])  # 只发布精确连续窗口的利润 TTM。
financial['quarter_lag_4_serial'] = company_groups['quarter_serial'].shift(4)  # 取得候选同比季度的连续序号。
financial['has_exact_year_lag'] = financial['quarter_serial'].sub(financial['quarter_lag_4_serial']).eq(4)  # 要求滞后项恰好相隔四个季度。
financial['equity_lag_4'] = company_groups['equity_parent_company'].shift(4).where(financial['has_exact_year_lag'])  # 仅保留精确同比期权益。
financial['equity_average'] = (financial['equity_parent_company'] + financial['equity_lag_4']) / 2  # 近似全年平均归母权益。
financial['single_info_ns'] = financial['single_info_date'].astype('int64')  # 将时间戳转为整数以执行滚动最大值。
financial['ttm_info_ns'] = company_groups['single_info_ns'].transform(lambda series: series.rolling(4, min_periods=4).max())  # 找到四个单季分量中最晚可得的日期。
financial['ttm_info_date'] = pd.to_datetime(financial['ttm_info_ns']).where(financial['has_valid_ttm_window'])  # 仅为有效连续窗口恢复可得时点。
financial['equity_lag_info_date'] = company_groups['info_date'].shift(4).where(financial['has_exact_year_lag'])  # 只接受精确同比期披露日。
financial['metric_info_date'] = financial[['ttm_info_date', 'info_date', 'equity_lag_info_date']].max(axis=1)  # 取全部输入都已可得的最晚日期。
financial['roe_ttm'] = financial['profit_ttm'] / financial['equity_average']  # 按 @eq-ch18-roe 计算滚动 ROE。
financial['net_margin_ttm'] = financial['profit_ttm'] / financial['revenue_ttm']  # 计算归母口径净利率。
financial['equity_multiplier'] = financial['total_assets'] / financial['equity_parent_company']  # 计算每单位归母权益支撑的资产规模。
financial['debt_to_assets'] = financial['total_liabilities'] / financial['total_assets']  # 计算债权融资占资产的比例。
financial_features = financial.dropna(subset=['metric_info_date', 'roe_ttm', 'net_margin_ttm']).copy()  # 只保留完成四季度窗口的可用指标。

表 18.7 使用一个含缺季、重复修订、跨年 Q1 和完整五季度路径的确定性夹具，核验上述守卫。它只验证算法机制，不替代真实样本的 表 18.5 与 quarter_gap_audit。

test_columns = ['order_book_id', 'quarter', 'info_date', 'operating_revenue', 'net_profit_parent_company', 'equity_parent_company']  # 声明最小测试合同。
test_rows = [('GAP', '2021q1', '2021-04-30', 100, 10, 100), ('GAP', '2021q2', '2021-08-01', 250, 25, 110), ('GAP', '2021q2', '2021-09-01', 260, 26, 111), ('GAP', '2021q4', '2022-03-31', 500, 50, 120), ('GAP', '2022q1', '2022-04-30', 120, 12, 125), ('FULL', '2021q1', '2021-04-30', 10, 1, 100), ('FULL', '2021q2', '2021-08-31', 30, 3, 110), ('FULL', '2021q3', '2021-10-31', 60, 6, 120), ('FULL', '2021q4', '2022-03-31', 100, 10, 130), ('FULL', '2022q1', '2022-04-30', 12, 2, 140)]  # 同时覆盖五类边界。
quarter_test_raw = pd.DataFrame(test_rows, columns=test_columns)  # 构造不依赖随机数的机制夹具。
quarter_test_raw['info_date'] = pd.to_datetime(quarter_test_raw['info_date'])  # 解析测试披露日。
quarter_test_raw['report_period'] = pd.PeriodIndex(quarter_test_raw['quarter'], freq='Q')  # 解析测试报告期。
quarter_test = quarter_test_raw.sort_values(['order_book_id', 'report_period', 'info_date']).drop_duplicates(['order_book_id', 'report_period'], keep='last')  # 模拟主流程的最新修订选择。
quarter_test = add_single_quarter_flows(quarter_test)  # 对夹具执行单季连续性守卫。
gap_q2 = quarter_test.loc[(quarter_test['order_book_id'].eq('GAP')) & (quarter_test['report_period'].eq(pd.Period('2021Q2')))].iloc[0]  # 提取修订后的 Q2。
gap_q4 = quarter_test.loc[(quarter_test['order_book_id'].eq('GAP')) & (quarter_test['report_period'].eq(pd.Period('2021Q4')))].iloc[0]  # 提取缺少 Q3 的 Q4。
gap_next_q1 = quarter_test.loc[(quarter_test['order_book_id'].eq('GAP')) & (quarter_test['report_period'].eq(pd.Period('2022Q1')))].iloc[0]  # 提取跨年重置的 Q1。
assert gap_q2['operating_revenue'] == 260 and gap_q2['revenue_single_quarter'] == 160  # 核验重复季度保留晚披露修订并正确差分。
assert gap_q2['single_info_date'] == pd.Timestamp('2021-09-01')  # 核验单季值等待两个累计输入中的较晚披露日。
assert pd.isna(gap_q4['profit_single_quarter']) and gap_q4['quarter_gap'] == 2  # 核验缺少中间季度时阻断差分。
assert gap_next_q1['profit_single_quarter'] == 12  # 核验跨年 Q1 不与上一年 Q4 作差。
test_groups = quarter_test.groupby('order_book_id', sort=False)  # 为 TTM 与同比滞后建立测试分组。
quarter_test['lag_3'] = test_groups['quarter_serial'].shift(3)  # 取得四条窗口起点。
quarter_test['valid_ttm'] = quarter_test['quarter_serial'].sub(quarter_test['lag_3']).eq(3) & test_groups['profit_single_quarter'].transform(lambda series: series.notna().rolling(4, min_periods=4).sum().eq(4))  # 复算 TTM 连续性合同。
quarter_test['lag_4'] = test_groups['quarter_serial'].shift(4)  # 取得候选同比季度。
quarter_test['valid_year_lag'] = quarter_test['quarter_serial'].sub(quarter_test['lag_4']).eq(4)  # 复算精确四季度滞后合同。
full_q4 = quarter_test.loc[(quarter_test['order_book_id'].eq('FULL')) & (quarter_test['report_period'].eq(pd.Period('2021Q4')))].iloc[0]  # 提取首个完整四季度终点。
full_next_q1 = quarter_test.loc[(quarter_test['order_book_id'].eq('FULL')) & (quarter_test['report_period'].eq(pd.Period('2022Q1')))].iloc[0]  # 提取精确同比终点。
assert full_q4['valid_ttm'] and not gap_q4.get('valid_ttm', False)  # 核验连续四季通过而缺季窗口失败。
assert full_next_q1['valid_year_lag']  # 核验同比滞后只在序号差恰为四时通过。
pd.DataFrame({'duplicate_revision_kept': [gap_q2['operating_revenue'] == 260], 'missing_quarter_blocked': [pd.isna(gap_q4['profit_single_quarter'])], 'cross_year_q1_reset': [gap_next_q1['profit_single_quarter'] == 12], 'availability_date_safe': [gap_q2['single_info_date'] == pd.Timestamp('2021-09-01')], 'contiguous_ttm_and_lag': [bool(full_q4['valid_ttm'] and full_next_q1['valid_year_lag'])]})  # 输出五项测试结果。
表 18.7: 季度连续性、修订与可得日的确定性测试
duplicate_revision_kept missing_quarter_blocked cross_year_q1_reset availability_date_safe contiguous_ttm_and_lag
0 True True True True True

18.4 估值与 point-in-time 合并

18.4.1 真实估值子集与文件格式边界

本地估值文件采用 Pandas fixed 格式，不能像 table 格式那样在磁盘层使用 where 或 columns。代码块 表 18.8 先核验原生 key 和格式，再一次读取约 66 MB 的固定表，立即缩减为样本证券、分析日期以及 PE、PB、市值三个字段。这里的全表读取是存储格式造成的可审计限制，不是默认做法。

valuation_path = STOCK_ROOT / 'valuation_factors_quarterly_15_years.h5'  # 定位本地季度估值文件。
with pd.HDFStore(valuation_path, mode='r') as valuation_store:  # 只读打开文件以检查原生存储合同。
    valuation_keys = valuation_store.keys()  # 获取真实 HDF5 key，而不猜测默认节点。
    valuation_format = valuation_store.get_storer(valuation_keys[0]).format_type  # 核验该节点是否支持选择性读取。
valuation_all = pd.read_hdf(valuation_path)  # fixed 格式要求整表读取，不能传入 columns 或 where。
valuation_columns = ['pe_ratio_ttm', 'pb_ratio_lf', 'market_cap']  # 仅保留估值解释所需字段。
valuation = valuation_all.loc[:, valuation_columns].reset_index()  # 在读取后立即缩减列宽并暴露多重索引。
is_sample_valuation = valuation['order_book_id'].isin(sample_company_codes)  # 限定预先冻结的长三角样本。
is_analysis_date = valuation['date'].between('2023-01-01', ANALYSIS_END)  # 限定可与财务和行情匹配的估值期。
valuation = valuation.loc[is_sample_valuation & is_analysis_date].rename(columns={'date': 'feature_as_of_date'}).copy()  # 形成公司—估值时点面板。
valuation_contract = pd.DataFrame({'native_keys': [', '.join(valuation_keys)], 'format_type': [valuation_format], 'file_rows': [len(valuation_all)], 'selected_rows': [len(valuation)], 'selected_companies': [valuation['order_book_id'].nunique()]})  # 记录从原始文件到分析子集的收缩。
del valuation_all  # 释放不再需要的固定格式全表，避免后续内存占用。
valuation_contract  # 输出格式和选择审计结果。
表 18.8: 季度估值文件的原生格式与内存子集审计
native_keys format_type file_rows selected_rows selected_companies
0 /valuation_factors fixed 207600 280 35

PE 与 PB 的经济定义分别由 式 18.5 给出：

\[ \mathrm{PE}_{i,t}=\frac{MarketValue_{i,t}}{Earnings_{i,t}^{\mathrm{TTM}}},\qquad \mathrm{PB}_{i,t}=\frac{MarketValue_{i,t}}{BookEquity_{i,t}^{\mathrm{LF}}}. \tag{18.5}\]

若盈利为负，常规 PE 失去“回收年数”的直觉；若账面权益很小或为负，PB 也会失真。即使两者均为正，较高估值既可能反映增长预期，也可能反映低风险、高无形资产或市场过度乐观，不能从比率本身识别因果。

18.4.2 用披露日向后合并

对估值日 \(t\)，允许使用的财务记录集合是

\[ \mathcal{I}_{i,t}=\{x_{i,q}:\mathrm{metric\_info\_date}_{i,q}\le t\}. \tag{18.6}\]

式 18.6 中的 point-in-time 特征取集合里可得日最近的一条。代码块 列表 18.5 用 merge_asof(direction='backward') 实现这个选择，而普通的同季度 merge 不带不等式约束，无法保证信息时点安全。

列表 18.5: 按 metric_info_date 执行向后 point-in-time 合并 
feature_columns = ['order_book_id', 'report_period', 'metric_info_date', 'roe_ttm', 'net_margin_ttm', 'equity_multiplier', 'debt_to_assets']  # 限定进入投资信息集的财务特征。
available_financial = financial_features.loc[:, feature_columns].sort_values(['metric_info_date', 'order_book_id'])  # 按 as-of 右键全局排序。
valuation_observations = valuation.sort_values(['feature_as_of_date', 'order_book_id'])  # 按 as-of 左键全局排序。
pit_panel = pd.merge_asof(valuation_observations, available_financial, by='order_book_id', left_on='feature_as_of_date', right_on='metric_info_date', direction='backward', allow_exact_matches=True)  # 只匹配估值时点已公开的最近指标。
pit_panel['financial_lag_days'] = (pit_panel['feature_as_of_date'] - pit_panel['metric_info_date']).dt.days  # 量化财务信息陈旧程度。
has_financial_match = pit_panel['metric_info_date'].notna()  # 区分合法未匹配与成功匹配观测。
assert (pit_panel.loc[has_financial_match, 'metric_info_date'] <= pit_panel.loc[has_financial_match, 'feature_as_of_date']).all()  # 以可执行断言排除财务前视偏差。

18.4.3 行情特征也只能向后取值

季度估值日可能落在周末或节假日。代码块 列表 18.6 先在真实交易记录上计算 63 个观测日的动量与年化波动率，再把估值日向后匹配到最近交易日。停牌日不被填成零收益，price_date 也必须不晚于估值日。

列表 18.6: 构造并向后匹配后复权市场特征 
market_features = post_adjusted.sort_values(['order_book_id', 'date']).copy()  # 以公司内时间顺序准备后复权行情。
market_groups = market_features.groupby('order_book_id', sort=False)  # 确保收益窗口不会跨公司。
market_features['daily_log_return'] = market_groups['close'].transform(lambda series: np.log(series / series.shift(1)))  # 相邻实际成交日计算对数收益。
market_features['momentum_63d'] = market_groups['close'].transform(lambda series: series.pct_change(63, fill_method=None))  # 计算约一季度的持有期动量。
market_features['volatility_63d'] = market_groups['daily_log_return'].transform(lambda series: series.rolling(63, min_periods=50).std() * np.sqrt(252))  # 用不少于五十个收益观测估计年化波动率。
market_columns = ['order_book_id', 'date', 'momentum_63d', 'volatility_63d']  # 限定行情匹配所需字段。
available_market = market_features.loc[:, market_columns].rename(columns={'date': 'price_date'}).sort_values(['price_date', 'order_book_id'])  # 将交易日显式命名为价格可得日。
pit_panel = pd.merge_asof(pit_panel.sort_values(['feature_as_of_date', 'order_book_id']), available_market, by='order_book_id', left_on='feature_as_of_date', right_on='price_date', direction='backward', allow_exact_matches=True)  # 周末估值日向后匹配最近交易日。
has_price_match = pit_panel['price_date'].notna()  # 标记成功取得历史行情的观测。
assert (pit_panel.loc[has_price_match, 'price_date'] <= pit_panel.loc[has_price_match, 'feature_as_of_date']).all()  # 排除行情时点前视偏差。

18.4.4 前视偏差的显式反证

表 18.9 同时报告安全合并的覆盖率、财务信息延迟和一个故意构造的错误基准。错误基准把估值日所属自然季度与财务报告季度相等视为可用；只要该报告的 info_date 晚于估值日，就构成前视观测。

valuation_quarters = valuation.assign(report_period=valuation['feature_as_of_date'].dt.to_period('Q'))  # 构造仅供反证的自然季度键。
unsafe_columns = ['order_book_id', 'report_period', 'info_date']  # 只取识别错误时点所需字段。
unsafe_panel = valuation_quarters.merge(financial.loc[:, unsafe_columns], on=['order_book_id', 'report_period'], how='left')  # 故意执行不带披露约束的同季度合并。
unsafe_future_rows = (unsafe_panel['info_date'] > unsafe_panel['feature_as_of_date']).sum()  # 统计错误方法提前使用报告的次数。
pit_audit = pd.DataFrame({'valuation_rows': [len(pit_panel)], 'financial_matches': [has_financial_match.sum()], 'financial_missing': [(~has_financial_match).sum()], 'median_financial_lag_days': [pit_panel.loc[has_financial_match, 'financial_lag_days'].median()], 'maximum_financial_lag_days': [pit_panel.loc[has_financial_match, 'financial_lag_days'].max()], 'unsafe_same_quarter_future_rows': [unsafe_future_rows]})  # 汇总安全与错误设计的差异。
assert unsafe_future_rows > 0  # 证明本地数据上普通同季度合并确实会产生前视偏差。
pit_audit  # 输出可复核的时点安全证据。
表 18.9: point-in-time 覆盖、延迟与错误同季度合并审计
valuation_rows financial_matches financial_missing median_financial_lag_days maximum_financial_lag_days unsafe_same_quarter_future_rows
0 280 252 28 165.0 615.0 280

易混淆概念辨析：报告期、披露日与特征日

report_period 回答“这组会计数字描述哪一段经营活动”，metric_info_date 回答“支持该指标的全部数字最早何时在本快照中可得”，feature_as_of_date 回答“投资决策站在哪一天”。向后合并比较的是后两者；报告期只用于计算累计差分和滚动窗口。把三者都称为“季度”是前视偏差最常见的语言根源。

18.5 截面结果、行业图形与样本审计

18.5.1 最新可审计截面

代码块 表 18.10 从每家公司最后一个估值观测构造截面，并展示四家锚点的估值、盈利、杠杆、行情和信息延迟。PE 采用本地估值文件的 TTM 口径，PB 采用最近账面值口径；财务 ROE 则完全由前述时点安全流程计算，因此二者的会计分母未必完全一致。

company_attributes = sample_companies[['order_book_id', 'symbol', 'province', 'citics_2019_l1_name']]  # 提取解释截面所需公司属性。
pit_panel = pit_panel.merge(company_attributes, on='order_book_id', how='left', validate='many_to_one')  # 以多对一约束关联真实地域和行业。
latest_cross_section = pit_panel.sort_values('feature_as_of_date').groupby('order_book_id', as_index=False).tail(1).copy()  # 每家公司只保留最后一个估值时点。
latest_cross_section['implied_roe_from_valuation'] = latest_cross_section['pb_ratio_lf'] / latest_cross_section['pe_ratio_ttm']  # 在正 PE/PB 条件下计算估值隐含的账面回报比。
anchor_columns = ['order_book_id', 'symbol', 'province', 'feature_as_of_date', 'report_period', 'metric_info_date', 'financial_lag_days', 'roe_ttm', 'net_margin_ttm', 'equity_multiplier', 'pe_ratio_ttm', 'pb_ratio_lf', 'momentum_63d', 'volatility_63d']  # 建立逐字段审计清单。
latest_cross_section.loc[latest_cross_section['order_book_id'].isin(anchor_codes), anchor_columns].sort_values('order_book_id')  # 输出四家锚点公司的完整证据链。
表 18.10: 四家长三角锚点公司的最新时点安全审计样本
order_book_id symbol province feature_as_of_date report_period metric_info_date financial_lag_days roe_ttm net_margin_ttm equity_multiplier pe_ratio_ttm pb_ratio_lf momentum_63d volatility_63d
267 002230.XSHE 科大讯飞 安徽省 2025-12-31 2023Q3 2025-04-22 253.0 0.014898 0.012802 2.156697 138.872426 6.532254 -0.104682 0.329885
276 002415.XSHE 海康威视 浙江省 2025-12-31 2023Q3 2025-04-19 256.0 0.187811 0.151655 1.782304 20.735910 3.488526 -0.033500 0.251225
278 600104.XSHG 上汽集团 上海市 2025-12-31 2023Q3 2025-04-30 245.0 0.053247 0.020697 3.331030 61.178140 0.592028 -0.113053 0.189347
279 600276.XSHG 恒瑞医药 江苏省 2025-12-31 2023Q3 2025-03-31 275.0 0.110139 0.188276 1.081603 52.942101 6.644534 -0.175730 0.281241

若 PE 与 PB 都为正且定义完全相容，则 \(PB/PE\) 与基于市场口径利润和账面权益的 ROE 接近；本章计算的 implied_roe_from_valuation 只用于一致性诊断。它与时点安全 ROE 的差异可能来自估值供应商口径、财务修订、少数股东权益处理或信息时点不同，不能把差异自动判为错误。

18.5.2 行业中位数而非“漂亮的平均数”

PE、PB 和 ROE 都可能有极端值。表 18.11 与 图 18.1 使用行业中位数，并同时报告总公司数和财务指标可用数。中位数降低单个极端公司的影响，却不会修复样本选择或行业内商业模式差异。

valid_cross_section = latest_cross_section.replace([np.inf, -np.inf], np.nan)  # 将无经济意义的无穷比率转为显式缺失。
valid_cross_section['positive_pe'] = valid_cross_section['pe_ratio_ttm'].where(valid_cross_section['pe_ratio_ttm'] > 0)  # 负盈利公司的 PE 不参与回收年数比较。
industry_summary = valid_cross_section.groupby('citics_2019_l1_name').agg(company_count=('order_book_id', 'nunique'), usable_roe_count=('roe_ttm', 'count'), median_roe=('roe_ttm', 'median'), median_net_margin=('net_margin_ttm', 'median'), median_debt_ratio=('debt_to_assets', 'median'), median_pe=('positive_pe', 'median'), median_pb=('pb_ratio_lf', 'median'), median_volatility=('volatility_63d', 'median')).reset_index()  # 以稳健中位数汇总多维指标。
industry_summary.sort_values('median_roe', ascending=False)  # 输出按盈利能力排序的真实行业截面。
表 18.11: 长三角样本行业的盈利、杠杆与估值中位数
citics_2019_l1_name company_count usable_roe_count median_roe median_net_margin median_debt_ratio median_pe median_pb median_volatility
0 医药 9 9 0.110139 0.040401 0.373100 25.603816 1.647368 0.254361
2 电子 9 9 0.095636 0.069989 0.397518 59.873240 3.644480 0.391484
1 汽车 9 9 0.090782 0.046782 0.520156 42.570439 2.276811 0.361958
3 计算机 8 8 0.018717 0.031846 0.413390 79.954721 3.033681 0.336058 
plot_metrics = [('median_roe', 'TTM ROE', 100), ('median_net_margin', 'TTM 净利率', 100), ('median_debt_ratio', '资产负债率', 100), ('median_pe', '正值 PE', 1)]  # 定义四个需要并列比较的经济指标。
plot_colors = ['#E3120B', '#008080', '#F0A700', '#2C3E50']  # 使用全书统一的高对比学术配色。
figure, axes = plt.subplots(2, 2, figsize=(11, 8))  # 创建四面板行业仪表板。
for axis, metric_specification, bar_color in zip(axes.flat, plot_metrics, plot_colors):  # 逐面板绑定指标及颜色。
    metric_name, axis_title, display_scale = metric_specification  # 解包列名、标题和百分比缩放。
    ordered_summary = industry_summary.sort_values(metric_name)  # 按当前指标排序以便比较。
    axis.barh(ordered_summary['citics_2019_l1_name'], ordered_summary[metric_name] * display_scale, color=bar_color)  # 绘制真实行业中位数。
    axis.set_title(axis_title)  # 标明每个面板的经济含义。
    axis.set_xlabel('%' if display_scale == 100 else '倍')  # 区分比率百分数与估值倍数。
    axis.grid(axis='x', alpha=0.2)  # 添加克制的辅助网格以支持数值比较。
figure.suptitle('最新可审计截面：盈利、杠杆与估值并读')  # 给出跨面板共同主题。
figure.tight_layout()  # 防止中文标签和面板互相遮挡。
plt.show()  # 显示由真实数据计算得到的图形。
图 18.1: 长三角样本四行业的盈利、杠杆与估值中位数

18.5.3 结果解释的顺序

阅读 图 18.1 时应遵循“盈利—杠杆—估值”的顺序。首先比较 ROE 与净利率，判断回报是否来自经营盈利；其次检查资产负债率和权益乘数，识别杠杆放大；最后观察 PE 与 PB，判断市场价格是否已经反映较高盈利。如果一个行业同时具有高 ROE、高杠杆和高 PB，不能仅凭图形断言市场高估，因为增长预期、资产轻重和风险暴露仍未被控制。

18.6 可复现性、审计清单与局限

这个端到端案例建立了以下可复核链条：文件 key 和存储格式被检查；大表在磁盘层选择性读取；累计利润表可重构；复权收益一致；停牌没有被静默填零；财务和行情均向后合并；两条时点断言必须通过；错误同季度合并在真实数据上被证明会前视。

仍需明确四项局限：

  • 财务快照可能只保留最新调整值，不能重建每一次历史修订；使用较晚 info_date 是保守处理，但会使财务特征显得陈旧；

  • 行业样本是确定性分层样本，不包含退市公司和 2020 年后上市公司，存在存续者与上市年限选择；

  • 季度估值文件是固定格式，读取时不能下推行列过滤；若扩展到更大文件，应重新存储为 table 或 Parquet 分区数据集；

  • 描述性截面不能识别因果。若研究“盈利能力是否导致估值上升”，还需控制增长、风险、公司和时间效应，并按时间划分训练与检验样本。

18.7 本章小结

财务—行情联合分析的难点不是 merge 的语法，而是定义每个数字的经济口径与最早可得时点。不复权价格用于核对真实报价，复权价格用于可比收益；累计利润表必须先还原单季流量，才能构造 TTM 指标；存量分母应与流量窗口相容；PE 与 PB 需要结合盈利、杠杆和会计口径解释；最终合并必须满足 metric_info_date <= feature_as_of_date。这些约束把一张“看起来完整”的表变成可审计的研究样本。

18.8 分层练习与完整解答

18.8.1 基础：复权收益为何一致

练习 18.1

由 式 18.1 推导前复权与后复权的单期收益相等，并用恒瑞医药真实行情验证；同时报告不复权收益与后复权收益差异最大的日期。再从共同交易日审计中找出缺失率最高的公司，分别报告缺失报价日与零成交日，并说明为何仅凭这两个计数不能断言每一天都是停牌。

完整解答

因为 \(P_t^{\mathrm{post}}=(A_T/A_0)P_t^{\mathrm{pre}}\)，两序列只相差常数 \(c=A_T/A_0\)，所以

\[ \frac{P_t^{\mathrm{post}}}{P_{t-1}^{\mathrm{post}}}-1 =\frac{cP_t^{\mathrm{pre}}}{cP_{t-1}^{\mathrm{pre}}}-1 =\frac{P_t^{\mathrm{pre}}}{P_{t-1}^{\mathrm{pre}}}-1. \tag{18.7}\]

exercise_adjustment = adjustment_audit.dropna(subset=['pre_adjusted_return', 'post_adjusted_return', 'no_adjusted_return']).copy()  # 只在三种收益都可观察的日期比较。
maximum_adjusted_difference = exercise_adjustment['adjusted_return_gap'].max()  # 计算前后复权收益的最大数值误差。
largest_raw_difference = exercise_adjustment.nlargest(1, 'raw_return_gap').reset_index()  # 找到公司行动影响最明显的真实交易日。
assert maximum_adjusted_difference < 5e-6  # 在真实文件的价格舍入精度内验证代数结论。
pd.DataFrame({'maximum_pre_post_return_difference': [maximum_adjusted_difference], 'largest_raw_gap_date': [largest_raw_difference.loc[0, 'date']], 'largest_raw_return_gap': [largest_raw_difference.loc[0, 'raw_return_gap']]})  # 输出关键结果而非只给提示。
表 18.12: 练习 18.1：复权收益恒等式的真实数据验证
maximum_pre_post_return_difference largest_raw_gap_date largest_raw_return_gap
0 0.000003 2021-06-10 0.17598

式 18.7 的代数结论和 表 18.12 的断言共同说明，两套复权数据在这份本地快照中只存在价格舍入量级的收益差异；不复权差异最大的日期应回查分红送转公告，不能仅凭价格跳空判断经营冲击。

表 表 18.13 汇总本节的计算或审计结果，解释时应遵循正文给出的口径与限制。

exercise_missing_audit = missing_audit.reset_index().sort_values(['missing_rate', 'zero_volume_days'], ascending=False)  # 按缺失率和零成交日排列真实公司审计结果。
highest_missing_company = exercise_missing_audit.head(1)  # 取得题目要求的最高缺失率公司及其证券代码。
assert highest_missing_company[['missing_quote_days', 'zero_volume_days']].ge(0).all().all()  # 核验两项公司日计数都满足非负业务约束。
highest_missing_company[['order_book_id', 'symbol', 'missing_quote_days', 'zero_volume_days', 'missing_rate']]  # 输出可评分的缺失与零成交证据。
表 18.13: 练习 18.1：缺失报价与零成交的公司级审计
order_book_id symbol missing_quote_days zero_volume_days missing_rate
13 000980.XSHE 众泰汽车 0 17 0.0

missing_quote_days 表示共同样本交易日历上没有该公司报价的日期，可能来自停牌、数据缺口或证券状态变化；zero_volume_days 则是存在行情行但成交量为零的记录。两者的数据结构不同，且都没有包含公告原因，所以答案只能标出待核查日期，不能把所有缺失或零成交机械判定为停牌。

18.8.2 进阶：累计数还原的会计边界

练习 18.2

证明一年四个单季流量之和等于四季度累计流量，并检查样本中是否存在不满足该恒等式的公司年度。

完整解答

将 式 18.2 四项相加，所有中间累计项两两抵消：

\[ Y_1^{\mathrm{YTD}}+(Y_2^{\mathrm{YTD}}-Y_1^{\mathrm{YTD}}) +(Y_3^{\mathrm{YTD}}-Y_2^{\mathrm{YTD}}) +(Y_4^{\mathrm{YTD}}-Y_3^{\mathrm{YTD}})=Y_4^{\mathrm{YTD}}. \tag{18.8}\]

annual_reconciliation = financial.groupby(['order_book_id', 'fiscal_year']).agg(quarter_signature=('quarter_number', lambda values: tuple(sorted(values.unique()))), single_quarters_complete=('profit_single_quarter', lambda values: values.notna().all()), summed_single_profit=('profit_single_quarter', 'sum'), fourth_quarter_ytd=('net_profit_parent_company', lambda series: series.iloc[-1])).reset_index()  # 汇总季度集合、单季完整性与年末累计利润。
complete_years = annual_reconciliation['quarter_signature'].map(lambda signature: signature == (1, 2, 3, 4)) & annual_reconciliation['single_quarters_complete']  # 逐组比较季度元组，只接受精确 Q1—Q4 且四个单季均可得的年度。
annual_reconciliation['absolute_error'] = (annual_reconciliation['summed_single_profit'] - annual_reconciliation['fourth_quarter_ytd']).abs()  # 计算望远镜求和的数值误差。
annual_reconciliation['relative_error'] = annual_reconciliation['absolute_error'] / annual_reconciliation['fourth_quarter_ytd'].abs().clip(lower=1)  # 形成跨规模可比误差。
assert annual_reconciliation.loc[complete_years, 'relative_error'].max() < 1e-10  # 完整公司年度必须通过累计口径核验。
annual_reconciliation.loc[complete_years].nlargest(5, 'relative_error')  # 输出误差最大的五个完整公司年度。
表 18.14: 练习 18.2：公司年度累计流量的望远镜恒等式
order_book_id fiscal_year quarter_signature single_quarters_complete summed_single_profit fourth_quarter_ytd absolute_error relative_error
23 000581.XSHE 2022 (1, 2, 3, 4) True 1.188198e+08 1.188198e+08 4.470348e-08 3.762291e-16
56 000868.XSHE 2020 (1, 2, 3, 4) True 9.958245e+07 9.958245e+07 2.980232e-08 2.992729e-16
158 002230.XSHE 2024 (1, 2, 3, 4) True 5.601627e+08 5.601627e+08 1.192093e-07 2.128119e-16
193 002331.XSHE 2024 (1, 2, 3, 4) True 3.691652e+07 3.691652e+07 7.450581e-09 2.018224e-16
143 002156.XSHE 2023 (1, 2, 3, 4) True 1.694385e+08 1.694385e+08 2.980232e-08 1.758887e-16

式 18.8 给出望远镜消项，表 18.14 则在每个完整公司年度执行同一检查。若断言失败，应先检查缺季、财政年度口径和报表修订，不应继续计算 TTM 盈利。

18.8.3 应用：量化错误合并造成的前视比例

练习 18.3

计算错误同季度合并中前视记录的比例，并验证安全面板的所有财务匹配都满足信息不等式。

完整解答

unsafe_matched = unsafe_panel['info_date'].notna()  # 识别错误方法实际取得财务记录的估值观测。
unsafe_is_future = unsafe_panel['info_date'] > unsafe_panel['feature_as_of_date']  # 标记披露日晚于决策日的非法匹配。
unsafe_future_share = unsafe_is_future.sum() / unsafe_matched.sum()  # 计算错误匹配中的前视比例。
safe_is_valid = pit_panel.loc[has_financial_match, 'metric_info_date'] <= pit_panel.loc[has_financial_match, 'feature_as_of_date']  # 逐行复核安全信息集不等式。
assert safe_is_valid.all()  # 任一未来财务记录都会阻断练习答案。
pd.DataFrame({'unsafe_matched_rows': [unsafe_matched.sum()], 'unsafe_future_rows': [unsafe_is_future.sum()], 'unsafe_future_share': [unsafe_future_share], 'safe_future_rows': [(~safe_is_valid).sum()]})  # 输出两种设计的关键差异。
表 18.15: 练习 18.3：安全与错误合并的前视偏差比较
unsafe_matched_rows unsafe_future_rows unsafe_future_share safe_future_rows
0 280 280 1.0 0

表 18.15 中安全流程的未来记录数必须为零；错误流程的比例由本地披露时点实际决定。这个比较说明，前视偏差不是抽象警告，而是可以被测试的样本属性。

18.8.4 综合挑战：行业结论的稳健性

练习 18.4

比较各行业 ROE 的简单中位数与市值加权平均数。解释为什么两者差异大时不能任选更“好看”的一个。最后分别写出报表修订历史不完整、行业内异质性和确定性样本选择会阻止哪些更强结论，以及还需什么证据。

完整解答

robustness_sample = valid_cross_section.dropna(subset=['roe_ttm', 'market_cap', 'citics_2019_l1_name']).copy()  # 限定盈利和市值同时可用的公司。
robustness_sample = robustness_sample.loc[robustness_sample['market_cap'] > 0]  # 排除不具经济意义的非正市值。
weighted_numerator = (robustness_sample['roe_ttm'] * robustness_sample['market_cap']).groupby(robustness_sample['citics_2019_l1_name']).sum()  # 汇总市值加权 ROE 分子。
weighted_denominator = robustness_sample.groupby('citics_2019_l1_name')['market_cap'].sum()  # 汇总各行业有效市值权重。
weighted_roe = (weighted_numerator / weighted_denominator).rename('market_cap_weighted_roe')  # 计算大公司影响更强的加权均值。
median_roe = robustness_sample.groupby('citics_2019_l1_name')['roe_ttm'].median().rename('median_roe')  # 计算典型公司的稳健中位数。
robustness_comparison = pd.concat([median_roe, weighted_roe], axis=1).reset_index()  # 对齐两种统计量。
robustness_comparison['difference'] = robustness_comparison['market_cap_weighted_roe'] - robustness_comparison['median_roe']  # 量化规模权重改变行业结论的程度。
robustness_comparison.sort_values('difference', key=lambda series: series.abs(), ascending=False)  # 优先展示口径最敏感的行业。
表 18.16: 练习 18.4：行业 ROE 的中位数与市值加权均值
citics_2019_l1_name median_roe market_cap_weighted_roe difference
1 汽车 0.090782 0.041714 -0.049068
2 电子 0.095636 0.134452 0.038816
3 计算机 0.018717 0.003758 -0.014960
0 医药 0.110139 0.114535 0.004396

表 18.16 中，中位数回答“典型样本公司如何”，市值加权均值回答“每一元市场价值对应的组合表现如何”。差异较大通常意味着行业由少数大公司主导；研究者应按问题预先选择统计量，并同时披露另一口径作为稳健性检查。

完整的证据边界如下：

  • 本地财务快照可能只保留最新修订，因而不能重建投资者在每个历史决策日看到的所有报表版本；要作严格历史回测，需要带版本号与实际发布时间的 point-in-time 数据库；
  • 行业内 ROE、杠杆、成长性和风险存在异质性，中位数或市值加权均值都不能证明“行业身份导致估值”；要作因果解释，需要预先指定控制变量、固定效应或其他识别设计；
  • 确定性样本排除了退市、特别处理和较晚上市公司，行业统计只描述本题冻结样本，不能直接外推到整个 A 股；要作总体推断，需要明确抽样框、纳入退出公司并进行样本选择稳健性检验。