# Day 26 (100天计划): 动量策略（横截面动量） # 核心：Jegadeesh & Titman(1993) —— 过去3-12个月强势股未来继续跑赢 # 横截面动量 = 同一时点，选相对其他股票表现最好的 # ==================== 可切换参数（在这里修改）==================== VERSION = 'v1' # 'v1': 12-1动量 | 'v2': 6-1动量 | 'v3': 动量+质量过滤 | 'v4': 动量+低波动 # ================================================================ import pandas as pd import numpy as np from jqdata import * """ 回测参数：2020-01-01 至 2026-02-01，100万，沪深300，月度调仓，持仓20只 基准：沪深300（14.88%） 横截面动量理论（Jegadeesh & Titman 1993）： 过去3~12个月收益排名靠前的股票，未来3~12个月仍会跑赢 原因：①信息扩散慢（投资者反应不足） ②机构资金追涨 ③盈利惊喜持续（PEAD效应） Fama-French五因子：UMD = Up Minus Down（动量因子） 关键细节——跳过最近1个月（Skip-1）： 计算动量时排除最近1个月（约20个交易日） 原因：最近1个月存在"短期反转效应"——涨太快的股票短期会回调 标准做法：用 T-252 到 T-21 这段收益率（12个月 skip 1个月） 横截面 vs 时序动量： 横截面动量：选相对排名靠前的股票（今天学的） 时序动量：某只股票是否比自身历史强（绝对动量，Day27再学） V1: 12-1动量（T-252到T-21，标准Jegadeesh配方） V2: 6-1动量（T-126到T-21，更短窗口，捕捉中期趋势） V3: 12-1动量 + ROE>10%（动量+质量过滤，排除问题股的技术性反弹） V4: 12-1动量 + 低波动（动量×60% + 低波动×40%，降低追涨波动） 注：换手率使用 get_fundamentals 稳定接口（同Day24-25） """ def initialize(context): g.stock_num = 20 g.index = '000300.XSHG' set_option('use_real_price', True) log.set_level('order', 'error') run_monthly(trade, 1) # ==================== 数据获取 ==================== def get_momentum(stocks, date, lookback=252, skip=21): """ 计算横截面动量（收益率） lookback: 回看窗口（交易日），默认252=12个月 skip: 跳过最近N天，默认21=1个月（避免短期反转） 公式：mom = Price(T-skip) / Price(T-lookback) - 1 """ try: count = lookback + skip + 5 # 多取几天防止缺数据 price_df = get_price( stocks, end_date=date, count=count, fields=['close'], panel=False, fq='pre' ) if price_df is None or price_df.empty: return pd.Series(dtype=float) close_p = price_df.pivot(index='time', columns='code', values='close') mom = {} for stock in close_p.columns: cl = close_p[stock].dropna() if len(cl) < lookback: continue # Skip-1：用 T-skip 的价格除以 T-lookback 的价格 price_end = cl.iloc[-(skip + 1)] # T-21 price_start = cl.iloc[0] # T-lookback if price_start > 0: mom[stock] = price_end / price_start - 1 return pd.Series(mom).dropna() except Exception as e: log.error(f"动量计算失败: {e}") return pd.Series(dtype=float) def get_volatility(stocks, date, window=60): """ 计算历史波动率（60日） 用于V4：动量+低波动组合 """ try: price_df = get_price( stocks, end_date=date, count=window + 5, fields=['close'], panel=False, fq='pre' ) if price_df is None or price_df.empty: return pd.Series(dtype=float) close_p = price_df.pivot(index='time', columns='code', values='close') vols = {} for stock in close_p.columns: cl = close_p[stock].dropna() if len(cl) >= window // 2: ret = cl.pct_change().dropna() vols[stock] = ret.std() * np.sqrt(252) # 年化波动率 return pd.Series(vols).dropna() except Exception as e: log.error(f"波动率计算失败: {e}") return pd.Series(dtype=float) def get_roe(stocks, date): """获取ROE（用于V3质量过滤）""" try: q = query( valuation.code, indicator.roe ).filter( valuation.code.in_(stocks) ) df = get_fundamentals(q, date=date) if df.empty: return pd.Series(dtype=float) return df.set_index('code')['roe'] except Exception as e: log.error(f"ROE获取失败: {e}") return pd.Series(dtype=float) def to_score(series, higher_is_better=True): """因子值 → 百分位分数 (0, 100]""" s = series.dropna() if s.empty or s.std() == 0: return pd.Series(50.0, index=s.index) if higher_is_better: return s.rank(pct=True) * 100 else: return s.rank(ascending=False, pct=True) * 100 # ==================== 交易 ==================== def trade(context): stocks = get_index_stocks(g.index) dt = context.current_dt target = [] if VERSION == 'v1': # ── V1：12-1动量（标准Jegadeesh & Titman配方）── # 过去12个月收益，跳过最近1个月，选涨幅最大Top20 mom = get_momentum(stocks, dt, lookback=252, skip=21) if mom.empty: log.warn("[V1] 动量数据为空，跳过") return target = mom.nlargest(g.stock_num).index.tolist() log.info(f"[V1] 12-1动量Top3: {mom.nlargest(3).round(3).to_dict()}") elif VERSION == 'v2': # ── V2：6-1动量（中期趋势，窗口更短）── # 过去6个月收益，跳过最近1个月 # 更短窗口 → 对近期趋势更敏感 → 换手更高，可能收益更大但波动也更大 mom = get_momentum(stocks, dt, lookback=126, skip=21) if mom.empty: log.warn("[V2] 动量数据为空，跳过") return target = mom.nlargest(g.stock_num).index.tolist() log.info(f"[V2] 6-1动量Top3: {mom.nlargest(3).round(3).to_dict()}") elif VERSION == 'v3': # ── V3：12-1动量 + ROE>10%（动量+质量）── # 排除：靠业绩暴雷后技术反弹的股票（动量强但基本面差） # 保留：真实盈利驱动的趋势 mom = get_momentum(stocks, dt, lookback=252, skip=21) roe = get_roe(stocks, dt) if mom.empty: log.warn("[V3] 动量数据为空，跳过") return quality_stocks = roe[roe > 10].index if not roe.empty else mom.index mom_filtered = mom.reindex(quality_stocks).dropna() if len(mom_filtered) < g.stock_num: log.warn(f"[V3] 质量过滤后仅{len(mom_filtered)}只，放宽到ROE>5%") quality_stocks = roe[roe > 5].index mom_filtered = mom.reindex(quality_stocks).dropna() target = mom_filtered.nlargest(g.stock_num).index.tolist() log.info(f"[V3] 动量+ROE>10%过滤后{len(mom_filtered)}只，选{len(target)}只") elif VERSION == 'v4': # ── V4：动量(60%) + 低波动(40%)（追涨但控波动）── # 动量策略的缺陷：追涨时容易买到高波动股，回撤大 # 加入低波动因子：在同等动量下，优先选波动小的股票 mom = get_momentum(stocks, dt, lookback=252, skip=21) vol = get_volatility(stocks, dt, window=60) if mom.empty: log.warn("[V4] 动量数据为空，跳过") return mom_s = to_score(mom, higher_is_better=True) # 动量大=分高 vol_s = to_score(vol, higher_is_better=False) # 波动小=分高 if vol_s.empty: log.warn("[V4] 波动率数据为空，退化为V1") target = mom.nlargest(g.stock_num).index.tolist() else: common = mom_s.index.intersection(vol_s.index) total = mom_s[common] * 0.6 + vol_s[common] * 0.4 target = total.sort_values(ascending=False).head(g.stock_num).index.tolist() log.info(f"[V4] 动量×0.6+低波动×0.4，候选{len(common)}只") else: return if not target: log.warn(f"[{VERSION}] 无符合条件股票，跳过") return # ── 执行交易 ── for stock in list(context.portfolio.positions.keys()): if stock not in target: order_target(stock, 0) w = 1.0 / len(target) for stock in target: order_target_value(stock, context.portfolio.total_value * w)