在业绩密集披露、市场情绪反复的阶段,很多做数据分析的同学都会遇到同一个问题:财报数据拿到了,估值指标也算了,但面对一堆数字就是讲不出“贵还是便宜”的逻辑。其实估值这件事,本质上是一套可以量化、可以复现的计算流程。本文结合财报季数据处理的实际场景,聊聊如何用 Python 完成从财务报表获取、估值指标计算到可视化分析的一整套实操方案。
1. 背景与核心概念
1.1 估值分析到底在算什么
估值分析是投资研究中最基础也最常用的工作之一。简单来说,估值就是回答一个问题:一家公司当前的价格,相对于它赚到的钱、拥有的资产、或者未来的增长潜力,到底是贵了还是便宜了。
这个问题的答案从来不是绝对的。同样是 30 倍市盈率,放在高速增长的软件公司和增长停滞的传统制造业公司身上,含义完全不同。所以估值分析的核心不是“记住哪个数字是合理的”,而是掌握一套可复用的计算框架,能够结合行业特征、公司阶段和宏观环境去做对比。
在财报季这个时间窗口,估值分析的工作量会明显增加。原因在于财报数据集中释放,分析师和投资者需要在短时间内处理大量报表,用统一的口径计算指标,然后横向对比同行业公司、纵向对比历史区间。这种场景非常适合用 Python 来做流程化处理。
1.2 常用估值指标及其含义
在实操中,以下指标是出场频率最高的。
市盈率(PE)等于股价除以每股收益,或者用总市值除以净利润。它衡量的是“为每 1 元利润支付多少价格”。市盈率分为静态市盈率、滚动市盈率和动态市盈率,区别在于分母用的是上一年度净利润、最近四个季度净利润还是预测净利润。财报季通常更关注滚动市盈率,因为它包含了最新一期财报的数据。
市净率(PB)等于股价除以每股净资产,衡量“为每 1 元净资产支付多少价格”。对于银行、保险、地产这类资产驱动型企业,市净率比市盈率更有参考价值。
PEG 指标是在市盈率基础上引入增长率,计算公式是市盈率除以盈利增长率。PEG 的引入是为了修正“高增长公司天然高市盈率”带来的误判,用增长去消化估值。
股息率等于每股分红除以股价,反映的是持有期内通过分红获得的回报率。对于现金流稳定的成熟企业,股息率是估值的重要锚点。
这四个指标各有盲区:PE 无法衡量亏损企业,PB 对轻资产公司意义有限,PEG 依赖预测增长率的准确性,股息率忽略了成长性本身的价值。所以实际分析中通常需要组合使用。
1.3 为什么用 Python 做估值分析
手工计算估值指标最大的问题是口径不一致和效率低。不同平台的数据源、不同人对“净利润”的取法,都会导致结果差异。Python 的优势在于可以把数据获取、清洗、计算、可视化全部固化为一套流程。
比如同一份财报数据,你可以用脚本同时计算 PE、PB、PEG 和股息率,然后输出对比表格。换一只股票、换一个季度,只需要改代码中的参数,完全不用重来一遍。对于研究多只股票、多个周期的场景,Python 的批处理能力非常实用。
2. 环境准备与版本说明
2.1 运行环境要求
本文的代码以 Python 3.8+ 为基础,推荐使用 conda 或 venv 创建独立环境,避免依赖冲突。
python -m venv fin_env source fin_env/bin/activate # Windows 下执行 fin_env\Scripts\activate2.2 依赖库安装
数据获取使用 akshare,这是国产开源财经数据接口库,覆盖 A 股和港美股的部分公开数据,无需申请 token。数据处理使用 pandas,数值计算使用 numpy,可视化使用 matplotlib。另外为了处理中文显示,需要额外配置字体。
pip install akshare pandas numpy matplotlib2.3 版本说明
需要说明的是,akshare 的接口变动比较频繁,不同版本的方法名和返回字段可能存在差异。本文以 1.x 常见接口为例演示完整的处理思路。如果你安装的版本接口有调整,重点理解数据清洗和指标计算的逻辑,数据字段名可以在 akshare 官方文档中按版本查询。
3. 核心语法与原理拆解
3.1 财务数据获取:理解财务摘要接口
akshare 提供了多层次的财务数据接口。做估值分析时,最常用的是一组财务摘要接口,能够获取到净利润、净资产、每股收益等关键字段。
import akshare as ak # 获取指定股票最近的财务摘要(示例代码,需按实际版本调整参数) # df = ak.stock_financial_abstract(symbol="600519")这段代码获取到的数据通常是以报告期为索引的宽表,每一行代表一个报告期,每一列代表一个财务指标。实际使用中需要注意以下几点:
第一,数据的单位。有的接口返回的单位是元,有的是万元,需要先统一换算,否则后续计算指标时量级会出错。
第二,字段口径。比如“净利润”可能有“归母净利润”和“净利润”两种口径。计算 PE 时,分子市值是包含少数股东权益的,分母如果用归母净利润,计算出来的 PE 会偏高。因此要么统一用净利润,要么统一用归母净利润,并记录自己的选择。
第三,数据缺失。次新股、B 股、部分境外上市公司的数据可能不完整,需要先做缺失值检查。
下面是完整的获取与预览代码。
import akshare as ak import pandas as pd pd.set_option("display.max_columns", None) pd.set_option("display.width", 200) try: # 获取财务摘要数据 df = ak.stock_financial_abstract(symbol="600519") print("数据形状:", df.shape) print("前两行数据:") print(df.head(2).T) except Exception as e: print("接口调用失败:", e)3.2 市值数据获取:理解股本与价格的结合
计算 PE 和 PB 的分子是总市值。总市值等于总股本乘以当前股价。市值的获取有两种途径:一种是直接使用接口返回的市值字段,一种是用股本和股价自行计算。
# 获取个股信息,通常包含总市值、流通市值等字段 # stock_individual_info_em 是东方财富接口,字段较稳定 info = ak.stock_individual_info_em(symbol="600519")这个接口返回的是键值对结构,通常包含股票代码、股票简称、总股本、流通股本、总市值、流通市值等字段。使用时可以转换为字典方便取值。
info_dict = dict(zip(info["item"], info["value"])) total_market_cap = float(info_dict["总市值"]) print("总市值:", total_market_cap)3.3 指标计算的边界条件处理
估值指标计算看起来只是除法,但实际工程中边界条件非常多。
净利润为 0 或为负时,PE 没有意义。要么显示为空,要么标记为“亏损”,不能直接丢弃。净资产为 0 或为负时,PB 没有意义,常见于资不抵债的公司。每股收益的数值极小时,PE 会异常放大,需要考虑是否数据单位错误。股本变动(增发、回购)会影响市值,进而影响所有以市值为基础的指标。
因此,在实际编码中要养成一个习惯:先判断分母是否合法,再执行除法,最后对结果做合理性检查。
def safe_divide(numerator, denominator): if denominator is None or abs(denominator) < 1e-9: return None return numerator / denominator3.4 技术指标的常见误区
关于估值指标,有几个高频误区必须澄清。
第一,不要混用不同时点的市值和净利润。严格来说,计算 PE 时分子和分母应尽量接近同一时点。虽然实践中常用“当前市值 / 最新年报净利润”的简化算法,但跨期混用会带来误差,尤其是在股价大幅波动或业绩急剧变化的阶段。
第二,不要盲目跨行业比较。不同行业的商业模式差异极大,PE 和 PB 的合理区间完全不同。同行业内比较才有意义。
第三,不要忽略会计口径差异。净利润有扣非净利润和包含非经常性损益的净利润之分,两者可能差异巨大。地产、金融等行业的利润确认规则复杂,直接用报表数值做指标,可能得出失真结论。
4. 完整实战案例
下面以一个完整案例串联整个流程。场景设定为:有一批股票代码,需要在财报季统一计算估值指标,并输出对比表格和可视化图表。
4.1 创建项目结构
estimator/ ├── data/ # 缓存数据 ├── output/ # 输出结果 ├── main.py # 主脚本 └── valuation.py # 估值计算模块4.2 编写估值计算模块
首先在valuation.py中封装数据获取和指标计算函数。
# 文件路径:estimator/valuation.py import akshare as ak import pandas as pd def get_financial_summary(symbol): """获取财务摘要数据,并转换为以报告期为索引的 DataFrame。""" df = ak.stock_financial_abstract(symbol=symbol) # 将第一列设置为索引,通常是报告期 first_col = df.columns[0] df = df.set_index(first_col) return df def get_market_data(symbol): """获取总市值、总股本等市场数据。""" info = ak.stock_individual_info_em(symbol=symbol) info_dict = dict(zip(info["item"], info["value"])) return info_dict def calculate_valuation_metrics(symbol, report_dates=None): """ 计算指定股票在全部报告期的估值指标。 返回包含净利润、净资产、市值、PE、PB 的 DataFrame。 """ summary = get_financial_summary(symbol) market = get_market_data(symbol) total_market_cap = float(market["总市值"]) total_share = float(market["总股本"]) # 假设财务摘要中有这几个字段,实际使用时需根据接口输出调整 result = pd.DataFrame(index=summary.index) result["净利润"] = pd.to_numeric(summary["净利润"], errors="coerce") result["净资产"] = pd.to_numeric(summary["净资产"], errors="coerce") result["每股收益"] = pd.to_numeric(summary["每股收益"], errors="coerce") # 市值可以按报告期调整,也可以使用当前市值。这里演示使用当前市值。 result["当前总市值"] = total_market_cap # 计算 PE result["PE"] = result.apply( lambda row: total_market_cap / row["净利润"] if row["净利润"] and row["净利润"] > 0 else None, axis=1, ) # 计算 PB result["PB"] = result.apply( lambda row: total_market_cap / row["净资产"] if row["净资产"] and row["净资产"] > 0 else None, axis=1, ) return result这段代码中的关键点是errors="coerce",它可以把财报接口中返回的字符串型数字统一转换为数值类型,无法转换的置为 NaN。
4.3 编写主脚本
主脚本main.py负责调用模块、缓存数据并输出结果。
# 文件路径:estimator/main.py import os import pandas as pd from valuation import calculate_valuation_metrics # 示例股票池,实际使用时可替换为其他代码 stock_pool = ["600519", "000858", "300750"] results = {} for symbol in stock_pool: print(f"正在处理:{symbol}") try: df = calculate_valuation_metrics(symbol) # 只保留最近 8 个报告期 df = df.head(8) results[symbol] = df # 将数据缓存为 CSV os.makedirs("data", exist_ok=True) df.to_csv(f"data/{symbol}_metrics.csv", encoding="utf-8-sig") except Exception as e: print(f"处理 {symbol} 失败:{e}") # 汇总最近一个报告期数据 summary = [] for symbol, df in results.items(): if not df.empty: latest = df.iloc[0] summary.append( { "股票代码": symbol, "报告期": df.index[0], "净利润": latest["净利润"], "PE": latest["PE"], "PB": latest["PB"], } ) summary_df = pd.DataFrame(summary) summary_df.to_csv("output/valuation_summary.csv", index=False, encoding="utf-8-sig") print("估值汇总表已输出到 output/valuation_summary.csv")4.4 运行与验证
在项目根目录执行:
python main.py正常输出的结果中,每个股票的处理过程会打印在控制台。最终生成的valuation_summary.csv可以使用 Excel 直接打开查看。
4.5 可视化对比
为了直观对比多只股票的估值水平,可以绘制柱状图。这里以 PE 对比图为例。
# 文件路径:estimator/plot_pe.py import pandas as pd import matplotlib.pyplot as plt plt.rcParams["font.sans-serif"] = ["SimHei", "Arial Unicode MS"] plt.rcParams["axes.unicode_minus"] = False summary_df = pd.read_csv("output/valuation_summary.csv") fig, ax = plt.subplots(figsize=(10, 6)) colors = ["#4C72B0" if v == summary_df["PE"].min() else "#DD8452" for v in summary_df["PE"]] ax.bar(summary_df["股票代码"], summary_df["PE"], color=colors) ax.set_title("当前报告期 PE 对比") ax.set_ylabel("市盈率") ax.axhline(0, color="black", linewidth=0.8) for i, v in enumerate(summary_df["PE"]): if pd.notna(v): ax.text(i, v + 0.5, f"{v:.1f}", ha="center") plt.tight_layout() plt.savefig("output/pe_compare.png", dpi=150) print("图表已保存至 output/pe_compare.png")4.6 结果说明
通过上面的流程,你可以得到至少两个层次的输出:
第一层是单只股票在不同报告期的 PE、PB 数据,可以用来观察估值的时间序列变化,判断当前估值处于历史高位还是低位。
第二层是同一时期多只股票的横截面对比,可以用来做行业内的估值分位排序。
需要注意,这里的案例是简化演示,实际研究中还需要加入行业分类、盈利预测、分红记录等更多维度的数据。
5. 常见问题与排查思路
5.1 接口返回字段不对
akshare 的不同版本、不同接口返回的字段名称存在差异。如果代码运行时报 KeyError,优先打印接口返回的字段列表。
df = ak.stock_financial_abstract(symbol="600519") print(df.columns.tolist())根据实际字段名称调整代码中的列名即可。
5.2 数据库表结构梳理
没有,这里不需要数据库表结构,但要注意财务数据的单位问题。很多财务接口返回的数据单位为“万元”,而市值接口返回的可能是“元”,直接相除会导致结果偏差巨大。
解决方案是统一除以 10000 或者统一乘以 10000,确保分子分母单位一致。
5.3 非交易日的市值波动
这个问题在批量处理中经常出现。如果不同股票的市值取到的时间点不一致(比如某只股票停牌,市值停留在停牌前的价格),会导致横向对比失真。
处理思路是:在数据获取时记录市值的时间戳,如果发现某只股票的市值时间明显滞后,可以视为数据异常,在汇总表中标注或剔除。
5.4 接口被限流或超时
akshare 底层依赖公开网页接口,访问频率过高时可能触发反爬机制。解决方案有两个方向:一是控制请求频率,在循环中增加time.sleep(1);二是将结果缓存到本地,第二次运行时直接读取缓存文件,减少请求次数。
5.5 一个完整的排查流程清单
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
| KeyError: 某字段不存在 | 接口字段名变更 | 打印 columns 列表核对 |
| PE 数值异常巨大 | 净利润单位或口径不一致 | 检查单位和是否使用归母口径 |
| 数据列全部为 NaN | 接口返回为字符串格式 | 增加 errors="coerce" 并检查原始数据 |
| 与财经网站指标不一致 | 指标时点或口径不同 | 确认使用滚动年报还是最新报告期 |
| 接口超时 | 请求频繁或网络问题 | 增加重试机制,使用本地缓存 |
| 图片中文乱码 | 缺少中文字体配置 | 设置 rcParams 字体,改用系统已有字体 |
6. 最佳实践与工程建议
6.1 指标口径统一
建议在代码中定义一个枚举或者常量类,把所有估值指标的计算口径固定下来。比如“PE = 当前总市值 / 最近一期报告期净利润年化值”,就把它写进配置,而不是散落在多个函数里。这能避免后续维护时出现同一字段两种计算逻辑的问题。
6.2 数据缓存与增量更新
财务数据不是高频变化的数据,对同一只股票在同一天内多次调用接口是浪费资源。建议第一次获取后保存到本地,后续优先读缓存,支持通过参数强制刷新。这样既减少接口压力,也方便离线调试。
import os import pandas as pd def load_or_fetch(symbol, fetch_func, cache_dir="data"): os.makedirs(cache_dir, exist_ok=True) cache_path = os.path.join(cache_dir, f"{symbol}.csv") if os.path.exists(cache_path): return pd.read_csv(cache_path) df = fetch_func(symbol) df.to_csv(cache_path, index=False, encoding="utf-8-sig") return df6.3 异常处理与边界保护
在企业层面做估值分析,数据质量直接影响决策质量。建议做到:计算指标时使用防护函数,杜绝除零错误;对无效数据保留 NaN 而不是填充 0,防止“假数据”污染结果;在报表中标记数据时点和口径,方便追溯。
6.4 关于未来函数与数据穿越
这个问题在量化分析中特别重要。如果在历史回测中使用了当前时刻的总市值去计算历史 PE,就会引入未来信息,导致回测结果过度乐观。正确的做法是:回测时使用当时时点的市值数据,即“时点市值 = 当时股价 × 当时总股本”。
如果你的数据源无法提供历史市值,至少要意识到这个问题,并在结论中说明偏差来源。
6.5 合规与风险提示
估值分析是研究工具,不是投资建议。在输出报告、代码、图表时,建议明确标注数据时间、数据来源和计算口径。报告中的结论要强调概率思维,避免绝对化表述。
7. 总结与学习路线
通过这篇文章的实战流程,你可以掌握几个核心能力:使用 Python 批量获取财务摘要和市场数据;统一口径计算 PE、PB 等估值指标;处理财报数据中的单位、缺失值和异常值问题;用图表对比多只股票的估值水平。
下一步建议从三个方向继续深入。
第一个方向是扩充指标维度。在现有基础上加入 PEG、股息率、市现率等指标,形成更完整的估值体系。
第二个方向是引入时间序列分析。把单只股票的 PE 历史分位数计算出来,判断当前估值处于历史什么位置,这比单纯看绝对数值更有参考意义。
第三个方向是搭建自己的估值数据中台。把数据获取、清洗、计算、展示封装成标准服务,支持命令行参数和定时任务,让估值分析从“手动跑脚本”变成“自动化流水线”。
最后补充一点:估值分析本质上是在处理不确定性,任何指标都只是对过去的描述和对未来的近似。写代码时保持严谨,看结果时保持理性,这才是做数据分析最稳妥的态度。如果本文对你有帮助,可以收藏备用,后续遇到财报季批量计算估值指标时,直接对照文章思路实现即可。