五年五倍的菜场大妈量化策略与风险平价(python代码解析)

在上一篇文章中十年73倍的菜场大妈量化策略与风险平价模型，是在聚宽平台上回测的，但是有球友希望在ptrade上也能回测，为后面模拟盘做准备。

按照日线回测，月度调仓，一年的回测结果如下，回测期间总收益为 60.07%，策略的最大回撤为 22.7%。原始股票池变成了中小板，收益率比之前提高了一点。回测速度比聚宽要慢很多，五年回测要将近20分钟。

策略整体逻辑

在每日运行中通过包装函数判断是否为月度首个交易日，若是，则触发选股流程。选股过程（select_stocks）是策略的阿尔法来源，它通过多层级过滤剔除不合格标的（如科创板/北交所、ST、停牌、涨跌停股），随后运用财务指标（股息率、PEG、股价）进行正面筛选，得到最终备选池。 得到股票池后，交易流程（trade_stocks）启动。首先，策略获取备选股票的历史收益率数据，计算其协方差矩阵，并输入到风险平价模型（get_risk_parity_weights）中求解最优权重。该模型的目标是使组合中每个资产对整体风险的贡献度相等，从而实现真正的风险分散。最后，策略根据计算出的目标权重执行调仓操作：卖出不在新组合中的持仓，并调整现有持仓至目标权重。

select_stocks(context)函数详解

1. 初始筛选与日志记录

date_str = context.previous_date.strftime('%Y%m%d')log.info(f"获取股票列表，日期: {date_str}")# stock_list = get_Ashares(date_str)stock_list = get_index_stocks('399101.XSHE')log.info(f"初始股票数量: {len(stock_list)}")

• 作用：获取指定日期的全部A股或特定指数成分股列表，并记录初始数量。
• 说明：实际环境中应使用get_Ashares()获取全市场数据，此处为示例简化处理。

2. 剔除高风险类别

stock_list = filter_kcbj_stock(stock_list)log.info(f"剔除科创北交后股票数量: {len(stock_list)}")stock_list = filter_st_stock(stock_list, context.previous_date)log.info(f"剔除ST股票后数量: {len(stock_list)}")stock_list = filter_paused_stock(stock_list, context.previous_date)log.info(f"剔除停牌股票后数量: {len(stock_list)}")

• 功能模块：依次调用三个过滤函数移除科创板/北交所股票、ST及退市预警股、当日停牌股。
• 必要性：降低非正常交易状态导致的不确定性风险。

3. 流动性管理

stock_list_before_limit = stock_list.copy()stock_list = filter_limitup_stock(context, stock_list)log.info(f"剔除涨停股票后数量: {len(stock_list)}")stock_list = filter_limitdown_stock(context, stock_list)log.info(f"剔除跌停股票后数量: {len(stock_list)}")

• 关键操作：创建副本保存原始列表用于后续回退机制；分别过滤涨跌停个股。
• 容错设计：若过滤后无剩余标的，则恢复至仅去除ST/停牌的状态继续流程。

4. 核心筛选逻辑

# PEG指标筛选stock_list = get_peg_filter_list(context, stock_list, g.peg_dn, g.peg_up)log.info(f"PEG筛选后股票数量: {len(stock_list)}")# 价格区间约束stock_list = get_price_filter_list(context, stock_list, g.price_dn, g.price_up)log.info(f"价格筛选后股票数量: {len(stock_list)}")

• 多维考量：结合收益性（股息率）、成长性（PEG）、安全性（价格波动）三大维度交叉验证。
• 排序规则：PEG筛选结果按市值升序排列，便于后续头寸管理。

trade_stocks(context)函数详解

1. 前置检查

ifnot hasattr(g, 'select_list') ornot g.select_list:    log.info("本月无选股，不进行交易")return

• 防御性编程：防止空仓状态下的错误操作。

2. 数据采集与预处理

df_list = []for stock in select_list:    hist_data = get_history(        count=g.cov_days+30,        frequency='1d',        field=['close'],        security_list=[stock],        fq='pre',        include=False    )ifnot hist_data.empty and stock in hist_data['code'].values:        stock_data = hist_data[hist_data['code']==stock]ifnot stock_data.empty:            close_prices = pd.to_numeric(stock_data['close'], errors='coerce')            df_list.append(pd.DataFrame({stock: close_prices.values}, index=stock_data.index))

• 批量抓取：循环请求每支股票的历史行情数据。
• 质量控制：跳过缺失值过多的异常条目，保证数据完整性。

3. 协方差矩阵构建

df = pd.concat(df_list, axis=1)returns = df.pct_change().dropna().tail(g.cov_days)

• 标准化转换：将原始价格转为百分比变动序列。
• 时效聚焦：仅保留最近g.cov_days天的观测值用于建模。

4. 风险平价建模

weights = get_risk_parity_weights(returns, thre_down=g.pos_dn, thre_up=g.pos_up)weights = weights.sort_values()

• 核心算法：调用独立函数完成风险贡献度相等化的数学优化。
• 排序目的：方便程序按权重从小到大依次执行买卖指令。

5. 仓位再平衡

# 卖出不在新名单中的持仓for s in context.portfolio.positions:if (s notin weights.index):        order_target(s, 0)# 买入目标权重对应的股票total_value = context.portfolio.portfolio_valuefor s in weights.index:    w = weights[s]    value = total_value * w     order_target_value(s, value)

• 两步走策略：先清空非入选股票仓位释放资金，再按新权重建仓。
• 动态适配：每次调仓均基于当前总资产重新计算绝对金额需求。

get_risk_parity_weights(df, thre_down=None, thre_up=None)函数详解

1. 输入参数校验

cov_matrix = df.cov()

• 基础假设：输入DataFrame的每一列代表一支资产的时间序列回报。

2. 目标函数定义

defrisk_parity(weights):    sigma = np.sqrt(np.dot(np.dot(weights, cov_matrix), weights))     mrc = np.dot(cov_matrix, weights) / sigma    trc = weights * mrc    diff_trc_list = [sum((v - trc) ** 2) for v in trc]    target = 10000 * sum(diff_trc_list)return target

• 数学本质：最小化各资产风险贡献度的平方误差之和。
• 放大系数：乘以常数因子加速收敛过程。

3. 约束条件设置

weights0 = np.ones(df.shape[1]) / df.shape[1]cons = [{'type': 'eq', 'fun': lambda w: np.sum(w) - 1}]bounds = tuple((thre_down, thre_up) for w in weights0)

• 初始化方案：等权分配作为起点。
• 边界管控：可选设置单只个股的最高/最低占比限制。

4. 数值求解

if thre_down isNone:    res = minimize(risk_parity, weights0, method='SLSQP', constraints=cons, options={'maxiter': 100})else:    res = minimize(risk_parity, weights0, method='SLSQP', constraints=cons, bounds=bounds, options={'maxiter': 100})

• 算法选择：采用带约束条件的非线性规划求解器SLSQP。
• 终止条件：最多迭代100次防止无限循环。

get_peg_filter_list(context, stocks, thre1, thre2)函数详解

1. 财务数据获取

df_pe = get_fundamentals(stocks, 'valuation', fields='pe_dynamic', date=date_str)df_growth = get_fundamentals(stocks, 'growth_ability', fields='np_parent_company_yoy', date=date_str)

• 接口调用：分别提取动态市盈率和母公司净利润同比增长率两项指标。
• 时间戳匹配：确保使用的是同一个交易日的数据快照。

2. 数据清洗与整合

# 统一索引格式if'secu_code'notin df_pe.columns and df_pe.index.name == 'secu_code':    df_pe = df_pe.reset_index()df_pe = df_pe.rename(columns={'secu_code': 'code'}).set_index('code')# 同理处理df_growth...# 内连接合并df_merged = pd.concat([df_pe, df_growth], axis=1, join='inner')

• 标准化流程：无论源数据库返回何种结构都能正确映射到code字段。
• 去重机制：inner join自动剔除任一表格不存在的共同键记录。

3. PEG计算与筛选

df_merged['peg'] = df_merged['pe_dynamic'] / df_merged['np_parent_company_yoy']df_filtered = df_merged[(df_merged['peg'] > thre1) & (df_merged['peg'] < thre2)]peg_stocks = list(df_filtered.index)

• 公式应用：严格遵循PEG=PE/GrowthRate的定义式。
• 双侧截断：同时设置下限和上限阈值过滤极端值。

4. 市值排序增强稳健性

df_cap = get_fundamentals(peg_stocks, 'valuation', fields='total_value', date=date_str)if df_cap isNoneor df_cap.empty:return peg_stocks  # 保持原有顺序df_cap = df_cap.rename(columns={'secu_code': 'code'}).sort_values(by='total_value', ascending=True)return list(df_cap['code'])

• 二次验证：引入总市值维度打破单纯依赖估值比率的潜在偏见。
• 降级处理：当无法获取市值数据时退化为默认排序方案。

PS: 源码下载，请移步知识星球！

星球中整理了文章中涉及到的源码，加入后即可以看到之前发的源码，三天内不满意可以退款。也会将收集到的机构研报，优秀策略源码，群友常咨询的问题整理到不同标签下，做为知识库沉淀下来。

所有策略仅用于学习和研究，不保证交易收益，不作为投资建议，风险自负，所有收益仅表示历史回测收益，不表示未来收益。大QMT和ptrade请充分使用模拟盘测试，miniQMT使用模拟账号测试。