Tsfresh + TA-Lib + LightGBM ：A 股市场量化投资策略实战入门

Tsfresh + TA-Lib + LightGBM ：A 股市场量化投资策略实战入门

本项目以 A 股市场为研究对象，通过量化技术对市场数据进行分析，构建量化投资策略，并利用历史数据回测验证策略的有效性。项目旨在为量化技术初学者提供一个系统的学习框架，帮助读者掌握从数据获取到策略评估的全流程操作。
文中内容仅限技术学习与代码实践参考，市场存在不确定性，技术分析需谨慎验证，不构成任何投资建议。适合量化新手建立系统认知，为策略开发打下基础。

‼️ 学习基础知识推荐阅读 Python 金融科技 技术专栏 🚀：

• tsfresh：时间序列特征自动提取与应用 🔥
• 深入TA-Lib：量化技术指标详解 🔥
• Pandas+PyArrow：股票数据存储 Parquet 入门指引 🔥
• Dask：Python高效并行计算利器 🔥

1. 项目概述

1.1 项目目标

1. 数据获取与存储
   使用 tushare 工具获取 A 股股票的历史日线数据，确保数据的准确性和完整性。借助 Dask 和 Parquet 技术进行数据存储，优化数据读写效率，为后续分析提供支持。

2. 特征提取与筛选
   利用 tsfresh 提取时间序列特征，从历史数据中挖掘与股票未来收益率相关的特征。通过相关性分析和特征筛选，确定对预测目标具有显著影响的特征集合。

3. 模型训练与优化
   基于筛选后的特征，使用 LightGBM 机器学习算法进行模型训练，并通过 Optuna 进行超参数优化。目标是构建一个能够有效预测股票未来收益率的模型。

4. 交易策略制定与回测评估
   根据模型预测结果，设计量化交易策略，并通过历史数据回测验证策略性能。优化策略的信号生成和买卖时机，计算策略评价指标（如夏普比率、最大回撤等），并绘制策略表现曲线。

5. 风险收益分析与策略优化
   分析策略的风险收益特征，结合回测结果优化策略参数。通过详细解读策略表现，帮助读者理解策略的优势与不足，为进一步优化提供方向。

1.2 项目意义

本项目旨在为量化技术初学者提供一个完整的实战案例，涵盖数据处理、特征工程、模型训练、策略回测和风险分析等关键环节。通过系统的学习和实践，读者可以掌握量化技术的基本方法，理解量化策略的构建过程，并为进一步深入研究打下坚实基础。

2. 技术介绍

2.1 Tushare

Tushare 是一个免费且开源的金融数据接口库，它主要服务于Python开发者。通过Tushare，用户可以轻松获取股票、期货、外汇等多种类型的金融市场数据。其特点包括：

• 丰富的数据源：支持从A股到美股、港股等全球多个市场的数据。
• 易于使用：提供了简洁的API接口，方便用户快速上手。
• 社区活跃：拥有庞大的用户群体和活跃的社区支持，不断更新维护。
• 文档详尽：官方提供了详细的文档教程，帮助新手入门。

2.2 Tsfresh

Tsfresh (Time Series Feature extraction based on scalable hypothesis tests) 是一个强大的时间序列特征提取工具，旨在自动识别出对于预测任务有用的特征。它的优势在于：

• 自动化：能够自动生成数百种可能对模型有用的特征，减少了手动选择特征的工作量。
• 高效性：利用并行处理技术提高特征提取的速度。
• 灵活性：不仅限于特定类型的时间序列数据，适用于多种场景下的数据分析。
• 集成友好：与Scikit-learn等流行机器学习库良好兼容。

2.3 Dask

Dask 是一种灵活的并行计算库，特别适合处理超出内存限制的大规模数据集。它的特性有：

• 扩展性强：既可以在单机上运行也可以部署在分布式集群中。
• 动态调度：根据任务需求智能调整资源分配。
• 接口熟悉：设计风格类似于Pandas和NumPy，降低了学习成本。
• 生态系统丰富：支持多种存储格式如HDF5、Parquet等，并与其他大数据框架（如Spark）具有良好的互操作性。

2.4 Parquet

Parquet 是一种列式存储格式，广泛应用于大数据领域。相比于传统的行式存储，Parquet的优势在于：

• 压缩效率高：采用列式存储可以更有效地压缩相同类型的数据。
• 查询速度快：仅读取需要分析的列，避免了不必要的I/O开销。
• 跨平台兼容：被许多主流的大数据处理系统支持，如Apache Spark, Apache Hadoop等。
• 元数据丰富：文件内嵌入了详细的统计信息，便于进行预处理优化。

2.5 LightGBM

LightGBM 是由微软开发的一种基于梯度提升决策树算法的机器学习框架，以其实现速度和准确性著称。主要优点如下：

• 高效性：采用了GOSS (Gradient-based One-Side Sampling) 和 EFB (Exclusive Feature Bundling) 等创新技术来加速训练过程。
• 内存占用小：相较于其他同类算法，在处理大规模数据时更加节省内存。
• 易于使用：提供了Python、R等多个语言版本的支持，并且配置选项丰富。
• 性能优越：在多项基准测试中表现出色，特别是在分类问题上。

2.6 Optuna

Optuna 是一款专为超参数优化设计的开源软件库，可以帮助研究人员或工程师更快地找到最优的模型配置。特色功能包括：

• 简单易用：只需几行代码即可启动搜索过程。
• 灵活性高：支持定义复杂的搜索空间以及自定义目标函数。
• 可视化好：内置了丰富的图表绘制功能，便于观察优化过程。
• 多后端支持：既可以单独运行也可以集成到现有的机器学习流水线中。

2.7 TA-Lib

TA-Lib (Technical Analysis Library) 是一个广泛使用的开源技术分析库，主要用于金融市场的技术指标计算。其特点包括：

• 全面的技术指标：提供了超过150种常用的技术分析指标，如移动平均线、MACD、RSI等。
• 高性能：底层实现是用C语言编写的，确保了计算速度。
• 多语言支持：除了原生的C语言外，还提供了Python、Java、C#等多种语言的绑定。
• 易于集成：可以很容易地与Pandas等数据处理库结合使用，适用于各种金融分析和交易策略开发。
• 社区活跃：拥有活跃的用户社区和丰富的文档资源，便于学习和使用。

3. 项目结构

project_root/
├── data/
│   ├── raw_data.parquet
│   ├── selected_features.parquet
│   └── target.parquet
├── src/
│   ├── data_loader.py
│   ├── feature_engineering.py
│   ├── final_model.txt
│   ├── model_training.py
│   ├── strategy_backtesting.py
│   └── utils.py
├── README.md
└── requirements.txt

详细说明

• data/：存放数据文件的目录。

  • raw_data.parquet：从 Tushare 获取的原始股票历史日线数据。
  • selected_features.parquet：经过特征工程处理后选择的特征数据。
  • target.parquet：目标变量数据（未来收益率）。

• src/：存放源代码文件的目录。

  • data_loader.py：用于从 Tushare 获取股票历史日线数据并保存为 Parquet 文件。
  • feature_engineering.py：用于提取时间序列特征并选择与目标变量相关的特征。
  • final_model.txt：训练好的 LightGBM 模型文件。
  • model_training.py：用于训练机器学习模型，并使用 Optuna 进行超参数优化。
  • strategy_backtesting.py：用于回测策略并评估策略性能。
  • utils.py：包含一些辅助函数，如计算未来收益率、技术指标等。

• README.md：项目说明文档，包含项目概述、安装步骤、使用方法等信息。

  # 安装项目依赖：
  pip install -r requirements.txt
  cd src
  # 数据获取
  python data_loader.py
  # 特征工程
  python feature_engineering.py
  # 模型训练
  python model_training.py
  # 策略回测
  python strategy_backtesting.py
  

• requirements.txt：项目依赖库列表，用于安装所需的 Python 包。

  tushare
  tsfresh
  lightgbm
  optuna
  pandas
  numpy
  scikit-learn
  ta-lib
  pyarrow
  matplotlib
  

3. 代码实现

3.1 数据加载 (data_loader.py)

# src/data_loader.py
import tushare as ts

def load_stock_data(stock_code, start_date, end_date, save_path="../data/raw_data.parquet"):
    """
    从Tushare API下载股票历史日线数据并保存。

    :param stock_code: 股票代码
    :param start_date: 开始日期
    :param end_date: 结束日期
    :param save_path: 保存路径
    """
    # Tushare Pro API token
    ts.set_token("your_token")  # 设置 Tushare API 的访问令牌
    pro = ts.pro_api()  # 初始化 Tushare Pro API 客户端

    df = pro.daily(ts_code=stock_code, start_date=start_date, end_date=end_date)  # 获取指定股票在指定日期范围内的日线数据
    df.sort_values(by="trade_date", inplace=True)  # 按交易日期排序

    df.to_parquet(
        path=save_path,
        engine="pyarrow",
        compression="snappy",
        index=False,
    )  # 将数据保存为 Parquet 文件，使用 Snappy 压缩

    print(f"Data saved to {save_path}")  # 打印保存路径

if __name__ == "__main__":
    load_stock_data("600519.SH", "20200101", "20240101")  # 贵州茅台（600519）

3.2 特征工程 (feature_engineering.py)

# src/feature_engineering.py
import pandas as pd
import talib
from sklearn.ensemble import RandomForestRegressor
from sklearn.feature_selection import RFE, SelectFromModel, SelectKBest, f_regression, mutual_info_regression
from sklearn.linear_model import Lasso
from tsfresh import extract_features
from tsfresh.utilities.dataframe_functions import impute

from utils import calculate_compound_returns, calculate_future_return_class, calculate_future_returns, calculate_technical_indicator_changes

def select_features_with_rfe(features, target, n_features_to_select=10):
    """
    使用递归特征消除选择特征
    """
    estimator = RandomForestRegressor(n_estimators=100, random_state=42)  # 初始化随机森林回归器
    selector = RFE(estimator, n_features_to_select=n_features_to_select)  # 初始化 RFE 选择器
    selector = selector.fit(features, target)  # 训练 RFE 选择器
    selected_features = features.iloc[:, selector.support_]  # 选择特征
    return selected_features

def select_features_with_lasso(features, target):
    """
    使用LASSO回归选择特征
    """
    lasso = Lasso(alpha=0.1)  # 初始化 LASSO 回归
    lasso.fit(features, target)  # 训练 LASSO 回归
    model = SelectFromModel(lasso, prefit=True)  # 初始化特征选择器
    selected_features = model.transform(features)  # 选择特征
    return pd.DataFrame(selected_features, columns=features.columns[model.get_support()])

def add_technical_indicators(df):
    """
    添加技术指标
    """
    df["SMA_5"] = talib.SMA(df["close"], timeperiod=5)  # 5日简单移动平均
    df["SMA_10"] = talib.SMA(df["close"], timeperiod=10)  # 10日简单移动平均
    df["RSI"] = talib.RSI(df["close"], timeperiod=14)  # 14日相对强弱指数
    df["MACD"], df["MACD_signal"], _ = talib.MACD(df["close"], fastperiod=12, slowperiod=26, signalperiod=9)  # MACD 指标
    df["EMA_12"] = talib.EMA(df["close"], timeperiod=12)  # 12日指数移动平均
    df["EMA_26"] = talib.EMA(df["close"], timeperiod=26)  # 26日指数移动平均
    df["ATR"] = talib.ATR(df["high"], df["low"], df["close"], timeperiod=14)  # 14日平均真实波动幅度
    df["ADX"] = talib.ADX(df["high"], df["low"], df["close"], timeperiod=14)  # 14日平均趋向指数
    return df

def add_time_window_features(df):
    """
    添加时间窗口特征
    """
    df["close_mean_5d"] = df["close"].rolling(window=5).mean()  # 5日收盘价均值
    df["close_std_5d"] = df["close"].rolling(window=5).std()  # 5日收盘价标准差
    df["close_mean_10d"] = df["close"].rolling(window=10).mean()  # 10日收盘价均值
    df["close_std_10d"] = df["close"].rolling(window=10).std()  # 10日收盘价标准差
    return df

def add_manual_features(df):
    """
    添加手工特征
    """
    df["volume_change"] = df["vol"].pct_change()  # 成交量变化率
    df["price_volatility"] = df["close"].rolling(window=5).std()  # 5日价格波动率
    df["price_momentum"] = (df["close"] - df["close"].shift(5)) / df["close"].shift(5)  # 5日价格动量
    return df

def extract_and_select_features(data_path, target_variable):
    """
    提取时间序列特征并选择与目标变量相关的特征。
    """
    df = pd.read_parquet(data_path)  # 读取 Parquet 文件
    df["date"] = pd.to_datetime(df["trade_date"], format="%Y%m%d")  # 将交易日期转换为 datetime 格式
    df.sort_values(by="date", inplace=True)  # 按日期排序

    df = calculate_future_returns(df, days_list=[3, 5, 7, 10])  # 计算多个时间窗口的未来收益率
    df = calculate_compound_returns(df, days=5)  # 计算复合收益率
    df = calculate_future_return_class(df, days=5)  # 计算分类目标
    df = calculate_technical_indicator_changes(df, indicator="RSI", period=14, days=5)  # 计算技术指标变化
    df = add_technical_indicators(df)  # 添加技术指标
    df = add_time_window_features(df)  # 添加时间窗口特征
    df = add_manual_features(df)  # 添加手工特征

    df.dropna(inplace=True)  # 删除包含 NaN 值的行
    df["id"] = range(0, len(df))  # 添加 id 列
    df.set_index("id", inplace=True)  # 设置 id 为索引
    df.reset_index(inplace=True)  # 重置索引

    target = pd.DataFrame(df[target_variable])  # 提取目标变量

    extracted_features = extract_features(df, column_id="id", column_sort="date", column_value="close")  # 提取时间序列特征
    impute(extracted_features)  # 填充缺失值

    selected_features = select_features_with_lasso(extracted_features, target)  # 选择特征

    return selected_features, target

if __name__ == "__main__":
    features, target = extract_and_select_features("../data/raw_data.parquet", "future_return_5d")  # 提取和选择特征
    features.to_parquet("../data/selected_features.parquet", index=False)  # 保存特征
    target.to_parquet("../data/target.parquet", index=False)  # 保存目标变量

3.3 模型训练 (model_training.py)

# src/model_training.py
import re
import lightgbm as lgb
import numpy as np
import optuna
import pandas as pd
from sklearn.metrics import mean_squared_error
from sklearn.model_selection import KFold, train_test_split
from sklearn.preprocessing import StandardScaler

def clean_feature_names(df):
    df.columns = [re.sub(r"[^a-zA-Z0-9_]", "", col) for col in df.columns]  # 清理特征名称，去除特殊字符
    return df

def preprocess_data(features, target):
    features = features.interpolate(method="linear").fillna(features.mean())  # 插值填充缺失值
    scaler = StandardScaler()  # 初始化标准化器
    features = pd.DataFrame(scaler.fit_transform(features), columns=features.columns)  # 标准化特征
    return features, target

def objective(trial, X_train, y_train, X_val, y_val):
    params = {
        "objective": "regression",  # 目标函数
        "metric": "mse",  # 评价指标
        "boosting_type": "gbdt",  # 提升类型
        "num_leaves": trial.suggest_int("num_leaves", 2, 256),  # 叶子节点数
        "learning_rate": trial.suggest_loguniform("learning_rate", 1e-5, 1e-1),  # 学习率
        "max_depth": trial.suggest_int("max_depth", 3, 15),  # 最大深度
        "min_child_samples": trial.suggest_int("min_child_samples", 5, 100),  # 最小样本数
        "subsample": trial.suggest_uniform("subsample", 0.5, 1.0),  # 子采样比例
        "colsample_bytree": trial.suggest_uniform("colsample_bytree", 0.5, 1.0),  # 列采样比例
        "reg_alpha": trial.suggest_loguniform("reg_alpha", 1e-5, 1e2),  # L1 正则化项
        "reg_lambda": trial.suggest_loguniform("reg_lambda", 1e-5, 1e2),  # L2 正则化项
    }

    dtrain = lgb.Dataset(X_train, label=y_train)  # 创建训练数据集
    dval = lgb.Dataset(X_val, label=y_val)  # 创建验证数据集

    callbacks = [
        lgb.early_stopping(stopping_rounds=50, first_metric_only=True, verbose=False),  # 早停回调
        lgb.log_evaluation(period=0),  # 禁用日志输出
    ]

    model = lgb.train(params, dtrain, valid_sets=[dtrain, dval], valid_names=["train", "val"], num_boost_round=1000, callbacks=callbacks)  # 训练模型

    y_pred = model.predict(X_val)  # 预测验证集
    mse = mean_squared_error(y_val, y_pred)  # 计算均方误差

    return mse

def train_model(features_path, target_path, n_trials=100):
    features = pd.read_parquet(features_path)  # 读取特征
    target = pd.read_parquet(target_path)  # 读取目标变量

    features = clean_feature_names(features)  # 清理特征名称
    features, target = preprocess_data(features, target)  # 预处理数据

    X_train, X_val, y_train, y_val = train_test_split(features, target, test_size=0.2, random_state=42)  # 划分训练集和验证集

    study = optuna.create_study(direction="minimize")  # 创建 Optuna 研究
    study.optimize(lambda trial: objective(trial, X_train, y_train, X_val, y_val), n_trials=n_trials)  # 优化超参数

    best_params = study.best_params  # 获取最优参数
    best_mse = study.best_value  # 获取最优 MSE

    print(f"Best MSE: {best_mse}")  # 打印最优 MSE
    print(f"Best Parameters: {best_params}")  # 打印最优参数

    return best_params

def train_final_model(best_params, features, target):
    features, target = preprocess_data(features, target)  # 预处理数据
    kf = KFold(n_splits=5, shuffle=True, random_state=42)  # 初始化 K-Fold 交叉验证
    mse_scores = []
    for train_index, val_index in kf.split(features):  # 进行交叉验证
        X_train, X_val = features.iloc[train_index], features.iloc[val_index]
        y_train, y_val = target.iloc[train_index], target.iloc[val_index]
        dtrain = lgb.Dataset(X_train, label=y_train)  # 创建训练数据集
        dval = lgb.Dataset(X_val, label=y_val)  # 创建验证数据集
        callbacks = [
            lgb.early_stopping(stopping_rounds=50, first_metric_only=True, verbose=False),  # 早停回调
            lgb.log_evaluation(period=0),  # 禁用日志输出
        ]
        model = lgb.train(best_params, dtrain, valid_sets=[dtrain, dval], valid_names=["train", "val"], num_boost_round=1000, callbacks=callbacks)  # 训练模型
        y_pred = model.predict(X_val)  # 预测验证集
        mse = mean_squared_error(y_val, y_pred)  # 计算均方误差
        mse_scores.append(mse)
    avg_mse = np.mean(mse_scores)  # 计算平均 MSE
    print(f"Final Model Average MSE: {avg_mse}")  # 打印平均 MSE
    return model

if __name__ == "__main__":
    features_path = "../data/selected_features.parquet"
    target_path = "../data/target.parquet"

    best_params = train_model(features_path, target_path, n_trials=100)  # 训练模型并优化超参数

    features = pd.read_parquet(features_path)  # 读取特征
    target = pd.read_parquet(target_path)  # 读取目标变量

    features = clean_feature_names(features)  # 清理特征名称

    final_model = train_final_model(best_params, features, target)  # 训练最终模型

    final_model.save_model("final_model.txt")  # 保存模型

3.4 策略回测 (strategy_backtesting.py)

# strategy_backtesting.py
import re
import lightgbm as lgb
import numpy as np
import pandas as pd
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from sklearn.preprocessing import StandardScaler

def clean_feature_names(df):
    df.columns = [re.sub(r"[^a-zA-Z0-9_]", "", col) for col in df.columns]  # 清理特征名称，去除特殊字符
    return df

def preprocess_data(features, target=None):
    features = features.fillna(features.mean())  # 填充缺失值
    scaler = StandardScaler()  # 初始化标准化器
    features = pd.DataFrame(scaler.fit_transform(features), columns=features.columns)  # 标准化特征

    if target is not None:
        return features, target
    else:
        return features

def load_model(model_path):
    model = lgb.Booster(model_file=model_path)  # 加载 LightGBM 模型
    return model

def backtest_strategy(features, target, model, threshold=0.5):
    features = preprocess_data(features)  # 预处理特征

    predictions = model.predict(features)  # 预测

    print(f"Predictions: Min={predictions.min()}, Max={predictions.max()}, Mean={predictions.mean()}")  # 打印预测值统计信息

    mse = mean_squared_error(target, predictions)  # 计算均方误差
    mae = mean_absolute_error(target, predictions)  # 计算平均绝对误差
    r2 = r2_score(target, predictions)  # 计算 R² 分数

    print(f"Mean Squared Error (MSE): {mse:.6f}")  # 打印均方误差
    print(f"Mean Absolute Error (MAE): {mae:.6f}")  # 打印平均绝对误差
    print(f"R² Score: {r2:.4f}")  # 打印 R² 分数

    signals = (predictions > threshold).astype(int)  # 生成交易信号

    print(f"Signals: Sum={signals.sum()}, Count={len(signals)}, Non-zero count={(signals > 0).sum()}")  # 打印信号统计信息

    daily_returns = target * signals  # 计算每日收益

    daily_returns_series = pd.Series(daily_returns)  # 将每日收益转换为 Pandas Series

    cumulative_returns = (1 + daily_returns_series).cumprod() - 1  # 计算累积收益

    total_return = cumulative_returns.iloc[-1]  # 计算总收益

    sharpe_ratio = np.sqrt(252) * (daily_returns_series.mean() / daily_returns_series.std())  # 计算夏普比率

    print(f"Total Return: {total_return:.4f}")  # 打印总收益
    print(f"Sharpe Ratio: {sharpe_ratio:.4f}")  # 打印夏普比率

    return cumulative_returns, total_return, sharpe_ratio

if __name__ == "__main__":
    features_path = "../data/selected_features.parquet"
    target_path = "../data/target.parquet"

    features = pd.read_parquet(features_path)  # 读取特征
    target = pd.read_parquet(target_path).values.flatten()  # 读取目标变量

    features = clean_feature_names(features)  # 清理特征名称

    model_path = "final_model.txt"
    model = load_model(model_path)  # 加载模型

    cumulative_returns, total_return, sharpe_ratio = backtest_strategy(features, target, model, threshold=0.001)  # 回测策略

    import matplotlib.pyplot as plt

    plt.figure(figsize=(14, 7))
    plt.plot(cumulative_returns, label="Cumulative Returns")  # 绘制累积收益曲线
    plt.title("Cumulative Returns Over Time")
    plt.xlabel("Time")
    plt.ylabel("Cumulative Returns")
    plt.legend()
    plt.show()

3.5 辅助函数 (utils.py)

# src/utils.py
import numpy as np
import pandas as pd
import talib


def calculate_future_returns(df, days_list=[3, 5, 7, 10]):
    """
    计算多个时间窗口的未来收益率。

    :param df: 包含收盘价的时间序列数据
    :param days_list: 未来天数列表
    :return: 包含未来收益率的DataFrame
    """
    for days in days_list:
        target_variable = f"future_return_{days}d"
        df[target_variable] = df["close"].pct_change(periods=-days).shift(days)
    
    df.dropna(inplace=True)
    return df


def calculate_compound_returns(df, days=5):
    """
    计算复合收益率。
    :param df: 包含收盘价的时间序列数据
    :param days: 未来天数
    :return: 包含复合收益率的DataFrame
    """
    target_variable = f"compound_return_{days}d"
    df[target_variable] = (df["close"].shift(-days) / df["close"]) - 1
    df.dropna(subset=[target_variable], inplace=True)
    return df


def calculate_future_return_class(df, days=5):
    """
    计算未来收益率并将其转换为分类目标。
    :param df: 包含收盘价的时间序列数据
    :param days: 未来天数
    :return: 包含分类目标的DataFrame
    """
    target_variable = f"future_return_{days}d"
    df[target_variable] = df["close"].pct_change(periods=-days).shift(days)

    # 将收益率转换为分类目标
    df[f"{target_variable}_class"] = pd.cut(
        df[target_variable],
        bins=[-np.inf, -0.01, 0.01, np.inf],
        labels=["down", "flat", "up"],
    )

    df.dropna(subset=[f"{target_variable}_class"], inplace=True)
    return df


def calculate_technical_indicator_changes(df, indicator="RSI", period=14, days=5):
    """
    计算技术指标的变化。
    :param df: 包含收盘价的时间序列数据
    :param indicator: 技术指标名称
    :param period: 技术指标的时间周期
    :param days: 未来天数
    :return: 包含技术指标变化的DataFrame
    """
    if indicator == "RSI":
        df[f"{indicator}_{period}"] = talib.RSI(df["close"], timeperiod=period)
    elif indicator == "MACD":
        macd, signal, _ = talib.MACD(
            df["close"], fastperiod=12, slowperiod=26, signalperiod=9
        )
        df[f"{indicator}_{period}"] = macd - signal
    else:
        raise ValueError(f"Unsupported indicator: {indicator}")

    target_variable = f"{indicator}_change_{days}d"
    df[target_variable] = (
        df[f"{indicator}_{period}"].pct_change(periods=-days).shift(days)
    )

    df.dropna(subset=[target_variable], inplace=True)
    return df

4. 关键技术参数说明

5. 策略评价指标

在量化投资策略的评估中，评价指标是衡量策略性能的关键工具。以下是一些常用的评价指标及其意义：

5.1 预测准确性指标

• 均方误差 (MSE)：衡量预测值与实际值之间的差异。较低的 MSE 表示模型预测更准确。计算公式为：

  $$\text{MSE}=\frac{1}{n}\sum _{i=1}^n(y_i-{\hat{y}}_i{)}^2$$
  其中，$y_i$ 是实际值，${\hat{y}}_i$ 是预测值，$n$ 是样本数量。

• 平均绝对误差 (MAE)：衡量预测值与实际值之间的绝对差异。较低的 MAE 表示模型预测更准确。计算公式为：

  $$\text{MAE}=\frac{1}{n}\sum _{i=1}^n|y_i-{\hat{y}}_i|$$
  其中，$y_i$ 是实际值，${\hat{y}}_i$ 是预测值，$n$ 是样本数量。

• R² 分数：衡量模型解释数据变异性的能力。较高的 R² 分数表示模型解释能力强。R² 的取值范围为 [0, 1]，值越接近 1 表示模型拟合效果越好。计算公式为：

  $$R^2=1-\frac{{\sum }_{i=1}^n(y_i-{\hat{y}}_i{)}^2}{{\sum }_{i=1}^n(y_i-\bar{y}{)}^2}$$
  其中，$y_i$ 是实际值，${\hat{y}}_i$ 是模型的预测值，$\bar{y}$ 是实际值的均值，$n$ 是样本数量。

5.2 风险与收益指标

• 夏普比率：衡量单位风险的超额回报，越高越好。计算公式为：

  $$\text{夏普比率}=\frac{R_p-R_f}{{\sigma }_p}$$
  其中，$R_p$ 是策略的预期收益率，$R_f$ 是无风险利率，${\sigma }_p$ 是策略收益率的标准差。较高的夏普比率表示策略在承担相同风险的情况下获得了更高的回报。

• 累积收益：反映策略的整体表现。通过计算策略在每个时间点的净值变化，绘制累积收益曲线。平滑上升的累积收益曲线表示策略表现良好，且波动较小。

• 总收益：反映策略的总体盈利能力，计算公式为：

  $$\text{总收益}=\frac{\text{最终资产价值}-\text{初始资产价值}}{\text{初始资产价值}}\times 100\%$$
  较高的总收益表示策略盈利能力强。

• 最大回撤 (MDD)：衡量策略在选定周期内可能出现的最大损失。计算公式为：

  $$\text{MDD}=\max \left(\frac{\text{高点净值}-\text{低点净值}}{\text{高点净值}}\right)$$
  较低的最大回撤表示策略抗风险能力较强。

• 卡玛比率 (Calmar Ratio)：衡量单位最大回撤的复合年化收益率，越高越好。计算公式为：

  $$\text{卡玛比率}=\frac{\text{复合年化收益率}}{\text{最大回撤}}$$

• 胜率：衡量策略在交易中盈利的频率，计算公式为：

  $$\text{胜率}=\frac{\text{盈利交易次数}}{\text{总交易次数}}\times 100\%$$
  较高的胜率表示策略在多数交易中能够获利。

5.3 如何判断是否达到了预期

• MSE 和 MAE：如果这两个指标低于预期阈值（如 MSE < 0.05，MAE < 0.1），则认为模型预测准确。
• R² 分数：如果 R² 分数高于预期阈值（如 0.8 或更高），则认为模型解释能力强。
• 夏普比率：如果夏普比率高于预期阈值（如 1.0 或更高），则认为策略在承担相同风险的情况下获得了更高的回报。
• 累积收益：如果累积收益曲线平滑上升且没有大幅波动，则认为策略表现稳定。
• 总收益：如果总收益高于预期阈值（如 10% 或更高），则认为策略盈利能力强。
• 最大回撤：如果最大回撤低于预期阈值（如 10% 或更低），则认为策略抗风险能力较强。
• 卡玛比率：如果卡玛比率高于预期阈值（如 1.0 或更高），则认为策略在控制回撤方面表现良好。
• 胜率：如果胜率高于预期阈值（如 50% 或更高），则认为策略在多数交易中能够获利。

风险提示与免责声明
本文内容基于公开信息研究整理，不构成任何形式的投资建议。历史表现不应作为未来收益保证，市场存在不可预见的波动风险。投资者需结合自身财务状况及风险承受能力独立决策，并自行承担交易结果。作者及发布方不对任何依据本文操作导致的损失承担法律责任。市场有风险，投资须谨慎。