找回密码
立即注册
搜索
热搜: Java Python Linux Go
发回帖 发新帖

977

积分

0

好友

139

主题
发表于 前天 11:44 | 查看: 3| 回复: 0

多因子模型是量化投资领域最广泛应用的策略之一。经典的CAPM模型为股票收益提供了基础解释,但市场异象的不断涌现揭示了超越市场因子的超额收益空间,这催生了更为复杂的多因子模型。Fama-French的三因子和五因子模型是其里程碑,后续研究者在此基础上不断扩展,形成了Carhart四因子、Hou-Xue-Zhang四因子等多种模型。核心争议在于,究竟包含多少因子是“合适”的,目前尚无定论。

F-Score模型则另辟蹊径,它并非旨在解释收益来源,而是设计了一个由9个财务指标构成的评分体系,用于评估股票的基本面健康状况。得分越高,表明公司财务质量越优质。这9个因子均可基于上市公司财报数据计算得出,具体清单与计算规则如下:

图片

模型规则简明:每个因子满足条件则得1分,否则得0分。累计满足N个条件即得N分,因此最高评分为9分。下面我们来详细解读每个因子的经济含义:

  • 资产收益率(ROA):衡量企业扣除非经常性损益后的真实盈利能力(如出售资产、投资收益等被剔除),大于零表示企业主营业务盈利。
  • 经营活动现金流/总资产:评估企业经营活动带来真实现金流入的能力。利润可能包含应收账款,此指标大于零意味着企业有真金白银的利润流入。
  • 资产收益率变化:反映企业盈利能力的变化趋势,大于零表示盈利加速,处于扩张期。
  • 应计收益率:该指标比现金流比率更进一步,通过对比经营现金流与会计利润,衡量盈利质量,大于零表示利润含金量高。
  • 长期负债率变化:衡量企业长期偿债能力的变化,下降(小于零)表示财务结构更稳健,长期偿债压力减轻。
  • 流动比率变化:反映企业短期变现能力的变化,上升(大于零)意味着短期偿债能力增强。
  • 股票是否增发:考察企业是否依赖股权再融资。未增发通常意味着企业能通过自身经营产生健康现金流,具备“造血”能力。
  • 毛利率变化:衡量企业产品竞争力与定价权。上升(大于零)可能意味着企业拥有护城河,能抵御竞争、维持或提升利润空间。
  • 资产周转率变化:评估企业营运效率与管理质量。上升(大于零)表明公司能更有效地利用资产创造收入。

那么,这套源自海外的模型在A股市场是否有效呢?华创金工曾进行过分组回测,将得分0-3、4-6、7-9的股票分别归为Low、Middle、High组,结果显示净值曲线分层明显,证明了模型具备基本面区分能力。

对于个人投资者,或许不需要投资一篮子股票,可以直接筛选出F-Score满分的“优等生”。基于这一朴素思想,我们使用Python和AKShare数据包实现一个完整的策略进行回测:每月初调仓,等权重投资所有得分为9的股票。

import akshare as ak
import pandas as pd
import numpy as np
import datetime
from datetime import timedelta
import matplotlib.pyplot as plt

# 设置中文显示
plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]

class FScoreStrategy:
    def __init__(self, start_date, end_date, initial_capital=1000000):
        """初始化策略参数"""
        self.start_date = start_date
        self.end_date = end_date
        self.initial_capital = initial_capital
        self.positions = {}  # 持仓字典
        self.history = []    # 历史记录
        self.current_date = None

    def get_trade_dates(self):
        """获取回测期间的交易日历"""
        try:
            # 获取A股交易日历
            trade_dates = ak.tool_trade_date_hist_sina()
            trade_dates = pd.to_datetime(trade_dates)
            # 筛选回测期间的交易日
            mask = (trade_dates >= self.start_date) & (trade_dates <= self.end_date)
            return trade_dates[mask].tolist()
        except Exception as e:
            print(f"获取交易日历失败: {e}")
            return []

    def get_stock_list(self, date):
        """获取指定日期的股票列表(沪深300成分股作为样本)"""
        try:
            # 转换日期格式
            str_date = date.strftime("%Y%m%d")
            # 获取沪深300成分股
            stock_list = ak.index_stock_cons(symbol="000300")
            # 保留股票代码
            return stock_list["品种代码"].tolist()
        except Exception as e:
            print(f"获取股票列表失败: {e}")
            return []

    def get_financial_data(self, stock_code, end_date):
        """获取股票的财务数据"""
        try:
            # 转换日期格式
            str_end_date = end_date.strftime("%Y-%m-%d")
            # 获取利润表数据
            income_df = ak.stock_financial_report_sina(
                symbol=stock_code, 
                symbol_type="income", 
                report_type="yearly"
            )
            # 获取资产负债表数据
            balance_df = ak.stock_financial_report_sina(
                symbol=stock_code, 
                symbol_type="balance", 
                report_type="yearly"
            )
            # 获取现金流量表数据
            cashflow_df = ak.stock_financial_report_sina(
                symbol=stock_code, 
                symbol_type="cashflow", 
                report_type="yearly"
            )

            return {
                "income": income_df,
                "balance": balance_df,
                "cashflow": cashflow_df
            }
        except Exception as e:
            print(f"获取{stock_code}财务数据失败: {e}")
            return None

    def calculate_f_score(self, financial_data):
        """计算F-Score得分"""
        if not financial_data:
            return 0

        income = financial_data["income"]
        balance = financial_data["balance"]
        cashflow = financial_data["cashflow"]

        # 确保数据不为空
        if income.empty or balance.empty or cashflow.empty:
            return 0

        score = 0

        try:
            # 1. 资产收益率(ROA) > 0
            # 扣非净利润/平均总资产
            net_profit = float(income[income["项目"] == "归属于母公司所有者的净利润扣除非经常性损益后的净额"].iloc[0, 1])
            total_assets = float(balance[balance["项目"] == "资产总计"].iloc[0, 1])
            if total_assets != 0 and net_profit / total_assets > 0:
                score += 1

            # 2. 经营活动产生的现金流量净额/总资产 > 0
            ocf = float(cashflow[cashflow["项目"] == "经营活动产生的现金流量净额"].iloc[0, 1])
            if total_assets != 0 and ocf / total_assets > 0:
                score += 1

            # 3. 资产收益率变化 > 0
            if len(income) > 1:
                prev_net_profit = float(income[income["项目"] == "归属于母公司所有者的净利润扣除非经常性损益后的净额"].iloc[0, 2])
                prev_total_assets = float(balance[balance["项目"] == "资产总计"].iloc[0, 2])
                if total_assets != 0 and prev_total_assets != 0:
                    current_roa = net_profit / total_assets
                    prev_roa = prev_net_profit / prev_total_assets
                    if current_roa - prev_roa > 0:
                        score += 1

            # 4. 应计收益率 > 0 (经营活动现金流/总资产 - ROA)
            if total_assets != 0:
                current_roa = net_profit / total_assets if total_assets != 0 else 0
                ocfoa = ocf / total_assets
                if ocfoa - current_roa > 0:
                    score += 1

            # 5. 长期负债率变化 < 0
            if len(balance) > 1:
                non_current_liability = float(balance[balance["项目"] == "非流动负债合计"].iloc[0, 1])
                prev_non_current_liability = float(balance[balance["项目"] == "非流动负债合计"].iloc[0, 2])

                current_ltd_ratio = non_current_liability / total_assets if total_assets != 0 else 0
                prev_ltd_ratio = prev_non_current_liability / prev_total_assets if prev_total_assets != 0 else 0

                if current_ltd_ratio - prev_ltd_ratio < 0:
                    score += 1

            # 6. 流动比率变化 > 0
            if len(balance) > 1:
                current_assets = float(balance[balance["项目"] == "流动资产合计"].iloc[0, 1])
                current_liability = float(balance[balance["项目"] == "流动负债合计"].iloc[0, 1])
                prev_current_assets = float(balance[balance["项目"] == "流动资产合计"].iloc[0, 2])
                prev_current_liability = float(balance[balance["项目"] == "流动负债合计"].iloc[0, 2])

                current_cr = current_assets / current_liability if current_liability != 0 else 0
                prev_cr = prev_current_assets / prev_current_liability if prev_current_liability != 0 else 0

                if current_cr - prev_cr > 0:
                    score += 1

            # 7. 未进行股票增发
            # 这里简化处理,实际需要查询增发记录
            score += 1 # 假设未增发

            # 8. 毛利率变化 > 0
            if len(income) > 1:
                operating_revenue = float(income[income["项目"] == "营业收入"].iloc[0, 1])
                operating_cost = float(income[income["项目"] == "营业成本"].iloc[0, 1])
                prev_operating_revenue = float(income[income["项目"] == "营业收入"].iloc[0, 2])
                prev_operating_cost = float(income[income["项目"] == "营业成本"].iloc[0, 2])

                current_gpm = (operating_revenue - operating_cost) / operating_revenue if operating_revenue != 0 else 0
                prev_gpm = (prev_operating_revenue - prev_operating_cost) / prev_operating_revenue if prev_operating_revenue != 0 else 0

                if current_gpm - prev_gpm > 0:
                    score += 1

            # 9. 资产周转率变化 > 0
            if len(income) > 1 and len(balance) > 1:
                current_tat = operating_revenue / total_assets if total_assets != 0 else 0
                prev_tat = prev_operating_revenue / prev_total_assets if prev_total_assets != 0 else 0

                if current_tat - prev_tat > 0:
                    score += 1

        except Exception as e:
            print(f"计算F-Score失败: {e}")

        return score

    def get_stock_price(self, stock_code, date):
        """获取股票在指定日期的价格"""
        try:
            str_date = date.strftime("%Y%m%d")
            # 获取股票行情数据
            df = ak.stock_zh_a_daily(symbol=stock_code, start_date="20000101", end_date=str_date)
            # 找到最近的价格
            df["date"] = pd.to_datetime(df["date"])
            df = df[df["date"] <= date]
            if not df.empty:
                return df.iloc[-1]["close"]
            return None
        except Exception as e:
            print(f"获取{stock_code}价格失败: {e}")
            return None

    def rebalance(self, date):
        """调仓逻辑:筛选F-Score=9的股票"""
        self.current_date = date
        print(f"开始调仓: {date.strftime('%Y-%m-%d')}")

        # 获取股票列表
        stock_list = self.get_stock_list(date)
        if not stock_list:
            return

        # 计算每只股票的F-Score
        high_score_stocks = []
        for code in stock_list[:50]: # 为了速度,只取前50只股票
            financial_data = self.get_financial_data(code, date)
            f_score = self.calculate_f_score(financial_data)
            if f_score == 9:
                high_score_stocks.append(code)
                print(f"{code} F-Score=9,加入持仓")

        # 获取当前总资产
        total_assets = self.initial_capital
        for stock, pos in self.positions.items():
            price = self.get_stock_price(stock, date)
            if price:
                total_assets += pos * price

        # 等权重分配资金
        if high_score_stocks:
            weight = 1.0 / len(high_score_stocks)
            for stock in high_score_stocks:
                target_value = total_assets * weight
                price = self.get_stock_price(stock, date)
                if price:
                    target_shares = int(target_value / price)
                    self.positions[stock] = target_shares

        # 记录当前状态
        self.history.append({
            "date": date,
            "positions": self.positions.copy(),
            "total_assets": total_assets
        })

        print(f"调仓完成,持仓股票数量: {len(self.positions)}")

    def run_backtest(self):
        """运行回测"""
        print("开始回测...")
        trade_dates = self.get_trade_dates()
        if not trade_dates:
            print("没有可用的交易日数据")
            return

        # 每月第一个交易日调仓
        rebalance_dates = []
        prev_month = None
        for date in trade_dates:
            if date.month != prev_month:
                rebalance_dates.append(date)
                prev_month = date.month

        # 执行调仓
        for date in rebalance_dates:
            self.rebalance(date)

        # 绘制回测结果
        self.plot_results()

    def plot_results(self):
        """绘制回测结果"""
        if not self.history:
            print("没有回测数据可绘制")
            return

        dates = [item["date"] for item in self.history]
        values = [item["total_assets"] for item in self.history]

        plt.figure(figsize=(12, 6))
        plt.plot(dates, values, label="策略净值")
        plt.axhline(y=self.initial_capital, color='r', linestyle='--', label="初始资金")
        plt.title("F-Score策略回测结果")
        plt.xlabel("日期")
        plt.ylabel("总资产(元)")
        plt.legend()
        plt.grid(True)
        plt.show()

if __name__ == "__main__":
    # 设置回测时间范围
    start = datetime.datetime(2018, 1, 1)
    end = datetime.datetime(2023, 1, 1)

    # 初始化并运行策略
    strategy = FScoreStrategy(start, end)
    strategy.run_backtest()

策略逻辑为每月第一个交易日调仓一次,筛选F-Score得分为9的股票进行等权重配置,最后绘制策略净值曲线与初始资金对比。

附1:F-Score因子计算表

因子序号 因子名称 计算方式/判断条件
1 资产收益率(ROA) 扣非净利润 ÷ 平均总资产 > 0
2 经营活动现金流/总资产 经营活动产生的现金流量净额 ÷ 总资产 > 0
3 资产收益率变化 当期ROA - 上期ROA > 0
4 应计收益率 (经营活动现金流/总资产) - ROA > 0
5 长期负债率变化 (当期非流动负债/总资产) - (上期非流动负债/总资产) < 0
6 流动比率变化 (当期流动资产/流动负债) - (上期流动资产/流动负债) > 0
7 股票是否增发 近1年未进行股票增发(通过增发记录查询)
8 毛利率变化 (当期毛利率) - (上期毛利率) > 0 (毛利率=(营业收入-营业成本)÷营业收入)
9 资产周转率变化 (当期总资产周转率) - (上期总资产周转率) > 0 (总资产周转率=营业收入÷总资产)

附2:单只股票F-Score计算示例

import akshare as ak
import pandas as pd

def calculate_f_score(stock_code, end_date):
    # 获取利润表、资产负债表、现金流量表(以akshare为例)
    income = ak.stock_financial_report_sina(symbol=stock_code, symbol_type="income", report_type="yearly")
    balance = ak.stock_financial_report_sina(symbol=stock_code, symbol_type="balance", report_type="yearly")
    cashflow = ak.stock_financial_report_sina(symbol=stock_code, symbol_type="cashflow", report_type="yearly")

    score = 0
    # 因子1:ROA > 0
    net_profit = income[income["项目"] == "归属于母公司所有者的净利润扣除非经常性损益后的净额"].iloc[0, 1]
    total_assets = balance[balance["项目"] == "资产总计"].iloc[0, 1]
    if (net_profit / total_assets) > 0:
        score += 1

    # 因子2:经营活动现金流/总资产 > 0
    ocf = cashflow[cashflow["项目"] == "经营活动产生的现金流量净额"].iloc[0, 1]
    if (ocf / total_assets) > 0:
        score += 1

    # 因子3:ROA变化 > 0(需对比上期数据)
    if len(income) > 1 and len(balance) > 1:
        prev_net_profit = income[income["项目"] == "归属于母公司所有者的净利润扣除非经常性损益后的净额"].iloc[0, 2]
        prev_total_assets = balance[balance["项目"] == "资产总计"].iloc[0, 2]
        current_roa = net_profit / total_assets
        prev_roa = prev_net_profit / prev_total_assets
        if (current_roa - prev_roa) > 0:
            score += 1

    # 因子4:应计收益率 > 0
    if (ocf / total_assets) - (net_profit / total_assets) > 0:
        score += 1

    # 因子5:长期负债率变化 < 0
    non_current_liability = balance[balance["项目"] == "非流动负债合计"].iloc[0, 1]
    prev_non_current_liability = balance[balance["项目"] == "非流动负债合计"].iloc[0, 2]
    current_ltd = non_current_liability / total_assets
    prev_ltd = prev_non_current_liability / prev_total_assets
    if (current_ltd - prev_ltd) < 0:
        score += 1

    # 因子6:流动比率变化 > 0
    current_assets = balance[balance["项目"] == "流动资产合计"].iloc[0, 1]
    current_liability = balance[balance["项目"] == "流动负债合计"].iloc[0, 1]
    prev_current_assets = balance[balance["项目"] == "流动资产合计"].iloc[0, 2]
    prev_current_liability = balance[balance["项目"] == "流动负债合计"].iloc[0, 2]
    current_cr = current_assets / current_liability if current_liability != 0 else 0
    prev_cr = prev_current_assets / prev_current_liability if prev_current_liability != 0 else 0
    if (current_cr - prev_cr) > 0:
        score += 1

    # 因子7:未增发(简化处理,实际需查增发记录)
    score += 1 # 假设未增发

    # 因子8:毛利率变化 > 0
    operating_rev = income[income["项目"] == "营业收入"].iloc[0, 1]
    operating_cost = income[income["项目"] == "营业成本"].iloc[0, 1]
    prev_operating_rev = income[income["项目"] == "营业收入"].iloc[0, 2]
    prev_operating_cost = income[income["项目"] == "营业成本"].iloc[0, 2]
    current_gpm = (operating_rev - operating_cost) / operating_rev if operating_rev != 0 else 0
    prev_gpm = (prev_operating_rev - prev_operating_cost) / prev_operating_rev if prev_operating_rev != 0 else 0
    if (current_gpm - prev_gpm) > 0:
        score += 1

    # 因子9:资产周转率变化 > 0
    current_tat = operating_rev / total_assets if total_assets != 0 else 0
    prev_tat = prev_operating_rev / prev_total_assets if prev_total_assets != 0 else 0
    if (current_tat - prev_tat) > 0:
        score += 1

    return score

附3:阈值选择与动态调整

原始研究采用“三分法”:0-3分为Low组(基本面差),4-6分为Middle组(基本面一般),7-9分为High组(基本面优质)。追求安全边际可直接选择High组作为股票池。

实际操作中,阈值可灵活调整:

  • 牛市/高估值:提高阈值(如8-9分),严控质量以规避泡沫。
  • 熊市/低估值:可适当放宽(如6-9分),捕捉更多可能被错杀的标的。
  • 行业特性:强周期行业(地产、能源)适用更严阈值(7-9分);弱周期行业(消费、医药)可适当放宽。

最科学的方式是基于历史回测寻找最优阈值,平衡收益与标的数量。

附4:因子权重优化思路

默认F-Score模型采用等权重。若想进一步提升策略表现,可探索加权方案,核心是让对收益预测能力更强的因子占据更高权重。

  1. 因子收益率回归法:以历史数据做回归,用因子得分预测未来收益,回归系数的大小可反映因子影响力,归一化后作为权重。
  2. 因子IC值法:计算每个因子的信息系数(IC),即因子得分与下期收益的相关系数。IC绝对值越高,预测能力越强,相应权重应更高。
  3. 机器学习优化法:使用线性回归、随机森林等算法,以因子为特征、未来收益为目标进行训练,自动学习最优权重组合,但需警惕过拟合。

对于初学者,建议从等权重开始,策略更稳定。调整权重后,务必通过严谨的样本外回测验证其有效性。

附5:策略表现评估框架

优化权重后,需通过回测系统评估其表现,核心是验证“新权重能否提升策略的收益风险比”。

评估步骤

  1. 设定基准:等权重模型、市场指数(如沪深300)。
  2. 关注核心指标
    • 收益能力:年化收益率、超额收益。
    • 风险控制:最大回撤、年化波动率。
    • 风险收益比夏普比率(核心),衡量每承担一单位风险获得的超额收益。
    • 稳定性:胜率(盈利月份占比)。
  3. 样本外验证(关键):将历史数据分为训练集和测试集。在训练集上优化得到的权重,必须在测试集上表现依然稳健(如夏普比率下降不明显),否则可能存在过拟合。
  4. 敏感性分析:微调权重,观察策略关键指标的变化幅度,变化越小说明权重越稳健。

示例结论:若回测发现,优化权重A相比等权重模型,在提升年化收益的同时保持了较高的夏普比率,且在样本外测试中表现稳定,则该权重调整可能是有效的。




上一篇:BlockLever集成测试实战:PancakeSwapV3 QuoterV3只读报价函数适配与修复
下一篇:Golang面试核心:中间件实战与系统设计要点,突破高并发场景
您需要登录后才可以回帖 登录 | 立即注册

手机版|小黑屋|网站地图|云栈社区 ( 苏ICP备2022046150号-2 )

GMT+8, 2025-12-17 08:41 , Processed in 0.167347 second(s), 39 queries , Gzip On.

Powered by Discuz! X3.5

© 2025-2025 云栈社区.

快速回复 返回顶部 返回列表