如何用Python计算全市场等权PE百分位？

Question

如何用Python计算全市场等权PE百分位时，常遇到的问题是：在获取大量股票PE数据后，如何处理缺失值与异常值？例如部分ST股或亏损企业PE为负或极大值，直接计算等权平均会导致偏差。此外，对全市场成分股动态调整（如新股、退市）的处理也影响历史分位准确性。如何结合Tushare或AKShare高效获取清洗数据，并用scipy.stats.percentileofscore合理计算当前等权PE所处分位？

Answer 1

1. 问题背景与数据获取挑战

在量化投资分析中，全市场等权PE（市盈率）是衡量整体股市估值水平的重要指标。计算其历史百分位有助于判断当前市场是否处于高估或低估状态。然而，在使用Python进行该指标构建时，常面临如下核心问题：

• 如何从Tushare或AKShare高效获取全市场个股的PE数据？
• 部分股票（如ST股、亏损企业）的PE为负值或极大异常值，影响等权平均的准确性。
• 缺失值处理不当会导致样本偏差。
• 成分股动态变化（新股上市、退市、暂停交易）对历史序列一致性构成挑战。

以A股市场为例，每日有超过5000只证券变动，若不妥善处理这些数据质量问题，最终得到的PE分位数将失去参考意义。

2. 数据获取：Tushare vs AKShare 对比分析

维度	Tushare	AKShare
数据覆盖	全面，需积分权限	开源免费，覆盖较全
更新频率	实时/日频	日频为主
API稳定性	较高（商业支持）	社区维护，偶有波动
安装便捷性	pip install tushare	pip install akshare
历史数据回溯	支持多年回溯	依赖接口实现

推荐策略：对于长期研究项目，可结合两者优势——用Tushare获取高质量历史快照，AKShare用于日常增量更新。

3. 异常值识别与清洗方法论

PE异常主要表现为：

1. PE ≤ 0：代表公司亏损或无盈利，不具备传统估值意义。
2. PE > 1000：极端炒作或财务异常导致，显著拉高均值。
3. NaN值：停牌、未披露财报或数据源缺失。

清洗逻辑应遵循以下流程：

import pandas as pd
import numpy as np

def clean_pe_series(pe_series):
    # 过滤负值和极大值（设定阈值）
    pe_clean = pe_series[(pe_series > 0) & (pe_series < 1000)]
    # 填补剩余缺失（如有）
    pe_clean = pe_clean.dropna()
    return pe_clean

进一步优化可引入Winsorization（缩尾处理），减少极端值影响而不完全剔除。

4. 成分股动态调整机制设计

全市场范围并非静态集合。新股上市、老股退市、ST标记变更都会改变有效样本池。为保证历史可比性，建议采用“交易日快照”方式存储每日有效的股票列表及对应PE。

graph TD A[获取当日全部正常交易股票] --> B[调用AKShare/Tushare获取PE] B --> C[清洗异常PE值] C --> D[计算等权PE均值] D --> E[存入历史时间序列] E --> F{是否为最新日？} F -- 是 --> G[输出当前分位] F -- 否 --> H[继续循环]

关键点在于每次计算都基于当日实际可交易且数据完整的股票集合，避免未来函数偏差。

5. 百分位计算实现：scipy.stats.percentileofscore 应用

假设已有历史等权PE序列 history_pe 和当前值 current_pe，可通过以下代码计算其所处分位：

from scipy import stats
import numpy as np

# 示例数据：过去一年的每日等权PE
history_pe = np.random.lognormal(mean=2.5, sigma=0.4, size=250)
current_pe = 48.6

# 计算当前PE在历史分布中的百分位
percentile = stats.percentileofscore(history_pe, current_pe, kind='weak')

print(f"当前等权PE: {current_pe:.2f}，位于历史 {percentile:.1f}% 分位")

参数 kind='weak' 表示 ≤ current_pe 的比例，符合金融语义下的“低于或等于”的分位定义。

6. 工程化建议与性能优化

面对高频批量请求，应注意以下最佳实践：

• 使用缓存机制（如Redis或本地Pickle）避免重复抓取相同日期数据。
• 并发获取多只股票数据（配合 asyncio + aiohttp 提升AKShare效率）。
• 建立数据库表结构存储每日 cleaned_pe_snapshot(date, stock_code, pe_ttm)，便于回测分析。
• 定期校验数据完整性，例如对比沪深两市总市值与Wind/同花顺基准。

完整流程应封装成模块化函数，支持配置化运行：

class MarketPEAnalyzer:
    def __init__(self, data_source='akshare'):
        self.source = data_source
    
    def fetch_daily_pe(self, trade_date):
        # 实现具体数据拉取逻辑
        pass
    
    def compute_equal_weighted_pe(self, pe_series):
        cleaned = clean_pe_series(pe_series)
        return np.mean(cleaned)
    
    def get_current_percentile(self, current_pe, historical_pe_list):
        return stats.percentileofscore(historical_pe_list, current_pe, kind='weak')