问题描述
请帮我写一个 Piotroski F-Score 选股策略
解决方案
这是一个基于 Piotroski F-Score (F-Score) 模型的选股策略。
策略原理
Piotroski F-Score 通过 9 个财务指标对公司进行评分（满分 9 分），分数越高代表公司财务状况越健康。策略通常买入 7-9 分的股票。
9 个评分维度（每项满足得 1 分）：
- 盈利能力 (Profitability)
- 资产回报率 (ROA) > 0
- 经营现金流 (CFO) > 0
- ROA 同比增长 (Current ROA > Previous ROA)
- 应计项 (Accruals): CFO > 净利润
- 财务杠杆与流动性 (Leverage, Liquidity and Source of Funds)
- 长期杠杆比率同比下降 (Current Leverage < Previous Leverage)
- 流动比率同比上升 (Current Ratio > Previous Ratio)
- 未增发新股 (Current Shares <= Previous Shares)
- 注：QMT中判断增发较复杂，本策略简化为总股本未增加
- 运营效率 (Operating Efficiency)
- 毛利率同比上升 (Current Gross Margin > Previous Gross Margin)
- 资产周转率同比上升 (Current Asset Turnover > Previous Asset Turnover)
策略代码实现
该策略设定为每年 5 月第一个交易日（确保年报已披露）进行调仓，选取沪深 300 成分股中 F-Score $\ge$ 7 分的股票进行等权买入。
# -*- coding: gbk -*-
import pandas as pd
import numpy as np
import time
def init(ContextInfo):
# 设置资金账号
ContextInfo.accid = '600000248'
ContextInfo.set_account(ContextInfo.accid)
# 设置股票池为沪深300
ContextInfo.index_code = '000300.SH'
ContextInfo.set_universe([ContextInfo.index_code])
# 策略参数
ContextInfo.score_threshold = 7 # 买入分值门槛
ContextInfo.rebalance_month = 5 # 调仓月份（5月，确保年报出齐）
ContextInfo.holding_list = [] # 当前持仓列表
def handlebar(ContextInfo):
# 获取当前K线的时间
bar_time = ContextInfo.get_bar_timetag(ContextInfo.barpos)
current_date = timetag_to_datetime(bar_time, '%Y%m%d')
current_month = int(current_date[4:6])
# 仅在每年的指定月份的第一个交易日进行调仓判断
if not ContextInfo.is_new_bar() or current_month != ContextInfo.rebalance_month:
return
# 简单的逻辑：如果本月已经调仓过，则跳过（这里简化处理，实际可增加标志位）
# 这里假设每年5月的第一根K线触发
last_date = timetag_to_datetime(ContextInfo.get_bar_timetag(ContextInfo.barpos - 1), '%Y%m%d')
if last_date[4:6] == str(ContextInfo.rebalance_month).zfill(2):
return
print(f"开始执行选股逻辑，当前日期: {current_date}")
# 1. 获取股票池
stock_list = ContextInfo.get_stock_list_in_sector(ContextInfo.index_code)
# 2. 确定财报日期：比较去年年报(Y-1)和前年年报(Y-2)
current_year = int(current_date[:4])
report_date_curr = f"{current_year - 1}1231" # 去年年报
report_date_prev = f"{current_year - 2}1231" # 前年年报
# 3. 获取财务数据
# 需要的字段列表
# net_profit: 净利润, tot_assets: 总资产, net_cash_flows_oper_act: 经营现金流
# long_term_loans: 长期借款, total_current_assets: 流动资产, total_current_liability: 流动负债
# total_capital: 总股本, revenue: 营业收入, operating_cost: 营业成本
fields = [
'ASHAREINCOME.net_profit_incl_min_int_inc', # 净利润
'ASHAREBALANCESHEET.tot_assets', # 总资产
'ASHARECASHFLOW.net_cash_flows_oper_act', # 经营现金流
'ASHAREBALANCESHEET.long_term_loans', # 长期借款
'ASHAREBALANCESHEET.total_current_assets', # 流动资产
'ASHAREBALANCESHEET.total_current_liability',# 流动负债
'CAPITALSTRUCTURE.total_capital', # 总股本
'ASHAREINCOME.revenue', # 营业收入
'ASHAREINCOME.operating_cost' # 营业成本
]
# 获取两年的数据
# 注意：get_financial_data 返回的数据结构需要处理
# 这里我们分别获取两个时间点的数据
df_curr = get_fundamental_data(ContextInfo, stock_list, fields, report_date_curr)
df_prev = get_fundamental_data(ContextInfo, stock_list, fields, report_date_prev)
if df_curr.empty or df_prev.empty:
print("财务数据获取失败")
return
# 4. 计算 F-Score
selected_stocks = []
for stock in stock_list:
if stock not in df_curr.index or stock not in df_prev.index:
continue
try:
# 提取当前和过去的数据
curr = df_curr.loc[stock]
prev = df_prev.loc[stock]
# 数据清洗，处理NaN
if curr.isnull().any() or prev.isnull().any():
continue
score = 0
# --- 盈利能力 (Profitability) ---
# 1. ROA > 0
roa_curr = curr['ASHAREINCOME.net_profit_incl_min_int_inc'] / curr['ASHAREBALANCESHEET.tot_assets']
if roa_curr > 0: score += 1
# 2. CFO > 0
cfo_curr = curr['ASHARECASHFLOW.net_cash_flows_oper_act']
if cfo_curr > 0: score += 1
# 3. dROA > 0 (ROA同比增加)
roa_prev = prev['ASHAREINCOME.net_profit_incl_min_int_inc'] / prev['ASHAREBALANCESHEET.tot_assets']
if roa_curr > roa_prev: score += 1
# 4. Accrual: CFO > Net Income
if cfo_curr > curr['ASHAREINCOME.net_profit_incl_min_int_inc']: score += 1
# --- 杠杆与流动性 (Leverage & Liquidity) ---
# 5. dLeverage < 0 (长期杠杆比率下降)
# 长期借款 / 总资产
lev_curr = curr['ASHAREBALANCESHEET.long_term_loans'] / curr['ASHAREBALANCESHEET.tot_assets']
lev_prev = prev['ASHAREBALANCESHEET.long_term_loans'] / prev['ASHAREBALANCESHEET.tot_assets']
if lev_curr < lev_prev: score += 1
# 6. dCurrentRatio > 0 (流动比率上升)
# 流动资产 / 流动负债
cr_curr = curr['ASHAREBALANCESHEET.total_current_assets'] / curr['ASHAREBALANCESHEET.total_current_liability']
cr_prev = prev['ASHAREBALANCESHEET.total_current_assets'] / prev['ASHAREBALANCESHEET.total_current_liability']
if cr_curr > cr_prev: score += 1
# 7. No New Shares (未增发股份)
if curr['CAPITALSTRUCTURE.total_capital'] <= prev['CAPITALSTRUCTURE.total_capital']: score += 1
# --- 运营效率 (Operating Efficiency) ---
# 8. dGrossMargin > 0 (毛利率上升)
# (营收 - 成本) / 营收
if curr['ASHAREINCOME.revenue'] > 0 and prev['ASHAREINCOME.revenue'] > 0:
gm_curr = (curr['ASHAREINCOME.revenue'] - curr['ASHAREINCOME.operating_cost']) / curr['ASHAREINCOME.revenue']
gm_prev = (prev['ASHAREINCOME.revenue'] - prev['ASHAREINCOME.operating_cost']) / prev['ASHAREINCOME.revenue']
if gm_curr > gm_prev: score += 1
# 9. dAssetTurnover > 0 (资产周转率上升)
# 营收 / 总资产
at_curr = curr['ASHAREINCOME.revenue'] / curr['ASHAREBALANCESHEET.tot_assets']
at_prev = prev['ASHAREINCOME.revenue'] / prev['ASHAREBALANCESHEET.tot_assets']
if at_curr > at_prev: score += 1
# 筛选
if score >= ContextInfo.score_threshold:
selected_stocks.append(stock)
except Exception as e:
# print(f"Error calculating score for {stock}: {e}")
pass
print(f"选出股票数量: {len(selected_stocks)}")
# 5. 执行交易
adjust_position(ContextInfo, selected_stocks)
def get_fundamental_data(ContextInfo, stock_list, fields, report_date):
"""
辅助函数：获取指定报告期的财务数据并转为DataFrame
"""
# report_type='report_time' 确保获取的是指定财报期的数据（如20221231）
# 注意：get_financial_data 对于多股单时间点，返回的是 DataFrame (index=stock, columns=field)
df = ContextInfo.get_financial_data(fields, stock_list, report_date, report_date, report_type='report_time')
return df
def adjust_position(ContextInfo, target_stocks):
"""
辅助函数：调仓逻辑
"""
# 获取当前持仓
positions = ContextInfo.get_trade_detail_data(ContextInfo.accid, 'stock', 'position')
current_holdings = [pos.m_strInstrumentID for pos in positions if pos.m_nVolume > 0]
# 卖出不在目标列表中的股票
for stock in current_holdings:
if stock not in target_stocks:
passorder(24, 1101, ContextInfo.accid, stock, 5, -1, 1, ContextInfo) # 1代表全卖(QMT特定逻辑，或需手动计算数量)
# 更稳健的写法是查询持仓数量后卖出
# order_target_value(stock, 0, ContextInfo, ContextInfo.accid)
print(f"卖出: {stock}")
# 买入目标股票
if len(target_stocks) > 0:
# 简单的等权分配
# 注意：回测中可用 ContextInfo.capital，实盘需用 get_trade_detail_data 获取可用资金
# 这里演示使用 order_target_percent
target_weight = 1.0 / len(target_stocks)
for stock in target_stocks:
# 使用 order_target_percent 需要在回测模式下，实盘建议计算金额后用 order_value
order_target_percent(stock, target_weight, ContextInfo, ContextInfo.accid)
print(f"买入/调仓: {stock}, 目标权重: {target_weight:.2%}")
def timetag_to_datetime(timetag, format_str):
"""
时间戳转日期字符串
"""
import time
return time.strftime(format_str, time.localtime(timetag / 1000))
代码关键点解析
-
数据获取 (
get_financial_data):- 使用了
report_type='report_time'。这是为了确保我们对比的是严格的“去年年报”和“前年年报”。例如在 2023 年 5 月，我们对比 20221231 和 20211231 的数据。 - 如果使用默认的
announce_time，可能会因为不同公司发布财报时间不同而导致数据对齐困难。
- 使用了
-
调仓时间:
- 策略设定在 5 月。因为 A 股年报披露截止日是 4 月 30 日。在 5 月初进行计算可以保证所有公司去年的年报数据都已经披露，避免用到“未来数据”。
-
评分逻辑:
- 严格遵循 Piotroski 的 9 点计分法。
- ROA: 使用
net_profit_incl_min_int_inc(包含少数股东损益的净利润) 除以tot_assets。 - CFO: 使用
net_cash_flows_oper_act。 - 杠杆: 使用
long_term_loans(长期借款) 作为主要负债指标。
-
交易执行:
- 使用了
order_target_percent进行等权买入。请注意，该函数主要用于回测。如果是实盘，建议先查询账户总资产，计算出单股金额，然后使用order_value或passorder。
- 使用了
Q&A
Q: 为什么选择在 5 月份调仓？
A: A 股上市公司必须在 4 月 30 日前披露上一年度的年报。如果在 1 月或 3 月运行此策略，很多公司的新年报还没出来，会导致数据缺失或使用了旧数据。
Q: 如何在实盘中使用此策略？
A: 实盘中 order_target_percent 可能无法直接获取准确的总资产（取决于 QMT 版本和券商接口）。建议在 adjust_position 函数中，使用 ContextInfo.get_trade_detail_data 获取账户的总资产 (m_dBalance)，然后手动计算每只股票应该买入的金额，再调用 order_value。
Q: 为什么有些股票数据获取不到？
A: 银行、保险等金融类企业的财务报表结构（如“营业成本”、“流动资产”）与普通企业不同，可能导致部分指标计算为 NaN。代码中包含了 try...except 和 isnull() 检查来跳过这些数据异常的股票。