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()