中金公司《高频因子手册》分享6：筹码分布因子
上一篇文章我们介绍了高频因子的流动性因子、量价相关性因子，这一篇继续介绍筹码分布因子，并在因子分析的基础上加入策略回测。
研究环境利用聚宽因子分析API，构建因子函数类；研究在日内高频分钟级数据中挖掘构建高频因子，对该因子进行有效性检验，并利用回测平台进行回测。
一 筹码分布因子
1.1 因子介绍
第六大类因子为筹码分布因子。筹码分布旨在刻画股票持有人的持仓成本分布情况。筹码分布能够直观地展示不同价格区间上的持仓数量，从而帮助投资者判断市场的平均持仓成本。如果大部分筹码集中在较低的价格区间，说明市场的平均成本较低，未来股价上涨的压力可能较小；反之，如果筹码集中在较高的价格区间，上涨的阻力可能较大。
我们据此构建了筹码分布的标准差、偏度、峰度等因子，以及不同盈亏水平的筹码分布共 12 个因子，并检验了这些因子在不同选股范围内的有效性。doc_vol_pdf90_std、doc_vol_pdf95_std 和 doc_vol_pdf90bi_std 因子在四个选股域中均具有良好有效性。基于相关性统计，我们大致将筹码分布因子分为以下两类：筹码分布形状因子和筹码占比因子。
- 筹码分布形状因子包括描述分布的二到四阶标准矩相关因子，即标准差（doc_std）、偏度（doc_skew）与峰度（doc_kurt），该类因子的有效性整体弱于筹码分布占比因子。其中，doc_skew_std 因子相比其他筹码分布形状因子性能优势较为明显。在全市场范围内采用月度频率的调仓方式，其 IC 均值为-5.86%，ICIR 均值为 0.71；在沪深 300 中，其ICIR 表现下降为 0.29。
- 筹码占比因子包括收益率分组筹码头部占比（例如 doc_vol50_ratio）以及分组筹码占比分位数（例如 doc_vol_pdf90）两个子类。从预测方向来看，这类因子更多表现为反转的含义，且分组筹码占比分位数因子彼此间存在较高互相关性。从因子有效性角度来看，doc_vol_pdf90_std、doc_vol_pdf95_std 和 doc_vol_pdf80_std 三个因子在四个选股域中均具有相对领先的表现。以 doc_vol_pdf90_std 为例，在全市场范围内，其 ICIR 表现为1.52，年化多空收益超过 28%；在沪深 300 中，其 IC 均值为-5.6%，ICIR 降为 0.83。
1.2 因子表现
综合来看，筹码占比因子整体表现优于筹码分布形状因子。其中 doc_vol_pdf90_std、doc_vol_pdf95_std 和 doc_vol_pdf90bi_std 因子整体表现优异，在不同选股域中均有着较强的预测能力和良好的单调性，值得关注。
筹码分布因子之间的相关性统计如下表：
全市场范围内，筹码占比因子多空头收益表现较为突出。其中，doc_vol_pdf90bi_o 和doc_vol_pdf95_o 因子年化多空收益显著跑赢基线；doc_vol10_ratio_o 和 doc_vol5_ratio_o因子多头超额收益保持较快增速，其在 2015 年底出现小幅回撤。从单调性的角度来看，上述因子分组年化超额收益单调性良好，组间存在显著区分度。
二 因子复现
下面我们选取一个具体因子来进行构建和回测，使用聚宽研究环境。选取因子为doc_vol_pdf60_std，构建方式为计算分钟收益率分组筹码（以成交量计为筹码量）60%占比收益率分位，最后的后缀“_std”表式对调仓周期取当期标准差。
使用因子分析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 doc_vol_pdf60(Factor):
# 设置因子名称
name = 'doc_vol_pdf60'
# 设置获取数据的时间窗口长度
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')
chip_return = pd.DataFrame()
chip_return_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)
chip_return[day] = temp['factor']
chip_return_temp['factor'] = chip_return.apply(np.std, axis=1)
return chip_return_temp['factor']
## 计算因子值
def cal_factor(self,group1,group2):
chips = group1.rank()*group2
qtl = chips.quantile(0.6,interpolation = 'nearest')
qtlret = group1[chips.loc[:,qtl.index] == qtl]
chip_lq = pd.DataFrame()
chip_lq['factor'] = qtlret.fillna(0).mean()
return chip_lq
检查60%占比收益率分位计算如下
三 因子分析
3.1 引擎初始化
下面我们对上面构建的因子进行回测分析，从各个方面考察因子的有效性。为保持与研报的一致性，我们主要考察周度和月度调仓频率下的表现。
参数设置如下：
（1）测试时间：2024-01-01 至 2025-06-25；
（2）分位数：十分位数；
（3）调仓周期：5日、30日；
（4）仓位配置：等权配置；
（5）股票池：沪深300
start = datetime.datetime.today()
far = analyze_factor(
doc_vol_pdf60(), 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)
3.2 因子IC分析
我们使用IC信息系数法对趋势强度因子进行检验。我们排除因子值大小的影响，使用RankIC来替代IC值进行分析。RankIC为当期因子值的排名与下一期因子值排名的相关性。
下面我们绘制因子5日、30日的RankIC时间序列图和月度均值图。
在 2024-01到2025-06这一年半中，5、30日IC均值分别为为-0.014、-0.048，说明因子值与下月收益率呈现一定的负相关性，即因子值越低，下月预期收益率越高。我们直观的看下收益分层分布的情况。30日收益分层单调性较好，5日收益分层在中间部分单调性较差。
我们考察5、30日因子的IR值，分别为0.115、0.125。可见，5日换仓和30日换仓因子稳定性相近，5日更好。
进一步的计算IC_IR值为：-0.120、-0.384。因此，在单位风险下，30日换仓因子具有最佳的有效性。
四 策略回测
4.1 编辑策略
根据上述分析，我们构建因子的回测组合，用于考察因子实际的选股能力。这一环节我们使用真实费率。
股票池：沪深300，剔除 ST、停牌、上市时间小于 6 个月的股票
回测时间：2020-01-01 至 2025-05-31
调仓期：每30天
费率：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5元
选股步骤可分为以下三部分：
（1）在每个调仓日第一天计算因子值
（2）对因子值根据从小到大的顺序进行排序，并将其等分为 10 组
（3）每个调仓日选取第一组（前10%）股票池进行调仓交易
（4）作为对比，每个调仓日选取第10组（后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, [doc_vol_pdf60()], security_list)
# 3. 对因子做线性加权处理， 并将结果进行排序。您在这一步可以研究自己的因子权重模型来优化策略结果。
# 对因子做 rank 是因为不同的因子间由于量纲等原因无法直接相加，这是一种去量纲的方法。
#final_factor = vol_return1min.rank(ascending=False)
shape_skratio = factor_values['doc_vol_pdf60']
# 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 doc_vol_pdf60(Factor):
# 设置因子名称
name = 'doc_vol_pdf60'
# 设置获取数据的时间窗口长度
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')
chip_return = pd.DataFrame()
chip_return_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)
chip_return[day] = temp['factor']
chip_return_temp['factor'] = chip_return.apply(np.std, axis=1)
return chip_return_temp['factor']
## 计算因子值
def cal_factor(self,group1,group2):
chips = group1.rank()*group2
qtl = chips.quantile(0.6,interpolation = 'nearest')
qtlret = group1[chips.loc[:,qtl.index] == qtl]
chip_lq = pd.DataFrame()
chip_lq['factor'] = qtlret.fillna(0).mean()
return chip_lq
"""
###################### 工具 ######################
"""
## 开盘前运行函数
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
4.2 回测结果
沪深300股票池前10%策略收益40.25%，同期基准收益-6.26%，超额收益率49.62。后10%策略收益-10.41%，超额收益率-4.43%。多头组合整体上明显跑赢空头组合，分层效果较好。
五 总结
我们通过对所选取的筹码分布因子的有效性分析以及分层回测检验，初步得到以下结论:在 2024-01到2025-06中，因子值与下月收益率呈现一定的负相关性，我们考察因子的IC值及IR值，30日换仓因子相关性较好，且在单位风险下，30日换仓因子具有最佳的有效性。
继续构建策略选取前10%股票进行回测，选取五年数据。从分组回测来看，分层效果较好。多头组合整体上明显跑赢空头组合，符合单因子有效性的检验。
评论