3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在量化多因子选股策略中,我们通常在每个调仓期(如每日或每周)选择因子得分最高的 $N$ 只股票买入。然而,在日频交易中,由于因子值的微小波动,处于排名边缘的股票(例如第 $N$ 名和第 $N+1$ 名)会频繁发生排名交替,导致策略频繁买入卖出。这种高换手率会带来高额的交易成本(印花税、佣金及滑点),严重侵蚀策略的实际净值。
**退出缓冲区(Buffer Zone)**是一种经典的降低换手率的方法。其核心思想是:
在聚宽平台中,我们可以利用 initialize 初始化策略,并在 run_daily 定时任务中,通过 get_factor_values 获取因子值,计算排名,并结合缓冲区逻辑进行调仓。
get_factor_values 获取目标股票池的因子值。以下是一个在聚宽平台运行的完整多因子缓冲区策略示例。我们以“Skewness60(个股收益的60日偏度)”因子为例,设置买入前 15 名,退出缓冲区为前 30 名。
import pandas as pd
import numpy as np
from jqdata import *
from jqfactor import get_factor_values
def initialize(context):
# 设定沪深300作为基准
set_benchmark('000300.XSHG')
# 开启动态复权模式(真实价格)
set_option('use_real_price', True)
# 股票类交易手续费:买入佣金万三,卖出佣金万三加千一印花税,最低消费5元
set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
# 策略参数设置
g.factor_name = 'Skewness60' # 使用的因子
g.N = 15 # 买入目标数量(严格买入线)
g.M = 30 # 退出缓冲区边界(宽松持有线,必须大于N)
# 每天 09:30 运行调仓
run_daily(market_open, time='09:30')
def market_open(context):
# 1. 获取沪深300成分股作为初始股票池
stocks = get_index_stocks('000300.XSHG')
# 2. 获取当前的因子值
yesterday = context.previous_date
factor_data = get_factor_values(
securities=stocks,
factors=[g.factor_name],
start_date=yesterday,
end_date=yesterday
)[g.factor_name]
if factor_data.empty:
return
# 提取昨天的因子值并降序排序(假设因子值越大越好)
current_factors = factor_data.loc[pd.to_datetime(yesterday)].dropna()
sorted_stocks = current_factors.sort_values(ascending=False).index.tolist()
# 3. 划分买入线与持有线
buy_list = sorted_stocks[:g.N] # 前 N 名,严格买入名单
keep_list = sorted_stocks[:g.M] # 前 M 名,允许持有名单
# 4. 获取当前持仓
current_positions = list(context.portfolio.positions.keys())
# 5. 卖出逻辑:如果持仓股跌出了前 M 名(缓冲区外),则执行卖出
for stock in current_positions:
if stock not in keep_list:
order_target(stock, 0)
log.info(f"股票 {stock} 跌出前 {g.M} 名缓冲区,执行卖出。")
# 重新获取卖出后的最新持仓
current_positions = list(context.portfolio.positions.keys())
# 6. 买入逻辑:如果持仓数量未达到 N,且有资金,从前 N 名中买入未持有的股票
vacant_slots = g.N - len(current_positions)
if vacant_slots > 0:
# 筛选出在前 N 名中,且当前未持有的股票
to_buy_candidates = [s for s in buy_list if s not in current_positions]
# 确定本次实际买入的股票
to_buy = to_buy_candidates[:vacant_slots]
if to_buy:
# 均分可用资金
cash_per_stock = context.portfolio.available_cash / len(to_buy)
for stock in to_buy:
order_value(stock, cash_per_stock)
log.info(f"股票 {stock} 进入前 {g.N} 名,执行买入。")