PTrade 风险预警：利用 Altman Z-Score 模型识别财务风险股

Altman Z-Score 模型是由爱德华·阿特曼（Edward Altman）于1968年提出的一种多变量统计分析方法，主要用于预测企业的破产风险。在量化交易中，它常被用作基本面因子的“排雷”指标，帮助策略剔除财务状况恶化的标的。

以下是关于如何在 PTrade 中实现 Altman Z-Score 模型的详细说明和策略代码。

1. Altman Z-Score 模型公式

针对上市公司的标准 Z-Score 公式如下：

$$Z = 1.2X_1 + 1.4X_2 + 3.3X_3 + 0.6X_4 + 1.0X_5$$

其中变量定义及 PTrade 对应字段如下：

• $X_1$ = (流动资产 - 流动负债) / 总资产
  • 反映营运资本占比。
  • PTrade字段：(total_current_assets - total_current_liability) / total_assets
• $X_2$ = 留存收益 / 总资产
  • 反映累积获利能力。
  • PTrade字段：retained_profit / total_assets
• $X_3$ = 息税前利润 (EBIT) / 总资产
  • 反映资产的经营回报率。
  • PTrade字段：ebit (在 profit_ability 表中) / total_assets
• $X_4$ = 股票市值 / 总负债
  • 反映财务杠杆和市场信心。
  • PTrade字段：total_value (在 valuation 表中) / total_liability
• $X_5$ = 营业收入 / 总资产
  • 反映资产周转率。
  • PTrade字段：operating_revenue / total_assets

判别标准：

• Z > 2.675：财务状况良好（安全区）。
• 1.81 < Z < 2.675：灰色区域（有一定的风险）。
• Z < 1.81：财务堪忧（危险区，高破产风险）。

2. PTrade 策略代码实现

以下策略代码实现了 Z-Score 的计算，并会在每日交易前筛选出 Z-Score 低于 1.81 的高风险股票进行预警（打印日志），同时在交易逻辑中卖出这些风险股，买入安全股。

import pandas as pd
import numpy as np

def initialize(context):
    """
    初始化函数
    """
    # 设置回测基准，例如沪深300
    set_benchmark('000300.SS')
    # 开启真实市价成交模式（回测用）
    set_limit_mode('UNLIMITED')
    # 设置手续费
    set_commission(commission_ratio=0.0003, min_commission=5.0, type="STOCK")
    
    # 定义全局变量：Z-Score 阈值
    g.z_score_safe = 2.675  # 安全阈值
    g.z_score_distress = 1.81 # 危险阈值
    
    # 设定要操作的股票池范围，这里示例使用沪深300
    g.index_code = '000300.SS'

def before_trading_start(context, data):
    """
    盘前处理：获取财务数据并计算 Z-Score
    """
    # 1. 获取股票池
    stocks = get_index_stocks(g.index_code)
    if not stocks:
        log.info("未获取到成分股")
        return

    # 2. 获取财务数据
    # 需要查询多张表：资产负债表、利润表、盈利能力表、估值表
    
    # (A) 资产负债表 (balance_statement)
    # 字段: 总资产, 流动资产, 流动负债, 总负债, 留存收益(未分配利润)
    q_bs = get_fundamentals(stocks, 'balance_statement', 
                            ['total_assets', 'total_current_assets', 'total_current_liability', 
                             'total_liability', 'retained_profit'],
                            date=context.blotter.current_dt.strftime("%Y%m%d"),
                            date_type=1) # date_type=1 表示获取最近报告期数据
    
    # (B) 利润表 (income_statement)
    # 字段: 营业收入
    q_is = get_fundamentals(stocks, 'income_statement', 
                            ['operating_revenue'],
                            date=context.blotter.current_dt.strftime("%Y%m%d"),
                            date_type=1)

    # (C) 盈利能力表 (profit_ability)
    # 字段: 息税前利润 (EBIT)
    q_pa = get_fundamentals(stocks, 'profit_ability', 
                            ['ebit'],
                            date=context.blotter.current_dt.strftime("%Y%m%d"),
                            date_type=1)
    
    # (D) 估值表 (valuation)
    # 字段: 总市值 (total_value)
    # 注意：valuation表只能按天查询，返回的是指定日期前一交易日的数据
    q_val = get_fundamentals(stocks, 'valuation', 
                             ['total_value'],
                             date=context.blotter.current_dt.strftime("%Y%m%d"))

    # 3. 数据合并与清洗
    # 将所有数据合并到一个 DataFrame 中
    if q_bs is None or q_is is None or q_pa is None or q_val is None:
        log.warning("财务数据获取失败，跳过今日计算")
        g.z_score_df = pd.DataFrame()
        return

    # 这里的 DataFrame 索引通常是股票代码
    df = pd.concat([q_bs, q_is, q_pa, q_val], axis=1, join='inner')
    
    # 去除包含 NaN 的行，防止计算报错
    df.dropna(inplace=True)
    
    # 4. 计算 Altman Z-Score 各项因子
    # X1 = (流动资产 - 流动负债) / 总资产
    df['X1'] = (df['total_current_assets'] - df['total_current_liability']) / df['total_assets']
    
    # X2 = 留存收益 / 总资产
    df['X2'] = df['retained_profit'] / df['total_assets']
    
    # X3 = EBIT / 总资产
    df['X3'] = df['ebit'] / df['total_assets']
    
    # X4 = 股票市值 / 总负债
    # 注意：total_liability 可能为0，需要处理除零异常，但一般上市公司总负债不为0
    df['X4'] = df['total_value'] / df['total_liability']
    
    # X5 = 营业收入 / 总资产
    df['X5'] = df['operating_revenue'] / df['total_assets']
    
    # 5. 计算 Z-Score
    # Z = 1.2*X1 + 1.4*X2 + 3.3*X3 + 0.6*X4 + 1.0*X5
    df['Z_Score'] = 1.2 * df['X1'] + 1.4 * df['X2'] + 3.3 * df['X3'] + 0.6 * df['X4'] + 1.0 * df['X5']
    
    # 保存计算结果到全局变量
    g.z_score_df = df
    
    # 6. 风险预警日志
    risky_stocks = df[df['Z_Score'] < g.z_score_distress].index.tolist()
    if len(risky_stocks) > 0:
        log.info("【财务风险预警】以下股票 Z-Score 小于 %.2f，存在财务风险: %s" % (g.z_score_distress, risky_stocks))

def handle_data(context, data):
    """
    盘中交易逻辑：卖出高风险股票，买入低风险股票
    """
    if not hasattr(g, 'z_score_df') or g.z_score_df.empty:
        return

    df = g.z_score_df
    
    # 筛选出安全股 (Z > 2.675) 和 危险股 (Z < 1.81)
    safe_stocks = df[df['Z_Score'] > g.z_score_safe].index.tolist()
    risky_stocks = df[df['Z_Score'] < g.z_score_distress].index.tolist()
    
    # 获取当前持仓
    current_positions = list(context.portfolio.positions.keys())
    
    # 1. 卖出逻辑：如果持仓股票变成了高风险股票，清仓
    for stock in current_positions:
        if stock in risky_stocks:
            log.info("股票 %s 财务风险较高 (Z=%.2f)，进行卖出" % (stock, df.loc[stock, 'Z_Score']))
            order_target(stock, 0)
    
    # 2. 买入逻辑：资金平均分配买入安全区的股票 (示例逻辑，仅买入前10只)
    # 过滤掉停牌、跌停的股票
    buy_list = []
    for stock in safe_stocks:
        if stock in data and data[stock].is_open == 1: # 确保开盘
             buy_list.append(stock)
    
    # 简单示例：只持有 Z-Score 最高的 10 只
    # 对安全股按 Z-Score 降序排列
    sorted_safe = df.loc[buy_list].sort_values(by='Z_Score', ascending=False)
    target_stocks = sorted_safe.head(10).index.tolist()
    
    if len(target_stocks) > 0:
        # 简单的等权重分配资金
        cash_per_stock = context.portfolio.portfolio_value / 10
        for stock in target_stocks:
            order_target_value(stock, cash_per_stock)

def after_trading_end(context, data):
    pass

3. 代码关键点解析

1. 数据获取 (get_fundamentals)：

   • Altman Z-Score 需要跨表查询数据。我们分别查询了 balance_statement（资产负债表）、income_statement（利润表）、profit_ability（盈利能力表）和 valuation（估值表）。
   • date_type=1：确保获取的是基于当前日期可见的最新财报数据，避免未来函数。
   • valuation 表比较特殊，它返回的是每日的市值数据，不需要 date_type 参数。

2. 数据对齐 (pd.concat)：

   • 由于 get_fundamentals 返回的是 DataFrame，索引为股票代码，使用 pd.concat([..., axis=1, join='inner']) 可以确保所有财务指标在同一只股票上对齐，并自动剔除数据缺失的股票。

3. 因子计算：

   • 严格按照 Altman 的原始公式进行加权求和。
   • 注意：$X_3$ 使用了 ebit（息税前利润），PTrade 的 profit_ability 表直接提供了该字段，这比手动用 净利润 + 所得税 + 利息费用 计算要方便且准确。

4. 风险预警与交易：

   • 在 before_trading_start 中计算完成后，直接筛选出 $Z < 1.81$ 的股票打印日志，这对于实盘或模拟盘监控非常有帮助。
   • 在 handle_data 中，策略会主动卖出落入“危险区”的持仓股票，并尝试买入“安全区”且 Z-Score 排名靠前的股票。

4. 注意事项

• 适用范围：原始的 Altman Z-Score 模型主要针对制造业企业设计。对于金融类（银行、券商）、房地产或公用事业类公司，该模型的适用性较弱，因为这些行业的资产负债结构与制造业差异巨大（例如银行的高负债是常态）。在使用时，建议剔除金融行业股票。
• 数据滞后性：财务数据（如总资产、留存收益）是低频数据（季频），而市值（$X_4$）是高频数据（日频）。这意味着 Z-Score 的剧烈波动通常由股价波动引起，而非基本面突然恶化。
• A股适配性：国内量化研究中有时会使用修正版的 Z-Score 模型（系数不同），或者使用 $Z''$-Score 模型（针对非制造业和新兴市场）。上述代码使用的是最经典的原始公式，您可以根据需要调整系数（例如：$Z = 6.56X_1 + 3.26X_2 + 6.72X_3 + 1.05X_4$ 是一种常见的修正版）。

Q&A

Q: 为什么在计算 $X_4$ 时使用 total_value 而不是 float_value（流通市值）？
A: Altman 原始公式中定义 $X_4$ 为“权益的市场价值 / 总负债”。权益的市场价值理论上应包含所有股权的价值，因此使用总市值 (total_value) 更符合原定义。但在 A 股实操中，也有策略使用流通市值，因为非流通股（或限售股）的定价可能不准确。

Q: 如何处理金融股的 Z-Score 异常值？
A: 可以在 before_trading_start 中使用 get_industry_stocks 获取金融行业代码，并在计算前将这些股票从列表中剔除。

Q: 这个策略可以直接用于实盘吗？
A: 该代码主要用于演示 Z-Score 的计算和预警逻辑。作为一个完整的交易策略，它还缺少仓位管理（如防止单只股票持仓过重）、止损逻辑以及针对特定行业的因子中性化处理。建议将其作为一个风控模块（Filter）嵌入到您的主策略中。