红利低波策略介绍
红利低波策略是一种结合了高股息（红利）和低波动（低波）因子的股票投资策略。该策略的核心是筛选出那些长期稳定派发股息、且股价波动相对较低的上市公司股票构建投资组合。
自然语言promat
通过简单俩句话生成红利低波策略，ai助手对于多模块连接任务表现良好，策略暂时无法一键跑通。经过多轮AI迭代后发现AI助手无法返回因子权重组合节点中的权重列表，改为等权可跑通。
策略收益与因子表现
从整体收益来看，策略年化收益率与市场主流红利低波指数基金水平相近，但在回撤控制方面表现一般，与“低波”特性的匹配度有待提升。
该红利低波策略在2024年表现较弱，因子选股能力不显著，分组收益未呈现理想单调性，风险调整后收益一般。建议深入检查因子构建逻辑、行业分布及市场环境适应性，并进行多维度优化以提升稳健性。如需进一步分析，可提供更长期的回测数据或因子构建细节。
策略代码实现逻辑分析
这段代码实现了一个结构完整、逻辑清晰的红利低波量化策略框架。其核心流程采用了指数成分股初筛 → 交易状态与基础因子过滤 → 红利与低波双因子计算 → 综合排序选股 → 定期等权调仓的经典范式。
主要优势在于模块化设计良好，从数据获取、因子计算到交易执行层层递进，风险控制（如过滤停牌、整手交易）考虑周全，代码可读性与可扩展性较高。
该代码为实现红利低波策略提供了一个可靠的基础骨架，但策略的有效性与超额收益高度依赖于因子合成公式、权重配置以及风控细节的优化。下一步应聚焦于因子模型（权重）的精细化构建与历史数据的深入回测，以验证并提升策略的稳健性与绩效。
def handle_data(context, data):
"""主策略逻辑：在调仓日进行选股并等权配置"""
# 仅在收盘后做调仓逻辑（日频 handle_data 一般就是收盘bar，可无需时间判断）
# 是否调仓日
if not _is_rebalance_day(context):
return
trade_date = context.now
print(f"\n===== 调仓日 {trade_date} 开始调仓 =====")
# 1. 获取基础股票池：指数成分股 ∩ 当日在售股票
base_list = _get_index_constituents(trade_date, context.index_symbol)
if not base_list:
print("指数成分为空，跳过调仓")
return
# 过滤停牌、退市：取当日在售股票
trading_list_df = panda_data.get_trading_stock_list(date=trade_date)
if trading_list_df is None or trading_list_df.empty:
print("在售股票列表为空，跳过调仓")
return
trading_set = set(trading_list_df["symbol"].astype(str).tolist())
universe = [s for s in base_list if s in trading_set]
if not universe:
print("基础股票池在当日无可交易标的，跳过调仓")
return
# 2. 获取基础因子数据（close、turnover、market_cap）
df_factor_panel = _get_factor_panel(trade_date, universe, context)
if df_factor_panel is None or df_factor_panel.empty:
print("因子数据为空，跳过调仓")
return
# 3. 计算流动性与市值过滤
liq_mv = _compute_liquidity_and_mv(df_factor_panel, trade_date, context)
if liq_mv is None or liq_mv.empty:
print("流动性/市值数据为空，跳过调仓")
return
# 4. 红利因子：股息率TTM
dy_df = _get_dividend_yield_ttm(trade_date, liq_mv["symbol"].tolist(), context)
# 5. 波动率因子
vol_df = _compute_volatility(df_factor_panel, trade_date, context)
# 6. 组合打分
score_df = _build_score(liq_mv, dy_df, vol_df, context)
if score_df is None or score_df.empty:
print("综合得分为空，跳过调仓")
return
# 7. 选取前 N 只股票
top_n = context.top_n
selected = score_df.head(top_n)["symbol"].tolist()
if not selected:
print("无标的通过筛选，跳过调仓")
return
print(f"选中标的数: {len(selected)}")
# 8. 目标持仓：等权配置（按股数）
stock_account = context.stock_account_dict.get(context.stock_account)
if stock_account is None:
print("未找到股票账户，跳过调仓")
return
total_value = stock_account.total_value
if total_value <= 0:
print("账户总价值<=0，跳过调仓")
return
# 查询当日收盘价
try:
mkt = panda_data.get_market_data(
symbol=selected,
start_date=trade_date,
end_date=trade_date,
type="stock",
fields=["symbol", "close"],
indicator="000300",
st=True,
)
except Exception as e:
print(f"获取价格失败: {e}")
return
if mkt is None or mkt.empty:
print("价格数据为空，跳过调仓")
return
price_map = mkt.set_index("symbol")["close"].to_dict()
# 等权价值
per_value = total_value / len(selected)
target_pos = {}
for s in selected:
price = price_map.get(s, None)
if price is None or price <= 0:
continue
# 按价值换算成股数，向下取整到手数（100股为一手）
shares = int(per_value // price // 100 * 100)
if shares <= 0:
continue
target_pos[s] = shares
if not target_pos:
print("目标股数全为0，跳过调仓")
return
print(f"目标持仓股票数: {len(target_pos)}")
# 9. 下单：使用目标持仓函数，一次性调仓
try:
target_stock_group_order(context.stock_account, target_pos)
print("目标持仓下单完成")
except Exception as e:
print(f"目标持仓下单失败: {e}")
return
# 更新最近调仓日
context.last_rebalance_date = trade_date
print(f"===== 调仓日 {trade_date} 调仓完成 =====\n")
未来展望
希望PandaAi推出像n8n一样多元化的组件或自定义组件
