中金公司《高频因子手册》分享5：流动性因子，量价相关性因子
上一篇文章我们介绍了高频因子的高阶特征因子，这一篇继续介绍流动性因子、量价相关性因子，并在因子分析的基础上加入策略回测。
研究环境利用聚宽因子分析API，构建因子函数类；研究在日内高频分钟级数据中挖掘构建高频因子，对该因子进行有效性检验，并利用回测平台进行回测。
一 流动性因子
1.1 因子介绍
第四大类因子为流动性因子。流动性刻画股票交易所需要的时间和成本，一般来说，流动性较差的个股通常有更高的预期收益，这是对流动性风险的风险补偿。因此，流动性因子通常表现为流动性越低，未来收益越高的特征（也会被称为为非流动性因子）。
本节利用价差宽度、价格深度等数据构建了 11 个高频流动性因子，并对其进行有效性检验。结合因子相关性和构建逻辑，我们大致将流动性因子分为两类：价格弹性因子和集合竞价因子。
- 价格弹性因子包括 Amihud 非流动性因子（liq_amihud_1min）、价差深度因子（如liq_avgDepthCct）和价格宽度因子（如 liq_spread）。在全市场范围内，liq_amihud_1min_o 和 liq_spread_std 相对其他流动性因子具有领先的有效性表现。在月度调仓频率下，liq_amihud_1min_o 因子的 ICIR 为 1.02，年化多空收益超过 33%，且该因子在中证 1000 和中证 500 的选股域中也表现良好。
- 集合竞价因子利用日内 tick 级别数据计算不同时间段内交易量占全体交易量的比例，能够反映流动性的分布情况，包括 liq_closevol，liq_openvol 和 liq_firstCallR 等因子。其中，全市场范围内，liq_firstCallR_o 和 liq_firstCallR_m 因子在周度调仓频率下具有不错的预测性，ICIR 均超过 0.70，且年化多空收益分别为 27.5%和 30.6%。沪深 300 选股域中，liq_closevol_z 因子在周度调仓频率下 ICIR 为 0.21，年化多头超额收益为 5.4%，领先于其他流动性因子。
总体来看，流动性因子整体表现较好，其中 liq_amihud_1min_o 因子在多个选股域中具有较为稳定且突出的表现，值得推荐。在大盘股中，也可以关注 liq_closevol_z 因子。
流动性因子之间的相关性统计如下表：
全市场范围内，liq_amihud_1min_o 因子具有领先的多空和多头收益，且单调性表现良好，分组年化超额收益组间区分度明晰。在常见因子相关性计算中，该因子与 Momentum_1M、DP、TURNOVER_1M、STD_1M 等因子的 IC 相关性位于 0.5 左右的水平。此外，集合竞价前的成交量（liq_closeprevol）和收盘前 3 分钟成交量（liq_closevol）相关因子在全市场中的多空收益表现良好，2021 年底出现了小幅回撤。
1.2 因子复现
下面我们选取一个具体因子来进行构建和回测，使用聚宽研究环境。选取因子为liq_amihud_1min_m，构建方式为日内分钟k线收益率的绝度值与成交额的比值，最后的后缀“_m”表式对调仓周期取当期均值。
使用因子分析API，在研究环境中调用以进行因子分析。API使用详情可参阅聚宽技术文档。
# 构建因子分析框架
from jqfactor import Factor,analyze_factor
from jqdata.apis import *
import datetime
import pandas as pd
from jqfactor import neutralize
from jqfactor import winsorize_med
from jqfactor import standardlize
# 构建高频因子计算函数
class liq_amihud_1min_m(Factor):
# 设置因子名称
name = 'liq_amihud_1min_m'
# 设置获取数据的时间窗口长度
max_window = 1
# 设置依赖的数据
dependencies = ['close'] #参考标的
# 计算因子的函数， 需要返回一个 pandas.Series, index 是股票代码，value 是因子值
def calc(self, data):
# 获取日期,股票池
date= data['close'].index[0] #获取(上一)交易日 实际上的逻辑是用上一交易日的数据在今天进行交易,时间戳是昨天的时间
security_list = data['close'].columns.tolist()
# 以下为获取更多数据
#获取分钟级数据,我们取前20日平均值作为当期的因子值
stock_bars = get_price(security_list,count =240 * 20,start_date=None, end_date=date,
frequency='1m', fields=['close','money'], skip_paused=False,
fq='none')
stock_bars['return'] = np.log(stock_bars['close']/stock_bars['close'].shift(1))
stock_rtn = stock_bars['return'].fillna(0)
stock_mon = stock_bars['money'].fillna(method='ffill')
#以日为单位迭代计算当日因子值
stock_rtn['time'] = [i.time() for i in stock_rtn.index]
stock_rtn.index = [i.date() for i in stock_rtn.index]
stock_rtn['day'] = stock_rtn.index
stock_mon['time'] = [i.time() for i in stock_mon.index]
stock_mon.index = [i.date() for i in stock_mon.index]
stock_mon['day'] = stock_mon.index
GroupBy = stock_rtn.groupby(by='day')
anti_moment = pd.DataFrame()
anti_moment_temp = pd.DataFrame()
for day, group in GroupBy:
group1 = group.drop(['day','time'], axis=1)
group2 = stock_mon.loc[stock_mon['day']==day].drop(['day','time'], axis=1)
temp = self.cal_factor(group1,group2)
anti_moment[day] = temp['factor']
anti_moment_temp['factor'] = anti_moment.apply(np.mean, axis=1)
return anti_moment_temp['factor']
## 计算因子值
def cal_factor(self,group1,group2):
stock_lq=abs(group1)/group2
fac_lq = pd.DataFrame()
fac_lq['factor']=stock_lq.fillna(0).mean()
return fac_lq
1.3 因子分析
- 引擎初始化
下面我们对上面构建的因子进行回测分析，从各个方面考察因子的有效性。为保持与研报的一致性，我们主要考察周度和月度调仓频率下的表现。
参数设置如下：
（1）测试时间：2024-01-01 至 2025-06-05；
（2）分位数：十分位数；
（3）调仓周期：5日、30日；
（4）仓位配置：等权配置；
（5）股票池：沪深300
start = datetime.datetime.today()
far = analyze_factor(
liq_amihud_1min_m(), start_date='2024-01-01', end_date='2025-06-25', universe='000300.XSHG',
quantiles=10,
periods=(5,30),
industry='jq_l1',
weight_method='avg',
use_real_price=True,
skip_paused=False,
max_loss=0.2
)
end = datetime.datetime.today()
print('程序耗时：', end - start)
- 因子IC分析
我们使用IC信息系数法对趋势强度因子进行检验。我们排除因子值大小的影响，使用RankIC来替代IC值进行分析。RankIC为当期因子值的排名与下一期因子值排名的相关性。
下面我们绘制因子5日、30日的RankIC时间序列图和月度均值图。
在 2024-01到2025-06这一年半中，5、30日IC均值分别为为0.013、0.021，说明因子值与下月收益率呈现一定的正相关性，即因子值越高，下月预期收益率越高。
我们考察5、30日因子的IR值，分别为0.133、0.121。可见，5日换仓和30日换仓因子稳定性相近，30日更好。
进一步的计算IC_IR值为：0.0977、0.173。因此，在单位风险下，30日换仓因子具有最佳的有效性。
二 量价相关性因子
2.1 因子介绍
第五大类因子为量价相关性因子。量价相关性因子刻画股票换手率与价格（或收益率）背离、同向程度。该类因子的有效性可从信息扩散理论的角度理解：信息扩散初期的特征应是量能领先于价格，此时量的变化主要由小部分理性交易者主导；而到信息扩散中期时，量价基本同步；到信息扩散末期时，更多成交可能来源于“羊群效应”的非理性行为。
我们将量价相关性因子在全市场、沪深 300、中证 500 和中证 1000 分别进行月度和周度调仓的有效性检验。根据因子间的相关性统计，图中所示的因子相互间均存在正相关性。从因子构建逻辑出发，大致将量价相关性因子分为两类：价量同步因子和领先滞后因子。
- 价量同步因子指的是价格（及其变化率）与成交量（及其变化率）的同期相关系数。corr_prv_std 和 corr_prvr_std 在全市场范围内表现较好，月度频率下，corr_prv_std 的IC 均值为-5.65%，ICIR 为 1.09；corr_prvr_std 的 IC 均值为-2.88%，ICIR 为 0.64。相比之下，它们在沪深 300 范围内表现均有所下降，corr_prv_std 的 ICIR 为 0.47，月度频率下年化多空收益率为 12.2%，说明该类因子在大市值股票中效果有所减弱。
- 领先滞后因子表示的是收益率与成交量（以及变化率）的异步相关系数，我们分别测试了领先成交量或领先价格的情况。其中，corr_pvl 的变体均值和标准差表现均较为有效，月度 ICIR 分别达 1.09 和 1.06，前额周度年化多空超 46%，后者多头月度超额达 8%。
综上所述，corr_pvl（分钟收盘价与领先成交量相关系数）和 corr_prv（分钟收盘价与成交量同步相关系数）总体表现较好，在多数选股范围内具有较强的选股能力，大盘股中有效性较有限。
量价相关性因子之间的相关性统计如下表：
全市场范围内， corr_pvl_m 和 corr_pv_m 因子多空收益表现较为领先，corr_pvl_std、corr_pv_std 和 corr_pv_m 因子多头收益表现显著跑赢基线，其中 corr_pv_m 因子自 2019 年起多头超额增长趋势放缓，2022 年出现明显回撤。上述因子分组年化超额收益均具有良好单调性。corr_pvl_std 因子与常见因子均不存在显著因子截面相关性及 IC 相关性。
2.2 因子复现
下面我们选取一个具体因子来进行构建和回测，使用聚宽研究环境。选取因子为corr_prv_m，构建方式为日内分钟k线收益率与成交量的相关系数，最后的后缀“_m”表式对调仓周期取当期均值。
使用因子分析API，在研究环境中调用以进行因子分析。API使用详情可参阅聚宽技术文档。
# 构建因子分析框架
from jqfactor import Factor,analyze_factor
from jqdata.apis import *
import datetime
import pandas as pd
from jqfactor import neutralize
from jqfactor import winsorize_med
from jqfactor import standardlize
# 构建高频因子计算函数
class corr_prv_m(Factor):
# 设置因子名称
name = 'corr_prv_m'
# 设置获取数据的时间窗口长度
max_window = 1
# 设置依赖的数据
dependencies = ['close'] #参考标的
# 计算因子的函数， 需要返回一个 pandas.Series, index 是股票代码，value 是因子值
def calc(self, data):
# 获取日期,股票池
date= data['close'].index[0] #获取(上一)交易日 实际上的逻辑是用上一交易日的数据在今天进行交易,时间戳是昨天的时间
security_list = data['close'].columns.tolist()
# 以下为获取更多数据
#获取分钟级数据,我们取前20日平均值作为当期的因子值
stock_bars = get_price(security_list,count =240 * 20,start_date=None, end_date=date,
frequency='1m', fields=['close','volume'], skip_paused=False,
fq='none')
stock_bars['return'] = np.log(stock_bars['close']/stock_bars['close'].shift(1))
stock_rtn = stock_bars['return'].fillna(0)
stock_mon = stock_bars['volume'].fillna(method='ffill')
#以日为单位迭代计算当日因子值
stock_rtn['time'] = [i.time() for i in stock_rtn.index]
stock_rtn.index = [i.date() for i in stock_rtn.index]
stock_rtn['day'] = stock_rtn.index
stock_mon['time'] = [i.time() for i in stock_mon.index]
stock_mon.index = [i.date() for i in stock_mon.index]
stock_mon['day'] = stock_mon.index
GroupBy = stock_rtn.groupby(by='day')
anti_moment = pd.DataFrame()
anti_moment_temp = pd.DataFrame()
for day, group in GroupBy:
group1 = group.drop(['day','time'], axis=1)
group2 = stock_mon.loc[stock_mon['day']==day].drop(['day','time'], axis=1)
temp = self.cal_factor(group1,group2)
anti_moment[day] = temp['factor']
anti_moment_temp['factor'] = anti_moment.apply(np.mean, axis=1)
return anti_moment_temp['factor']
## 计算因子值
def cal_factor(self,group1,group2):
stock_fa=group1.corrwith(group2)
fac_re = pd.DataFrame()
fac_re['factor']=stock_fa.fillna(0)
return fac_re
2.3 因子分析
- 引擎初始化
下面我们对上面构建的因子进行回测分析，从各个方面考察因子的有效性。为保持与研报的一致性，我们主要考察周度和月度调仓频率下的表现。
参数设置如下：
（1）测试时间：2024-01-01 至 2025-06-05；
（2）分位数：十分位数；
（3）调仓周期：5日、30日；
（4）仓位配置：等权配置；
（5）股票池：沪深300
start = datetime.datetime.today()
far = analyze_factor(
corr_prv_m(), start_date='2024-01-01', end_date='2025-06-25', universe='000300.XSHG',
quantiles=10,
periods=(5,30),
industry='jq_l1',
weight_method='avg',
use_real_price=True,
skip_paused=False,
max_loss=0.2
)
end = datetime.datetime.today()
print('程序耗时：', end - start)
- 因子IC分析
我们使用IC信息系数法对因子进行检验。我们排除因子值大小的影响，使用RankIC来替代IC值进行分析。RankIC为当期因子值的排名与下一期因子值排名的相关性。
下面我们绘制因子5日、30日的RankIC时间序列图和月度均值图。
在 2024-01到2025-06这一年半中，5、30日IC均值分别为为-0.003、-0.009，说明因子值与下月收益率呈现一定的负相关性，即因子值越低，下月预期收益率越高。我们直观的看下收益分层分布的情况。总的来看越靠后的收益越高，符合预期。
我们考察5、30日因子的IR值，分别为0.156、0.141。可见，5日换仓和30日换仓因子稳定性相近，30日更好。
进一步的计算IC_IR值为：-0.019、-0.063。因此，在单位风险下，30日换仓因子具有最佳的有效性。
三 策略回测
根据上述分析，我们构建因子的回测组合，用于考察因子实际的选股能力。这一环节我们使用真实费率。本节选量价相关性因子，流动性因子读者可自行测试。
股票池：沪深300，剔除 ST、停牌、上市时间小于 6 个月的股票
回测时间：2020-01-01 至 2025-05-31
调仓期：每30天
费率：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5元
选股步骤可分为以下三部分：
（1）在每个调仓日第一天计算因子值
（2）对因子值根据从小到大的顺序进行排序，并将其等分为 10 组
（3）每个调仓日选取第n组（前n*10%）股票池进行调仓交易
# 导入函数库
from jqdata import *
import datetime
import pandas as pd
import statsmodels.api as sm
from statsmodels import regression
from statsmodels.formula.api import ols
from jqfactor import Factor, calc_factors
from jqfactor import neutralize
from jqfactor import winsorize_med
from jqfactor import standardlize
# 初始化函数，设定基准等等
def initialize(context):
# 设定沪深300作为基准
set_benchmark('000300.XSHG')
# 开启动态复权模式(真实价格)
set_option('use_real_price', True)
# 输出内容到日志 log.info()
log.info('初始函数开始运行且全局只运行一次')
# 过滤掉order系列API产生的比error级别低的log
# log.set_level('order', 'error')
### 股票相关设定 ###
# 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
set_order_cost(OrderCost(close_tax=0.001, open_commission=0.00025, close_commission=0.00025, min_commission=5),type='stock')
## 运行函数：每月第一天开盘后进行调仓
run_monthly(market_open, monthday=1, time='open', reference_security='000300.XSHG')
#run_daily(market_open, time='open', reference_security='000300.XSHG')
### 变量初始化 ###
g.index = ['000300.XSHG']#['000852.XSHG','000906.XSHG']
g.N = 6
g.num = 1 #选取第num组
g.R = 20
g.timer = 1
'''
######################策略的交易逻辑######################
每周计算因子值， 并买入前 20 支股票
'''
# 每周或每月开盘运行一次， 按照上一交易日的因子值进行调仓
def market_open(context):
if g.timer % 10 == 1:
log.info('函数运行时间(market_open):'+str(context.current_dt.time()))
# 1. 定义计算因子的 universe，
# 建议使用与 benchmark 相同的指数，方便判断选股带来的 alpha
#universe = get_index_stocks('000300.XSHG')
security_list = get_qualified_stocks(context)
# 2. 获取因子值
# get_factor_values 有三个参数，context、因子列表、股票池，
# 返回值是一个 dict，key 是因子类的 name 属性，value 是 pandas.Series
# Series 的 index 是股票代码，value 是当前日期能看到的最新因子值
factor_values = get_factor_values(context, [corr_prv_m()], security_list)
# 3. 对因子做线性加权处理， 并将结果进行排序。您在这一步可以研究自己的因子权重模型来优化策略结果。
# 对因子做 rank 是因为不同的因子间由于量纲等原因无法直接相加，这是一种去量纲的方法。
#final_factor = vol_return1min.rank(ascending=False)
shape_skratio = factor_values['corr_prv_m']
# 4. 由因子确定每日持仓的股票列表：
# 采用因子值由大到小排名前 10% 股票作为目标持仓
stock_list, buy_stock_count = select_stock_universe(context, shape_skratio, g.num)
# 5. 根据股票列表进行调仓：
# 这里采取所有股票等额买入的方式，您可以使用自己的风险模型自由发挥个股的权重搭配
rebalance_position(context, stock_list, buy_stock_count)
g.timer += 1
'''
######################下面是策略中使用的因子######################
可以先使用因子分析功能生产出理想的因子， 再加入到策略中
因子分析：https://www.joinquant.com/algorithm/factor/list
'''
# 构建高频因子计算函数
class corr_prv_m(Factor):
# 设置因子名称
name = 'corr_prv_m'
# 设置获取数据的时间窗口长度
max_window = 1
# 设置依赖的数据
dependencies = ['close'] #参考标的
# 计算因子的函数， 需要返回一个 pandas.Series, index 是股票代码，value 是因子值
def calc(self, data):
# 获取日期,股票池
date= data['close'].index[0] #获取(上一)交易日 实际上的逻辑是用上一交易日的数据在今天进行交易,时间戳是昨天的时间
security_list = data['close'].columns.tolist()
# 以下为获取更多数据
#获取分钟级数据,我们取前20日平均值作为当期的因子值
stock_bars = get_price(security_list,count =240 * 20,start_date=None, end_date=date,
frequency='1m', fields=['close','volume'], skip_paused=False,
fq='none')
stock_bars['return'] = np.log(stock_bars['close']/stock_bars['close'].shift(1))
stock_rtn = stock_bars['return'].fillna(0)
stock_mon = stock_bars['volume'].fillna(method='ffill')
#以日为单位迭代计算当日因子值
stock_rtn['time'] = [i.time() for i in stock_rtn.index]
stock_rtn.index = [i.date() for i in stock_rtn.index]
stock_rtn['day'] = stock_rtn.index
stock_mon['time'] = [i.time() for i in stock_mon.index]
stock_mon.index = [i.date() for i in stock_mon.index]
stock_mon['day'] = stock_mon.index
GroupBy = stock_rtn.groupby(by='day')
anti_moment = pd.DataFrame()
anti_moment_temp = pd.DataFrame()
for day, group in GroupBy:
group1 = group.drop(['day','time'], axis=1)
group2 = stock_mon.loc[stock_mon['day']==day].drop(['day','time'], axis=1)
temp = self.cal_factor(group1,group2)
anti_moment[day] = temp['factor']
anti_moment_temp['factor'] = anti_moment.apply(np.mean, axis=1)
return anti_moment_temp['factor']
## 计算因子值
def cal_factor(self,group1,group2):
stock_fa=group1.corrwith(group2)
fac_re = pd.DataFrame()
fac_re['factor']=stock_fa.fillna(0)
return fac_re
"""
###################### 工具 ######################
"""
## 开盘前运行函数
def before_trading_start(context):
# 输出运行时间
log.info('函数运行时间(before_market_open)：'+str(context.current_dt.time()))
# 给微信发送消息（添加模拟交易，并绑定微信生效）
# send_message('美好的一天~')
## 收盘后运行函数
def after_trading_end(context):
log.info(str('函数运行时间(after_market_close):'+str(context.current_dt.time())))
#得到当天所有成交记录
trades = get_trades()
for _trade in trades.values():
log.info('成交记录：'+str(_trade))
log.info('一天结束')
log.info('##############################################################')
## 选股函数
def get_qualified_stocks(context):
#获取前一个交易日日期
previous_date = context.previous_date
#获取全体A股
stock_list = []
for index in g.index:
stock_list += get_index_stocks(index_symbol=index, date=previous_date)
#剔除ST
st_list = get_extras('is_st', stock_list, count=1, end_date=previous_date)
stock_list = [stock for stock in stock_list if not st_list[stock][0]]
#剔除上市不足N个月新股
stock_list_fn = []
for stock in stock_list:
start_date = get_security_info(stock).start_date
if start_date < previous_date - datetime.timedelta(days=g.N*30):
stock_list_fn.append(stock)
#剔除停牌股票
paused_series = get_price(stock_list_fn,end_date=context.current_dt,count=1,fields='paused')['paused'].iloc[0]
stock_list_fn = paused_series[paused_series==False].index.tolist()
return stock_list_fn
## 待选股票池
def select_stock_universe(context, factor_df, num):
length = int(len(factor_df.index) / 10)
stock_list = factor_df.index[length*(num-1):length*num]
return (stock_list.tolist(),length)
"""
调仓：
先卖出持仓中不在 stock_list 中的股票
再等价值买入 stock_list 中的股票
"""
## 调仓
def rebalance_position(context, buy_stocks, buy_stock_count):
# 现持仓的股票，如果不在“目标池”中，且未涨停，就卖出
if len(context.portfolio.positions)>0:
last_prices = history(1, '1m', 'close', security_list=context.portfolio.positions.keys())
for stock in context.portfolio.positions.keys():
if stock not in buy_stocks :
curr_data = get_current_data()
if last_prices[stock][-1] < curr_data[stock].high_limit:
order_target_value(stock, 0)
# 依次买入“目标池”中的股票
for stock in buy_stocks:
position_count = len(context.portfolio.positions)
curr_data = get_current_data()
last_prices = history(1, '1m', 'close', security_list=buy_stocks)
#if get_growth_rate(g.indexNew,10) >= 0.01:
if buy_stock_count > position_count:
value = context.portfolio.cash / (buy_stock_count - position_count)
if context.portfolio.positions[stock].total_amount == 0 and last_prices[stock][-1] < curr_data[stock].high_limit:
order_target_value(stock, value)
"""
# 策略中获取因子数据的函数
每日返回上一日的因子数据
详见 帮助-单因子分析
"""
def get_factor_values(context,factor_list, universe):
"""
输入： 因子、股票池
返回： 前一日的因子值
"""
# 取因子名称
factor_name = list(factor.name for factor in factor_list)
# 计算因子值
values = calc_factors(universe,
factor_list,
context.previous_date,
context.previous_date)
# 装入 dict
factor_dict = {i:values[i].iloc[0] for i in factor_name}
return factor_dict
沪深300股票池前10%策略收益-7.72%，低于同期基准收益-4.12%。后10%策略收益42.28%，高于同期基准收益47.83%。可以看出2021年以前差别不大，2021年以后第十组明显优于第一组。
四 总结
本篇我们介绍了流动性因子、量价相关性因子，并对因子有效性进行了分析以及策略回测。在所测时间范围内，因子具有显著的有效性，对收益率分层效果明显。
评论