🚀 AI 一键生成 joinquant 策略代码

JoinQuant 价值投资策略：如何用 Python 构建皮尔托斯基 F-Score 模型？

joinquant | 发布时间: 2025/12/16
以下内容由 EasyQuant 生成。

问题描述

请帮我用财务数据编写一个皮尔托斯基 F-Score 选股策略。

解决方案

这是一个基于 皮尔托斯基 F-Score (Piotroski F-Score) 模型的选股策略实现。

策略原理

Piotroski F-Score 是一种通过财务指标评估公司财务状况的评分系统（0-9分），通常用于在低市净率（低 P/B）的股票池中筛选出基本面强劲的公司。

评分标准（共9分）：

  1. 盈利能力 (4分)
    • 资产回报率 (ROA) > 0
    • 经营现金流 (CFO) > 0
    • ROA 同比增长 (Current ROA > Previous ROA)
    • 经营现金流 > 净利润 (Accruals: CFO > Net Income)
  2. 财务杠杆与流动性 (3分)
    • 长期负债比率同比下降 (Leverage 下降)
    • 流动比率同比上升 (Current Ratio 上升)
    • 未增发新股 (股本数 $\le$ 上一年)
  3. 运营效率 (2分)
    • 毛利率同比上升 (Gross Margin 上升)
    • 资产周转率同比上升 (Asset Turnover 上升)

策略逻辑

  1. 初筛：选取全市场市净率 (P/B) 最低的 20% 股票（剔除停牌、ST）。
  2. 评分：对初筛股票计算 F-Score。
  3. 买入：选取 F-Score $\ge$ 7 分的股票。
  4. 调仓：按月调仓，等权重买入。

策略代码

# -*- coding: utf-8 -*-
from jqdata import *
import pandas as pd
import numpy as np
import datetime

def initialize(context):
    # 设定基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # 设定手续费
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 设定全局变量
    g.stock_num = 20  # 最终持仓数量
    
    # 按月运行，每月第一个交易日
    run_monthly(trade, 1)

def trade(context):
    # 1. 获取股票池：全市场 P/B 最低的 20%
    # 剔除停牌、ST、上市不满一年的股票
    current_date = context.current_dt.date()
    all_stocks = list(get_all_securities(['stock'], date=current_date).index)
    
    # 过滤停牌和ST
    current_data = get_current_data()
    all_stocks = [stock for stock in all_stocks 
                  if not current_data[stock].paused 
                  and not current_data[stock].is_st
                  and '退' not in current_data[stock].name
                  and (current_date - get_security_info(stock).start_date).days > 365]
    
    # 获取市净率数据
    q = query(valuation.code, valuation.pb_ratio).filter(
        valuation.code.in_(all_stocks)
    ).order_by(valuation.pb_ratio.asc())
    
    df_pb = get_fundamentals(q)
    # 取前20%低P/B的股票作为候选池
    limit_idx = int(len(df_pb) * 0.2)
    candidate_stocks = list(df_pb['code'][:limit_idx])
    
    # 2. 计算 F-Score 并选股
    # 获取当前和一年前的财务数据
    # 注意：为了简化计算，这里对比的是"当前可见的最新财报"与"一年前可见的财报"
    # 这种方式在回测中是无未来函数的
    
    # 获取当前数据
    df_curr = get_financial_data(candidate_stocks, current_date)
    # 获取一年前数据
    last_year_date = current_date - datetime.timedelta(days=365)
    df_prev = get_financial_data(candidate_stocks, last_year_date)
    
    if df_curr.empty or df_prev.empty:
        return

    # 合并数据以便比较
    # 后缀 _curr 代表当前，_prev 代表去年
    df_merge = pd.merge(df_curr, df_prev, on='code', suffixes=('_curr', '_prev'))
    
    # 计算 F-Score
    scores = calculate_f_score(df_merge)
    
    # 筛选 F-Score >= 7 的股票
    buy_list = scores[scores['f_score'] >= 7].index.tolist()
    
    # 如果选出的股票过多，按市值从小到大排序取前 g.stock_num 只
    if len(buy_list) > g.stock_num:
        q_cap = query(valuation.code).filter(
            valuation.code.in_(buy_list)
        ).order_by(valuation.market_cap.asc()).limit(g.stock_num)
        buy_list = list(get_fundamentals(q_cap)['code'])
    
    # 3. 执行交易
    do_rebalance(context, buy_list)

def get_financial_data(stock_list, date):
    """
    获取计算 F-Score 所需的财务数据
    """
    q = query(
        valuation.code,
        # 1. 盈利能力指标
        indicator.roa, # 资产回报率
        income.net_profit, # 净利润
        cash_flow.net_operate_cash_flow, # 经营现金流
        balance.total_assets, # 总资产
        
        # 2. 杠杆与流动性指标
        balance.long_term_borrowing, # 长期借款 (近似长期负债)
        balance.total_current_assets, # 流动资产
        balance.total_current_liability, # 流动负债
        valuation.capitalization, # 总股本
        
        # 3. 运营效率指标
        indicator.gross_profit_margin, # 毛利率
        indicator.operation_revenue, # 营业收入
    ).filter(
        valuation.code.in_(stock_list)
    )
    
    df = get_fundamentals(q, date=date)
    
    # 数据清洗，填充空值为0
    df = df.fillna(0)
    
    # 计算衍生指标
    # 资产周转率 = 营业收入 / 总资产
    # 注意处理分母为0的情况
    df['asset_turnover'] = df['operation_revenue'] / df['total_assets'].replace(0, np.nan)
    df['asset_turnover'] = df['asset_turnover'].fillna(0)
    
    # 流动比率 = 流动资产 / 流动负债
    df['current_ratio'] = df['total_current_assets'] / df['total_current_liability'].replace(0, np.nan)
    df['current_ratio'] = df['current_ratio'].fillna(0)
    
    # 长期负债比率 = 长期借款 / 总资产 (简化版)
    df['leverage'] = df['long_term_borrowing'] / df['total_assets'].replace(0, np.nan)
    df['leverage'] = df['leverage'].fillna(0)
    
    return df

def calculate_f_score(df):
    """
    计算 Piotroski F-Score (0-9分)
    """
    df['f_score'] = 0
    
    # --- 1. 盈利能力 (Profitability) ---
    # ROA > 0
    df.loc[df['roa_curr'] > 0, 'f_score'] += 1
    # CFO > 0
    df.loc[df['net_operate_cash_flow_curr'] > 0, 'f_score'] += 1
    # Delta ROA > 0 (ROA 同比增长)
    df.loc[df['roa_curr'] > df['roa_prev'], 'f_score'] += 1
    # Accruals: CFO > Net Profit (经营现金流大于净利润，说明盈余质量高)
    df.loc[df['net_operate_cash_flow_curr'] > df['net_profit_curr'], 'f_score'] += 1
    
    # --- 2. 杠杆、流动性与资金来源 (Leverage, Liquidity and Source of Funds) ---
    # Delta Leverage < 0 (长期负债占比下降，越低越好)
    df.loc[df['leverage_curr'] < df['leverage_prev'], 'f_score'] += 1
    # Delta Current Ratio > 0 (流动比率上升)
    df.loc[df['current_ratio_curr'] > df['current_ratio_prev'], 'f_score'] += 1
    # No New Equity (未增发新股: 当前股本 <= 去年股本)
    # 考虑到送转股等因素，严格相等比较难，这里放宽为股本没有显著增加
    df.loc[df['capitalization_curr'] <= df['capitalization_prev'] * 1.01, 'f_score'] += 1
    
    # --- 3. 运营效率 (Operating Efficiency) ---
    # Delta Gross Margin > 0 (毛利率上升)
    df.loc[df['gross_profit_margin_curr'] > df['gross_profit_margin_prev'], 'f_score'] += 1
    # Delta Asset Turnover > 0 (资产周转率上升)
    df.loc[df['asset_turnover_curr'] > df['asset_turnover_prev'], 'f_score'] += 1
    
    return df[['code', 'f_score']].set_index('code')

def do_rebalance(context, buy_list):
    """
    调仓函数
    """
    # 获取当前持仓
    current_positions = list(context.portfolio.positions.keys())
    
    # 卖出不在买入列表中的股票
    for stock in current_positions:
        if stock not in buy_list:
            order_target_value(stock, 0)
            
    # 买入新股票
    if len(buy_list) > 0:
        # 等权重分配资金
        position_count = len(buy_list)
        value_per_stock = context.portfolio.total_value / position_count
        
        for stock in buy_list:
            order_target_value(stock, value_per_stock)
            
    log.info("调仓完成，当前持仓数量: %d" % len(buy_list))

关键点说明

  1. 数据获取 (get_fundamentals)：

    • 策略使用了 get_fundamentals 获取财务数据。
    • 为了计算 F-Score 中的“同比变化”，我们分别获取了 current_date（当前）和 current_date - 365 days（一年前）的数据进行对比。
    • JoinQuant 的 get_fundamentals 默认返回指定日期能看到的最新财报数据，因此这种写法避免了未来函数。

  2. 指标计算细节：

    • ROA (资产回报率)：直接使用 indicator.roa。
    • CFO (经营现金流)：使用 cash_flow.net_operate_cash_flow。
    • Accruals (应计项目)：比较经营现金流和净利润，现金流大于净利润通常意味着收益质量更高。
    • Leverage (杠杆)：使用 long_term_borrowing / total_assets 近似计算。
    • 新股增发：比较 capitalization (总股本)。

  3. 选股逻辑：

    • 先筛选低估值（低 P/B），这是 Piotroski 策略的核心前提，即在“垃圾股”或“被低估股”中寻找基本面改善的标的。
    • 然后计算 F-Score，只买入得分高的（$\ge$ 7分）。

  4. 风险控制：

    • 剔除了 ST 股和停牌股。
    • 剔除了上市不满一年的次新股（数据可能不稳定）。
    • 如果选出的股票太多，代码中增加了一个按市值从小到大排序取前 N 只的逻辑（小市值因子通常在 A 股有效）。

常见问题 Q&A

Q: 为什么使用市净率 (P/B) 进行初筛？
A: Piotroski F-Score 最初的设计目的就是为了区分“真正有价值的低估值股票”和“即将破产的垃圾股”。如果不先限制在低 P/B 股票池中，F-Score 的超额收益效果通常会减弱。

Q: 策略的回测频率为什么是按月？
A: 财务数据通常按季度更新，按月调仓可以及时捕捉到新发布的财报数据。如果希望降低换手率，可以改为按季度调仓（如 run_monthly(trade, 1) 改为每3个月运行一次）。

Q: 如何处理财报数据的滞后性？
A: 代码中使用了 get_fundamentals(..., date=current_date)。在回测中，聚宽平台会自动匹配该日期之前已发布的最新财报，确保不会用到未来数据（例如在4月1日只能看到去年年报或今年一季报，看不到未来的数据）。

相关推荐

聚宽回测技巧：通过 order_volume_ratio 模拟真实成交量限制

学习如何使用 JoinQuant 的 set_option('order_volume_ratio') 函数，根据市场实际成交量限制订单成交，避免策略高估收益，让您的量化回测结果更

聚宽模拟交易高级技巧：处理重启与代码更新

深入理解 JoinQuant 模拟交易的生命周期。本指南讲解了如何在模拟盘每日重启时使用 `process_initialize` 初始化非持久化对象，以及在更新代码后使用 `after_code_c...

聚宽ATR 移动止损策略实现

使用 jqlib.technical_analysis 库中的 ATR 函数计算平均真实波幅，并结合当前价格构建动态的吊灯止损（Chandelier Exit）策略。

聚宽交易分析：get_trades 获取成交记录详解

订单委托后，实际成交情况如何？本教程教你使用 get_trades 函数获取每一笔成交的时间、价格、数量等详细信息，帮助你进行精准的交易成本分析和策略复盘。

聚宽因子投资：获取并使用因子数据

通过 get_factor_values 函数，轻松获取 JoinQuant 因子库中的数百个因子数据，如估值、成长、质量因子。本教程将指导你如何利用这些数据构建一个简单的多因子选股策略。

聚宽事件驱动编程：on_event 函数处理分红与强平

学习 JoinQuant 中的事件驱动编程模式。本指南将介绍如何通过实现 `on_event` 回调函数，来处理持仓股票发生的分红送股（DividendsEvent）和强行平仓（ForcedLiqui...