
2026年重磅升级已全面落地!欢迎加入专注财经数据与量化投研的【数据科学实战】知识星球!您将获取持续更新的《财经数据宝典》与《量化投研宝典》,双典协同提供系统化指引;星球内含300篇以上独有高质量文章,深度覆盖策略开发、因子分析、风险管理等核心领域,内容基本每日更新;同步推出的「量化因子专题教程」系列(含完整可运行代码与实战案例),系统详解因子构建、回测与优化全流程,并实现日更迭代。我们持续扩充独家内容资源,全方位赋能您的投研效率与专业成长。无论您是量化新手还是资深研究者,这里都是助您少走弯路、事半功倍的理想伙伴,携手共探数据驱动的投资未来!
在量化交易领域,如何捕捉股价的长期趋势一直是交易者关注的核心问题。传统的线性回归模型往往难以准确描述股票价格的非线性变化规律,而多项式回归则提供了一种更灵活的解决方案。
本文将基于 Palantir Technologies(PLTR)的股票数据,详细介绍如何使用 Python 构建一个基于多项式回归的趋势跟踪交易策略。我们将从数据获取、模型选择、策略构建到回测验证,完整展示整个流程,并特别关注如何避免回测中的前瞻性偏差(Lookahead Bias)。
多项式回归是线性回归的扩展形式。它通过将输入特征 x 转换为多项式特征来捕捉数据中的非线性关系。其数学表达式如下:
其中,N 是多项式的阶数,w0 到 wN 是模型需要学习的参数。
核心挑战:选择合适的多项式阶数 N 至关重要。阶数太低会导致欠拟合,无法捕捉数据的复杂性;阶数太高则会过拟合,把噪声当作信号。
# 导入必要的库import pandas as pdimport numpy as npimport matplotlib.pyplot as pltfrom sklearn.preprocessing import PolynomialFeaturesfrom sklearn.linear_model import LinearRegression# --- 参数设置 ---symbol = "PLTR"start_date = "2021-01-01"end_date = "2026-01-31"# --- 从 EODHD 下载价格数据 ---# 注意:需要替换为你自己的 API 密钥from eodhd import APIClientapi = APIClient("YOUR_API_CODE")resp = api.get_eod_historical_stock_market_data(symbol=symbol, period='d')# 将数据整理为 DataFrame 格式data = pd.DataFrame(resp)data['date'] = pd.to_datetime(data['date'])data.set_index('date', inplace=True)data = data[['open', 'high', 'low', 'close', 'volume']]data.sort_index(inplace=True)# 查看数据结构print(data.info())我们分别测试 1 阶(线性)到 4 阶多项式,观察不同阶数的拟合效果。
from sklearn.metrics import r2_score, mean_absolute_error, mean_squared_errordef fit_polynomial_regression(df, degree): """ 拟合指定阶数的多项式回归模型 参数: df: 包含 close 价格的 DataFrame degree: 多项式阶数 返回: 拟合后的预测值 """ # 创建时间索引作为特征 df = df.copy() df["t"] = np.arange(len(df)) X = df[["t"]] y = df["close"] # 转换为多项式特征 poly = PolynomialFeatures(degree=degree) X_poly = poly.fit_transform(X) # 拟合线性回归模型 model = LinearRegression() model.fit(X_poly, y) # 返回预测结果 return model.predict(X_poly)# 测试不同阶数并计算评估指标results = []for N in [1, 2, 3, 4]: df_test = data.copy().reset_index() df_test["poly_fit"] = fit_polynomial_regression(df_test, N) # 计算评估指标 r2 = r2_score(df_test["close"], df_test["poly_fit"]) mae = mean_absolute_error(df_test["close"], df_test["poly_fit"]) rmse = np.sqrt(mean_squared_error(df_test["close"], df_test["poly_fit"])) results.append({ "阶数": N, "R2": f"{r2:.4f}", "MAE": f"${mae:.2f}", "RMSE": f"${rmse:.2f}" }) print(f"N={N}: R2={r2:.4f}, MAE=${mae:.2f}, RMSE=${rmse:.2f}")不同阶数的对比结果:
从结果可以看出,三次多项式(N=3)达到了最佳平衡点:R² 高达 0.96,MAE 最低。四次多项式的改进微乎其微,反而有过拟合的风险。
如果我们用全部历史数据拟合回归模型,然后基于拟合结果生成交易信号,这就存在前瞻偏差——因为模型在训练时「偷看」了未来的数据。
我们采用扩展窗口(Expanding Window)方法:在每个时间点,只使用该时间点之前的数据来拟合模型。
def bias_free_polynomial_fit(df, degree=3, window=None): """ 无前瞻偏差的多项式回归拟合 参数: df: 包含 close 价格的 DataFrame degree: 多项式阶数,默认为 3(三次) window: 滚动窗口大小,None 表示扩展窗口 返回: 添加了 poly_fit 和 reg_slope 列的 DataFrame """ df = df.copy() df["poly_fit"] = np.nan # 遍历每个时间点,只使用历史数据 for i in range(degree + 1, len(df)): if window: start = max(0, i - window) else: start = 0 # 扩展窗口:使用所有历史数据 # 准备训练数据 X = np.arange(start, i).reshape(-1, 1) y = df["close"].iloc[start:i].values # 拟合多项式回归 poly = PolynomialFeatures(degree=degree) X_poly = poly.fit_transform(X) model = LinearRegression() model.fit(X_poly, y) # 只预测当前时间点的值 X_current = poly.transform(np.array([[i]])) df.at[df.index[i], "poly_fit"] = model.predict(X_current)[0] # 计算回归线斜率(趋势方向) df["reg_slope"] = df["poly_fit"].diff() return df# 应用无偏差拟合df = data.copy().reset_index()df = bias_free_polynomial_fit(df, degree=3)我们的交易逻辑非常简单:
def generate_trading_signals(df): """ 生成交易信号 参数: df: 包含 poly_fit 和 reg_slope 的 DataFrame 返回: 添加了交易信号列的 DataFrame """ df = df.copy() # 做多信号:价格高于回归线且斜率为正 df["reg_long_signal"] = ( (df["close"] > df["poly_fit"]) & (df["reg_slope"] > 0) ) # 将布尔信号转换为持仓状态(1 表示持有,0 表示空仓) df["position"] = df["reg_long_signal"].astype(int) # 标记入场和出场点 df["entry"] = (df["position"] == 1) & (df["position"].shift(1) == 0) df["exit"] = (df["position"] == 0) & (df["position"].shift(1) == 1) return dfdf = generate_trading_signals(df)print(f"总入场次数: {df['entry'].sum()}")print(f"总出场次数: {df['exit'].sum()}")def backtest_strategy(df, transaction_cost=0.001): """ 回测交易策略 参数: df: 包含交易信号的 DataFrame transaction_cost: 单次交易成本,默认 0.1% 返回: 添加了收益计算列的 DataFrame """ df = df.copy() # 计算每日收益率 df["daily_return"] = df["close"].pct_change().fillna(0) # 策略收益 = 每日收益 × 持仓状态 df["strategy_return"] = df["daily_return"] * df["position"] # 扣除交易成本 df["trade"] = df["position"].diff().abs().fillna(0) df["strategy_return_tc"] = df["strategy_return"] - df["trade"] * transaction_cost # 计算累计收益 df["cum_strategy"] = (1 + df["strategy_return_tc"]).cumprod() df["cum_buy_hold"] = (1 + df["daily_return"]).cumprod() return dfdf = backtest_strategy(df)def calculate_performance_metrics(df, return_col="strategy_return_tc", cum_col="cum_strategy"): """ 计算策略绩效指标 参数: df: 回测结果 DataFrame return_col: 日收益列名 cum_col: 累计收益列名 返回: 绩效指标字典 """ # 总收益 total_return = df[cum_col].iloc[-1] - 1 # 年化收益率(CAGR) days = len(df) annual_factor = 252 / days cagr = (1 + total_return) ** annual_factor - 1 # 年化波动率 volatility = df[return_col].std() * np.sqrt(252) # 夏普比率(假设无风险利率为 0) sharpe = cagr / volatility if volatility > 0 else 0 # 最大回撤 roll_max = df[cum_col].cummax() drawdown = (df[cum_col] - roll_max) / roll_max max_drawdown = drawdown.min() return { "总收益": f"{total_return:.2%}", "年化收益率": f"{cagr:.2%}", "年化波动率": f"{volatility:.2%}", "夏普比率": f"{sharpe:.2f}", "最大回撤": f"{max_drawdown:.2%}" }# 计算策略绩效strategy_metrics = calculate_performance_metrics(df)print("=== 策略绩效 ===")for k, v in strategy_metrics.items(): print(f"{k}: {v}")# 计算买入持有绩效bh_metrics = calculate_performance_metrics(df, "daily_return", "cum_buy_hold")print("\n=== 买入持有绩效 ===")for k, v in bh_metrics.items(): print(f"{k}: {v}")为了进一步验证策略的有效性,我们进行样本外测试:
def out_of_sample_test(df, split_ratio=0.9): """ 样本外测试 参数: df: 原始数据 DataFrame split_ratio: 训练集占比 返回: 样本外测试结果 """ # 按时间划分数据 split_index = int(len(df) * split_ratio) df_oos = df.iloc[split_index:].copy() # 重新拟合和生成信号 df_oos = bias_free_polynomial_fit(df_oos.reset_index(drop=True), degree=3) df_oos = generate_trading_signals(df_oos) df_oos = backtest_strategy(df_oos) return df_oosdf_oos = out_of_sample_test(data.reset_index())# 输出样本外测试结果print("=== 样本外测试结果 ===")oos_metrics = calculate_performance_metrics(df_oos)for k, v in oos_metrics.items(): print(f"{k}: {v}")样本外测试结果:
即使在样本外测试期间市场下跌的情况下,策略仍然实现了正收益,并且回撤控制得更好。
为什么胜率低但收益高?
本策略的胜率约为 25%,看起来很低。但这正是趋势跟踪策略的特点:
本文详细介绍了如何使用多项式回归构建趋势跟踪交易策略:
关键要点:
注意:本文仅供学习交流,不构成任何投资建议。实际交易中还需考虑滑点、流动性、市场冲击等因素。
2026年全面升级已落地!【数据科学实战】知识星球核心权益如下:
星球已沉淀丰富内容生态——涵盖量化文章专题教程库、因子日更系列、高频数据集、PyBroker实战课程、专家深度分享与实时答疑服务。无论您是初探量化的学习者,还是深耕领域的从业者,这里都是助您少走弯路、高效成长的理想平台。诚邀加入,共探数据驱动的投资未来!
好文推荐
1. 用 Python 打造股票预测系统:Transformer 模型教程(一)
2. 用 Python 打造股票预测系统:Transformer 模型教程(二)
3. 用 Python 打造股票预测系统:Transformer 模型教程(三)
4. 用 Python 打造股票预测系统:Transformer 模型教程(完结)
6. YOLO 也能预测股市涨跌?计算机视觉在股票市场预测中的应用
9. Python 量化投资利器:Ridge、Lasso 和 Elastic Net 回归详解
好书推荐