量化系统开发全流程

# Python 数据清洗示例
import pandas as pd
import numpy as np

# 读取数据
df = pd.read_csv('stock_data.csv', parse_dates=['date'])
df.set_index('date', inplace=True)

# 前复权处理
df['adj_close'] = df['close'] * df['adj_factor']

# 处理缺失值（前向填充最多3天）
df.fillna(method='ffill', limit=3, inplace=True)

# 去除停牌日（成交量为0的日期）
df = df[df['volume'] > 0]

# 识别异常值（3倍标准差）
mean = df['return'].mean()
std = df['return'].std()
df = df[np.abs(df['return'] - mean) <= 3 * std]

# 构建动量因子
def momentum_factor(df, window=20):
    """计算过去N天的动量"""
    df['momentum'] = df['adj_close'] / df['adj_close'].shift(window) - 1
    return df

# 因子标准化（Z-score）
def zscore_standardize(df, column):
    mean = df[column].mean()
    std = df[column].std()
    df[column + '_z'] = (df[column] - mean) / std
    return df

# 市值中性化（回归法）
from sklearn.linear_model import LinearRegression

def neutralize(df, factor_col, market_cap_col):
    model = LinearRegression()
    X = df[[market_cap_col]].values
    y = df[factor_col].values
    model.fit(X, y)
    residuals = y - model.predict(X)
    df[factor_col + '_neutral'] = residuals
    return df

# 使用 backtrader 进行回测
import backtrader as bt

class MyStrategy(bt.Strategy):
    def __init__(self):
        self.sma = bt.indicators.SimpleMovingAverage(
            self.data.close, period=20)
        self.rsi = bt.indicators.RSI(self.data.close)

    def next(self):
        if self.position.size == 0:
            if self.data.close[0] > self.sma[0] and self.rsi < 70:
                self.buy()
        else:
            if self.data.close[0] < self.sma[0] or self.rsi > 80:
                self.sell()

cerebro = bt.Cerebro()
cerebro.addstrategy(MyStrategy)
# ... 添加数据、设置初始资金、佣金等
results = cerebro.run()

# 基于ATR的动态仓位管理
def calculate_position_size(account_balance, risk_percent, entry_price, stop_loss_price):
    """根据账户余额和止损距离计算仓位"""
    risk_amount = account_balance * (risk_percent / 100)
    stop_distance = abs(entry_price - stop_loss_price)
    position_size = risk_amount / stop_distance
    return position_size

# 凯利公式
def kelly_criterion(win_rate, avg_win, avg_loss):
    """计算最优下注比例"""
    b = avg_win / avg_loss # 盈亏比
    p = win_rate # 胜率
    kelly = (b * p - (1 - p)) / b
    return max(0, kelly) # 半凯利：kelly * 0.5

# 绩效归因分析示例
import pandas as pd
import empyrical as ep

# 计算滚动夏普比率
returns = pd.Series(daily_returns)
rolling_sharpe = ep.rolling_sharpe_ratio(returns, window=60)

# 计算最大回撤
max_drawdown = ep.max_drawdown(returns)

# 计算信息比率（相对基准）
excess_returns = returns - benchmark_returns
information_ratio = ep.information_ratio(excess_returns)

# 月度收益统计
monthly_returns = ep.monthly_returns(returns)
print(monthly_returns)