# Day 15: Alpha-191因子库精选 # 国泰君安2017《基于短周期价量特征的多因子选股体系》 # ==================== 可切换参数（在这里修改）==================== VERSION = 'v1' # 'v1': Alpha#001量价背离 | 'v2': Alpha#002影线因子 | 'v3': Alpha#028类KDJ-J | 'v4': 等权合成 # ================================================================ import pandas as pd import numpy as np from jqdata import * """ 回测参数：2020-01-01 至 2026-02-01，100万，沪深300，月度调仓，持仓20只 Alpha-191 vs Alpha-101： Alpha-101: WorldQuant 2015，截面rank操作为主 Alpha-191: 国泰君安 2017，短周期时序操作，引入SMA(x,n,m)算子 V1: Alpha#001 = -corr(rank(delta(log(vol),1)), rank((close-open)/open), 6) 量价背离反转因子：成交量加速度排名 与 当日振幅排名 负相关 V2: Alpha#002 = -delta(((close-low)-(high-close))/(high-low), 1) 影线因子：CTR（收盘位置指标）变化的负值 V3: Alpha#028 = 3*SMA(KD值,3,1) - 2*SMA(SMA(KD值,3,1),3,1) 类KDJ-J值 超卖反弹因子：J值越低=超卖越严重=均值回归买入信号 V4: 等权合成（方向统一后相加） """ 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 winsorize(series): return series.clip(series.quantile(0.05), series.quantile(0.95)) def standardize(series): std = series.std() if std == 0: return series * 0 return (series - series.mean()) / std def calc_sma(series, n, m): """ Alpha-191 SMA算子: SMA(x, n, m) = m/n * x + (1 - m/n) * SMA_prev 等价于 EWM(alpha=m/n, adjust=False) """ return series.ewm(alpha=m / n, adjust=False).mean() # ==================== Alpha因子计算 ==================== def get_alpha001(stocks, date): """ Alpha#001: -corr(rank(delta(log(vol),1)), rank((close-open)/open), 6) 核心逻辑：成交量加速度的时序排名 与 当日振幅的时序排名 负相关 放量大涨（量价同步）→ corr高 → 因子值低 → 短期动量透支 → 反转信号 """ try: price_df = get_price(stocks, end_date=date, count=8, fields=['open', 'close', 'volume'], panel=False, fq='pre') if price_df is None or price_df.empty: return pd.Series(dtype=float) open_p = price_df.pivot(index='time', columns='code', values='open') close_p = price_df.pivot(index='time', columns='code', values='close') vol_p = price_df.pivot(index='time', columns='code', values='volume') alpha = {} for stock in close_p.columns: op = open_p[stock].dropna() cl = close_p[stock].dropna() vo = vol_p[stock].dropna() if len(cl) < 7: continue # delta(log(volume), 1) log_vol = np.log(vo.replace(0, np.nan)).dropna() delta_lv = log_vol.diff(1).dropna() # (close - open) / open，与delta_lv时序对齐 ret_open = ((cl - op) / op).iloc[1:] if len(delta_lv) < 6 or len(ret_open) < 6: continue x = delta_lv.iloc[-6:].rank() y = ret_open.iloc[-6:].rank() corr_val = x.corr(y) if not pd.isna(corr_val): alpha[stock] = -corr_val result = pd.Series(alpha) if result.empty: return result return standardize(winsorize(result)) except Exception as e: log.error(f"Alpha#001失败: {e}") return pd.Series(dtype=float) def get_alpha002(stocks, date): """ Alpha#002: -delta(((close-low)-(high-close))/(high-low), 1) 核心逻辑：CTR（收盘位置指标）一阶差分的负值 CTR ∈ [-1,+1]，衡量收盘价在日内高低区间的位置 昨日CTR高今日CTR低 → 上影线出现 → 上方压力增大 → 因子值正 → 短期方向信号 """ try: price_df = get_price(stocks, end_date=date, count=3, fields=['high', 'low', 'close'], panel=False, fq='pre') if price_df is None or price_df.empty: return pd.Series(dtype=float) high_p = price_df.pivot(index='time', columns='code', values='high') low_p = price_df.pivot(index='time', columns='code', values='low') close_p = price_df.pivot(index='time', columns='code', values='close') alpha = {} for stock in close_p.columns: hi = high_p[stock].dropna() lo = low_p[stock].dropna() cl = close_p[stock].dropna() if len(hi) < 2: continue hl_range = (hi - lo).replace(0, np.nan) # CTR = [(close-low) - (high-close)] / (high-low) # = 2*(close-low)/(high-low) - 1，范围[-1, +1] ctr = ((cl - lo) - (hi - cl)) / hl_range delta_ctr = ctr.diff(1).dropna() if delta_ctr.empty: continue alpha[stock] = -delta_ctr.iloc[-1] result = pd.Series(alpha) if result.empty: return result return standardize(winsorize(result)) except Exception as e: log.error(f"Alpha#002失败: {e}") return pd.Series(dtype=float) def get_alpha028(stocks, date): """ Alpha#028: 3*SMA(KD,3,1) - 2*SMA(SMA(KD,3,1),3,1) KD = (close - LLV(low,9)) / (HHV(high,9) - LLV(low,9)) * 100 核心逻辑：类KDJ中的J值 J < 0 → 超卖区间 → 均值回归买入信号 J > 100 → 超买区间 → 避开 注意：此因子使用 ascending=True 排序（选J值最低的超卖股） """ try: price_df = get_price(stocks, end_date=date, count=15, fields=['high', 'low', 'close'], panel=False, fq='pre') if price_df is None or price_df.empty: return pd.Series(dtype=float) high_p = price_df.pivot(index='time', columns='code', values='high') low_p = price_df.pivot(index='time', columns='code', values='low') close_p = price_df.pivot(index='time', columns='code', values='close') alpha = {} for stock in close_p.columns: hi = high_p[stock].dropna() lo = low_p[stock].dropna() cl = close_p[stock].dropna() if len(cl) < 9: continue hhv9 = hi.rolling(9).max() llv9 = lo.rolling(9).min() hl_range = (hhv9 - llv9).replace(0, np.nan) kd = ((cl - llv9) / hl_range * 100).dropna() if len(kd) < 3: continue sma1 = calc_sma(kd, 3, 1) sma2 = calc_sma(sma1, 3, 1) j_val = 3 * sma1.iloc[-1] - 2 * sma2.iloc[-1] if not pd.isna(j_val): alpha[stock] = j_val result = pd.Series(alpha) if result.empty: return result return standardize(winsorize(result)) except Exception as e: log.error(f"Alpha#028失败: {e}") return pd.Series(dtype=float) # ==================== 因子合成 ==================== def combine_equal(factors): common = None for f in factors.values(): if common is None: common = set(f.index) else: common = common.intersection(set(f.index)) common = list(common) if not common: return pd.Series(dtype=float) result = pd.Series(0.0, index=common) for f in factors.values(): result += f.loc[common] / len(factors) return result # ==================== 交易 ==================== def trade(context): stocks = get_index_stocks(g.index) if VERSION == 'v1': factor = get_alpha001(stocks, context.current_dt) ascending = False # 因子值高 = 量价背离强 = 买入 elif VERSION == 'v2': factor = get_alpha002(stocks, context.current_dt) ascending = False # 因子值高 = CTR下降（多头结构延续） elif VERSION == 'v3': factor = get_alpha028(stocks, context.current_dt) ascending = True # 因子值低 = J值低 = 超卖 = 均值回归买入 elif VERSION == 'v4': a1 = get_alpha001(stocks, context.current_dt) a2 = get_alpha002(stocks, context.current_dt) a28 = get_alpha028(stocks, context.current_dt) # a28取负值统一方向：低J值 → 高分数 → ascending=False a28_inv = -a28 if not a28.empty else a28 valid = {} if not a1.empty: valid['a1'] = a1 if not a2.empty: valid['a2'] = a2 if not a28_inv.empty: valid['a28'] = a28_inv if not valid: return factor = combine_equal(valid) ascending = False else: factor = get_alpha001(stocks, context.current_dt) ascending = False if factor.empty: return target = factor.sort_values(ascending=ascending).head(g.stock_num).index.tolist() for stock in list(context.portfolio.positions.keys()): if stock not in target: order_target(stock, 0) if target: w = 1.0 / len(target) for stock in target: order_target_value(stock, context.portfolio.total_value * w)