历史公认有效的因子和回测结果-固定因子系数

因子是会衰减的而这里设死了若干个因子和对应权重在长时间窗口去回测其实是不太合理的所以在长时间窗口下回测结果不太理想也是合理的。更好的方案是根据不同市场情况或因子拥挤度去选择不同的加权系数或因子。历史公认/统计有效的因子以下数据来自全市场 5487 只股票、188 个月度调仓周期的实证检验所有因子均已做市值中性化处理因子类型方向IC 均值IR胜率核心逻辑1月反转价量负向0.04~0.050.7~0.8~65%A 股散户主导短期过度反应后回调EP盈利收益率价值正向0.030.5~60%买得便宜安全边际高ROE质量正向0.020.4~58%盈利能力强护城河深低波动IVOL风险负向0.02~0.030.4~0.5~58%低波动异象A 股显著分析师预期ANA情绪正向0.05670.891.9%一致预期上调信息优势残差动量价量正向0.0150.15~55%剥离市场 Beta 后的特质动量关键结论来自 2015-2025 年全量数据A 股不存在传统价格动量而是显著的短期反转效应2024 年防御风格红利、低波、价值主导2025 年进攻风格小盘、成长、质量发生镜像反转市值中性化必须做否则所有因子都是 Size Proxy伪市值因子平台选择建议平台网址优势适用场景JoinQuant聚宽 ⭐joinquant.com国内最主流社区策略丰富免费版够用API 成熟新手入门、策略研究RiceQuant米筐ricequant.com数据质量高内置因子库get_factor机构常用专业因子研究BigQuantbigquant.comAI 量化集成可视化好有因子研究工具AI量化FactorHubfactorhub.cn专注因子研究支持 MCP 协议 AI 调用因子挖掘可粘贴到聚宽直接回测的代码from jqdata import * import numpy as np import pandas as pd def initialize(context): set_benchmark(000300.XSHG) set_option(use_real_price, True) set_option(avoid_future_data, True) set_order_cost(OrderCost( open_tax0, close_tax0.001, open_commission0.0003, close_commission0.0003, close_today_commission0, min_commission5 ), typestock) set_slippage(FixedSlippage(0.002)) g.stock_num 20 g.candidate_pool 000300.XSHG g.factor_weights { ep: 0.30, roe: 0.25, reversal: 0.25, lowvol: 0.20 } run_monthly(rebalance, monthday1, time9:31, reference_security000300.XSHG) log.info( 策略初始化完成 | 持仓%d只 | 月度调仓 % g.stock_num) def get_stock_pool(context): 获取候选股票池沪深300过滤ST/停牌/科创板/北交所 stocks get_index_stocks(g.candidate_pool, context.previous_date) stocks [s for s in stocks if not s.startswith(688) and not s.startswith(8) and not s.startswith(4)] if len(stocks) 0: st_df get_extras(is_st, stocks, end_datecontext.previous_date, count1) if st_df is not None and len(st_df) 0: st_series st_df.iloc[-1] st_stocks st_series[st_series True].index.tolist() stocks [s for s in stocks if s not in st_stocks] return stocks def get_factor_data(context, stocks): 获取并计算因子数据 if len(stocks) 0: return pd.DataFrame() q query( valuation.code, valuation.market_cap, valuation.pe_ratio, indicator.roe, ).filter( valuation.code.in_(stocks), valuation.market_cap 0, valuation.pe_ratio 0, valuation.pe_ratio 200, indicator.roe 0 ) df get_fundamentals(q, datecontext.previous_date) if len(df) 50: return pd.DataFrame() df.set_index(code, inplaceTrue) df[ep] 1.0 / df[pe_ratio] price_data get_price( stocks, end_datecontext.previous_date, frequencydaily, fields[close], fqpre, count30 ) df[reversal] 0.0 df[lowvol] 0.0 if price_data is None or len(price_data) 0: return df[[market_cap, ep, roe, reversal, lowvol]].dropna() price_df price_data if len(price_df) 21: window price_df.iloc[-21:-1] valid_mask window.notna().sum(axis0) 15 valid_stocks valid_mask[valid_mask].index.tolist() window window[valid_stocks] if len(window.columns) 0: ret window.pct_change().iloc[1:] ret ret.dropna(axis1, howany) if len(ret.columns) 0: cum_ret (ret 1).prod() - 1 common df.index.intersection(cum_ret.index) if len(common) 0: df.loc[common, ret_20] cum_ret[common] df.loc[common, reversal] -cum_ret[common] df.loc[common, volatility] ret[common].std() df.loc[common, lowvol] -ret[common].std() df df.replace([np.inf, -np.inf], np.nan) return df[[market_cap, ep, roe, reversal, lowvol]].dropna() def neutralize_size(factor_series, market_cap_series): 市值中性化 common_idx factor_series.dropna().index.intersection(market_cap_series.dropna().index) if len(common_idx) 10: return factor_series x np.log(market_cap_series[common_idx].values) y factor_series[common_idx].values x_mean, y_mean np.mean(x), np.mean(y) denominator np.sum((x - x_mean) ** 2) if denominator 0: return factor_series b np.sum((x - x_mean) * (y - y_mean)) / denominator a y_mean - b * x_mean residual y - (a b * x) return pd.Series(residual, indexcommon_idx) def get_industry_limit(df, max_per_industry2): 行业分散 industries {} selected [] for stock in df.index: ind_name 其他 try: ind_info get_industry(stock, dateNone) if stock in ind_info and sw_l1 in ind_info[stock]: ind_name ind_info[stock][sw_l1].get(industry_name, 其他) except: pass if industries.get(ind_name, 0) max_per_industry: selected.append(stock) industries[ind_name] industries.get(ind_name, 0) 1 if len(selected) g.stock_num: break return selected def rebalance(context): 月度调仓主函数 stocks get_stock_pool(context) df get_factor_data(context, stocks) if len(df) g.stock_num * 2: log.info(有效股票数量不足(%d只)跳过调仓 % len(df)) return for col in [ep, roe, reversal, lowvol]: df[col _neu] neutralize_size(df[col], df[market_cap]) rank_cols [ep_neu, roe_neu, reversal_neu, lowvol_neu] for col in rank_cols: if col in df.columns: df[col _rank] df[col].rank(pctTrue) else: df[col _rank] 0.5 df[score] ( g.factor_weights[ep] * df.get(ep_neu_rank, 0.5) g.factor_weights[roe] * df.get(roe_neu_rank, 0.5) g.factor_weights[reversal] * df.get(reversal_neu_rank, 0.5) g.factor_weights[lowvol] * df.get(lowvol_rank, 0.5) ) df df.sort_values(score, ascendingFalse) buy_list get_industry_limit(df, max_per_industry2) if len(buy_list) g.stock_num // 2: buy_list df.head(g.stock_num).index.tolist() log.info(本月选股TOP5: %s % str(buy_list[:5])) holdings list(context.portfolio.positions.keys()) # 卖出不在买入列表的股票 for stock in holdings: if stock not in buy_list: order_target(stock, 0) # 买入用 order_target_value 代替 order_target_percent if len(buy_list) 0: total_value context.portfolio.total_value for stock in buy_list: target_value total_value * (1.0 / len(buy_list)) order_target_value(stock, target_value) def handle_data(context, data): pass参数调优建议回测结果