Python 量化金融与算法交易实战指南

https://www.python.org/static/community_logos/python-logo-master-v3-TM.png

金融数据获取与处理

使用yfinance获取市场数据

python

复制

下载

import yfinance as yf
import pandas as pd# 下载苹果公司股票数据
aapl = yf.Ticker("AAPL")
hist = aapl.history(period="5y")# 计算移动平均线
hist['MA50'] = hist['Close'].rolling(window=50).mean()
hist['MA200'] = hist['Close'].rolling(window=200).mean()# 可视化
hist[['Close', 'MA50', 'MA200']].plot(figsize=(12, 6))
plt.title('Apple Stock Price with Moving Averages')
plt.ylabel('Price (USD)')
plt.savefig('aapl_ma.png')
plt.show()

https://matplotlib.org/stable/_images/sphx_glr_plot_001.png

使用pandas处理高频数据

python

复制

下载

# 重采样高频数据
intraday = yf.download("AAPL", period="1d", interval="1m")
intraday_5min = intraday.resample('5T').agg({'Open': 'first','High': 'max','Low': 'min','Close': 'last','Volume': 'sum'
})# 计算波动率
intraday_5min['Returns'] = intraday_5min['Close'].pct_change()
intraday_5min['Volatility'] = intraday_5min['Returns'].rolling(12).std() * np.sqrt(252*78)  # 78个5分钟交易日print(intraday_5min.tail())

技术指标实现

常用指标计算

python

复制

下载

import talib# RSI指标
hist['RSI14'] = talib.RSI(hist['Close'], timeperiod=14)# MACD指标
hist['MACD'], hist['MACD_Signal'], hist['MACD_Hist'] = talib.MACD(hist['Close'], fastperiod=12, slowperiod=26, signalperiod=9
)# Bollinger Bands
hist['UpperBand'], hist['MiddleBand'], hist['LowerBand'] = talib.BBANDS(hist['Close'], timeperiod=20, nbdevup=2, nbdevdn=2
)# 可视化
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8), gridspec_kw={'height_ratios': [3, 1]})
hist[['Close', 'MA50', 'MA200']].plot(ax=ax1)
hist[['RSI14']].plot(ax=ax2)
ax2.axhline(70, color='r', linestyle='--')
ax2.axhline(30, color='g', linestyle='--')
plt.savefig('technical_indicators.png')
plt.show()

回测框架实现

事件驱动回测引擎

python

复制

下载

class BacktestEngine:def __init__(self, data, initial_capital=100000):self.data = dataself.initial_capital = initial_capitalself.positions = []self.current_cash = initial_capitalself.portfolio_values = []def run_backtest(self, strategy):for i, (index, row) in enumerate(self.data.iterrows()):# 获取当前持仓current_position = self.positions[-1] if self.positions else 0# 执行策略signal = strategy(row, current_position, i)# 执行交易if signal > 0 and current_position <= 0:  # 买入信号position_size = int(self.current_cash / row['Close'])self.positions.append(position_size)self.current_cash -= position_size * row['Close']elif signal < 0 and current_position > 0:  # 卖出信号self.current_cash += current_position * row['Close']self.positions.append(0)else:  # 保持持仓self.positions.append(current_position)# 记录组合价值portfolio_value = self.current_cash + current_position * row['Close']self.portfolio_values.append(portfolio_value)return self.portfolio_values# 双均线策略
def dual_moving_average_strategy(data, current_position, index):if index < 200:  # 确保有足够数据计算MA200return 0if data['MA50'] > data['MA200'] and current_position <= 0:return 1  # 买入信号elif data['MA50'] < data['MA200'] and current_position > 0:return -1  # 卖出信号else:return 0  # 无信号# 运行回测
engine = BacktestEngine(hist)
portfolio_values = engine.run_backtest(dual_moving_average_strategy)

https://matplotlib.org/stable/_images/sphx_glr_plot_002.png

量化交易策略

均值回归策略

python

复制

下载

def mean_reversion_strategy(data, lookback=20, z_score_threshold=1.0):# 计算滚动统计量data['RollingMean'] = data['Close'].rolling(lookback).mean()data['RollingStd'] = data['Close'].rolling(lookback).std()data['Z-Score'] = (data['Close'] - data['RollingMean']) / data['RollingStd']# 生成交易信号data['Signal'] = 0data.loc[data['Z-Score'] < -z_score_threshold, 'Signal'] = 1  # 买入data.loc[data['Z-Score'] > z_score_threshold, 'Signal'] = -1  # 卖出return data# 应用策略
mr_data = mean_reversion_strategy(hist.copy())
mr_data[['Close', 'RollingMean', 'Z-Score', 'Signal']].plot(secondary_y=['Z-Score', 'Signal'], figsize=(12, 6),style=['-', '--', '-', 'o-']
)
plt.title('Mean Reversion Strategy Signals')
plt.savefig('mean_reversion.png')
plt.show()

动量策略

python

复制

下载

def momentum_strategy(data, lookback=3, hold_period=1):# 计算过去lookback个月的收益率data['Momentum'] = data['Close'].pct_change(lookback * 21)  # 假设21个交易日/月# 生成信号 (每月初调仓)data['Signal'] = 0data.loc[data['Momentum'] > 0, 'Signal'] = 1data.loc[data['Momentum'] <= 0, 'Signal'] = -1# 保持持仓hold_period个月data['Signal'] = data['Signal'].shift(1).rolling(hold_period * 21).mean()return data# 应用策略
momentum_data = momentum_strategy(hist.copy())
momentum_data[['Close', 'Momentum', 'Signal']].plot(secondary_y=['Momentum', 'Signal'], figsize=(12, 6)
)
plt.title('Momentum Strategy Signals')
plt.savefig('momentum.png')
plt.show()

风险管理

投资组合优化

python

复制

下载

import numpy as np
from scipy.optimize import minimize# 获取多资产收益率
tickers = ['AAPL', 'MSFT', 'GOOG', 'AMZN', 'META']
data = yf.download(tickers, start='2020-01-01', end='2023-01-01')['Adj Close']
returns = data.pct_change().dropna()# 计算协方差矩阵
cov_matrix = returns.cov() * 252  # 年化# 投资组合优化
def portfolio_volatility(weights, cov_matrix):return np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))def optimize_portfolio(returns, cov_matrix):num_assets = len(returns.columns)args = (cov_matrix,)constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})bounds = tuple((0, 1) for asset in range(num_assets))initial_guess = num_assets * [1./num_assets]result = minimize(portfolio_volatility, initial_guess, args=args,method='SLSQP', bounds=bounds, constraints=constraints)return result.xoptimal_weights = optimize_portfolio(returns, cov_matrix)
print("最优权重:", dict(zip(tickers, optimal_weights)))

风险价值(VaR)计算

python

复制

下载

from scipy.stats import normdef calculate_var(returns, confidence_level=0.95):mean = returns.mean()std_dev = returns.std()# 参数法VaRvar_parametric = norm.ppf(1-confidence_level, mean, std_dev)# 历史模拟法VaRvar_historical = np.percentile(returns, (1-confidence_level)*100)return var_parametric, var_historicalaapl_returns = returns['AAPL']
var_p, var_h = calculate_var(aapl_returns)
print(f"参数法VaR(95%): {var_p:.4f}")
print(f"历史模拟法VaR(95%): {var_h:.4f}")

实盘交易接口

使用CCXT连接交易所

python

复制

下载

import ccxt
import pandas as pd# 初始化交易所连接
exchange = ccxt.binance({'apiKey': 'YOUR_API_KEY','secret': 'YOUR_SECRET','enableRateLimit': True
})# 获取K线数据
ohlcv = exchange.fetch_ohlcv('BTC/USDT', '1d', limit=100)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')# 下订单示例
def place_limit_order(symbol, side, amount, price):try:order = exchange.create_order(symbol=symbol,type='limit',side=side,amount=amount,price=price)print(f"订单已提交: {order['id']}")return orderexcept Exception as e:print(f"下单失败: {str(e)}")return None# 获取账户余额
balance = exchange.fetch_balance()
print("USDT余额:", balance['USDT']['free'])

https://camo.githubusercontent.com/9e37b5d39a9e0e4b6b6a3f7b5e9d5d5e5b5e5d5e5/68747470733a2f2f636378742e696f2f696d672f65786368616e6765732e706e67

高频交易策略

订单簿分析

python

复制

下载

import numpy as npdef analyze_order_book(order_book, depth=10):bids = np.array(order_book['bids'][:depth])asks = np.array(order_book['asks'][:depth])# 计算买卖价差spread = asks[0][0] - bids[0][0]mid_price = (asks[0][0] + bids[0][0]) / 2# 计算市场深度bid_depth = bids[:, 0] * bids[:, 1]ask_depth = asks[:, 0] * asks[:, 1]return {'spread': spread,'mid_price': mid_price,'bid_depth': bid_depth.sum(),'ask_depth': ask_depth.sum(),'imbalance': (bid_depth.sum() - ask_depth.sum()) / (bid_depth.sum() + ask_depth.sum())}# 获取订单簿数据
order_book = exchange.fetch_order_book('BTC/USDT')
ob_metrics = analyze_order_book(order_book)
print("订单簿指标:", ob_metrics)

简单做市策略

python

复制

下载

class MarketMaker:def __init__(self, exchange, symbol, position_limit=10):self.exchange = exchangeself.symbol = symbolself.position_limit = position_limitself.orders = []def run_strategy(self, spread_pct=0.001, order_size=0.1):# 获取当前市场价格ticker = self.exchange.fetch_ticker(self.symbol)mid_price = (ticker['bid'] + ticker['ask']) / 2# 计算买卖价格bid_price = mid_price * (1 - spread_pct)ask_price = mid_price * (1 + spread_pct)# 获取当前持仓balance = self.exchange.fetch_balance()position = balance.get(self.symbol.split('/')[0], {}).get('free', 0)# 取消所有未成交订单self.cancel_all_orders()# 下新订单if position < self.position_limit:bid_order = self.exchange.create_limit_buy_order(self.symbol, order_size, bid_price)self.orders.append(bid_order['id'])if position > -self.position_limit:ask_order = self.exchange.create_limit_sell_order(self.symbol, order_size, ask_price)self.orders.append(ask_order['id'])def cancel_all_orders(self):for order_id in self.orders:try:self.exchange.cancel_order(order_id, self.symbol)except:continueself.orders = []# 使用示例
mm = MarketMaker(exchange, 'BTC/USDT')
mm.run_strategy()

机器学习在量化中的应用

特征工程

python

复制

下载

from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_splitdef create_features(data, lags=5):# 基础特征data['Returns'] = data['Close'].pct_change()data['Volatility'] = data['Returns'].rolling(21).std()data['Momentum'] = data['Close'] / data['Close'].shift(21) - 1# 滞后特征for lag in range(1, lags+1):data[f'Return_lag_{lag}'] = data['Returns'].shift(lag)# 技术指标data['RSI14'] = talib.RSI(data['Close'], timeperiod=14)data['MACD'], _, _ = talib.MACD(data['Close'])# 目标变量 (未来5天收益率)data['Target'] = data['Close'].shift(-5) / data['Close'] - 1return data.dropna()# 准备数据
featured_data = create_features(hist.copy())
X = featured_data.drop(['Target', 'Open', 'High', 'Low', 'Close', 'Volume'], axis=1)
y = np.where(featured_data['Target'] > 0, 1, 0)  # 分类问题# 标准化
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)# 划分数据集
X_train, X_test, y_train, y_test = train_test_split(X_scaled, y, test_size=0.2, shuffle=False)

预测模型构建

python

复制

下载

from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, accuracy_score
import xgboost as xgb# 随机森林模型
rf = RandomForestClassifier(n_estimators=100, random_state=42)
rf.fit(X_train, y_train)
y_pred_rf = rf.predict(X_test)
print("随机森林准确率:", accuracy_score(y_test, y_pred_rf))# XGBoost模型
xgb_model = xgb.XGBClassifier(n_estimators=100, learning_rate=0.1)
xgb_model.fit(X_train, y_train)
y_pred_xgb = xgb_model.predict(X_test)
print("XGBoost准确率:", accuracy_score(y_test, y_pred_xgb))# 特征重要性
plt.figure(figsize=(10, 6))
pd.Series(xgb_model.feature_importances_, index=X.columns).sort_values().plot(kind='barh')
plt.title('Feature Importance')
plt.savefig('feature_importance.png')
plt.show()