一、核心框架篇
1.1 量化交易四大支柱
数据 → 策略 → 回测 → 风控

1.2 完整工作流

# 典型量化流程
1. 数据获取 (pandas-datareader, tushare, akshare)
2. 数据清洗 (pandas, numpy)
3. 策略开发 (backtrader, zipline)
4. 回测分析 (pyfolio, empyrical)
5. 实盘执行 (vn.py, easytrader)



二、关键技术点

2.1 数据获取与处理

import pandas as pd
import numpy as np

# 数据标准化处理技巧
def normalize_data(df):
    """数据标准化"""
    df = df.copy()
    # 处理缺失值
    df.fillna(method='ffill', inplace=True)
    df.fillna(method='bfill', inplace=True)
    
    # 去除异常值 (3σ原则)
    for col in df.columns:
        mean = df[col].mean()
        std = df[col].std()
        df[col] = np.clip(df[col], mean-3*std, mean+3*std)
    
    return df

# 特征工程
def create_features(df):
    """创建技术指标特征"""
    # 移动平均
    df['MA5'] = df['close'].rolling(5).mean()
    df['MA20'] = df['close'].rolling(20).mean()
    
    # 布林带
    df['BB_mid'] = df['close'].rolling(20).mean()
    df['BB_std'] = df['close'].rolling(20).std()
    df['BB_upper'] = df['BB_mid'] + 2*df['BB_std']
    df['BB_lower'] = df['BB_mid'] - 2*df['BB_std']
    
    # RSI
    delta = df['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
    rs = gain / loss
    df['RSI'] = 100 - (100 / (1 + rs))
    
    return df


2.2 经典策略实现
双均线策略
class DualMAStrategy:
    """双均线策略"""
    def __init__(self, short_window=5, long_window=20):
        self.short_window = short_window
        self.long_window = long_window
        self.short_ma = None
        self.long_ma = None
        
    def generate_signals(self, prices):
        signals = pd.DataFrame(index=prices.index)
        signals['price'] = prices
        
        # 计算均线
        signals['short_ma'] = prices.rolling(self.short_window).mean()
        signals['long_ma'] = prices.rolling(self.long_window).mean()
        
        # 生成信号
        signals['signal'] = 0
        signals['signal'][self.short_window:] = np.where(
            signals['short_ma'][self.short_window:] > 
            signals['long_ma'][self.short_window:], 1, 0
        )
        
        # 交易信号
        signals['positions'] = signals['signal'].diff()
        
        return signals



均值回归策略

class MeanReversionStrategy:
    """均值回归策略"""
    def __init__(self, lookback=20, zscore_threshold=2):
        self.lookback = lookback
        self.zscore_threshold = zscore_threshold
        
    def calculate_zscore(self, series):
        """计算Z分数"""
        return (series - series.mean()) / series.std()
    
    def generate_signals(self, prices):
        signals = pd.DataFrame(index=prices.index)
        signals['price'] = prices
        
        # 计算移动平均和标准差
        signals['mean'] = prices.rolling(self.lookback).mean()
        signals['std'] = prices.rolling(self.lookback).std()
        
        # 计算Z分数
        signals['zscore'] = (prices - signals['mean']) / signals['std']
        
        # 生成信号
        signals['signal'] = 0
        signals.loc[signals['zscore'] > self.zscore_threshold, 'signal'] = -1  # 卖出
        signals.loc[signals['zscore'] < -self.zscore_threshold, 'signal'] = 1   # 买入
        signals.loc[abs(signals['zscore']) < 0.5, 'signal'] = 0                 # 平仓
        
        return signals


2.3 风险控制模块

class RiskManager:
    """风险管理系统"""
    def __init__(self, max_position_size=0.1, 
                 max_portfolio_risk=0.02,
                 stop_loss_pct=0.05):
        self.max_position_size = max_position_size
        self.max_portfolio_risk = max_portfolio_risk
        self.stop_loss_pct = stop_loss_pct
        
    def calculate_position_size(self, capital, risk_per_trade, 
                                entry_price, stop_loss_price):
        """凯利公式调整的头寸计算"""
        # 计算风险金额
        risk_amount = capital * risk_per_trade
        
        # 计算单笔亏损
        loss_per_share = abs(entry_price - stop_loss_price)
        
        # 计算头寸大小
        position_size = risk_amount / loss_per_share
        
        # 应用最大仓位限制
        max_shares = capital * self.max_position_size / entry_price
        position_size = min(position_size, max_shares)
        
        return int(position_size)
    
    def check_stop_loss(self, current_price, entry_price, position):
        """止损检查"""
        if position > 0:  # 多头
            loss_pct = (entry_price - current_price) / entry_price
            return loss_pct >= self.stop_loss_pct
        elif position < 0:  # 空头
            loss_pct = (current_price - entry_price) / entry_price
            return loss_pct >= self.stop_loss_pct
        return False



三、Backtrader实战示例
3.1 完整策略模板

import backtrader as bt

class MyStrategy(bt.Strategy):
    params = (
        ('ma_period', 20),
        ('rsi_period', 14),
        ('rsi_overbought', 70),
        ('rsi_oversold', 30),
    )
    
    def __init__(self):
        # 数据引用
        self.dataclose = self.datas[0].close
        
        # 指标计算
        self.sma = bt.indicators.SimpleMovingAverage(
            self.datas[0], period=self.params.ma_period
        )
        self.rsi = bt.indicators.RSI(
            self.datas[0], period=self.params.rsi_period
        )
        
        # 交易状态跟踪
        self.order = None
        self.buyprice = None
        self.buycomm = None
        
    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            return
            
        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(f'买入执行, 价格: {order.executed.price:.2f}')
            elif order.issell():
                self.log(f'卖出执行, 价格: {order.executed.price:.2f}')
                
        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log('订单取消/保证金不足/拒绝')
            
        self.order = None
        
    def next(self):
        # 如果有挂单，不进行新交易
        if self.order:
            return
            
        # 如果没有持仓
        if not self.position:
            # RSI超卖且价格在均线上方
            if (self.rsi < self.params.rsi_oversold and 
                self.dataclose[0] > self.sma[0]):
                self.order = self.buy()
                
        else:
            # RSI超买或价格跌破均线
            if (self.rsi > self.params.rsi_overbought or 
                self.dataclose[0] < self.sma[0]):
                self.order = self.sell()
    
    def log(self, txt, dt=None):
        dt = dt or self.datas[0].datetime.date(0)
        print(f'{dt.isoformat()}, {txt}')



3.2 回测分析

def run_backtest():
    cerebro = bt.Cerebro()
    
    # 添加数据
    data = bt.feeds.PandasData(dataname=get_stock_data('000001'))
    cerebro.adddata(data)
    
    # 添加策略
    cerebro.addstrategy(MyStrategy)
    
    # 设置初始资金
    cerebro.broker.setcash(100000.0)
    
    # 设置手续费
    cerebro.broker.setcommission(commission=0.001)
    
    # 添加分析器
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
    
    # 运行回测
    results = cerebro.run()
    
    # 打印结果
    strat = results[0]
    print('夏普比率:', strat.analyzers.sharpe.get_analysis())
    print('最大回撤:', strat.analyzers.drawdown.get_analysis())
    print('总收益率:', strat.analyzers.returns.get_analysis())
    
    # 绘制图表
    cerebro.plot()



四、关键经验总结
4.1 常见陷阱与解决方案
未来函数：确保不使用未来数据
过拟合：使用交叉验证，简化策略
幸存者偏差：使用完整历史数据
手续费忽略：必须计入交易成本
滑点问题：设置合理的滑点模型


4.2 绩效评估指标

def evaluate_performance(returns, risk_free_rate=0.03):
    """综合绩效评估"""
    metrics = {}
    
    # 年化收益率
    metrics['annual_return'] = returns.mean() * 252
    
    # 年化波动率
    metrics['annual_volatility'] = returns.std() * np.sqrt(252)
    
    # 夏普比率
    metrics['sharpe_ratio'] = (metrics['annual_return'] - risk_free_rate) / metrics['annual_volatility']
    
    # 最大回撤
    cum_returns = (1 + returns).cumprod()
    running_max = np.maximum.accumulate(cum_returns)
    drawdown = (cum_returns - running_max) / running_max
    metrics['max_drawdown'] = drawdown.min()
    
    # 卡尔玛比率
    metrics['calmar_ratio'] = metrics['annual_return'] / abs(metrics['max_drawdown'])
    
    # 胜率
    metrics['win_rate'] = len(returns[returns > 0]) / len(returns)
    
    return metrics


4.3 实盘注意事项
数据延迟：使用实时数据API
订单执行：考虑成交概率和速度
资金管理：严格执行仓位控制
异常处理：添加健壮的错误处理
监控系统：实时监控策略表现

五、进阶技巧
5.1 多因子策略

class MultiFactorStrategy:
    """多因子选股策略"""
    def __init__(self, factors=['value', 'growth', 'momentum']):
        self.factors = factors
        
    def calculate_factors(self, stock_data):
        """计算因子值"""
        scores = {}
        
        for stock in stock_data:
            # 价值因子（PE, PB）
            value_score = self.calculate_value_factor(stock)
            
            # 成长因子（营收增长率）
            growth_score = self.calculate_growth_factor(stock)
            
            # 动量因子（价格动量）
            momentum_score = self.calculate_momentum_factor(stock)
            
            # 综合得分
            total_score = (0.4 * value_score + 
                          0.3 * growth_score + 
                          0.3 * momentum_score)
            
            scores[stock] = total_score
            
        return scores



5.2 机器学习集成

from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler

class MLStrategy:
    """机器学习策略"""
    def __init__(self):
        self.model = RandomForestClassifier(n_estimators=100)
        self.scaler = StandardScaler()
        
    def prepare_features(self, data):
        """准备特征数据"""
        features = []
        labels = []
        
        for i in range(len(data) - 60):
            # 特征窗口
            feature_window = data.iloc[i:i+30]
            
            # 标签（未来5天涨跌）
            future_return = (data.iloc[i+35] - data.iloc[i+30]) / data.iloc[i+30]
            label = 1 if future_return > 0.02 else 0  # 2%为阈值
            
            # 提取特征
            features.append(self.extract_features(feature_window))
            labels.append(label)
            
        return np.array(features), np.array(labels)



六、资源推荐
必读书籍
《主动投资组合管理》
《打开量化投资的黑箱》
《算法交易：制胜策略与原理》

实用工具
数据源：Tushare、AKShare、Baostock
可视化：Plotly、Pyecharts
优化：Zipline-reloaded（Zipline维护版）
实盘：EasyTrader、VNPY

学习路径建议
第一阶段：Python基础 + Pandas/Numpy
第二阶段：Backtrader基础策略
第三阶段：风险管理和组合优化
第四阶段：机器学习在量化中的应用
第五阶段：实盘系统开发