# Day 17 (100天计划): 多因子选股模型（回归法） # Regression-based multi-factor model: 历史数据训练 → 预测未来收益 # ==================== 可切换参数（在这里修改）==================== VERSION = 'v1' # 'v1': OLS | 'v2': Ridge | 'v3': Lasso | 'v4': Walk-Forward Ridge # ================================================================ import pandas as pd import numpy as np from jqdata import * from sklearn.linear_model import LinearRegression, Ridge, Lasso from sklearn.preprocessing import StandardScaler """ 回测参数：2020-01-01 至 2026-02-01，100万，沪深300，月度调仓，持仓20只 回归法核心：历史因子值 → 训练回归模型 → 预测未来收益 → 选Top20 vs Day 16打分法：因子值 → 百分位排名 → 加权合成 4个核心因子（复用Day16）： PCF（低PCF好）、20日动量（高涨幅好）、低换手率（低换手好）、资产周转率（高周转好） V1: OLS线性回归（普通最小二乘法） V2: Ridge回归（L2正则化，α=1.0） V3: Lasso回归（L1正则化，α=0.1，可能特征选择） V4: Walk-Forward Ridge（滚动24月训练窗口） 理论预期： - OLS可能过拟合（训练好测试差） - Ridge缓解过拟合（正则化压缩权重） - Lasso可能过度稀疏（重要因子权重被置0） - Walk-Forward最贴近实战（但计算慢） """ def initialize(context): g.stock_num = 20 g.index = '000300.XSHG' g.train_months = 24 # 训练窗口（V1/V2/V3固定训练期，V4滚动） # 模型存储（V1/V2/V3一次性训练，V4每月重训） g.model = None g.scaler = None g.trained = False set_option('use_real_price', True) log.set_level('order', 'error') run_monthly(trade, 1) # ==================== 因子计算（复用Day16） ==================== def get_factor_data(stocks, date): """ 获取4个因子的原始值（非打分） 返回：DataFrame，index=股票代码，columns=['pcf', 'mom', 'turn', 'at'] 因子方向处理： - PCF低好 → 取负号（-PCF越大越好） - 动量高好 → 保持正（动量越大越好） - 换手低好 → 取负号（-换手率越大越好） - 资产周转高好 → 保持正（周转率越大越好） """ # 初始化4列DataFrame result = pd.DataFrame(index=stocks, columns=['pcf', 'mom', 'turn', 'at']) # 1. PCF（低PCF好 → 取负号） try: q = query(valuation.code, valuation.pcf_ratio).filter( valuation.code.in_(stocks), valuation.pcf_ratio > 0 ) df = get_fundamentals(q, date=date) if not df.empty: pcf = df.set_index('code')['pcf_ratio'] result['pcf'] = -pcf # 取负号：PCF低的变成高分 except: pass # 2. 20日动量（高涨幅好） try: price_df = get_price(stocks, end_date=date, count=21, fields=['close'], panel=False, fq='pre') if price_df is not None and not price_df.empty: 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) >= 2: mom[stock] = cl.iloc[-1] / cl.iloc[0] - 1 result['mom'] = pd.Series(mom) except: pass # 3. 换手率（低换手好 → 取负号） try: price_df = get_price(stocks, end_date=date, count=20, fields=['turnover_ratio'], panel=False, fq='pre') if price_df is not None and not price_df.empty: tr_pivot = price_df.pivot(index='time', columns='code', values='turnover_ratio') avg_tr = tr_pivot.mean(axis=0).dropna() result['turn'] = -avg_tr[avg_tr > 0] # 取负号 except: pass # 4. 资产周转率（高周转好） try: q = query( valuation.code, income.total_operating_revenue, balance.total_assets ).filter( valuation.code.in_(stocks), income.total_operating_revenue > 0, balance.total_assets > 0 ) df = get_fundamentals(q, date=date) if not df.empty: df = df.set_index('code') at = df['total_operating_revenue'] / df['total_assets'] result['at'] = at[at > 0] except: pass # 删除全NaN的行（4列都是NaN的股票） result = result.dropna(how='all') # 确保返回的DataFrame有4列（缺失值保留为NaN） return result[['pcf', 'mom', 'turn', 'at']] def get_future_return(stocks, start_date, months=1): """ 计算未来N个月的收益率（标签y） 用于训练时计算真实收益率 """ try: # 计算N个月后的日期 end_date = start_date + pd.DateOffset(months=months) # 获取起始和结束价格 price_start = get_price(stocks, end_date=start_date, count=1, fields=['close'], panel=False, fq='pre') price_end = get_price(stocks, end_date=end_date, count=1, fields=['close'], panel=False, fq='pre') if price_start is None or price_end is None: return pd.Series(dtype=float) p_start = price_start.groupby('code')['close'].last() p_end = price_end.groupby('code')['close'].last() # 计算收益率 ret = p_end / p_start - 1 return ret.dropna() except: return pd.Series(dtype=float) # ==================== 训练函数 ==================== def train_model(context, current_date): """ 训练回归模型（V1/V2/V3一次性训练，V4每月调用） 流程： 1. 回溯过去N个月（g.train_months=24） 2. 每个月：获取因子值（X）+ 下月收益率（y） 3. 合并所有月份数据 4. 标准化X 5. 训练回归模型 6. 返回训练好的模型和scaler """ log.info(f"[{VERSION}] 开始训练模型...") # 获取历史N个月的数据 train_data = [] stocks = get_index_stocks(g.index) # 回溯N个月 for i in range(g.train_months): # 计算该月日期（往前推i+1个月，因为需要用当月预测下月收益） date = current_date - pd.DateOffset(months=g.train_months - i) # 获取因子数据（X） factors = get_factor_data(stocks, date) if factors.empty: continue # 获取未来1月收益率（y） future_ret = get_future_return(factors.index.tolist(), date, months=1) if future_ret.empty: continue # 对齐 common = factors.index.intersection(future_ret.index) if len(common) < 10: # 至少10只股票 continue X = factors.loc[common].fillna(0) y = future_ret.loc[common] # 合并到训练集 for stock in common: row = X.loc[stock].tolist() + [y.loc[stock]] train_data.append(row) if len(train_data) < 50: # 至少50个样本 log.warn(f"训练样本不足: {len(train_data)}") return False # 转为DataFrame cols = ['pcf', 'mom', 'turn', 'at', 'ret'] df_train = pd.DataFrame(train_data, columns=cols) # 去除极端值（y的3倍标准差外） y_mean = df_train['ret'].mean() y_std = df_train['ret'].std() df_train = df_train[ (df_train['ret'] >= y_mean - 3*y_std) & (df_train['ret'] <= y_mean + 3*y_std) ] X_train = df_train[['pcf', 'mom', 'turn', 'at']].values y_train = df_train['ret'].values # 标准化（重要：不同因子量纲不同） g.scaler = StandardScaler() X_train = g.scaler.fit_transform(X_train) # 训练模型 if VERSION == 'v1': g.model = LinearRegression() elif VERSION in ['v2', 'v4']: g.model = Ridge(alpha=1.0) # L2正则化 elif VERSION == 'v3': g.model = Lasso(alpha=0.1, max_iter=10000) # L1正则化 g.model.fit(X_train, y_train) # 打印权重（诊断用） weights = g.model.coef_ intercept = g.model.intercept_ log.info(f"[{VERSION}] 训练完成，样本数={len(df_train)}") log.info(f"因子权重: PCF={weights[0]:.4f}, 动量={weights[1]:.4f}, " f"换手率={weights[2]:.4f}, 资产周转={weights[3]:.4f}, 截距={intercept:.4f}") # 权重诊断（理论预期：PCF+/动量+/换手+/资产周转+） expected_signs = [1, 1, 1, 1] # 因为已经对PCF和换手率取负号 actual_signs = [1 if w > 0 else -1 for w in weights] if expected_signs != actual_signs: log.warn(f"权重符号异常！预期[+,+,+,+]，实际{actual_signs}") return True # ==================== 交易 ==================== def trade(context): stocks = get_index_stocks(g.index) dt = context.current_dt # V1/V2/V3: 第一次运行时训练（固定权重） if VERSION in ['v1', 'v2', 'v3']: if not g.trained: success = train_model(context, dt) if not success: return g.trained = True # V4: 每月重新训练（滚动窗口） elif VERSION == 'v4': success = train_model(context, dt) if not success: return # ──── 获取当前因子值并预测 ──── factors = get_factor_data(stocks, dt) if factors.empty or g.model is None or g.scaler is None: return # 标准化（使用训练时的scaler） X = factors.fillna(0).values X = g.scaler.transform(X) # 预测未来收益率 pred_ret = g.model.predict(X) pred_series = pd.Series(pred_ret, index=factors.index) # 选择预测收益最高的Top20 target = pred_series.sort_values(ascending=False).head(g.stock_num).index.tolist() # 打印Top5预测收益（诊断用） top5 = pred_series.sort_values(ascending=False).head(5) log.info(f"[{VERSION}] Top5预测收益: {top5.to_dict()}") # 卖出不在目标池的 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)