因子合成与应用

本文讲解多因子合成方法，以及如何把验证通过的因子落地到期货策略中。

因子合成

本节沿用因子分析实操：以持仓量变化率为例中得到的 df，并假设已按同样方式计算出 oi_change 与 rsi 两个因子。

可以把单个因子想象成乐队里的独奏乐器：有人负责节奏，有人负责旋律，但没有人能独自演出一首完整的交响曲。单个因子往往只在某些市场阶段奏效，或只能从一个侧面描述资产。把多个因子合在一起，能互补短板、降低"跑调"风险，提升整体稳定性。通过加权平均、PCA、回归等方法，把多个因子转换成一个新的"合成因子"，类似把不同乐器的轨道用不同的方式混音成一条更具表达力的主旋律，再交给策略去下最终的交易指令。

多因子合成主要有三种方式：等权细分因子、利用 PCA 对高相关因子进行降维、利用逐步回归筛选细分因子。

以 PCA 合成因子为例，这里选择持仓量变化率以及 RSI 两个因子，在计算出两个因子后，导入所需要的库

import numpy as np
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler

提取因子值，进行 PCA 分析

# 准备PCA输入数据
# 提取所有因子值
factors = df[["oi_change", "rsi"]].values

# 标准化（重要：PCA对尺度敏感，且不同因子的量纲不同）
scaler = StandardScaler()
factors_scaled = scaler.fit_transform(factors)

# 应用PCA，提取第一主成分作为合成因子
pca = PCA(n_components=1)
pca_factor = pca.fit_transform(factors_scaled)

# 将合成因子添加到数据框
df["pca_factor"] = pca_factor.flatten()

# 检查PCA结果
print("PCA合成因子完成！")
print(f"\n合成因子统计信息：")
print(df["pca_factor"].describe())

# 显示PCA解释的方差比例和主成分系数
pca_full = PCA(n_components=2)
pca_full.fit(factors_scaled)
print(f"\n第一主成分解释的方差比例: {pca_full.explained_variance_ratio_[0]:.4f}")
print(f"第二主成分解释的方差比例: {pca_full.explained_variance_ratio_[1]:.4f}")
print(f"累计解释方差比例: {pca_full.explained_variance_ratio_.sum():.4f}")

准备 Alphalens 所需数据格式对合成后的因子进行分析，并生成报告。这样就完成了多因子的合成。

# 兼容多次运行：若此前已将 (date, asset) 设为索引，先还原为列
if "date" not in df.columns:
    try:
        df = df.reset_index()
    except Exception:
        pass

# 将日期转换为 datetime 格式
df["date"] = pd.to_datetime(df["date"]) 

# 先对 (date, asset) 去重聚合，避免重复索引导致 unstack 报错
df = (
    df.groupby(["date", "asset"], as_index=False)
      .agg({"close": "last", "pca_factor": "last"})
)

# 设置多级索引（date, asset），这是 alphalens 所要求的格式
df = df.set_index(["date", "asset"])

# 提取因子数据和价格数据
factor = df["pca_factor"]  # PCA合成因子
prices = df["close"].unstack("asset").ffill()  # 价格数据，转为宽表格式并前向填充缺失值

print("开始进行PCA合成因子分析...")
# 清洗因子数据并计算未来收益率
factor_data = alphalens.utils.get_clean_factor_and_forward_returns(
    factor,
    prices,
    quantiles=2,  # 将因子分为两组：高值组和低值组
    periods=(1, 5, 10),  # 分析 1 天、5 天和 10 天的预测能力
)

print("因子数据统计概览：")
print(factor_data.head())

print("生成PCA合成因子分析报告...")
# 创建因子分析摘要报告
alphalens.tears.create_summary_tear_sheet(factor_data)

从因子到策略

因子与策略是两个本质不同的事物，因子 是一个特征或信号，它说明资产在某些方面是"好"还是"坏"（例如，动量强还是弱），本身不构成买卖指令。策略是一套完整的规则，它利用一个或多个因子，并结合具体的仓位管理、风险控制等逻辑，最终生成买卖指令。

利用因子分析筛选出有效的因子后，可以将该因子应用到策略中。将因子计算方法保存到本地后，再在策略中进行读取，然后以新的数据进行因子值计算，以下是在策略中读取并使用因子的计算方法。

# 读取因子计算方法
rsi_fun = get_indicator_function("compute_factor.dill")
# 传入获取的K线，调用该方法计算因子值
rsi = rsi_fun(get_kline(symbol, period, 600))

单因子选择品种策略

在这我们简单实现一个单因子策略，策略逻辑为：每多少天/每多少根K线进行一次调仓，调仓逻辑为：根据因子值大小对品种进行排序，更换当前追踪的前N名的品种，新进来的进行开仓，被排出去的进行平仓。

需要注意的是：

根据因子是否正向，可以选择不同的开仓方向，我们在这个策略中默认因子是正向的因子。

因子还有着很多使用的方式，例如多因子可以将每个因子作为机器学习或深度学习的一个特征传入，得到最后的开平仓信号。

# ------------------------------------------#
# 文件类型:量化策略
# 帮助文档:https://qmfquant.com/static/doc/code/strategyEdit.html
# 期魔方，为您提供专业的量化服务
# ------------------------------------------#

"""
单因子调仓策略
策略说明：
1. 计算每个品种的因子值（持仓量变化率）
2. 根据因子值对品种进行排序
3. 选择因子值最优的N个品种进行持仓
4. 定期调仓：平掉不在持仓列表中的品种，开仓新的品种
"""

from pydantic import BaseModel, Field


class Params(BaseModel, validate_assignment=True):
    """策略外置参数配置"""

    QMF_SUBSCRIBE_SYMBOLS: str = Field(default="j2601:D1&ag2601:D1&rb2601:D1", title="订阅合约")
    volume: int = Field(default=1, title="每个品种交易手数")
    hold_count: int = Field(default=2, title="持仓品种数量")
    rebalance_period: int = Field(default=1, title="调仓周期(根K线)")


def on_init(context):
    """策略启动时运行一次，用于初始化全局变量、合约信息、订单管理等"""
    # 初始化全局变量
    context.all = {}
    context.symbol_list = []
    context.rebalance_counter = 0  # 调仓计数器
    context.current_hold_symbols = []  # 当前持仓的品种列表

    # 解析订阅合约
    symbol_periods = QMF_SUBSCRIBE_SYMBOLS.split("&")
    for symbol_period in symbol_periods:
        parts = symbol_period.split(":")
        if len(parts) == 2:
            symbol = parts[0]
            period = parts[1]
            if symbol not in context.symbol_list:
                context.symbol_list.append(symbol)

            # 初始化每个品种的数据结构
            context.all[symbol] = {
                "period": period,
                "PreOpenInterest": None,  # 上期持仓量
                "factor_value": 0.0,  # 因子值
                "position": 0,
            }


def on_tick(context):
    """每次报价时调用"""
    on_bar_run(on_bar, context)


def on_bar(context):
    """每根K线收线时调用"""
    # 获取K线数据
    get_k_line_data(context)

    # 计算因子值
    calc_factor(context)

    # 调仓逻辑
    rebalance_positions(context)


def get_k_line_data(context):
    """获取K线数据"""
    for symbol in context.all:
        period = context.all[symbol]["period"]
        # 获取K线数据（获取最近的数据用于计算）
        klines = get_kline(symbol, period, 600)
        if klines and len(klines.get("close", [])) > 0:
            context.all[symbol]["klines"] = klines
            # 获取tick数据
            tick_data = get_tick(symbol)
            if tick_data:
                context.all[symbol]["tick"] = tick_data


def calc_factor(context):
    """计算每个品种的因子值"""
    for symbol in context.all:
        if "tick" not in context.all[symbol] or "klines" not in context.all[symbol]:
            context.all[symbol]["factor_value"] = 0.0
            continue

        tick_data = context.all[symbol]["tick"]

        # 计算持仓量变化率
        prev_oi = float(tick_data.get("PreOpenInterest", 0))
        # put_log(f"前一日持仓量: {prev_oi}", level="USER_LOG")

        if context.all[symbol]["PreOpenInterest"] is not None and prev_oi > 0:
            # 持仓量变化率 = (前一日持仓量 - 上期持仓量) / 上期持仓量
            oi_change_rate = (prev_oi - context.all[symbol]["PreOpenInterest"]) / context.all[symbol]["PreOpenInterest"]
            context.all[symbol]["factor_value"] = oi_change_rate
        else:
            context.all[symbol]["factor_value"] = 0.0

        # 更新上期持仓量
        context.all[symbol]["PreOpenInterest"] = prev_oi


def rebalance_positions(context):
    """调仓逻辑：根据因子值选择最优品种进行持仓"""
    # 检查是否到达调仓周期
    context.rebalance_counter += 1
    if context.rebalance_counter < context.rebalance_period:
        return

    # 重置调仓计数器
    context.rebalance_counter = 0

    # 1. 计算所有品种的因子值并排序
    symbol_factor_list = []
    for symbol in context.all:
        factor_value = context.all[symbol].get("factor_value", 0.0)
        symbol_factor_list.append((symbol, factor_value))

    # 按因子值降序排序（因子值越大越好）
    symbol_factor_list.sort(key=lambda x: x[1], reverse=True)
    put_log(f"{time_current()} 因子值排序: {symbol_factor_list}", level="USER_LOG")

    # 2. 选择因子值最优的N个品种
    target_symbols = [item[0] for item in symbol_factor_list[: context.hold_count]]
    put_log(f"目标品种: {target_symbols}", level="USER_LOG")

    # 3. 找出需要平仓的品种（当前持仓但不在目标列表中的）
    symbols_to_close = [s for s in context.current_hold_symbols if s not in target_symbols]
    put_log(f"需要平仓品种: {symbols_to_close}", level="USER_LOG")

    # 4. 找出需要开仓的品种（在目标列表中但当前未持仓的）
    symbols_to_open = [
        s for s in target_symbols if (s not in context.current_hold_symbols) and (context.all[s]["factor_value"] != 0)
    ]
    put_log(f"需要开仓品种: {symbols_to_open}", level="USER_LOG")

    # 5. 执行平仓
    for symbol in symbols_to_close:
        # 检查是否有持仓
        position = context.all[symbol]["position"]

        if position > 0:
            # 平多仓
            sell_close(position, symbol)
            context.all[symbol]["position"] = 0
            put_log(f"调仓: 平多 {symbol} {position}手", level="USER_LOG")

    # 6. 执行开仓（统一开多仓，可根据策略需求修改）
    for symbol in symbols_to_open:
        # 开多仓
        buy_open(context.volume, symbol)
        context.all[symbol]["position"] += context.volume
        put_log(
            f"调仓: 开多 {symbol} {context.volume}手, 因子值: {context.all[symbol]['factor_value']:.4f}",
            level="USER_LOG",
        )

    # 7. 更新当前持仓品种列表
    if symbols_to_close or symbols_to_open:
        context.current_hold_symbols = target_symbols.copy()
        put_log(f"调仓完成，当前持仓品种: {context.current_hold_symbols}", level="USER_LOG")

        # 输出因子值排名
        factor_info = ", ".join([f"{s}({context.all[s]['factor_value']:.4f})" for s in target_symbols])
        put_log(f"因子值排名前{context.hold_count}名: {factor_info}", level="USER_LOG")


def on_order(context, order):
    """每次接收到报单时调用"""
    pass


def on_trade(context, trade):
    """每次接收到成交单时调用"""
    pass


def on_error(context, err):
    """策略运行出错时调用"""
    put_log(f"错误: {err}", level="ERROR")

---

相关阅读

• • 因子分析概述
• • 因子分析实操：以持仓量变化率为例