如果你觉得红利策略只能“吃股息、慢慢涨”，可以试试我的这套策略。
回测区间为 2020-06-01 至 2026-08-10，策略累计收益 1022.30%，同期上证红利基准收益 36.78%；按净值相对计算，累计超额收益 720.51%，最大回撤约 23.51%。
策略主要有四层：
- 红利质量筛选：先剔除新股、ST、停牌及科创/创业等高波动标的，再从盈利能力、营收增长、利润增长、估值等维度过滤，避开单纯高股息但基本面恶化的“价值陷阱”。
- 红利 + 动量双因子：正常市场环境下，股息率权重 85%，50 日动量权重 15%。既不放弃高分红的防御性，也不去接持续走弱的“便宜货”。
- 市场环境识别：当红利风格短期走弱、市场内部动量恶化时，策略会提高低波动因子权重，并在特定情形切换至质量动量池，避免只抱着红利标签不动。
- 多层风控与现金管理：个股 15% 止损、组合高水位回撤 20% 后主动减仓、指数趋势仓位调整，并将闲置资金停放在银华日利 ETF，而不是无效躺在现金里。
这套策略最值得复制的地方，是它解决了很多红利策略的老问题：
- 不迷信高股息。高股息只是起点，增长和趋势必须过关。
- 不靠频繁交易。策略按月调仓，规则清晰、执行成本低。
- 不把风控停留在口号。个股、组合、市场风格三层都设有应对机制。
# 克隆自聚宽文章：https://www.joinquant.com/post/77776
# 标题：红利稳稳的幸福，年化50%，今年依旧表现亮眼
# 作者：淡然
import json
import pandas as pd
from jqdata import *
# 初始化函数
def initialize(context):
# 系统设置
set_option(“avoid_future_data”, True)
set_option(‘use_real_price’, True)
set_benchmark(‘000015.XSHG’)
# 将滑点设置为0.005
set_slippage(FixedSlippage(0.02))
set_slippage(FixedSlippage(0), type=’fund’)
# 设置交易成本万分之三
set_order_cost(OrderCost(open_tax=0, close_tax=0.0005, open_commission=0.0003, close_commission=0.0003, close_today_commission=0, min_commission=5),type=’stock’)
# 银华日利作为现金替代品，买卖不计佣金、税费和滑点
set_order_cost(OrderCost(open_tax=0, close_tax=0, open_commission=0,
close_commission=0, close_today_commission=0,
min_commission=0), type=’fund’)
# 过滤order中低于error级别的日志
log.set_level(‘order’, ‘error’)
log.set_level(‘history’, ‘error’)
log.set_level(‘system’, ‘error’)
# 全局变量
g.sell_list = []
g.sell_reasons = {}
g.buy_df = []
g.target_weights = {}
g.stock_num = 8
g.high_limit_list = []
g.strategy = ‘hongli_momentum’
g.filename = ‘红利动量增强.csv’ #入金配置文件名称
g.first = 1
g.out_cash = 0
g.high_water_mark = 0
g.portfolio_risk_off = False
g.portfolio_risk_off_days = 0
g.cash_etf = ‘511880.XSHG’ # 银华日利ETF，作为闲置现金替代品
g.state_filename = ‘hongli_core_satellite_live_state_v2.json’
g.state_key = ‘hongli_core_satellite_v2′
g.min_market_cap = 0 #筛选最小市值
g.max_market_cap = 100000 #筛选最大市值
g.run_stoploss = True # 是否进行止损
g.stoploss_strategy = 1 # 1为止损线止损，2为市场趋势止损, 3为联合1、2策略
g.stoploss_limit = 0.15 # 止损线
g.stoploss_market = 0.05 # 市场趋势止损参数（上证指数）
# ========== 动量因子参数（可调整）==========
g.momentum_days = 50 # 动量计算天数，默认20日
g.dividend_weight = 0.85 # 股息率权重
g.momentum_weight = 0.15 # 动量权重
g.min_select = 1 # 最少选中数量
load_runtime_state(context)
# 交易时间
run_daily(prepare_stock_list, ’09:00′)
run_monthly(get_stock_list, 1, ’09:01′) #选股
run_monthly(sell_cash_etf, 1, ’09:31′) #先卖出现金ETF，释放股票买入资金
run_monthly(my_trade, 1 ,’10:20′) #买卖操作
run_daily(check_limit_up, ’10:00′) #检查涨停情况并卖出
run_daily(check_portfolio_drawdown, time=’14:40′)
run_daily(stoploss_stocks, time=’14:50′) # 止损函数
run_daily(park_idle_cash, time=’14:55’)
def after_code_changed(context):
“””
已有模拟交易替换代码时执行。
聚宽替换代码不会重新运行initialize，因此只补充本次新增的日志状态。
本版本没有修改定时任务，不调用unschedule_all，避免重复或遗漏调度。
“””
if not hasattr(g, ‘sell_reasons’):
g.sell_reasons = {}
log.info(‘代码替换完成：已启用证券名称、得分排序及交易原因日志’)
def is_backtest(context):
run_type = str(getattr(getattr(context, ‘run_params’, None), ‘type’, ”)).lower()
return ‘backtest’ in run_type
def load_runtime_state(context):
“””实盘重启时恢复组合高水位和风险状态；回测不读取持久化文件。”””
current_value = context.portfolio.total_value
g.high_water_mark = max(g.high_water_mark, current_value)
if is_backtest(context):
return
try:
raw = read_file(g.state_filename)
if hasattr(raw, ‘decode’):
raw = raw.decode(‘utf-8’)
state = json.loads(raw)
if state.get(‘strategy’) != g.state_key:
log.info(‘忽略不属于当前策略的持久化状态’)
return
g.high_water_mark = max(float(state.get(‘high_water_mark’, current_value)),
current_value)
g.portfolio_risk_off = bool(state.get(‘portfolio_risk_off’, False))
g.portfolio_risk_off_days = max(0, int(state.get(‘portfolio_risk_off_days’, 0)))
log.info(‘恢复实盘风控状态：高水位%.2f，风险状态%s，持续%d日’ %
(g.high_water_mark, g.portfolio_risk_off,
g.portfolio_risk_off_days))
except Exception as error:
log.info(‘未恢复持久化状态，使用当前净值初始化：%s’ % error)
def save_runtime_state(context):
“””只在模拟盘/实盘保存状态，避免不同回测之间互相污染。”””
if is_backtest(context):
return
state = {
‘version’: 1,
‘strategy’: g.state_key,
‘saved_at’: str(context.current_dt),
‘high_water_mark’: float(g.high_water_mark),
‘portfolio_risk_off’: bool(g.portfolio_risk_off),
‘portfolio_risk_off_days’: int(g.portfolio_risk_off_days),
}
try:
write_file(g.state_filename, json.dumps(state), append=False)
except Exception as error:
log.error(‘保存实盘风控状态失败：%s’ % error)
def sell_cash_etf(context):
“””买入股票前清空银华日利；无法成交时暂停新开股票仓位。”””
if g.cash_etf not in context.portfolio.positions:
return True
position = context.portfolio.positions[g.cash_etf]
if position.total_amount <= 0:
return True
data = get_current_data()[g.cash_etf]
if data.paused or data.last_price <= data.low_limit:
log.info(‘银华日利当前无法卖出，本次暂停新增股票仓位’)
return False
return submit_target_value_order(
context,
g.cash_etf,
0,
‘现金管理：月度调仓前释放现金’,
) is not None
def park_idle_cash(context):
“””收盘前将可用现金停放到银华日利，并保存当天风控状态。”””
try:
data = get_current_data()[g.cash_etf]
if data.paused or data.last_price >= data.high_limit:
log.info(‘银华日利当前无法买入，保留现金’)
return
cash = context.portfolio.available_cash
reserve = max(1000.0, context.portfolio.total_value * 0.005)
investable_cash = cash – reserve
minimum_lot_value = data.last_price * 100
if investable_cash >= minimum_lot_value:
order_obj = order_value(g.cash_etf, investable_cash)
estimated_amount = (
int(investable_cash / data.last_price / 100) * 100
)
if order_obj is None or getattr(order_obj, ‘error’, ”):
log.warn(
‘[现金管理委托未成功] %s %s | 买入约%d股 | 金额%.2f’
% (
g.cash_etf,
get_security_name(g.cash_etf),
estimated_amount,
investable_cash,
)
)
else:
log.info(
‘[现金管理已提交] 买入 | %s %s | 委托约%d股 | 金额%.2f’
% (
g.cash_etf,
get_security_name(g.cash_etf),
get_order_amount(order_obj, estimated_amount),
investable_cash,
)
)
finally:
save_runtime_state(context)
# 准备股票池
def prepare_stock_list(context):
# 获取昨日涨停列表
g.high_limit_list = []
g.hold_list = [s for s in context.portfolio.positions if s != g.cash_etf]
if len(g.hold_list) != 0:
df = get_price(g.hold_list, end_date=context.previous_date, frequency=’daily’, fields=[‘close’,’high_limit’], count=1, panel=False, fill_paused=False, skip_paused=False).dropna()
df = df[df[‘close’] == df[‘high_limit’]]
g.high_limit_list = list(df.code)
# ========== 核心：计算动量因子 ==========
def get_momentum_factor(context, stock_list, days=20):
“””
计算指定天数的收益率动量因子
返回：DataFrame，index为code，columns为[‘momentum’]
“””
if not stock_list:
return pd.DataFrame(columns=[‘momentum’])
yesterday = context.previous_date
# 获取前days+5天的收盘价（多取5天防止停牌缺失）
df = get_price(stock_list, end_date=yesterday, frequency=’daily’,
fields=[‘close’], count=days+5, panel=False,
skip_paused=False, fill_paused=True)
if df.empty:
return pd.DataFrame(columns=[‘momentum’])
# 按股票分组，取最近days个有效交易日计算收益率
momentum_list = []
for code in stock_list:
code_df = df[df[‘code’] == code].dropna()
if len(code_df) >= days:
start_price = code_df[‘close’].iloc[-days] # days天前的收盘价
end_price = code_df[‘close’].iloc[-1] # 最新收盘价
momentum = (end_price / start_price – 1) * 100 # 百分比收益率
else:
momentum = -999 # 数据不足，给个极低值排在后面
momentum_list.append({‘code’: code, ‘momentum’: momentum})
result = pd.DataFrame(momentum_list).set_index(‘code’)
return result
# ========== 候选股波动率因子 ==========
def get_volatility_factor(context, stock_list, days=60):
if not stock_list:
return pd.DataFrame(columns=[‘volatility’])
price_df = get_price(stock_list, end_date=context.previous_date, frequency=’daily’,
fields=[‘close’], count=days + 1, panel=False,
skip_paused=False, fill_paused=True)
if price_df.empty:
return pd.DataFrame(columns=[‘volatility’])
close_df = price_df.pivot(index=’time’, columns=’code’, values=’close’)
volatility = close_df.pct_change().std().dropna()
return volatility.to_frame(‘volatility’)
# ========== 逆波动风险预算 ==========
def get_inverse_vol_weights(context, stock_list, lookback=60):
“””按过去60个交易日波动率的倒数分配仓位，降低高波动个股对组合回撤的贡献。”””
if not stock_list:
return {}
price_df = get_price(stock_list, end_date=context.previous_date, frequency=’daily’,
fields=[‘close’], count=lookback + 1, panel=False,
skip_paused=False, fill_paused=True)
if price_df.empty:
equal_weight = 1.0 / len(stock_list)
return {s: equal_weight for s in stock_list}
close_df = price_df.pivot(index=’time’, columns=’code’, values=’close’)
volatility = close_df.pct_change().std().replace(0, float(‘nan’)).dropna()
if len(volatility) != len(stock_list):
equal_weight = 1.0 / len(stock_list)
return {s: equal_weight for s in stock_list}
inverse_vol = 1.0 / volatility
return (inverse_vol / inverse_vol.sum()).to_dict()
# ========== 评分加权与单股上限 ==========
def get_capped_score_weights(score_series, exposure=1.0, max_weight=0.25):
if score_series is None or len(score_series) == 0 or exposure <= 0:
return {}
scores = score_series.astype(float).clip(lower=0)
if scores.sum() <= 0:
scores[:] = 1.0
weights = pd.Series(0.0, index=scores.index)
remaining = float(exposure)
active = list(scores.index)
while active and remaining > 1e-10:
active_scores = scores.loc[active]
proposed = remaining * active_scores / active_scores.sum()
over = [s for s in active if proposed[s] > max_weight]
if not over:
weights.loc[active] = proposed
remaining = 0
else:
for s in over:
weights[s] = max_weight
remaining -= max_weight
active.remove(s)
return weights[weights > 0].to_dict()
# ========== 核心-卫星权重 ==========
def get_core_satellite_weights(stock_list, exposure=1.0, core_count=2, core_weight=0.43):
if not stock_list or exposure <= 0:
return {}
if len(stock_list) <= core_count:
equal_weight = exposure / len(stock_list)
return {s: equal_weight for s in stock_list}
core_each = min(core_weight, exposure / core_count)
core_total = core_each * core_count
satellite_weight = max(0, exposure – core_total) / (len(stock_list) – core_count)
weights = {s: core_each for s in stock_list[:core_count]}
weights.update({s: satellite_weight for s in stock_list[core_count:]})
return weights
# ========== 组合级趋势风控 ==========
def get_market_exposure(context, index_code=’000015.XSHG’, lookback=120, risk_off_exposure=0.95):
“””指数低于120日均线时将组合仓位降至95%，控制系统性回撤。”””
price_df = get_price(index_code, end_date=context.previous_date, frequency=’daily’,
fields=[‘close’], count=lookback, panel=False,
skip_paused=False, fill_paused=True)
if price_df.empty:
return 1.0
close = price_df[‘close’].dropna()
if len(close) < lookback:
return 1.0
return 1.0 if close.iloc[-1] >= close.mean() else risk_off_exposure
# ========== 红利指数短期回撤状态 ==========
def is_dividend_style_weak(context, lookback=20, drawdown_limit=0.03):
price_df = get_price(‘000015.XSHG’, end_date=context.previous_date, frequency=’daily’,
fields=[‘close’], count=lookback, panel=False,
skip_paused=False, fill_paused=True)
if price_df.empty:
return False
close = price_df[‘close’].dropna()
if len(close) < lookback:
return False
return close.iloc[-1] / close.max() – 1 < -drawdown_limit
# 选股
def get_stock_list(context):
# 基础信息
g.buy_df = pd.DataFrame(index=[], columns=[‘name’, ‘price’, ‘amount’, ‘value’])
yesterday = str(context.previous_date)
today = context.current_dt
# 初始过滤
initial_list = get_all_securities(‘stock’, today).index.tolist()
initial_list = filter_new_stock(context, initial_list)
initial_list = filter_kcb_stock(initial_list)
initial_list = filter_st_stock(initial_list)
initial_list = filter_paused_stock(initial_list)
initial_list = filter_by_market_cap_range(initial_list, g.min_market_cap, g.max_market_cap)
# ========== 红利价值财务过滤 ==========
stock_list = initial_list
df = get_fundamentals(query(
valuation.code,
).filter(
valuation.code.in_(stock_list),
# 合理的财务指标，既避免价值陷阱，也防止畸高收益
valuation.pe_ratio.between(5, 200), # 市盈率
indicator.inc_return.between(5, 100), # 净资产收益率(扣除非经常损益)(%)
indicator.inc_total_revenue_year_on_year.between(5, 100), # 营业总收入同比增长率(%)
indicator.inc_net_profit_year_on_year.between(10, 100), # 净利润同比增长率(%)
), date=context.previous_date)
stock_list = list(df.code)
# ========== 获取股息率数据 ==========
dividend_list = get_dividend_ratio_filter_list(context, stock_list, False, 0.00, 1.0, 0.00)
# 注意：这里先不做阈值过滤，保留所有有股息率的股票用于综合打分
if not dividend_list:
print(“警告：无满足条件的股票，本次不交易”)
g.sell_list = [s for s in g.hold_list if s not in g.high_limit_list]
g.sell_reasons = {s: ‘选股结果为空’ for s in g.sell_list}
g.buy_df = pd.DataFrame(index=[], columns=[‘name’, ‘price’, ‘amount’, ‘value’])
g.target_weights = {}
return
# ========== 计算动量因子 ==========
momentum_df = get_momentum_factor(context, dividend_list, days=g.momentum_days)
volatility_df = get_volatility_factor(context, dividend_list, days=60)
# 风格回撤但候选池仍有广度时，切换到全股票质量动量池
pre_breadth = (momentum_df[‘momentum’] > 0).mean() if not momentum_df.empty else 0
if is_dividend_style_weak(context) and pre_breadth >= 0.40:
risk_momentum = get_momentum_factor(context, stock_list, days=g.momentum_days)
risk_volatility = get_volatility_factor(context, stock_list, days=60)
risk_score = risk_momentum.join(risk_volatility, how=’inner’)
risk_score = risk_score[risk_score[‘momentum’] > 0]
if not risk_score.empty:
risk_score[‘momentum_rank’] = risk_score[‘momentum’].rank(pct=True, ascending=True)
risk_score[‘low_vol_rank’] = risk_score[‘volatility’].rank(pct=True, ascending=False)
risk_score[‘total_score’] = (0.70 * risk_score[‘momentum_rank’] +
0.30 * risk_score[‘low_vol_rank’])
risk_score = risk_score.sort_values(‘total_score’, ascending=False)
target_list = list(risk_score.index)[:min(8, len(risk_score))]
g.sell_list = [s for s in g.hold_list if s not in target_list and s not in g.high_limit_list]
g.sell_reasons = {s: ‘不在本月质量动量目标池’ for s in g.sell_list}
g.target_exposure = 0.90
g.target_weights = get_core_satellite_weights(target_list, g.target_exposure)
risk_log_df = risk_score[
[‘momentum’, ‘volatility’, ‘total_score’]
].head(10).copy()
risk_log_df.insert(
0,
‘name’,
[
get_security_name(code)
for code in risk_log_df.index
],
)
risk_log_df.insert(
0,
‘rank’,
range(1, len(risk_log_df) + 1),
)
print(‘========== 质量动量得分排序（前10名）==========’)
print(
risk_log_df.to_string(
float_format=lambda value: ‘%.2f’ % value
)
)
print(‘红利回撤且内部广度尚强，切换质量动量池’)
print_selected_targets(
target_list,
g.target_weights,
title=’质量动量选中标的列表’,
)
return
# ========== 获取股息率详细数据用于打分 ==========
time1 = context.previous_date
time0 = time1 – datetime.timedelta(days=365)
interval = 1000
list_len = len(dividend_list)
q = query(finance.STK_XR_XD.code, finance.STK_XR_XD.a_registration_date, finance.STK_XR_XD.bonus_amount_rmb
).filter(
finance.STK_XR_XD.a_registration_date >= time0,
finance.STK_XR_XD.a_registration_date <= time1,
finance.STK_XR_XD.code.in_(dividend_list[:min(list_len, interval)]))
div_df = finance.run_query(q)
if list_len > interval:
df_num = list_len // interval
for i in range(df_num):
q = query(finance.STK_XR_XD.code, finance.STK_XR_XD.a_registration_date, finance.STK_XR_XD.bonus_amount_rmb
).filter(
finance.STK_XR_XD.a_registration_date >= time0,
finance.STK_XR_XD.a_registration_date <= time1,
finance.STK_XR_XD.code.in_(dividend_list[interval*(i+1):min(list_len,interval*(i+2))]))
temp_df = finance.run_query(q)
div_df = div_df.append(temp_df)
dividend = div_df.fillna(0)
dividend = dividend.set_index(‘code’)
dividend = dividend.groupby(‘code’).sum()
temp_list = list(dividend.index)
q = query(valuation.code, valuation.market_cap).filter(valuation.code.in_(temp_list))
cap = get_fundamentals(q, date=time1)
cap = cap.set_index(‘code’)
div_ratio_df = pd.concat([dividend, cap], axis=1, sort=False)
div_ratio_df[‘dividend_ratio’] = (div_ratio_df[‘bonus_amount_rmb’]/10000) / div_ratio_df[‘market_cap’]
div_ratio_df = div_ratio_df[[‘dividend_ratio’]]
# ========== 综合打分 ==========
# 合并股息率和动量数据
score_df = div_ratio_df.join(momentum_df, how=’inner’).join(volatility_df, how=’inner’)
if score_df.empty:
print(“警告：综合打分数据为空，本次不交易”)
g.sell_list = [s for s in g.hold_list if s not in g.high_limit_list]
g.sell_reasons = {s: ‘综合打分数据为空’ for s in g.sell_list}
g.buy_df = pd.DataFrame(index=[], columns=[‘name’, ‘price’, ‘amount’, ‘value’])
g.target_weights = {}
return
# 分别计算排名分（0-1之间，越高越好）
# 股息率：越高越好，直接排名
score_df[‘dividend_rank’] = score_df[‘dividend_ratio’].rank(pct=True, ascending=True)
# 动量：越高越好，但过滤掉极端负值（停牌等异常）
score_df[‘momentum_rank’] = score_df[‘momentum’].rank(pct=True, ascending=True)
score_df[‘low_vol_rank’] = score_df[‘volatility’].rank(pct=True, ascending=False)
# 综合得分 = 股息率权重 * 股息率排名分 + 动量权重 * 动量排名分
momentum_breadth = (score_df[‘momentum’] > 0).mean()
strategy_risk_off = (momentum_breadth < 0.40 and is_dividend_style_weak(context))
if strategy_risk_off:
score_df[‘total_score’] = (0.40 * score_df[‘dividend_rank’] +
0.20 * score_df[‘momentum_rank’] +
0.40 * score_df[‘low_vol_rank’])
else:
score_df[‘total_score’] = (g.dividend_weight * score_df[‘dividend_rank’] +
g.momentum_weight * score_df[‘momentum_rank’])
# 按综合得分降序排列
score_df = score_df.sort_values(‘total_score’, ascending=False)
# 再做一次股息率阈值过滤（保留核心红利属性）
score_df = score_df[score_df[‘dividend_ratio’] > 0.03] # 股息率>3%
# 保证至少选中min_select只
target_stock_num = 10 if strategy_risk_off else g.stock_num
select_count = max(g.min_select, min(target_stock_num, len(score_df)))
target_list = list(score_df.index)[:select_count]
score_log_df = score_df[
[‘dividend_ratio’, ‘momentum’, ‘total_score’]
].head(10).copy()
score_log_df.insert(
0,
‘name’,
[get_security_name(code) for code in score_log_df.index],
)
score_log_df.insert(
0,
‘rank’,
range(1, len(score_log_df) + 1),
)
print(‘========== 综合得分排序（前10名）==========’)
print(
score_log_df.to_string(
float_format=lambda value: ‘%.2f’ % value
)
)
print(‘==========================================’)
g.sell_list = [s for s in g.hold_list if s not in target_list and s not in g.high_limit_list]
g.sell_reasons = {s: ‘不在本月最新目标列表’ for s in g.sell_list}
g.target_exposure = 0.85 if strategy_risk_off else get_market_exposure(context)
g.target_weights = get_core_satellite_weights(target_list, g.target_exposure)
# 盘前打印
print_selected_targets(
target_list,
g.target_weights,
title=’选中标的列表’,
)
print(
‘计划卖出:’,
[
‘%s %s（%s）’
% (
code,
get_security_name(code),
g.sell_reasons.get(code, ‘不在本月最新目标列表’),
)
for code in g.sell_list
],
)
print(
‘正动量广度：%.2f | 弱势模式：%s | 目标持仓数：%d’
% (
momentum_breadth,
strategy_risk_off,
target_stock_num,
)
)
# 日频组合回撤保护：只在状态切换时交易，避免每天重复减仓
def check_portfolio_drawdown(context):
total_value = context.portfolio.total_value
g.high_water_mark = max(getattr(g, ‘high_water_mark’, 0), total_value)
if g.high_water_mark <= 0:
return
portfolio_drawdown = 1 – total_value / g.high_water_mark
current_data = get_current_data()
if g.portfolio_risk_off:
g.portfolio_risk_off_days += 1
if (not g.portfolio_risk_off) and portfolio_drawdown >= 0.20:
for stock, position in context.portfolio.positions.items():
if stock == g.cash_etf:
continue
if (not current_data[stock].paused and
current_data[stock].last_price < current_data[stock].high_limit):
submit_target_value_order(
context,
stock,
position.value * 0.50,
‘组合回撤达到%.2f%%，仓位降至当前的一半’
% (portfolio_drawdown * 100),
)
g.portfolio_risk_off = True
g.portfolio_risk_off_days = 0
log.info(‘组合回撤达到%.2f%%，仓位降至当前的一半’ % (portfolio_drawdown * 100))
save_runtime_state(context)
elif (g.portfolio_risk_off and portfolio_drawdown <= 0.12) or (g.portfolio_risk_off and g.portfolio_risk_off_days >= 60 and get_market_exposure(context) >= 1.0):
g.portfolio_risk_off = False
log.info(‘组合回撤恢复至%.2f%%，解除组合风险状态’ % (portfolio_drawdown * 100))
save_runtime_state(context)
# ========== 日志辅助函数：仅增强可读性，不参与选股和权重计算 ==========
def get_security_name(security):
“””获取证券名称，行情对象不可用时回退到证券基础信息。”””
try:
name = get_current_data()[security].name
if name:
return name
except Exception:
pass
try:
security_info = get_security_info(security)
name = getattr(security_info, ‘display_name’, None)
if name:
return name
except Exception:
pass
return ‘未知名称’
def print_selected_targets(target_list, target_weights, title=’选中标的列表’):
“””以逐行编号列表展示标的；权重统一保留两位小数。”””
print(‘========== %s ==========’ % title)
if not target_list:
print(‘（无）’)
return
for rank, security in enumerate(target_list, 1):
weight = float(target_weights.get(security, 0))
print(
‘%d. %s %s | 目标权重 %.2f%%’
% (
rank,
security,
get_security_name(security),
weight * 100,
)
)
print(‘==================================’)
def get_position_amount(context, security):
if security in context.portfolio.positions:
return int(context.portfolio.positions[security].total_amount)
return 0
def get_order_amount(order_obj, fallback_amount):
“””优先读取聚宽订单对象中的委托数量，读取不到时使用计算值。”””
if order_obj is not None:
for attribute in (‘amount’, ‘_amount’):
try:
amount = getattr(order_obj, attribute)
if amount is not None and int(amount) != 0:
return abs(int(amount))
except Exception:
pass
return abs(int(fallback_amount))
def submit_target_value_order(context, security, target_value, reason):
“””按原策略提交目标市值订单，并输出统一的订单明细日志。”””
name = get_security_name(security)
current_amount = get_position_amount(context, security)
current_data = get_current_data()
price = current_data[security].last_price
if price is None or pd.isnull(price) or price <= 0:
estimated_target_amount = current_amount
elif target_value <= 0:
estimated_target_amount = 0
else:
estimated_target_amount = int(float(target_value) / float(price))
if target_value > (
float(context.portfolio.positions[security].value)
if security in context.portfolio.positions
else 0.0
):
estimated_target_amount = (
int(estimated_target_amount / 100) * 100
)
estimated_order_amount = abs(
estimated_target_amount – current_amount
)
if estimated_target_amount > current_amount:
direction = ‘买入/增仓’
elif estimated_target_amount < current_amount:
direction = ‘卖出/减仓’
else:
direction = ‘无需调整’
# 目标数量与当前数量相同，无须调用聚宽下单接口。
# 这只过滤必然为0的无效委托，不改变任何实际持仓。
if estimated_order_amount <= 0:
log.info(
‘[无需委托] %s %s | 原因：%s | 当前%d股 | 目标约%d股’
% (
security,
name,
reason,
current_amount,
estimated_target_amount,
)
)
return None
# 非清仓减仓不足100股时，聚宽会报“平仓数量不能小于100”。
# 完整清仓（target_value=0）不受此限制，仍按原逻辑提交。
if (
estimated_target_amount < current_amount
and target_value > 0
and estimated_order_amount < 100
):
log.info(
‘[跳过委托] 卖出/减仓 | %s %s | ‘
‘原因：目标减仓不足100股 | 当前%d股 | ‘
‘目标约%d股 | 预计减仓%d股 | 目标市值%.2f’
% (
security,
name,
current_amount,
estimated_target_amount,
estimated_order_amount,
float(target_value),
)
)
return None
order_obj = order_target_value(security, target_value)
error_message = (
getattr(order_obj, ‘error’, ”)
if order_obj is not None
else ‘订单对象为空’
)
if order_obj is None or error_message:
log.warn(
‘[委托未成功] %s | %s %s | 原因：%s | ‘
‘当前%d股 | 预计委托%d股 | 错误：%s’
% (
direction,
security,
name,
reason,
current_amount,
estimated_order_amount,
error_message,
)
)
return order_obj
actual_order_amount = get_order_amount(
order_obj,
estimated_order_amount,
)
log.info(
‘[已提交委托] %s | %s %s | 原因：%s | ‘
‘当前%d股 | 目标约%d股 | 委托%d股 | 目标市值%.2f’
% (
direction,
security,
name,
reason,
current_amount,
estimated_target_amount,
actual_order_amount,
float(target_value),
)
)
return order_obj
# 仅用于拦截必然失败的买入委托；不改变选股、权重、调仓时间或卖出逻辑
def safe_order_target_value(
context,
security,
target_value,
reason=’按目标权重再平衡’,
):
current_data = get_current_data()
data = current_data[security]
price = data.last_price
name = get_security_name(security)
if price is None or pd.isnull(price) or price <= 0:
log.info(
‘[跳过委托] %s %s | 原因：价格无效’
% (security, name)
)
return None
if security in context.portfolio.positions:
position = context.portfolio.positions[security]
current_amount = int(position.total_amount)
current_value = float(position.value)
else:
current_amount = 0
current_value = 0.0
# 减仓和清仓完全沿用原来的order_target_value逻辑，只保护开仓/增仓
if target_value > current_value:
target_amount = int(float(target_value) / float(price) / 100) * 100
buy_amount = target_amount – current_amount
# 目标增持不足一手时，该委托在聚宽中必然失败，直接跳过
if buy_amount < 100:
log.info(
‘[跳过委托] 买入/增仓 | %s %s | 原因：目标增持不足100股 | ‘
‘当前%d股 | 预计增持%d股 | 目标市值%.2f’
% (
security,
name,
current_amount,
buy_amount,
target_value,
)
)
return None
# 若可用现金连一手都买不起，平台会把订单缩成不足100股并报错
one_lot_cash = float(price) * 100 * 1.01 + 5.0
if float(context.portfolio.available_cash) < one_lot_cash:
log.info(
‘[跳过委托] 买入/增仓 | %s %s | ‘
‘原因：可用现金%.2f不足一手所需约%.2f’
% (
security,
name,
float(context.portfolio.available_cash),
one_lot_cash,
)
)
return None
return submit_target_value_order(
context,
security,
target_value,
reason,
)
# 交易
def my_trade(context):
# 基础信息，获取当前单位时间（当天/当前分钟）的涨跌停价, 是否停牌，当天的开盘价等。
current_data = get_current_data()
cash_etf_sold = sell_cash_etf(context)
# 卖出
for s in g.sell_list:
if current_data[s].last_price < current_data[s].high_limit:
submit_target_value_order(
context,
s,
0,
g.sell_reasons.get(
s,
‘月度调仓：不在本月最新目标列表’,
),
)
if not cash_etf_sold:
return
# 按等权对全部目标持仓再平衡；仅在回撤超过22%时降低仓位
total_value = context.portfolio.total_value
g.high_water_mark = max(getattr(g, ‘high_water_mark’, 0), total_value)
portfolio_drawdown = 0 if g.high_water_mark <= 0 else 1 – total_value / g.high_water_mark
risk_multiplier = 0.50 if g.portfolio_risk_off else (0.40 if portfolio_drawdown >= 0.22 else 1.0)
for s, weight in g.target_weights.items():
if (not current_data[s].paused and
current_data[s].last_price < current_data[s].high_limit):
target_value = total_value * weight * risk_multiplier
safe_order_target_value(
context,
s,
target_value,
‘月度调仓：按目标权重%.2f%%再平衡’
% (weight * 100),
)
# 调整昨日涨停股票
def check_limit_up(context):
current_data = get_current_data()
# 对昨日涨停股票观察到尾盘如不涨停则提前卖出，如果涨停即使不在应买入列表仍暂时持有
if g.high_limit_list != []:
for s in g.high_limit_list:
if current_data[s].last_price < current_data[s].high_limit:
submit_target_value_order(
context,
s,
0,
‘昨日涨停、今日涨停打开’,
)
print(‘———————————————————————————————————’)
else:
print(
s,
get_security_name(s),
‘继续涨停，继续持有’,
)
print(‘———————————————————————————————————’)
############################################################################################################################################################################
# 过滤函数
def filter_paused_stock(stock_list):
current_data = get_current_data()
return [stock for stock in stock_list if not current_data[stock].paused]
def filter_st_stock(stock_list):
current_data = get_current_data()
return [stock for stock in stock_list
if not current_data[stock].is_st
and ‘ST’ not in current_data[stock].name
and ‘*’ not in current_data[stock].name
and ‘退’ not in current_data[stock].name]
def filter_kcb_stock(stock_list):
return [stock for stock in stock_list if ((stock[0] != ‘4’) and (stock[0] != ‘8’) and (stock[0] != ’30’) and (stock[0:2] != ’68’))]
def filter_new_stock(context, stock_list):
yesterday = context.previous_date
return [stock for stock in stock_list if not yesterday – get_security_info(stock).start_date < datetime.timedelta(days=250)]
# 根据最近一年分红除以当前总市值计算股息率并筛选
def get_dividend_ratio_filter_list(context, stock_list, sort, p1, p2, threshold):
time1 = context.previous_date
time0 = time1 – datetime.timedelta(days=365)
# 获取分红数据，由于finance.run_query最多返回4000行，以防未来数据超限，最好把stock_list拆分后查询再组合
interval = 1000 # 某只股票可能一年内多次分红，导致其所占行数大于1，所以interval不要取满4000
list_len = len(stock_list)
# 截取不超过interval的列表并查询
q = query(finance.STK_XR_XD.code, finance.STK_XR_XD.a_registration_date, finance.STK_XR_XD.bonus_amount_rmb
).filter(
finance.STK_XR_XD.a_registration_date >= time0,
finance.STK_XR_XD.a_registration_date <= time1,
finance.STK_XR_XD.code.in_(stock_list[:min(list_len, interval)]))
df = finance.run_query(q)
# 对interval的部分分别查询并拼接
if list_len > interval:
df_num = list_len // interval
for i in range(df_num):
q = query(finance.STK_XR_XD.code, finance.STK_XR_XD.a_registration_date, finance.STK_XR_XD.bonus_amount_rmb
).filter(
finance.STK_XR_XD.a_registration_date >= time0,
finance.STK_XR_XD.a_registration_date <= time1,
finance.STK_XR_XD.code.in_(stock_list[interval*(i+1):min(list_len,interval*(i+2))]))
temp_df = finance.run_query(q)
df = df.append(temp_df)
dividend = df.fillna(0)
dividend = dividend.set_index(‘code’)
dividend = dividend.groupby(‘code’).sum()
temp_list = list(dividend.index) # query查询不到无分红信息的股票，所以temp_list长度会小于stock_list
# 获取市值相关数据
q = query(valuation.code,valuation.market_cap).filter(valuation.code.in_(temp_list))
cap = get_fundamentals(q, date=time1)
cap = cap.set_index(‘code’)
# 计算股息率
df = pd.concat([dividend, cap] ,axis=1, sort=False)
df[‘dividend_ratio’] = (df[‘bonus_amount_rmb’]/10000) / df[‘market_cap’]
# 排序并筛选
df = df.sort_values(by=[‘dividend_ratio’], ascending=sort)
df = df[int(p1*len(df)):int(p2*len(df))]
df = df[df[‘dividend_ratio’] > threshold]
return list(df.index)
#市值过滤函数
def filter_by_market_cap_range(initial_list, min_market_cap, max_market_cap):
final_list = []
q = query(valuation.code,valuation.market_cap).filter(valuation.code.in_(initial_list),valuation.market_cap.between(min_market_cap,max_market_cap)).order_by(valuation.market_cap.asc())
df_fun = get_fundamentals(q)
final_list = list(df_fun.code)
return final_list
#止损函数
def stoploss_stocks(context):
if g.run_stoploss:
current_positions = context.portfolio.positions
if g.stoploss_strategy == 1 or g.stoploss_strategy == 3:
for stock in current_positions.keys():
if stock == g.cash_etf:
continue
price = current_positions[stock].price
avg_cost = current_positions[stock].avg_cost
# 个股止损
if price < avg_cost * (1 – g.stoploss_limit):
submit_target_value_order(
context,
stock,
0,
‘个股止损：现价%.2f低于成本%.2f的%.2f%%’
% (
price,
avg_cost,
(1 – g.stoploss_limit) * 100,
),
)
if g.stoploss_strategy == 2 or g.stoploss_strategy == 3:
stock_df = get_price(security=get_index_stocks(‘000001.XSHG’), end_date=context.previous_date, frequency=’daily’, fields=[‘close’, ‘open’], count=1, panel=False)
down_ratio = (1-(stock_df[‘close’] / stock_df[‘open’]).mean())
# 市场大跌止损 ##参考哪个指数大跌 ##大盘大跌
if down_ratio >= g.stoploss_market:
log.debug(“大盘惨跌，全仓卖出,平均降幅{:.2%}”.format(down_ratio))
for stock in current_positions.keys():
if stock != g.cash_etf:
submit_target_value_order(
context,
stock,
0,
‘市场趋势止损：平均跌幅%.2f%%’
% (down_ratio * 100),
)
评论(0)