NumPy助力量化交易从入门到精通掌握高效数据处理与策略回测的核心技能让您的交易策略更加科学精准实现稳定盈利

引言：NumPy与量化交易的完美结合

量化交易作为现代金融领域的重要组成部分，依赖于数据分析、统计模型和计算机技术来实现交易决策的自动化与科学化。在这一过程中，高效的数据处理能力是构建成功量化策略的基石。NumPy（Numerical Python）作为Python科学计算的核心库，以其强大的多维数组对象、广播功能以及丰富的数学函数库，为量化交易提供了不可或缺的技术支持。

本文将全面介绍如何利用NumPy从入门到精通地掌握量化交易中的数据处理与策略回测技能，帮助您构建更加科学精准的交易策略，实现稳定盈利。

第一部分：NumPy基础入门

1.1 NumPy简介与安装

NumPy是Python语言的一个扩展程序库，支持大量的维度数组与矩阵运算，此外也针对数组运算提供大量的数学函数库。在量化交易中，NumPy的效率优势使其成为处理大规模金融数据的首选工具。

安装NumPy非常简单，只需使用pip命令：

pip install numpy

安装完成后，我们可以通过以下方式导入NumPy库：

import numpy as np

1.2 NumPy核心数据结构：ndarray

NumPy的核心是ndarray（N-dimensional array，N维数组）对象，它是一个快速而灵活的大数据集容器。与Python列表相比，NumPy数组在存储数据时更为紧凑，且提供了更高效的数学运算。

创建NumPy数组的基本方法：

# 创建一维数组 arr1 = np.array([1, 2, 3, 4, 5]) # 创建二维数组 arr2 = np.array([[1, 2, 3], [4, 5, 6]]) # 创建特定数组 zeros_arr = np.zeros((3, 4)) # 创建3行4列的全零数组 ones_arr = np.ones((2, 3)) # 创建2行3列的全1数组 random_arr = np.random.rand(3, 3) # 创建3行3列的随机数数组

1.3 NumPy数组操作与属性

NumPy数组提供了丰富的操作方法和属性，方便我们处理金融数据：

# 数组属性 arr = np.array([[1, 2, 3], [4, 5, 6]]) print(arr.shape) # 输出数组维度：(2, 3) print(arr.dtype) # 输出数据类型：int64 print(arr.size) # 输出元素总数：6 # 数组索引与切片 print(arr[0, 1]) # 输出第一行第二列的元素：2 print(arr[:, 1]) # 输出所有行的第二列元素：[2, 5] print(arr[0, :]) # 输出第一行的所有元素：[1, 2, 3] # 数组运算 arr1 = np.array([1, 2, 3]) arr2 = np.array([4, 5, 6]) print(arr1 + arr2) # 数组相加：[5, 7, 9] print(arr1 * 2) # 数组标量乘法：[2, 4, 6] print(np.dot(arr1, arr2)) # 点积：32

第二部分：NumPy在金融数据处理中的应用

2.1 金融数据的导入与预处理

在量化交易中，我们通常需要处理各种金融数据，如股票价格、交易量、财务指标等。NumPy可以高效地处理这些数据。

import numpy as np import pandas as pd # 假设我们从CSV文件导入了股票数据 # 使用pandas读取数据，然后转换为NumPy数组 data = pd.read_csv('stock_data.csv') prices = data['Close'].values # 将收盘价转换为NumPy数组 volumes = data['Volume'].values # 将交易量转换为NumPy数组 # 数据清洗：处理缺失值 prices = np.nan_to_num(prices, nan=np.nanmean(prices)) # 用均值替换NaN值 # 数据标准化 normalized_prices = (prices - np.mean(prices)) / np.std(prices)

2.2 计算金融指标

NumPy可以高效计算各种技术分析指标，如移动平均线、相对强弱指数(RSI)等。

def simple_moving_average(data, window): """计算简单移动平均线""" weights = np.repeat(1.0, window) / window sma = np.convolve(data, weights, 'valid') return sma def exponential_moving_average(data, alpha): """计算指数移动平均线""" ema = np.zeros_like(data) ema[0] = data[0] for i in range(1, len(data)): ema[i] = alpha * data[i] + (1 - alpha) * ema[i-1] return ema def rsi(prices, window=14): """计算相对强弱指数(RSI)""" deltas = np.diff(prices) seed = deltas[:window+1] up = seed[seed >= 0].sum()/window down = -seed[seed < 0].sum()/window rs = up/down rsi = np.zeros_like(prices) rsi[:window] = 100. - (100./(1.+rs)) for i in range(window, len(prices)): delta = deltas[i-1] if delta > 0: upval = delta downval = 0. else: upval = 0. downval = -delta up = (up*(window-1) + upval)/window down = (down*(window-1) + downval)/window rs = up/down rsi[i] = 100. - (100./(1.+rs)) return rsi

2.3 收益率计算与波动率分析

收益率和波动率是量化交易中重要的风险和收益度量指标。

def calculate_returns(prices): """计算对数收益率""" returns = np.log(prices[1:] / prices[:-1]) return returns def calculate_volatility(returns, window=252): """计算年化波动率""" volatility = np.std(returns) * np.sqrt(window) return volatility # 示例使用 prices = np.array([100, 101, 102, 101, 100, 99, 98, 99, 100, 101]) returns = calculate_returns(prices) volatility = calculate_volatility(returns) print(f"日收益率: {returns}") print(f"年化波动率: {volatility:.2%}")

第三部分：基于NumPy的交易策略开发

3.1 均值回归策略

均值回归策略基于价格会回归其历史均值的假设，当价格偏离均值时进行交易。

def mean_reversion_strategy(prices, window=20, threshold=1.5): """ 均值回归策略 参数: prices - 价格序列 window - 移动平均窗口 threshold - 触发交易的阈值(标准差倍数) 返回: signals - 交易信号(1:买入, -1:卖出, 0:持有) """ # 计算移动平均和标准差 rolling_mean = np.zeros_like(prices) rolling_std = np.zeros_like(prices) for i in range(window, len(prices)): rolling_mean[i] = np.mean(prices[i-window:i]) rolling_std[i] = np.std(prices[i-window:i]) # 计算z-score z_score = (prices - rolling_mean) / rolling_std # 生成交易信号 signals = np.zeros_like(prices) signals[z_score < -threshold] = 1 # 买入信号 signals[z_score > threshold] = -1 # 卖出信号 return signals

3.2 动量策略

动量策略基于价格会延续其趋势的假设，在价格上涨时买入，下跌时卖出。

def momentum_strategy(prices, short_window=10, long_window=30): """ 动量策略 参数: prices - 价格序列 short_window - 短期移动平均窗口 long_window - 长期移动平均窗口 返回: signals - 交易信号(1:买入, -1:卖出, 0:持有) """ # 计算短期和长期移动平均 short_ma = np.zeros_like(prices) long_ma = np.zeros_like(prices) for i in range(short_window, len(prices)): short_ma[i] = np.mean(prices[i-short_window:i]) for i in range(long_window, len(prices)): long_ma[i] = np.mean(prices[i-long_window:i]) # 生成交易信号 signals = np.zeros_like(prices) signals[short_ma > long_ma] = 1 # 买入信号 signals[short_ma < long_ma] = -1 # 卖出信号 return signals

3.3 配对交易策略

配对交易是一种市场中性策略，通过交易两个相关性高的资产来获利。

def pairs_trading_strategy(price_a, price_b, window=20, threshold=1.5): """ 配对交易策略 参数: price_a - 资产A的价格序列 price_b - 资产B的价格序列 window - 移动窗口 threshold - 触发交易的阈值 返回: signals_a - 资产A的交易信号 signals_b - 资产B的交易信号 """ # 计算价格比率 ratio = price_a / price_b # 计算移动平均和标准差 rolling_mean = np.zeros_like(ratio) rolling_std = np.zeros_like(ratio) for i in range(window, len(ratio)): rolling_mean[i] = np.mean(ratio[i-window:i]) rolling_std[i] = np.std(ratio[i-window:i]) # 计算z-score z_score = (ratio - rolling_mean) / rolling_std # 生成交易信号 signals_a = np.zeros_like(price_a) signals_b = np.zeros_like(price_b) # 当比率过高时，卖出A买入B signals_a[z_score > threshold] = -1 signals_b[z_score > threshold] = 1 # 当比率过低时，买入A卖出B signals_a[z_score < -threshold] = 1 signals_b[z_score < -threshold] = -1 return signals_a, signals_b

第四部分：使用NumPy进行策略回测

4.1 回测框架搭建

策略回测是评估交易策略历史表现的重要环节。下面我们使用NumPy构建一个简单的回测框架。

def backtest_strategy(prices, signals, initial_capital=10000, commission=0.001): """ 简单的回测框架 参数: prices - 价格序列 signals - 交易信号序列(1:买入, -1:卖出, 0:持有) initial_capital - 初始资金 commission - 交易手续费率 返回: portfolio_value - 组合价值序列 positions - 持仓序列 cash - 现金序列 """ n = len(prices) # 初始化 arrays cash = np.zeros(n) positions = np.zeros(n) portfolio_value = np.zeros(n) # 设置初始值 cash[0] = initial_capital portfolio_value[0] = initial_capital # 回测循环 for i in range(1, n): # 更新现金和组合价值 cash[i] = cash[i-1] positions[i] = positions[i-1] portfolio_value[i] = cash[i] + positions[i] * prices[i] # 执行交易 if signals[i] == 1 and positions[i] == 0: # 买入信号 # 计算可买入的股票数量 buy_amount = cash[i] * (1 - commission) positions[i] = buy_amount / prices[i] cash[i] = 0 elif signals[i] == -1 and positions[i] > 0: # 卖出信号 # 卖出所有持仓 sell_amount = positions[i] * prices[i] * (1 - commission) cash[i] += sell_amount positions[i] = 0 return portfolio_value, positions, cash

4.2 性能评估指标

回测完成后，我们需要计算各种性能指标来评估策略的表现。

def calculate_performance_metrics(portfolio_value, benchmark=None, risk_free_rate=0.02/252): """ 计算策略性能指标 参数: portfolio_value - 组合价值序列 benchmark - 基准指数序列(可选) risk_free_rate - 无风险利率(默认为2%年化，转换为日利率) 返回: metrics - 包含各种性能指标的字典 """ # 计算日收益率 returns = np.diff(portfolio_value) / portfolio_value[:-1] # 计算累计收益率 cumulative_return = portfolio_value[-1] / portfolio_value[0] - 1 # 计算年化收益率 annualized_return = (1 + cumulative_return) ** (252 / len(returns)) - 1 # 计算年化波动率 annualized_volatility = np.std(returns) * np.sqrt(252) # 计算夏普比率 sharpe_ratio = (annualized_return - risk_free_rate * 252) / annualized_volatility # 计算最大回撤 peak = np.maximum.accumulate(portfolio_value) drawdown = (portfolio_value - peak) / peak max_drawdown = np.min(drawdown) # 计算胜率 positive_returns = returns[returns > 0] win_rate = len(positive_returns) / len(returns) # 构建结果字典 metrics = { 'Cumulative Return': cumulative_return, 'Annualized Return': annualized_return, 'Annualized Volatility': annualized_volatility, 'Sharpe Ratio': sharpe_ratio, 'Max Drawdown': max_drawdown, 'Win Rate': win_rate } # 如果提供了基准，计算相对指标 if benchmark is not None: benchmark_returns = np.diff(benchmark) / benchmark[:-1] # 计算信息系数 information_coefficient = np.corrcoef(returns, benchmark_returns)[0, 1] # 计算跟踪误差 tracking_error = np.std(returns - benchmark_returns) * np.sqrt(252) # 计算信息比率 active_return = annualized_return - ((benchmark[-1] / benchmark[0] - 1) ** (252 / len(benchmark_returns)) - 1) information_ratio = active_return / tracking_error # 添加到结果字典 metrics['Information Coefficient'] = information_coefficient metrics['Tracking Error'] = tracking_error metrics['Information Ratio'] = information_ratio metrics['Alpha'] = active_return return metrics

4.3 完整回测示例

下面是一个完整的策略回测示例，包括数据生成、策略应用、回测执行和性能评估。

import numpy as np import matplotlib.pyplot as plt # 生成模拟价格数据 np.random.seed(42) n_days = 252 # 一年的交易日 returns = np.random.normal(0.001, 0.02, n_days) # 日收益率 prices = 100 * np.exp(np.cumsum(returns)) # 价格序列 # 应用均值回归策略 signals = mean_reversion_strategy(prices, window=20, threshold=1.5) # 执行回测 portfolio_value, positions, cash = backtest_strategy(prices, signals) # 计算性能指标 metrics = calculate_performance_metrics(portfolio_value) # 打印性能指标 print("策略回测结果:") for key, value in metrics.items(): if isinstance(value, float): print(f"{key}: {value:.4f}") else: print(f"{key}: {value}") # 绘制结果 plt.figure(figsize=(12, 8)) plt.subplot(2, 1, 1) plt.plot(prices, label='价格') plt.title('价格走势') plt.legend() plt.subplot(2, 1, 2) plt.plot(portfolio_value, label='组合价值') plt.title('组合价值变化') plt.legend() plt.tight_layout() plt.show()

第五部分：高级技巧与性能优化

5.1 向量化操作提升性能

NumPy的向量化操作可以显著提高代码执行效率，特别是在处理大规模金融数据时。

# 非向量化实现(慢) def calculate_returns_non_vectorized(prices): returns = np.zeros(len(prices)-1) for i in range(1, len(prices)): returns[i-1] = (prices[i] - prices[i-1]) / prices[i-1] return returns # 向量化实现(快) def calculate_returns_vectorized(prices): returns = (prices[1:] - prices[:-1]) / prices[:-1] return returns # 性能比较 import time prices = np.random.lognormal(0, 0.01, 100000) # 非向量化版本 start = time.time() returns_non_vec = calculate_returns_non_vectorized(prices) non_vec_time = time.time() - start # 向量化版本 start = time.time() returns_vec = calculate_returns_vectorized(prices) vec_time = time.time() - start print(f"非向量化执行时间: {non_vec_time:.4f}秒") print(f"向量化执行时间: {vec_time:.4f}秒") print(f"性能提升: {non_vec_time/vec_time:.2f}倍")

5.2 使用NumPy的ufuncs进行高效计算

NumPy的通用函数(ufuncs)可以对数组进行元素级操作，无需编写循环。

# 示例：计算多个技术指标 def calculate_technical_indicators(prices): """计算多个技术指标""" n = len(prices) # 初始化结果数组 sma_10 = np.zeros(n) sma_30 = np.zeros(n) ema_12 = np.zeros(n) ema_26 = np.zeros(n) rsi_14 = np.zeros(n) # 计算简单移动平均 for i in range(10, n): sma_10[i] = np.mean(prices[i-10:i]) for i in range(30, n): sma_30[i] = np.mean(prices[i-30:i]) # 计算指数移动平均 ema_12[0] = prices[0] ema_26[0] = prices[0] for i in range(1, n): ema_12[i] = 0.15 * prices[i] + 0.85 * ema_12[i-1] ema_26[i] = 0.075 * prices[i] + 0.925 * ema_26[i-1] # 计算RSI deltas = np.diff(prices) seed = deltas[:15] up = seed[seed >= 0].sum()/14 down = -seed[seed < 0].sum()/14 rs = up/down rsi_14[:15] = 100. - (100./(1.+rs)) for i in range(15, n): delta = deltas[i-1] if delta > 0: upval = delta downval = 0. else: upval = 0. downval = -delta up = (up*13 + upval)/14 down = (down*13 + downval)/14 rs = up/down rsi_14[i] = 100. - (100./(1.+rs)) return { 'SMA_10': sma_10, 'SMA_30': sma_30, 'EMA_12': ema_12, 'EMA_26': ema_26, 'RSI_14': rsi_14 }

5.3 使用NumPy的广播机制

广播机制是NumPy的强大功能，允许不同形状的数组进行算术运算。

# 示例：使用广播计算多只股票的收益率矩阵 # 假设我们有5只股票，252个交易日的价格数据 n_stocks = 5 n_days = 252 # 生成随机价格数据 np.random.seed(42) daily_returns = np.random.normal(0.001, 0.02, (n_days, n_stocks)) prices = 100 * np.exp(np.cumsum(daily_returns, axis=0)) # 计算每只股票的累计收益率 initial_prices = prices[0, :] # 初始价格(1维数组) cumulative_returns = (prices / initial_prices) - 1 # 广播操作 # 计算每只股票的年化波动率 annualized_volatility = np.std(daily_returns, axis=0) * np.sqrt(252) # 计算相关系数矩阵 correlation_matrix = np.corrcoef(daily_returns, rowvar=False) print("年化波动率:") for i, vol in enumerate(annualized_volatility): print(f"股票 {i+1}: {vol:.2%}") print("n相关系数矩阵:") print(correlation_matrix)

第六部分：实际案例分析

6.1 多因子选股策略

多因子选股策略是量化投资中常用的方法，通过综合多个因子来选择股票。

def multi_factor_strategy(stock_data, factors_weights): """ 多因子选股策略 参数: stock_data - 包含多只股票因子数据的字典 factors_weights - 各因子的权重 返回: scores - 每只股票的综合得分 selected_stocks - 选中的股票 """ # 获取股票列表和因子列表 stocks = list(stock_data.keys()) factors = list(factors_weights.keys()) # 标准化因子值 normalized_factors = {} for factor in factors: factor_values = np.array([stock_data[stock][factor] for stock in stocks]) normalized_values = (factor_values - np.mean(factor_values)) / np.std(factor_values) normalized_factors[factor] = normalized_values # 计算综合得分 scores = np.zeros(len(stocks)) for factor, weight in factors_weights.items(): scores += weight * normalized_factors[factor] # 选择得分最高的前20%股票 n_selected = int(len(stocks) * 0.2) selected_indices = np.argsort(scores)[-n_selected:] selected_stocks = [stocks[i] for i in selected_indices] return scores, selected_stocks # 示例使用 # 假设我们有100只股票的数据 n_stocks = 100 stock_symbols = [f"Stock_{i}" for i in range(1, n_stocks+1)] # 生成随机因子数据 np.random.seed(42) stock_data = {} for symbol in stock_symbols: stock_data[symbol] = { 'PE': np.random.normal(15, 5), # 市盈率 'PB': np.random.normal(2, 0.5), # 市净率 'ROE': np.random.normal(0.12, 0.03), # 净资产收益率 'Momentum': np.random.normal(0.05, 0.1) # 动量因子 } # 设置因子权重 factors_weights = { 'PE': -0.25, # 市盈率越低越好 'PB': -0.25, # 市净率越低越好 'ROE': 0.3, # 净资产收益率越高越好 'Momentum': 0.2 # 动量因子越高越好 } # 应用多因子策略 scores, selected_stocks = multi_factor_strategy(stock_data, factors_weights) print(f"选中的股票: {selected_stocks}") print(f"选中股票的平均得分: {np.mean([scores[stock_symbols.index(s)] for s in selected_stocks]):.2f}")

6.2 统计套利策略

统计套利策略利用统计模型识别价格偏离，并在价格回归时获利。

def statistical_arbitrage_strategy(price_a, price_b, window=60, entry_threshold=2.0, exit_threshold=0.5): """ 统计套利策略 参数: price_a - 资产A的价格序列 price_b - 资产B的价格序列 window - 用于计算统计关系的窗口 entry_threshold - 开仓阈值 exit_threshold - 平仓阈值 返回: hedge_ratio - 对冲比率序列 spread - 价差序列 signals_a - 资产A的交易信号 signals_b - 资产B的交易信号 """ n = len(price_a) # 初始化结果数组 hedge_ratio = np.zeros(n) spread = np.zeros(n) signals_a = np.zeros(n) signals_b = np.zeros(n) # 计算滚动对冲比率和价差 for i in range(window, n): # 获取窗口数据 window_a = price_a[i-window:i] window_b = price_b[i-window:i] # 计算对冲比率(使用简单线性回归) X = np.vstack([window_b, np.ones(len(window_b))]).T beta, alpha = np.linalg.lstsq(X, window_a, rcond=None)[0] hedge_ratio[i] = beta # 计算价差 spread[i] = price_a[i] - beta * price_b[i] - alpha # 计算价差的z-score spread_mean = np.mean(spread[i-window:i]) spread_std = np.std(spread[i-window:i]) z_score = (spread[i] - spread_mean) / spread_std # 生成交易信号 if z_score > entry_threshold: # 价差过高，卖出A买入B signals_a[i] = -1 signals_b[i] = 1 elif z_score < -entry_threshold: # 价差过低，买入A卖出B signals_a[i] = 1 signals_b[i] = -1 elif abs(z_score) < exit_threshold: # 价差回归，平仓 signals_a[i] = 0 signals_b[i] = 0 return hedge_ratio, spread, signals_a, signals_b # 示例使用 np.random.seed(42) n_days = 252 # 生成两个相关的价格序列 common_factor = np.cumsum(np.random.normal(0, 0.01, n_days)) price_a = 100 * np.exp(common_factor + np.random.normal(0, 0.005, n_days)) price_b = 100 * np.exp(common_factor * 0.8 + np.random.normal(0, 0.005, n_days)) # 应用统计套利策略 hedge_ratio, spread, signals_a, signals_b = statistical_arbitrage_strategy(price_a, price_b) # 回测策略 portfolio_value_a, _, _ = backtest_strategy(price_a, signals_a) portfolio_value_b, _, _ = backtest_strategy(price_b, signals_b) portfolio_value_total = portfolio_value_a + portfolio_value_b # 计算性能指标 metrics = calculate_performance_metrics(portfolio_value_total) print("统计套利策略回测结果:") for key, value in metrics.items(): if isinstance(value, float): print(f"{key}: {value:.4f}") else: print(f"{key}: {value}") # 绘制结果 plt.figure(figsize=(12, 10)) plt.subplot(3, 1, 1) plt.plot(price_a, label='资产A价格') plt.plot(price_b, label='资产B价格') plt.title('资产价格走势') plt.legend() plt.subplot(3, 1, 2) plt.plot(spread, label='价差') plt.title('价差序列') plt.legend() plt.subplot(3, 1, 3) plt.plot(portfolio_value_total, label='组合价值') plt.title('组合价值变化') plt.legend() plt.tight_layout() plt.show()

6.3 机器学习增强的交易策略

NumPy与机器学习库(如scikit-learn)结合使用，可以构建更复杂的交易策略。

from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score, precision_score, recall_score def ml_enhanced_strategy(prices, features, look_ahead=5, threshold=0.01): """ 机器学习增强的交易策略 参数: prices - 价格序列 features - 特征矩阵(每行是一个时间点的特征) look_ahead - 预测未来多少个时间点的收益率 threshold - 定义上涨/下跌的阈值 返回: signals - 交易信号 model - 训练好的模型 feature_importance - 特征重要性 """ n = len(prices) # 计算未来收益率 future_returns = np.zeros(n) for i in range(n - look_ahead): future_returns[i] = (prices[i + look_ahead] - prices[i]) / prices[i] # 创建标签(1: 上涨, 0: 持平, -1: 下跌) labels = np.zeros(n) labels[future_returns > threshold] = 1 labels[future_returns < -threshold] = -1 # 准备训练数据 X = features[:-look_ahead] # 特征 y = labels[:-look_ahead] # 标签 # 分割训练集和测试集 X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42, shuffle=False) # 训练随机森林模型 model = RandomForestClassifier(n_estimators=100, random_state=42) model.fit(X_train, y_train) # 评估模型 y_pred = model.predict(X_test) accuracy = accuracy_score(y_test, y_pred) precision = precision_score(y_test, y_pred, average='weighted') recall = recall_score(y_test, y_pred, average='weighted') print(f"模型准确率: {accuracy:.4f}") print(f"精确率: {precision:.4f}") print(f"召回率: {recall:.4f}") # 获取特征重要性 feature_importance = model.feature_importances_ # 使用模型预测信号 predictions = model.predict(features) signals = np.zeros(n) signals[predictions == 1] = 1 # 买入信号 signals[predictions == -1] = -1 # 卖出信号 return signals, model, feature_importance # 示例使用 np.random.seed(42) n_days = 252 # 生成模拟价格数据 returns = np.random.normal(0.001, 0.02, n_days) prices = 100 * np.exp(np.cumsum(returns)) # 创建特征矩阵 # 特征1: 过去5天的平均收益率 feature1 = np.zeros(n_days) for i in range(5, n_days): feature1[i] = np.mean(returns[i-5:i]) # 特征2: 过去10天的波动率 feature2 = np.zeros(n_days) for i in range(10, n_days): feature2[i] = np.std(returns[i-10:i]) # 特征3: RSI指标 feature3 = rsi(prices, window=14) # 特征4: MACD指标 feature4 = np.zeros(n_days) ema12 = exponential_moving_average(prices, 2/(12+1)) ema26 = exponential_moving_average(prices, 2/(26+1)) feature4 = ema12 - ema26 # 组合特征矩阵 features = np.column_stack((feature1, feature2, feature3, feature4)) # 应用机器学习增强策略 signals, model, feature_importance = ml_enhanced_strategy(prices, features) # 回测策略 portfolio_value, _, _ = backtest_strategy(prices, signals) # 计算性能指标 metrics = calculate_performance_metrics(portfolio_value) print("n机器学习增强策略回测结果:") for key, value in metrics.items(): if isinstance(value, float): print(f"{key}: {value:.4f}") else: print(f"{key}: {value}") print("n特征重要性:") for i, importance in enumerate(feature_importance): print(f"特征 {i+1}: {importance:.4f}") # 绘制结果 plt.figure(figsize=(12, 8)) plt.subplot(2, 1, 1) plt.plot(prices, label='价格') plt.title('价格走势与交易信号') plt.scatter(np.where(signals == 1)[0], prices[np.where(signals == 1)[0]], color='g', marker='^', label='买入') plt.scatter(np.where(signals == -1)[0], prices[np.where(signals == -1)[0]], color='r', marker='v', label='卖出') plt.legend() plt.subplot(2, 1, 2) plt.plot(portfolio_value, label='组合价值') plt.title('组合价值变化') plt.legend() plt.tight_layout() plt.show()

第七部分：风险管理与资金管理

7.1 使用NumPy实现风险管理

有效的风险管理是量化交易成功的关键。NumPy可以帮助我们计算各种风险指标。

def calculate_risk_metrics(returns, benchmark_returns=None, confidence_level=0.95): """ 计算风险指标 参数: returns - 策略收益率序列 benchmark_returns - 基准收益率序列(可选) confidence_level - VaR和CVaR的置信水平 返回: risk_metrics - 包含各种风险指标的字典 """ # 计算波动率 volatility = np.std(returns) # 计算下行波动率 downside_returns = returns[returns < 0] downside_volatility = np.std(downside_returns) if len(downside_returns) > 0 else 0 # 计算最大回撤 cumulative_returns = np.cumprod(1 + returns) peak = np.maximum.accumulate(cumulative_returns) drawdown = (cumulative_returns - peak) / peak max_drawdown = np.min(drawdown) # 计算VaR(Value at Risk) var = np.percentile(returns, (1 - confidence_level) * 100) # 计算CVaR(Conditional Value at Risk) cvar = np.mean(returns[returns <= var]) # 计算偏度和峰度 skewness = np.mean(((returns - np.mean(returns)) / np.std(returns)) ** 3) kurtosis = np.mean(((returns - np.mean(returns)) / np.std(returns)) ** 4) - 3 # 构建结果字典 risk_metrics = { 'Volatility': volatility, 'Downside Volatility': downside_volatility, 'Max Drawdown': max_drawdown, 'VaR': var, 'CVaR': cvar, 'Skewness': skewness, 'Kurtosis': kurtosis } # 如果提供了基准，计算相对风险指标 if benchmark_returns is not None: # 计算跟踪误差 tracking_error = np.std(returns - benchmark_returns) # 计算信息比率 active_return = np.mean(returns) - np.mean(benchmark_returns) information_ratio = active_return / tracking_error if tracking_error != 0 else 0 # 计算Beta covariance = np.cov(returns, benchmark_returns)[0, 1] benchmark_variance = np.var(benchmark_returns) beta = covariance / benchmark_variance if benchmark_variance != 0 else 0 # 计算Alpha alpha = np.mean(returns) - beta * np.mean(benchmark_returns) # 计算R平方 correlation = np.corrcoef(returns, benchmark_returns)[0, 1] r_squared = correlation ** 2 # 添加到结果字典 risk_metrics['Tracking Error'] = tracking_error risk_metrics['Information Ratio'] = information_ratio risk_metrics['Beta'] = beta risk_metrics['Alpha'] = alpha risk_metrics['R-squared'] = r_squared return risk_metrics

7.2 资金管理策略

合理的资金管理可以控制风险并提高长期收益。

def kelly_criterion(win_rate, avg_win, avg_loss): """ 凯利公式计算最优仓位 参数: win_rate - 胜率 avg_win - 平均盈利 avg_loss - 平均亏损 返回: kelly_fraction - 凯利最优仓位比例 """ if avg_loss == 0: return 0 kelly_fraction = (win_rate * avg_win - (1 - win_rate) * avg_loss) / avg_win return max(0, min(kelly_fraction, 1)) # 限制在0到1之间 def fixed_fractional_position_sizing(capital, risk_per_trade, stop_loss_pct): """ 固定分数仓位管理 参数: capital - 总资金 risk_per_trade - 每笔交易风险占总资金的比例 stop_loss_pct - 止损百分比 返回: position_size - 仓位大小 """ position_size = (capital * risk_per_trade) / stop_loss_pct return position_size def volatility_adjusted_position_sizing(capital, target_volatility, current_volatility, time_period=252): """ 波动率调整仓位管理 参数: capital - 总资金 target_volatility - 目标波动率 current_volatility - 当前波动率 time_period - 波动率的计算周期 返回: position_size - 仓位大小 """ # 年化波动率调整 annualized_current_vol = current_volatility * np.sqrt(time_period) # 计算仓位调整系数 adjustment_factor = target_volatility / annualized_current_vol if annualized_current_vol > 0 else 0 # 限制调整系数在合理范围内 adjustment_factor = max(0.1, min(adjustment_factor, 2.0)) # 计算仓位大小 position_size = capital * adjustment_factor return position_size # 示例使用 # 假设我们有一系列交易结果 trade_results = np.array([100, -50, 150, -75, 200, -100, 175, -60, 120, -80]) # 计算交易统计 winning_trades = trade_results[trade_results > 0] losing_trades = trade_results[trade_results < 0] win_rate = len(winning_trades) / len(trade_results) avg_win = np.mean(winning_trades) if len(winning_trades) > 0 else 0 avg_loss = -np.mean(losing_trades) if len(losing_trades) > 0 else 0 # 使用凯利公式 kelly_fraction = kelly_criterion(win_rate, avg_win, avg_loss) print(f"凯利最优仓位比例: {kelly_fraction:.2%}") # 使用固定分数仓位管理 capital = 100000 risk_per_trade = 0.02 # 每笔交易风险2% stop_loss_pct = 0.05 # 止损5% position_size = fixed_fractional_position_sizing(capital, risk_per_trade, stop_loss_pct) print(f"固定分数仓位大小: ${position_size:.2f}") # 使用波动率调整仓位管理 target_volatility = 0.15 # 目标年化波动率15% current_volatility = 0.02 # 当前日波动率2% position_size = volatility_adjusted_position_sizing(capital, target_volatility, current_volatility) print(f"波动率调整仓位大小: ${position_size:.2f}")

7.3 投资组合优化

NumPy可以用于实现现代投资组合理论，优化资产配置。

def portfolio_optimization(returns, target_return=None, risk_free_rate=0.02): """ 投资组合优化 参数: returns - 各资产收益率矩阵(每列代表一个资产) target_return - 目标收益率(可选) risk_free_rate - 无风险利率 返回: weights - 最优权重 expected_return - 期望收益率 expected_volatility - 期望波动率 sharpe_ratio - 夏普比率 """ n_assets = returns.shape[1] # 计算期望收益率和协方差矩阵 expected_returns = np.mean(returns, axis=0) cov_matrix = np.cov(returns, rowvar=False) # 如果没有指定目标收益率，则寻找最大夏普比率组合 if target_return is None: # 定义目标函数(负的夏普比率) def objective(weights): portfolio_return = np.sum(expected_returns * weights) portfolio_volatility = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights))) sharpe_ratio = (portfolio_return - risk_free_rate) / portfolio_volatility return -sharpe_ratio # 约束条件: 权重之和为1 constraints = {'type': 'eq', 'fun': lambda weights: np.sum(weights) - 1} # 边界条件: 每个权重在0到1之间 bounds = tuple((0, 1) for _ in range(n_assets)) # 初始权重(等权重) initial_weights = np.array([1 / n_assets] * n_assets) # 使用SLSQP方法优化 from scipy.optimize import minimize result = minimize(objective, initial_weights, method='SLSQP', bounds=bounds, constraints=constraints) # 获取最优权重 weights = result.x # 计算组合指标 portfolio_return = np.sum(expected_returns * weights) portfolio_volatility = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights))) sharpe_ratio = (portfolio_return - risk_free_rate) / portfolio_volatility else: # 定义目标函数(最小化波动率) def objective(weights): portfolio_volatility = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights))) return portfolio_volatility # 约束条件: 权重之和为1，且达到目标收益率 constraints = ( {'type': 'eq', 'fun': lambda weights: np.sum(weights) - 1}, {'type': 'eq', 'fun': lambda weights: np.sum(expected_returns * weights) - target_return} ) # 边界条件: 每个权重在0到1之间 bounds = tuple((0, 1) for _ in range(n_assets)) # 初始权重(等权重) initial_weights = np.array([1 / n_assets] * n_assets) # 使用SLSQP方法优化 from scipy.optimize import minimize result = minimize(objective, initial_weights, method='SLSQP', bounds=bounds, constraints=constraints) # 获取最优权重 weights = result.x # 计算组合指标 portfolio_return = np.sum(expected_returns * weights) portfolio_volatility = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights))) sharpe_ratio = (portfolio_return - risk_free_rate) / portfolio_volatility return weights, portfolio_return, portfolio_volatility, sharpe_ratio def efficient_frontier(returns, n_points=100, risk_free_rate=0.02): """ 计算有效前沿 参数: returns - 各资产收益率矩阵(每列代表一个资产) n_points - 有效前沿上的点数 risk_free_rate - 无风险利率 返回: frontier_returns - 有效前沿上的收益率 frontier_volatilities - 有效前沿上的波动率 frontier_weights - 有效前沿上的权重 max_sharpe_weights - 最大夏普比率组合的权重 """ # 计算期望收益率和协方差矩阵 expected_returns = np.mean(returns, axis=0) cov_matrix = np.cov(returns, rowvar=False) # 计算最小方差组合和最大收益率组合 min_var_weights, _, min_var_volatility, _ = portfolio_optimization(returns, risk_free_rate=risk_free_rate) # 计算各资产的最大和最小期望收益率 min_return = np.min(expected_returns) max_return = np.max(expected_returns) # 生成目标收益率序列 target_returns = np.linspace(min_return, max_return, n_points) # 初始化结果数组 frontier_volatilities = np.zeros(n_points) frontier_weights = np.zeros((n_points, len(expected_returns))) # 计算有效前沿上的点 for i, target_return in enumerate(target_returns): weights, _, volatility, _ = portfolio_optimization(returns, target_return, risk_free_rate) frontier_volatilities[i] = volatility frontier_weights[i] = weights # 计算最大夏普比率组合 max_sharpe_weights, max_sharpe_return, max_sharpe_volatility, _ = portfolio_optimization(returns, risk_free_rate=risk_free_rate) return target_returns, frontier_volatilities, frontier_weights, max_sharpe_weights # 示例使用 np.random.seed(42) n_assets = 5 n_obs = 252 # 生成随机收益率数据 expected_returns = np.random.normal(0.001, 0.005, n_assets) cov_matrix = np.random.uniform(0.0001, 0.001, (n_assets, n_assets)) cov_matrix = (cov_matrix + cov_matrix.T) / 2 # 确保对称 np.fill_diagonal(cov_matrix, np.random.uniform(0.001, 0.005, n_assets)) # 设置对角线 # 生成收益率矩阵 returns = np.random.multivariate_normal(expected_returns, cov_matrix, n_obs) # 计算有效前沿 frontier_returns, frontier_volatilities, frontier_weights, max_sharpe_weights = efficient_frontier(returns) # 计算最大夏普比率组合的指标 max_sharpe_weights, max_sharpe_return, max_sharpe_volatility, max_sharpe_ratio = portfolio_optimization(returns) print("最大夏普比率组合:") print(f"权重: {max_sharpe_weights}") print(f"期望收益率: {max_sharpe_return:.4f}") print(f"期望波动率: {max_sharpe_volatility:.4f}") print(f"夏普比率: {max_sharpe_ratio:.4f}") # 绘制有效前沿 plt.figure(figsize=(10, 6)) plt.plot(frontier_volatilities, frontier_returns, 'b-', label='有效前沿') plt.plot(max_sharpe_volatility, max_sharpe_return, 'ro', label='最大夏普比率组合') plt.xlabel('波动率') plt.ylabel('收益率') plt.title('投资组合有效前沿') plt.legend() plt.grid(True) plt.show()