Python量化投资实战
所属分类 quant
浏览量 9
量化投资的核心：用数据和模型替代主观判断
Python 凭借其丰富的第三方库、易上手的语法、高效的数据分析能力，
成为量化投资领域最主流的编程语言之一。
从数据获取、策略研究到回测、实盘交易，Python 几乎覆盖了量化投资的全流程
Python 降低了量化投资的门槛，从入门级的均线策略到进阶的机器学习模型，都能高效实现。
核心是先掌握 pandas、numpy 等基础工具，再结合回测框架验证策略，最后通过实盘接口落地。
注意：量化策略的核心是 逻辑有效 + 风险可控，而非单纯的代码技巧。
一、核心应用场景
环节 核心目标
数据获取与清洗 获取高质量的金融数据
因子挖掘与特征工程 筛选能预测收益的有效因子（如 PE、动量）
策略研发 构建交易规则（如均线策略、多因子）
回测 验证策略历史表现
风险控制 控制最大回撤、仓位等 风险指标（VaR、夏普率）计算
实盘交易 自动执行交易 对接券商 API、订单管理
可视化与报告 展示策略效果 动态图表、自动化报告生成
二、量化投资必备的 Python库
1. 数据获取类
tushare：
免费获取 A 股、期货、基金等数据（需注册），适合入门；
akshare：
开源财经数据接口，覆盖股票、期货、外汇、宏观经济等，实时 / 历史数据均可获取；
baostock：
免费获取沪深 A 股全部历史数据（分笔、日线、财务数据）
yfinance：
获取美股、港股等海外市场数据（雅虎财经接口）
Wind/Pytdx：
对接万得、通达信等专业金融终端（需付费 / 权限）
2. 数据处理与分析类
pandas：量化核心库，处理时间序列数据（如 K 线）、计算指标（均线、RSI）、数据清洗；
numpy：高性能数值计算，支撑因子矩阵、矩阵运算；
scipy：统计分析（如假设检验、因子显著性检验）
3. 策略回测类
Backtrader：轻量级回测框架，支持多资产、多周期，易上手，适合中小策略；
VNPY：国内主流的量化框架，支持回测 + 实盘，对接多家券商 / 期货公司 API；
QuantLib：复杂衍生品定价（如期权、利率产品）；
JoinQuant / 聚宽：线上量化平台，内置 Python 环境，支持回测和实盘
4. 可视化类
matplotlib/seaborn：绘制 K 线图、策略收益曲线、因子分布图；
plotly：交互式图表，适合展示策略绩效（如回撤曲线、持仓分布）；
pyecharts：生成可视化网页报告
5. 机器学习 / 深度学习类（进阶）
scikit-learn：构建量化模型（如线性回归、随机森林预测股价、因子筛选）；
TensorFlow/PyTorch：深度学习模型（如 LSTM 预测价格、CNN 分析 K 线形态）；
xgboost/lightgbm：梯度提升树，用于多因子模型的因子加权
三、Python 量化投资实战流程（以 A 股均线策略为例）
步骤 1：数据获取（以 akshare 为例）
获取某只股票的日线数据（如贵州茅台 600519）：
import akshare as ak
# 获取日线数据
stock_df = ak.stock_zh_a_hist(symbol="600519", period="daily", start_date="20200101", end_date="20250101", adjust="qfq")
# 数据清洗：保留核心列，转换日期格式
stock_df = stock_df[["日期", "开盘价", "最高价", "最低价", "收盘价", "成交量"]]
stock_df["日期"] = pd.to_datetime(stock_df["日期"])
stock_df.set_index("日期", inplace=True)
print(stock_df.head())
步骤 2：因子 / 指标计算（以 5 日均线、20 日均线为例）
均线策略核心逻辑：
金叉（5 日均线上穿 20 日均线）买入，死叉（5 日均线下穿 20 日均线）卖出。
import pandas as pd
# 计算5日和20日均线
stock_df["ma5"] = stock_df["收盘价"].rolling(window=5).mean()
stock_df["ma20"] = stock_df["收盘价"].rolling(window=20).mean()
# 生成交易信号：1=买入，0=卖出，NaN=无信号
stock_df["signal"] = 0
# 金叉：前一天ma5 < ma20，当天ma5 > ma20
golden_cross = (stock_df["ma5"].shift(1) < stock_df["ma20"].shift(1)) & (stock_df["ma5"] > stock_df["ma20"])
# 死叉：前一天ma5 > ma20，当天ma5 < ma20
death_cross = (stock_df["ma5"].shift(1) > stock_df["ma20"].shift(1)) & (stock_df["ma5"] < stock_df["ma20"])
stock_df.loc[golden_cross, "signal"] = 1
stock_df.loc[death_cross, "signal"] = 0
# 填充持仓状态：1=持有，0=空仓
stock_df["position"] = stock_df["signal"].fillna(method="ffill")
步骤 3：回测（计算策略收益）
python
# 计算每日收益率
stock_df["daily_return"] = stock_df["收盘价"].pct_change()
# 计算策略收益率（持仓时才产生收益）
stock_df["strategy_return"] = stock_df["daily_return"] * stock_df["position"].shift(1)
# 计算累计收益
stock_df["cum_market_return"] = (1 + stock_df["daily_return"]).cumprod()
stock_df["cum_strategy_return"] = (1 + stock_df["strategy_return"]).cumprod()
# 打印核心指标
total_market_return = stock_df["cum_market_return"].iloc[-1] - 1
total_strategy_return = stock_df["cum_strategy_return"].iloc[-1] - 1
sharp_ratio = (stock_df["strategy_return"].mean() / stock_df["strategy_return"].std()) * np.sqrt(252) # 年化夏普率
max_drawdown = (stock_df["cum_strategy_return"] / stock_df["cum_strategy_return"].cummax() - 1).min()
print(f"市场累计收益：{total_market_return:.2%}")
print(f"策略累计收益：{total_strategy_return:.2%}")
print(f"夏普率：{sharp_ratio:.2f}")
print(f"最大回撤：{max_drawdown:.2%}")
步骤 4：可视化策略效果
import matplotlib.pyplot as plt
plt.rcParams["font.sans-serif"] = ["SimHei"] # 显示中文
plt.figure(figsize=(12, 6))
plt.plot(stock_df["cum_market_return"], label="市场收益")
plt.plot(stock_df["cum_strategy_return"], label="策略收益")
plt.title("茅台均线策略回测（2020-2025）")
plt.xlabel("日期")
plt.ylabel("累计收益")
plt.legend()
plt.grid(True)
plt.show()
四、进阶应用方向
多因子模型：
结合价值因子（PE/PB）、动量因子（收益率）、质量因子（ROE）等，
用回归 / 机器学习加权因子，构建更稳定的策略；
算法交易：
通过 Python 对接券商 API（如中信证券、华泰证券的量化接口），实现自动下单、仓位动态调整；
高频交易：
结合 NumPy 向量化运算、Cython 加速，处理分笔 / 逐笔数据，实现毫秒级交易决策
风险模型：
计算 VaR（风险价值）、跟踪误差、行业暴露度，控制策略的系统性风险
事件驱动策略：
爬取财报、公告、新闻等文本数据，用 NLP（如 jieba、BERT）分析情绪，触发交易信号。
五、注意事项
数据偏差：
回测时需注意幸存者偏差（只选存活的股票）、前视偏差（用未来数据），避免过度优化
实盘适配：
回测≠实盘，需考虑滑点（买卖价差）、手续费、市场冲击成本
合规性：
量化交易需遵守券商规则，高频交易可能受交易所限制；
库的选择：
入门用 Backtrader+akshare，实盘优先 VNPY / 券商官方 API
六 典型实战案例
入门级均线策略、进阶多因子模型、事件驱动策略和机器学习预测
案例 1：入门级 —— 双均线交叉策略（A 股）
核心逻辑
短期均线上穿长期均线（金叉）：买入；
短期均线下穿长期均线（死叉）：卖出；
对比策略收益与市场基准（持有不动）
import akshare as ak
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
# 1. 数据获取（贵州茅台 600519，2020-2025日线）
def get_stock_data(symbol="600519", start="20200101", end="20250101"):
stock_df = ak.stock_zh_a_hist(
symbol=symbol, period="daily", start_date=start, end_date=end, adjust="qfq"
)
# 数据清洗
stock_df = stock_df[["日期", "开盘价", "最高价", "最低价", "收盘价", "成交量"]]
stock_df["日期"] = pd.to_datetime(stock_df["日期"])
stock_df.set_index("日期", inplace=True)
stock_df.rename(columns={"收盘价": "close"}, inplace=True)
return stock_df
# 2. 计算均线和交易信号
def generate_signal(stock_df, short_window=5, long_window=20):
# 计算均线
stock_df["ma_short"] = stock_df["close"].rolling(window=short_window).mean()
stock_df["ma_long"] = stock_df["close"].rolling(window=long_window).mean()
# 生成交易信号：1=买入，0=卖出
stock_df["signal"] = np.nan
# 金叉：前一天短期 < 长期，当天短期 > 长期
golden_cross = (stock_df["ma_short"].shift(1) < stock_df["ma_long"].shift(1)) & (stock_df["ma_short"] > stock_df["ma_long"])
# 死叉：前一天短期 > 长期，当天短期 < 长期
death_cross = (stock_df["ma_short"].shift(1) > stock_df["ma_long"].shift(1)) & (stock_df["ma_short"] < stock_df["ma_long"])
stock_df.loc[golden_cross, "signal"] = 1
stock_df.loc[death_cross, "signal"] = 0
# 填充持仓状态（前向填充）
stock_df["position"] = stock_df["signal"].fillna(method="ffill")
# 初始无信号时设为0（空仓）
stock_df["position"].fillna(0, inplace=True)
return stock_df
# 3. 回测计算收益
def backtest(stock_df):
# 每日收益率
stock_df["daily_return"] = stock_df["close"].pct_change()
# 策略收益率（持仓时才产生收益）
stock_df["strategy_return"] = stock_df["daily_return"] * stock_df["position"].shift(1)
# 累计收益
stock_df["cum_market"] = (1 + stock_df["daily_return"]).cumprod()
stock_df["cum_strategy"] = (1 + stock_df["strategy_return"]).cumprod()
# 计算核心指标
total_market = (stock_df["cum_market"].iloc[-1] - 1) * 100
total_strategy = (stock_df["cum_strategy"].iloc[-1] - 1) * 100
# 年化夏普率（无风险利率按3%算）
sharpe = (stock_df["strategy_return"].mean() - 0.03/252) / stock_df["strategy_return"].std() * np.sqrt(252)
# 最大回撤
drawdown = (stock_df["cum_strategy"] / stock_df["cum_strategy"].cummax() - 1).min() * 100
print(f"市场累计收益：{total_market:.2f}%")
print(f"策略累计收益：{total_strategy:.2f}%")
print(f"年化夏普率：{sharpe:.2f}")
print(f"最大回撤：{drawdown:.2f}%")
return stock_df
# 4. 可视化
def plot_result(stock_df):
plt.rcParams["font.sans-serif"] = ["SimHei"] # 显示中文
plt.figure(figsize=(12, 8))
# 子图1：价格+均线
ax1 = plt.subplot(211)
ax1.plot(stock_df["close"], label="收盘价", color="blue")
ax1.plot(stock_df["ma_short"], label="5日均线", color="red")
ax1.plot(stock_df["ma_long"], label="20日均线", color="green")
ax1.scatter(stock_df[stock_df["signal"]==1].index,
stock_df[stock_df["signal"]==1]["close"],
marker="^", color="red", label="买入")
ax1.scatter(stock_df[stock_df["signal"]==0].index,
stock_df[stock_df["signal"]==0]["close"],
marker="v", color="green", label="卖出")
ax1.set_title("双均线策略（茅台 2020-2025）")
ax1.set_ylabel("价格（元）")
ax1.legend()
ax1.grid(True)
# 子图2：累计收益
ax2 = plt.subplot(212)
ax2.plot(stock_df["cum_market"], label="市场收益", color="blue")
ax2.plot(stock_df["cum_strategy"], label="策略收益", color="red")
ax2.set_ylabel("累计收益（倍）")
ax2.set_xlabel("日期")
ax2.legend()
ax2.grid(True)
plt.tight_layout()
plt.show()
# 主流程
if __name__ == "__main__":
df = get_stock_data()
df = generate_signal(df)
df = backtest(df)
plot_result(df)
输出效果
打印策略收益、夏普率、最大回撤等核心指标；
可视化股价 + 均线 + 买卖点、策略收益 vs 市场收益曲线
案例 2：进阶 —— 多因子选股策略（沪深 300 成分股）
核心逻辑
筛选沪深 300 成分股，选取 3 个经典因子：
价值因子：市盈率（PE）倒数（越高越有价值）；
动量因子：近 20 日收益率（越高动量越强）；
质量因子：净资产收益率（ROE）（越高公司质量越好）；
对因子标准化后加权打分，每月选前 50 只股票买入，下月调仓。
import akshare as ak
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# 1. 获取沪深300成分股列表
def get_hs300_stocks():
hs300_df = ak.index_stock_cons_csindex(symbol="000300")
return hs300_df["成分券代码"].tolist()
# 2. 获取单只股票的因子数据
def get_stock_factor(stock_code, date):
try:
# 财务数据（PE、ROE）
finance_df = ak.stock_financial_analysis_indicator(stock=stock_code)
pe = finance_df["市盈率(TTM)"].iloc[0] if not pd.isna(finance_df["市盈率(TTM)"].iloc[0]) else 100
roe = finance_df["净资产收益率(ROE)"].iloc[0] if not pd.isna(finance_df["净资产收益率(ROE)"].iloc[0]) else 0
# 动量因子（近20日收益率）
end_date = date.strftime("%Y%m%d")
start_date = (date - timedelta(days=20)).strftime("%Y%m%d")
price_df = ak.stock_zh_a_hist(stock_code, period="daily", start_date=start_date, end_date=end_date, adjust="qfq")
momentum = (price_df["收盘价"].iloc[-1] / price_df["收盘价"].iloc[0] - 1) * 100
return {
"code": stock_code,
"pe": pe,
"roe": roe,
"momentum": momentum,
"pe_score": 1/pe if pe != 0 else 0, # PE倒数（越小PE得分越高）
}
except:
return None
# 3. 因子标准化+打分
def factor_score(factor_df):
# 标准化（z-score）：(值-均值)/标准差
for col in ["pe_score", "roe", "momentum"]:
factor_df[col + "_norm"] = (factor_df[col] - factor_df[col].mean()) / factor_df[col].std()
# 加权打分（等权重）
factor_df["total_score"] = (factor_df["pe_score_norm"] + factor_df["roe_norm"] + factor_df["momentum_norm"]) / 3
# 按分数排序，取前50
factor_df = factor_df.sort_values("total_score", ascending=False).head(50)
return factor_df["code"].tolist()
# 4. 月度调仓回测（简化版）
def backtest_multi_factor():
# 回测时间：2022-2024，每月最后一个交易日调仓
start_date = datetime(2022, 1, 1)
end_date = datetime(2024, 12, 31)
hs300_stocks = get_hs300_stocks()[:50] # 简化：只取前50只（实际取全部）
# 月度调仓
month_dates = pd.date_range(start=start_date, end=end_date, freq="M")
portfolio = [] # 持仓股票列表
total_return = 1.0 # 累计收益
for date in month_dates:
print(f"调仓日期：{date.strftime('%Y-%m-%d')}")
# 获取因子数据
factor_list = []
for code in hs300_stocks:
factor = get_stock_factor(code, date)
if factor:
factor_list.append(factor)
if not factor_list:
continue
# 因子打分，选前50
factor_df = pd.DataFrame(factor_list)
portfolio = factor_score(factor_df)
# 计算当月持仓收益（简化：取持仓股票平均涨幅）
month_return = 0.0
count = 0
for code in portfolio:
try:
# 当月收益率
month_price = ak.stock_zh_a_hist(code, period="daily",
start_date=date.strftime("%Y%m%d"),
end_date=(date + timedelta(days=30)).strftime("%Y%m%d"),
adjust="qfq")
if len(month_price) > 0:
ret = (month_price["收盘价"].iloc[-1] / month_price["收盘价"].iloc[0] - 1)
month_return += ret
count += 1
except:
continue
# 当月平均收益
if count > 0:
month_return = month_return / count
total_return *= (1 + month_return)
print(f"当月收益：{month_return*100:.2f}%，累计收益：{(total_return-1)*100:.2f}%")
print(f"多因子策略累计收益：{(total_return-1)*100:.2f}%")
if __name__ == "__main__":
backtest_multi_factor()
注意事项
实际应用需处理因子缺失值、行业中性化（避免单一行业集中）；
需考虑交易成本（手续费、滑点），本例为简化版未计入
案例 3：事件驱动 —— 财报超预期策略
核心逻辑
爬取 A 股公司财报，筛选 “净利润同比增长 > 50% 且营收增长 > 30%” 的公司；
财报发布后次日买入，持有 10 个交易日卖出；
统计该策略的收益表现。
import akshare as ak
import pandas as pd
import numpy as np
# 1. 获取财报数据（全部A股）
def get_finance_report():
# 利润表（净利润）
profit_df = ak.stock_profit_sheet_all(symbol="年报") # 可改"中报"/"季报"
profit_df = profit_df[["股票代码", "净利润同比增长率(%)"]]
# 资产负债表（营收）
revenue_df = ak.stock_income_sheet_all(symbol="年报")
revenue_df = revenue_df[["股票代码", "营业总收入同比增长率(%)"]]
# 合并数据
finance_df = pd.merge(profit_df, revenue_df, on="股票代码", how="inner")
# 筛选超预期：净利润 > 50%，营收 > 30%
finance_df = finance_df[
(finance_df["净利润同比增长率(%)"] > 50) &
(finance_df["营业总收入同比增长率(%)"] > 30)
]
return finance_df["股票代码"].tolist()
# 2. 事件驱动回测
def event_backtest():
# 获取超预期股票列表
target_stocks = get_finance_report()[:20] # 简化：取前20只
total_strategy_return = []
for code in target_stocks:
try:
# 获取财报发布后10个交易日数据
stock_df = ak.stock_zh_a_hist(code, period="daily", start_date="20230101", end_date="20240101", adjust="qfq")
stock_df["日期"] = pd.to_datetime(stock_df["日期"])
stock_df.set_index("日期", inplace=True)
# 假设财报发布日为2023-04-30（实际需对接财报发布日期接口）
publish_date = pd.to_datetime("2023-04-30")
# 找到发布日后的第一个交易日
trade_date = stock_df[stock_df.index >= publish_date].index[0]
# 取发布日后10个交易日的收益
trade_idx = stock_df.index.get_loc(trade_date)
if trade_idx + 10 >= len(stock_df):
continue
# 买入价（发布次日开盘价），卖出价（10日后收盘价）
buy_price = stock_df["开盘价"].iloc[trade_idx + 1]
sell_price = stock_df["收盘价"].iloc[trade_idx + 10]
ret = (sell_price / buy_price - 1) * 100
total_strategy_return.append(ret)
print(f"股票{code}：10日收益 {ret:.2f}%")
except:
continue
# 策略平均收益
if total_strategy_return:
avg_return = np.mean(total_strategy_return)
win_rate = len([r for r in total_strategy_return if r > 0]) / len(total_strategy_return) * 100
print(f"\n策略平均10日收益：{avg_return:.2f}%")
print(f"胜率：{win_rate:.2f}%")
if __name__ == "__main__":
event_backtest()
案例 4：高阶 —— 机器学习预测股价（LSTM）
核心逻辑
用 LSTM 神经网络预测股票次日收盘价；
特征：前 60 日收盘价、成交量、均线等；
训练集：2020-2023 年数据，测试集：2024 年数据
import akshare as ak
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense
# 1. 数据预处理
def preprocess_data(symbol="600519"):
# 获取数据
df = ak.stock_zh_a_hist(symbol, period="daily", start_date="20200101", end_date="20241231", adjust="qfq")
df = df[["日期", "收盘价", "成交量", "最高价", "最低价"]]
df["日期"] = pd.to_datetime(df["日期"])
df.set_index("日期", inplace=True)
# 归一化（0-1）
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(df["收盘价"].values.reshape(-1, 1))
# 构建训练集：前60天数据预测第61天
x_train, y_train = [], []
for i in range(60, len(scaled_data)):
x_train.append(scaled_data[i-60:i, 0])
y_train.append(scaled_data[i, 0])
x_train, y_train = np.array(x_train), np.array(y_train)
# LSTM输入格式：[样本数, 时间步, 特征数]
x_train = np.reshape(x_train, (x_train.shape[0], x_train.shape[1], 1))
return df, scaler, x_train, y_train
# 2. 构建LSTM模型
def build_lstm_model(input_shape):
model = Sequential()
model.add(LSTM(units=50, return_sequences=True, input_shape=input_shape))
model.add(LSTM(units=50, return_sequences=False))
model.add(Dense(units=25))
model.add(Dense(units=1)) # 预测次日收盘价
model.compile(optimizer="adam", loss="mean_squared_error")
return model
# 3. 训练+预测
def train_and_predict():
df, scaler, x_train, y_train = preprocess_data()
# 划分训练/测试集（2024年为测试集）
train_size = int(len(x_train) * 0.8)
x_train, x_test = x_train[:train_size], x_train[train_size:]
y_train, y_test = y_train[:train_size], y_train[train_size:]
# 训练模型
model = build_lstm_model((x_train.shape[1], 1))
model.fit(x_train, y_train, batch_size=32, epochs=50)
# 预测
predictions = model.predict(x_test)
# 反归一化
predictions = scaler.inverse_transform(predictions)
y_test = scaler.inverse_transform(y_test.reshape(-1, 1))
# 可视化
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.figure(figsize=(12, 6))
plt.plot(y_test, label="真实收盘价", color="blue")
plt.plot(predictions, label="预测收盘价", color="red")
plt.title("LSTM股价预测（茅台）")
plt.xlabel("时间")
plt.ylabel("价格（元）")
plt.legend()
plt.show()
# 计算误差
rmse = np.sqrt(np.mean(((predictions - y_test) ** 2)))
print(f"测试集RMSE（均方根误差）：{rmse:.2f}元")
if __name__ == "__main__":
train_and_predict()
案例扩展与注意事项
实盘适配：
回测需加入滑点（如 0.1%）、手续费（如万 3）；
对接券商 API（如 VNPY、同花顺 iFinD）实现自动交易；
策略优化：
多因子模型可加入因子正交化、行业中性化；
机器学习模型可加入更多特征（RSI、MACD、情绪因子）；
风险控制：
单只股票仓位不超过 10%；
最大回撤超过 20% 时强制平仓；
数据来源：
免费：akshare、tushare、baostock；
付费：Wind、Choice、聚宽 / 米筐平台。
这些案例覆盖了量化投资的核心场景，从入门到高阶，
可根据需求调整因子、参数或模型结构，
建议先在回测平台（如 JoinQuant）验证策略有效性后再考虑实盘。
上一篇
《宽客：华尔街顶级数量金融大师的另类人生》精华笔记
量化策略分类
量化策略评估指标