案例：量化交易策略设计与回测 - 大数据金融(Big Data in Finance)

本案例研究的目标是使用基于分类的模型，根据短期价格和长期价格预测当前信号是买入还是卖出。基于特征工程，创造出大量的新特征，以期待增加模型的学习和预测能力。最后，我们对于策略进行了回测。

1. 问题定义¶

在金融市场中，交易者设计量化交易策略往往面临如下挑战：

如何准确预测价格走势，并生成买入或卖出的交易信号
单纯依赖价格数据难以全面捕捉市场动态，如何提取有用特征是一个挑战
选择合适的机器学习模型并对其进行优化是构建有效交易策略的关键
验证策略在历史数据上的表现，以确保其稳定性和盈利能力。如何有效地回测策略并评估其性能？

本案例的解决流程：

根据短期和长期价格数据生成有效的交易信号
通过特征工程，该案例研究从历史价格数据提取了大量新特征。这些特征的引入显著提高了模型的学习和预测能力，帮助模型更好地理解市场状况。
应用决策树作为分类器，进行样本的训练和预测，对比训练集和测试集的表现，并对特征重要性进行排序
最后，通过回测，该案例研究评估了策略的累计收益指标，以确保策略的可行性和有效性。

为了进行案例研究，我们从日均交易量最大的比特币交易所之一Bitstamp 获取数据。数据可在以下网址找到： https://www.kaggle.com/mczielinski/bitcoin-historical-data

2. 读取数据集¶

import pandas as pd
dataset = pd.read_csv('datasets/BitstampData_sample.csv',index_col=0)

2.1 描述性统计¶

# describe data
pd.set_option('float_format', '{:.3f}'.format)
dataset.describe()

3. 数据准备¶

3.1 数据清洗¶

#Checking for any null values and removing the null values'''
print('Null Values =',dataset.isnull().values.any())

Null Values = True

鉴于存在空值，我们需要用最后可用的值填充 NaN，从而清理数据。

dataset[dataset.columns.values] = dataset[dataset.columns.values].ffill()

dataset=dataset.drop(columns=['Timestamp'])

3.2 交易信号¶

我们为每一个走势贴上标签：

如果信号显示短期价格将比长期价格上涨，则标记为 1；
如果信号显示短期价格将比长期价格下跌，则标记为 0。

可以设计为：如果短期移动平均线超过长期移动平均线，则信号为 1.0，否则为 0.0

import numpy as np

# 创建短期简单移动平均线，窗口为短期窗口
dataset['short_mavg'] = dataset['Close'].rolling(window=60, min_periods=60, center=False).mean()

# 创建长期简单移动平均线，窗口为长期窗口
dataset['long_mavg'] = dataset['Close'].rolling(window=200, min_periods=200, center=False).mean()

# 创建交易信号
dataset['signal'] = np.where(dataset['short_mavg'] > dataset['long_mavg'], 1.0, 0.0)

dataset.tail()

3.3 特征工程——构建技术指标¶

首先，我们要构建一个数据集，其中包含用于预测的预测因子和输出变量。

双币的当前数据包括日期、开盘价、最高价、最低价、收盘价和成交量。利用这些数据，我们可以计算出以下技术指标：

移动平均线：移动平均线通过减少价格图表上的 "噪音 "量来显示价格变动的趋势。
随机震荡指标 %K 和 %D：随机震荡指标是一种动量指标，将证券的特定收盘价与一定时期内的价格范围进行比较。%K 和 %D 是慢速和快速指标。
相对强弱指数（RSI）：这是一种动量指标，用于衡量近期价格变化的幅度，以评估股票或其他资产价格的超买或超卖情况。
变化率（ROC）：这是一种动量震荡指标，用于测量当前价格与过去 N 期价格之间的百分比变化。
动量（MOM）：是指证券价格或成交量的加速率，即价格变化的速度。

计算指数移动平均线

def EMA(df, n):
    # 计算指数移动平均线并将结果存储为序列，使用给定的窗口大小 n
    EMA = pd.Series(df['Close'].ewm(span=n, min_periods=n).mean(), name='EMA_' + str(n))
    return EMA

# 计算并添加10日指数移动平均线到数据集
dataset['EMA10'] = EMA(dataset, 10)

# 计算并添加30日指数移动平均线到数据集
dataset['EMA30'] = EMA(dataset, 30)

# 计算并添加200日指数移动平均线到数据集
dataset['EMA200'] = EMA(dataset, 200)

计算变动率（Rate of Change）

def ROC(df, n):  
    # 计算价格变动值 M
    M = df.diff(n - 1)  
    # 获取 n 期前的价格 N
    N = df.shift(n - 1)  
    # 计算变动率 ROC，并将结果存储为序列
    ROC = pd.Series(((M / N) * 100), name='ROC_' + str(n))   
    return ROC

# 计算并添加10日变动率到数据集
dataset['ROC10'] = ROC(dataset['Close'], 10)

# 计算并添加30日变动率到数据集
dataset['ROC30'] = ROC(dataset['Close'], 30)

计算价格动量（Price Momentum）

def MOM(df, n):   
    # 计算价格变化值 MOM
    MOM = pd.Series(df.diff(n), name='Momentum_' + str(n))   
    return MOM

# 计算并添加10日价格动量到数据集
dataset['MOM10'] = MOM(dataset['Close'], 10)

# 计算并添加30日价格动量到数据集
dataset['MOM30'] = MOM(dataset['Close'], 30)

计算相对强度指数（Relative Strength Index，RSI）

def RSI(series, period):
    delta = series.diff().dropna()
    u = delta * 0
    d = u.copy()
    u[delta > 0] = delta[delta > 0]
    d[delta < 0] = -delta[delta < 0]
    u[u.index[period-1]] = np.mean( u[:period] ) # 第一个值是平均增益的总和
    u = u.drop(u.index[:(period-1)])
    d[d.index[period-1]] = np.mean( d[:period] ) # 第一个值是平均损失的总和
    d = d.drop(d.index[:(period-1)])
    rs = u.ewm(com=period-1, adjust=False).mean() / \
         d.ewm(com=period-1, adjust=False).mean()
    return 100 - 100 / (1 + rs)

# 计算并添加10日相对强度指数到数据集
dataset['RSI10'] = RSI(dataset['Close'], 10)

# 计算并添加30日相对强度指数到数据集
dataset['RSI30'] = RSI(dataset['Close'], 30)

# 计算并添加200日相对强度指数到数据集
dataset['RSI200'] = RSI(dataset['Close'], 200)

计算随机振荡指标（Stochastic Oscillator）

# 计算 %K 值
def STOK(close, low, high, n): 
    STOK = ((close - low.rolling(n).min()) / (high.rolling(n).max() - low.rolling(n).min())) * 100
    return STOK

# 计算 %D 值
def STOD(close, low, high, n):
    STOK = ((close - low.rolling(n).min()) / (high.rolling(n).max() - low.rolling(n).min())) * 100
    STOD = STOK.rolling(3).mean()
    return STOD

# 计算并添加10日随机振荡指标到数据集
dataset['%K10'] = STOK(dataset['Close'], dataset['Low'], dataset['High'], 10)
dataset['%D10'] = STOD(dataset['Close'], dataset['Low'], dataset['High'], 10)

# 计算并添加30日随机振荡指标到数据集
dataset['%K30'] = STOK(dataset['Close'], dataset['Low'], dataset['High'], 30)
dataset['%D30'] = STOD(dataset['Close'], dataset['Low'], dataset['High'], 30)

# 计算并添加200日随机振荡指标到数据集
dataset['%K200'] = STOK(dataset['Close'], dataset['Low'], dataset['High'], 200)
dataset['%D200'] = STOD(dataset['Close'], dataset['Low'], dataset['High'], 200)

计算移动平均线（Moving Average）

# 计算移动平均线
def MA(df, n):
    MA = pd.Series(df['Close'].rolling(n, min_periods=n).mean(), name='MA_' + str(n))
    return MA

# 计算并添加21日移动平均线到数据集
dataset['MA21'] = MA(dataset, 10)

# 计算并添加63日移动平均线到数据集
dataset['MA63'] = MA(dataset, 30)

# 计算并添加252日移动平均线到数据集
dataset['MA252'] = MA(dataset, 200)

# 显示数据集的末尾
dataset.tail()

删除不需要用于预测的列

dataset=dataset.drop(['High','Low','Open', 'Volume_(Currency)','short_mavg','long_mavg'], axis=1)

dataset = dataset.dropna(axis=0)

dataset.tail()

3.4 数据可视化¶

import matplotlib.pyplot as plt
from matplotlib_inline import backend_inline
backend_inline.set_matplotlib_formats('svg') 
dataset[['Weighted_Price']].plot(grid=True)
plt.show()

# histograms
dataset.hist(sharex=False, sharey=False, xlabelsize=1, ylabelsize=1, figsize=(12,12))
plt.show()

fig = plt.figure()
plot = dataset.groupby(['signal']).size().plot(kind='barh', color='red')
plt.show()

买入信号的数量多于卖出信号的数量

计算并可视化相关性矩阵

import seaborn as sns
correlation = dataset.corr()
plt.figure(figsize=(15,15))
plt.title('Correlation Matrix')
sns.heatmap(correlation, vmax=1, square=True,annot=True,cmap='cubehelix')

<Axes: title={'center': 'Correlation Matrix'}>

4. 建模和评估算法¶

4.1 训练集/测试集划分¶

分割出用于最后验证的数据集

from sklearn.model_selection import train_test_split

# 提取目标变量 Y
Y = dataset["signal"]

# 提取特征变量 X
X = dataset.loc[:, dataset.columns != 'signal']

# 定义验证集的大小
test_size = 0.2

# 将数据集划分为训练集和测试集
X_train, X_test, Y_train, Y_test = train_test_split(X, Y, train_size=0.8,random_state=42)

4.2 训练集表现¶

应用决策树分类器模型。决策树是一种基于树形结构的分类算法，通过对特征进行递归划分来进行决策。在这里，DecisionTreeClassifier类的实例被创建。

from sklearn.tree import DecisionTreeClassifier
model = DecisionTreeClassifier(max_depth=3, splitter='best')

model.fit(X_train, Y_train)

Y_train_predicted = model.predict(X_train)

accuracy = model.score(X_train, Y_train)

from sklearn import metrics
precision = metrics.precision_score(Y_train, Y_train_predicted)
recall = metrics.recall_score(Y_train, Y_train_predicted)
f1_score = metrics.f1_score(Y_train, Y_train_predicted)

print(accuracy, precision,recall,f1_score)

0.8249055116139368 0.8323128804276335 0.832945386221794 0.8326290132041332

from sklearn.tree import plot_tree
plt.figure(figsize=(24,6))
plot_tree(model,
          max_depth=3,
         class_names=['No', 'Yes'],
         feature_names=X_train.columns.to_list(),
          rounded=True,
          fontsize=12,
         filled=True)
plt.title('DecisionTree') 
plt.show()

4.3 测试集表现¶

Y_test_predicted = model.predict(X_test)

accuracy = model.score(X_test, Y_test)

from sklearn import metrics
precision = metrics.precision_score(Y_test, Y_test_predicted)
recall = metrics.recall_score(Y_test, Y_test_predicted)
f1_score = metrics.f1_score(Y_test, Y_test_predicted)

print(accuracy, precision,recall,f1_score)

0.8227770477441962 0.8294068304373876 0.8309003601440577 0.8301529235382309

4.4 特征重要性排序¶

Importance = pd.DataFrame({'Importance':model.feature_importances_*100}, index=X.columns)
Importance.sort_values('Importance', axis=0, ascending=True).plot(kind='barh', color='r' )
plt.xlabel('Variable Importance')

通过绘制特征重要性的排序结果，能够看到哪些指标在模型的训练过程中扮演了重要的决策。

5. 策略回测¶

通过将前一天的持仓与当天的日收益率相乘，创建策略收益率列

backtestdata = pd.DataFrame(index=X.index)
backtestdata['Market Returns'] = X['Close'].pct_change()   # 市场收益率

backtestdata['signal_actual'] = Y                    # 实际交易信号
backtestdata['signal_pred'] = model.predict(X)       # 基于决策树模型的预测信号

计算基于交易信号的收益率

backtestdata['Strategy Returns'] = backtestdata['Market Returns'] * backtestdata['signal_actual'].shift(1)  # 实际收益率
backtestdata['TreeModel Predicted Returns'] = backtestdata['Market Returns'] * backtestdata['signal_pred'].shift(1)   # 模型预测信号的收益率

为了对比该交易策略的表现，我们使用基准模型，例如买入并持有（Buy&Hold）策略

backtestdata['Buy&Hold Returns'] = backtestdata['Market Returns']       # 买入并持有的策略收益等于价格收益率
backtestdata = backtestdata.dropna()
backtestdata.head()

绘制策略收益率和实际收益率的累积直方图

backtestdata[['Strategy Returns',
              'TreeModel Predicted Returns',
             'Buy&Hold Returns']].cumsum().hist()

array([[<Axes: title={'center': 'Strategy Returns'}>,
        <Axes: title={'center': 'TreeModel Predicted Returns'}>],
       [<Axes: title={'center': 'Buy&Hold Returns'}>, <Axes: >]],
      dtype=object)

绘制策略收益率和实际收益率的累积曲线图

backtestdata[['Strategy Returns','TreeModel Predicted Returns']].cumsum().plot()

<Axes: >

结论¶

我们展示了使用特征工程的效率，它可以创建与价格变动趋势和动量相关的直观特征，并提高模型的预测能力。最后，我们演示了如何进行回测，回测允许我们使用历史数据模拟交易策略，以生成结果，并在投入任何实际资金之前分析风险和盈利能力。