共计 2693 个字符，预计需要花费 7 分钟才能阅读完成。
数据层构建
行情数据获取
使用 yfinance 获取美股数据（或 akshare 获取 A 股数据）是最快捷的方式。以下是日级数据获取示例：
import yfinance as yf
import pandas as pd
# 获取苹果公司三年历史数据
data = yf.download('AAPL',
start='2020-01-01',
end='2023-01-01',
interval='1d',
auto_adjust=True) # 自动复权关键注意事项：
- 复权处理：建议优先使用
auto_adjust=True参数 - 停牌处理：用
fillna(method='ffill')向前填充 - 分钟数据：需注意 API 调用频率限制
数据清洗流程
- 处理缺失值：对停牌日期的 OHLC 数据采用前值填充
- 异常值检测：使用 3σ 原则过滤异常价格波动
- 数据标准化：将不同量纲的特征缩放到相同区间
# 数据清洗示例
def clean_data(df):
# 处理缺失值
df = df.fillna(method='ffill')
# 去除极端值
mean = df['Close'].mean()
std = df['Close'].std()
df = df[(df['Close'] > mean - 3*std) &
(df['Close'] < mean + 3*std)]
return df特征工程
技术指标计算
使用 TA-Lib 计算常见技术指标：
import talib
# 计算 MACD
data['macd'], data['signal'], _ = talib.MACD(data['Close'],
fastperiod=12,
slowperiod=26,
signalperiod=9)
# 计算布林带
data['upper'], data['middle'], data['lower'] = talib.BBANDS(data['Close'],
timeperiod=20)特征标准化
不同量纲的特征需要统一标准化：
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
features = ['macd', 'upper', 'lower']
data[features] = scaler.fit_transform(data[features])模型构建
监督学习模型
适合趋势跟踪类策略：
from xgboost import XGBClassifier
from sklearn.model_selection import train_test_split
# 构造标签：次日涨跌
data['target'] = (data['Close'].shift(-1) > data['Close']).astype(int)
# 划分数据集
X = data[features]
y = data['target']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False)
# 训练模型
model = XGBClassifier(n_estimators=100)
model.fit(X_train, y_train)强化学习模型
适合高频交易场景：
import gym
from stable_baselines3 import PPO
# 创建交易环境
env = gym.make('stocks-v0', df=data)
# 训练 PPO 智能体
model = PPO('MlpPolicy', env, verbose=1)
model.learn(total_timesteps=10000)回测实现
Backtrader 框架
import backtrader as bt
class MyStrategy(bt.Strategy):
def __init__(self):
self.sma = bt.indicators.SMA(period=15)
def next(self):
if self.data.close[0] > self.sma[0]:
self.buy()
elif self.data.close[0] < self.sma[0]:
self.sell()
# 运行回测
cerebro = bt.Cerebro()
datafeed = bt.feeds.PandasData(dataname=data)
cerebro.adddata(datafeed)
cerebro.addstrategy(MyStrategy)
results = cerebro.run()避免前视偏差
关键措施：
- 严格按时间戳顺序处理数据
- 在特征计算中使用
.shift(1)避免数据泄露 - 回测时禁用
peek模式
生产部署
Docker 容器化
FROM python:3.8
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "main.py"]风控系统设计
class RiskManager:
def __init__(self, max_drawdown=0.2):
self.max_drawdown = max_drawdown
def check_risk(self, portfolio):
current_dd = 1 - portfolio.value / portfolio.max_value
return current_dd < self.max_drawdown风险控制
过拟合检测
采用 Walk-Forward 验证：
- 将数据划分为多个训练 / 测试窗口
- 滚动训练并记录各窗口表现
- 观察测试集表现的稳定性
交易成本模拟
class Slippage(bt.Slippage):
def __init__(self, slip=0.001):
self.slip = slip
def _getslippage(self, size, price):
return price * (1 + np.sign(size) * self.slip)延伸思考
策略失效应对
- 持续监控策略表现（设置预警阈值）
- 保留多个备选策略
- 建立市场状态分类器
高频交易优化
# Latency 测量示例
import time
def measure_latency():
start = time.perf_counter_ns()
# 执行交易指令
end = time.perf_counter_ns()
return (end - start) / 1e6 # 毫秒经验总结
- 数据质量决定策略上限
- 简单策略往往更鲁棒
- 风险控制比收益更重要
- 回测结果需打折扣看待
建议从日频策略开始实践，逐步过渡到更高频次。记住：市场上没有圣杯策略，持续迭代才是王道。
正文完
