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 = data
self.initial_capital = initial_capital
self.positions = []
self.current_cash = initial_capital
self.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: # 确保有足够数据计算MA200
return 0
if 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'] = 0
data.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'] = 0
data.loc[data['Momentum'] > 0, 'Signal'] = 1
data.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.x
optimal_weights = optimize_portfolio(returns, cov_matrix)
print("最优权重:", dict(zip(tickers, optimal_weights)))
风险价值(VaR)计算
python
复制
下载
from scipy.stats import norm
def calculate_var(returns, confidence_level=0.95):
mean = returns.mean()
std_dev = returns.std()
# 参数法VaR
var_parametric = norm.ppf(1-confidence_level, mean, std_dev)
# 历史模拟法VaR
var_historical = np.percentile(returns, (1-confidence_level)*100)
return var_parametric, var_historical
aapl_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 order
except Exception as e:
print(f"下单失败: {str(e)}")
return None
# 获取账户余额
balance = exchange.fetch_balance()
print("USDT余额:", balance['USDT']['free'])
高频交易策略
订单簿分析
python
复制
下载
import numpy as np
def 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 = exchange
self.symbol = symbol
self.position_limit = position_limit
self.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:
continue
self.orders = []
# 使用示例
mm = MarketMaker(exchange, 'BTC/USDT')
mm.run_strategy()
机器学习在量化中的应用
特征工程
python
复制
下载
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
def 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'] - 1
return 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()
结语与学习路径
https://www.python.org/static/community_logos/python-powered-h-140x182.png
通过这九篇系列教程，你已经掌握了：
金融数据获取与处理技术
技术指标实现与可视化
回测框架设计与策略评估
经典量化交易策略实现
投资组合优化与风险管理
实盘交易接口使用
高频交易策略基础
机器学习在量化中的应用
进阶学习方向：
深入量化领域：
市场微观结构研究
期权定价与波动率交易
套利策略开发
技术深化：
C++/Rust扩展性能关键部分
分布式回测系统构建
强化学习在交易中的应用
专业认证：
CFA (特许金融分析师)
FRM (金融风险管理师)
CMT (特许市场技术分析师)
实盘经验：
从小资金开始实盘测试
参与量化交易比赛
加入量化对冲基金团队
量化交易是金融与技术的完美结合，Python作为这一领域的核心工具，将持续发挥重要作用。保持学习，你将成为市场的敏锐捕手！