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

1次阅读

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

相比于成熟市场，A 股量化交易有几个关键限制需要特别注意：

T+ 1 交易制度：当日买入的股票需隔日才能卖出，直接影响日内策略设计
涨跌停限制：普通股票±10%、ST 股±5% 的波动限制，需在回测中模拟
流动性差异：小盘股的买卖价差较大，需考虑冲击成本
政策敏感度：A 股受政策影响显著，需在因子中纳入宏观指标

免费版限制：500 积分 / 日，单次最多提取 5000 条
优势：数据质量稳定，包含财务因子

典型调用：

import tushare as ts
pro = ts.pro_api('your_token')
df = pro.daily(ts_code='600519.SH', start_date='20200101', end_date='20201231')

完全免费：无调用次数限制
更新及时：包含北向资金等特色数据

示例代码：

import akshare as ak
df = ak.stock_zh_a_daily(symbol="sh600519", adjust="hfq")

计算 20 日波动率因子，包含异常值处理：

import pandas as pd
import numpy as np

def calc_volatility(df, window=20):
    """df 需包含 ['close'] 列
    返回: 添加了 volatility 因子的 DataFrame
    """returns = np.log(df['close'] / df['close'].shift(1))

    # 处理首行 NaN 和极端值
    returns = returns.replace([np.inf, -np.inf], np.nan).fillna(0)

    # 滚动计算标准差，年化处理
    df['volatility'] = returns.rolling(window).std() * np.sqrt(252)
    return df

Backtrader 框架示例，含佣金设置：

import backtrader as bt

class MultiFactorStrategy(bt.Strategy):
    params = (('factor1_weight', 0.6),
        ('factor2_weight', 0.4),
    )

    def __init__(self):
        self.factor1 = ...  # 因子 1 计算逻辑
        self.factor2 = ...  # 因子 2 计算逻辑

    def next(self):
        # 综合因子计算
        combined_score = (self.p.factor1_weight * self.factor1 +
                          self.p.factor2_weight * self.factor2)

        # 交易逻辑
        if combined_score > self.threshold:
            self.order_target_percent(target=0.9)
        elif combined_score < -self.threshold:
            self.order_target_percent(target=0.1)

# 佣金设置 (万 2.5 + 印花税)
cerebro.broker.setcommission(commission=0.00025)  # 佣金
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="ta")  # 印花税通过分析器处理

三种检测方法：

时间戳验证：确保所有数据点仅使用当前及之前的信息
滚动回测：采用 walk-forward 方式逐步验证
数据滞后：关键指标延迟 1 个周期使用

方法	实现复杂度	真实度
固定比例	低	中
随机区间	中	高
盘口重构	高	极高

示例代码（固定比例）：

class Slippage(bt.Slippage):
    def _getslippage(self, size, price, execprice):
        return execprice * 1.0003  # 固定 0.03% 滑点

测试案例：计算 1000 只股票 250 天的移动平均

方法	执行时间(ms)
pandas 滚动计算	42
numpy 向量化	38
Python 循环	2100

ma_calculator.pyx文件：

# cython: language_level=3
import numpy as np
cimport numpy as np

def rolling_mean(np.ndarray[double] arr, int window):
    cdef int n = arr.shape[0]
    cdef np.ndarray[double] out = np.empty(n)
    cdef double current_sum = 0.0

    for i in range(n):
        current_sum += arr[i]
        if i >= window:
            current_sum -= arr[i-window]
        out[i] = current_sum / min(i+1, window)

    return out