3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在量化多因子选股策略(如红利低波、红利价值策略)中,频繁的调仓会导致高昂的交易成本(佣金和印花税),从而蚕食策略的超额收益。为了解决这一问题,引入**双阈值排名缓冲机制(Rank Buffer)**是一种非常经典的工程优化方法。
其核心逻辑是:
这种“宽出严进”的机制能有效过滤掉排名在临界点微幅波动导致的无谓交易,大幅降低换手率。
以下是一个在聚宽平台运行的红利增强策略框架,展示了如何结合 get_fundamentals 获取红利因子(以股息率/市净率等代入),并实现买入前 N 名(假设 N=10)、持仓保留前 2N 名(2N=20)的缓冲逻辑。
import pandas as pd
from jqdata import *
def initialize(context):
# 开启真实价格模式
set_option('use_real_price', True)
# 设定沪深300作为基准
set_benchmark('000300.XSHG')
# 股票类交易税费设置
set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
# 策略参数设置
g.N = 10 # 买入阈值:前10名
g.limit_2N = 20 # 保留阈值:前20名
# 每月第一个交易日开盘运行调仓
run_monthly(handle_dividend_strategy, monthday=1, time='09:30')
def handle_dividend_strategy(context):
# 1. 获取全市场可交易股票池(排除ST、停牌)
current_data = get_current_data()
all_stocks = list(get_all_securities(['stock'], context.current_dt.date()).index)
feasible_stocks = [s for s in all_stocks if not current_data[s].paused and not current_data[s].is_st]
# 2. 查询红利价值因子(以股息率/市净率倒数为例,这里使用估值表中的 pb_ratio 倒数作为价值红利替代演示)
# 实际中可结合聚宽财务库或因子库获取更精准的 dividend_yield
q = query(
valuation.code,
valuation.pb_ratio
).filter(
valuation.code.in_(feasible_stocks),
valuation.pb_ratio > 0
)
df = get_fundamentals(q)
# 计算账面市值比(BP = 1/PB),BP越大代表价值红利属性越强
df['bp_ratio'] = 1.0 / df['pb_ratio']
# 按价值红利得分从大到小排序
df_sorted = df.sort_values(by='bp_ratio', ascending=False).reset_index(drop=True)
# 建立股票到排名的映射字典
rank_map = {row['code']: index + 1 for index, row in df_sorted.iterrows()}
# 3. 获取当前持仓
current_holdings = list(context.portfolio.positions.keys())
# 4. 核心排名缓冲筛选逻辑
to_sell = []
to_keep = []
for stock in current_holdings:
rank = rank_map.get(stock, 9999) # 若无排名则视为极差
if rank > g.limit_2N:
# 跌出 2N 名,卖出
to_sell.append(stock)
else:
# 保留在 2N 名内,继续持有
to_keep.append(stock)
# 执行卖出操作
for stock in to_sell:
order_target(stock, 0)
log.info(f"股票 {stock} 排名 {rank_map.get(stock, 9999)} 跌破 {g.limit_2N} 名,执行卖出。")
# 5. 补充买入逻辑
# 计算还需要买入多少只股票以达到 N 只
slots_needed = g.N - len(to_keep)
if slots_needed > 0:
# 候选买入池:排名在前 N 名,且当前未持有的股票
candidates = [row['code'] for _, row in df_sorted.head(g.N).iterrows() if row['code'] not in to_keep]
# 挑选最靠前的候选股票买入
to_buy = candidates[:slots_needed]
# 资金等分分配
available_cash = context.portfolio.available_cash
if len(to_buy) > 0:
cash_per_stock = available_cash / len(to_buy)
for stock in to_buy:
order_value(stock, cash_per_stock)
log.info(f"新买入排名前 {g.N} 的强势红利股: {stock}")