AI量化交易A股入门指南：从数据获取到策略回测全流程实战

1次阅读

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

T+ 1 交易制度 ：A 股实行 T + 1 制度，即当日买入的股票需次日才能卖出，这限制了日内高频策略的实施。
涨跌停限制 ：A 股个股单日涨跌幅通常限制在 10%（ST 股 5%），导致极端行情下流动性枯竭。
政策市特征 ：宏观政策、产业规划等非市场因素对 A 股影响显著，需在模型中纳入事件驱动因子。

Tushare Pro：适合个人开发者，基础数据免费但需积分兑换，分钟级数据需付费。接口稳定，社区支持好。
AkShare：完全开源免费，但接口偶有不稳定，适合作为备用数据源。
WindAPI：机构级数据，年费数万元起，数据覆盖最全但成本高。

import pandas as pd

# 读取 CSV 数据（含停牌标识）df = pd.read_csv('stock_data.csv')

# 处理停牌日：前复权价格填充
df['close'] = df['close'].fillna(method='ffill')
df['volume'] = df['volume'].fillna(0)  # 停牌日成交量为 0

# 计算对数收益率
df['log_return'] = np.log(df['close'] / df['close'].shift(1))

import talib

# 计算 MACD 指标
df['macd'], df['signal'], _ = talib.MACD(df['close'],
    fastperiod=12,
    slowperiod=26,
    signalperiod=9
)

# 布林带计算
df['upper'], df['middle'], df['lower'] = talib.BBANDS(df['close'],
    timeperiod=20,
    nbdevup=2,
    nbdevdn=2
)

import torch
import torch.nn as nn

class LSTMModel(nn.Module):
    def __init__(self, input_dim=10, hidden_dim=64):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size=input_dim,
            hidden_size=hidden_dim,
            batch_first=True
        )
        self.fc = nn.Linear(hidden_dim, 1)  # 输出涨跌概率

    def forward(self, x):
        out, _ = self.lstm(x)
        return torch.sigmoid(self.fc(out[:, -1]))

# 委托价格不超过涨停价
current_limit_up = df['close'].iloc[-1] * 1.1  # 计算涨停价
order_price = min(intend_price, current_limit_up)

在财报发布前后 20 个交易日禁用基本面因子
引入事件因子标记财报公告日

# 按成交量的 0.1% 模拟滑点
executed_price = original_price * (1 + 0.001 * 
    (trade_volume / df['volume'].rolling(5).mean().iloc[-1]))

import backtrader as bt

class MyStrategy(bt.Strategy):
    params = (('ma_period', 20),)

    def __init__(self):
        self.sma = bt.indicators.SMA(
            self.data.close, 
            period=self.p.ma_period
        )

    def next(self):
        if self.data.close[0] > self.sma[0]:
            self.buy()
        elif self.data.close[0] < self.sma[0]:
            self.sell()

# 回测结果分析
print('夏普比率:', results[0].analyzers.sharpe.get_analysis()['sharperatio'])
print('最大回撤:', results[0].analyzers.drawdown.get_analysis()['max']['drawdown'])