AI量化源码入门指南：从零搭建你的第一个量化交易模型

1次阅读

没有评论

共计 2961 个字符，预计需要花费 8 分钟才能阅读完成。

在开始写代码前，我们需要了解几个核心概念：

Alpha：策略跑赢市场的超额收益，比如大盘涨 5% 你的策略赚 8%，那 3% 就是 Alpha
Beta：策略与市场波动的关联性，Beta= 1 表示完全跟随大盘
回测：用历史数据模拟交易，检验策略效果

举个生活化的例子：Alpha 就像外卖小哥在基础工资外赚的跑单奖励，Beta 则像电动车电量——电量足 (Beta 高) 时接单更猛，但风险也更大。

根据我带新人的经验，这些坑最容易踩：

数据获取难：免费源不稳定，付费 API 贵
过拟合陷阱：在历史数据表现完美，实盘一塌糊涂
交易延迟：网速慢导致成交价偏离预期
手续费吞噬：没算清交易成本导致实际收益为负

最近有个学员的策略回测年化 60%，实盘却亏损，就是因为忽略了每笔 0.2% 的手续费——频繁交易 200 次后，40% 收益就被吃掉了。

import pandas as pd
import yfinance as yf  # 免费股票数据源

# 获取茅台近 3 年数据
data = yf.download('600519.SS', start='2021-01-01', end='2024-01-01')

# 基础特征工程
data['5 日均线'] = data['Close'].rolling(5).mean()
data['20 日均线'] = data['Close'].rolling(20).mean()
data['金叉'] = (data['5 日均线'] > data['20 日均线']).astype(int)

def trend_strategy(df):
    signals = []
    position = 0  # 0 空仓 1 持仓

    for i in range(1, len(df)):
        # 金叉且空仓时买入
        if df['金叉'].iloc[i] and position == 0:
            signals.append(1)
            position = 1
        # 死叉且持仓时卖出
        elif df['金叉'].iloc[i] == 0 and position == 1:
            signals.append(-1)
            position = 0
        else:
            signals.append(0)

    return pd.Series(signals, index=df.index[1:])

class Backtester:
    def __init__(self, data, signals):
        self.data = data.iloc[1:]  # 对齐信号
        self.signals = signals

    def run(self, initial_capital=100000):
        positions = self.signals.cumsum()
        portfolio = initial_capital + (positions * self.data['Close']).cumsum()
        return portfolio

# 使用示例
signals = trend_strategy(data)
tester = Backtester(data, signals)
equity_curve = tester.run()

向量化操作：用 pandas 内置函数替代循环
数据分块：大数据集按时间切片处理
缓存机制：重复使用的中间结果存为 pickle

实测案例：把特征计算的 for 循环改为df.rolling().apply()，2000 行数据处理时间从 3.2 秒降到 0.15 秒。

建议在回测中设置 2 -3‰的滑点参数，比如：

# 在 Backtester 类中添加
self.slippage = 0.003  # 0.3% 滑点

def adjusted_price(self, signal_price):
    return signal_price * (1 + self.slippage) if signal > 0 \
           else signal_price * (1 - self.slippage)

用 Walk-Forward 分析：

把数据分成 5 段
用前 4 段训练，最后 1 段验证
滚动向前重复

如果验证段收益暴跌，说明策略可能过拟合。

# 环境准备
!pip install yfinance pandas

import pandas as pd
import yfinance as yf
import matplotlib.pyplot as plt

# 数据获取与预处理
def get_data(ticker):
    raw = yf.download(ticker)
    df = raw[['Close']].copy()
    df['5 日均线'] = df['Close'].rolling(5).mean()
    df['20 日均线'] = df['Close'].rolling(20).mean()
    df['金叉'] = (df['5 日均线'] > df['20 日均线']).astype(int)
    return df.dropna()

# 策略实现
class TrendStrategy:
    @staticmethod
    def generate_signals(df):
        signals = [0]
        position = 0

        for i in range(1, len(df)):
            current_signal = 0

            if df['金叉'].iloc[i] and position == 0:
                current_signal = 1
                position = 1
            elif df['金叉'].iloc[i] == 0 and position == 1:
                current_signal = -1
                position = 0

            signals.append(current_signal)

        return pd.Series(signals, index=df.index)

# 回测引擎
class BacktestEngine:
    def __init__(self, df, signals):
        self.df = df.copy()
        self.signals = signals

    def run(self, initial_capital=100000):
        self.df['持仓'] = self.signals.cumsum()
        self.df['市值'] = initial_capital + (self.df['持仓'] * self.df['Close']).cumsum()
        return self.df

# 主流程
if __name__ == '__main__':
    data = get_data('600519.SS')
    signals = TrendStrategy.generate_signals(data)
    engine = BacktestEngine(data, signals)
    result = engine.run()

    # 可视化
    result[['Close', '市值']].plot(secondary_y='市值')
    plt.show()