Python实战：A股量化交易入门指南与避坑手册

1次阅读

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

A 股市场与海外市场相比有诸多独特规则，直接影响策略设计：

T+ 1 交易制度：当日买入的股票需次日才能卖出，策略中需特别处理持仓周期
涨跌停限制：普通股票±10%（ST 股±5%），需在订单执行逻辑中加入价格校验
手续费计算：包含印花税（卖出 0.1%）+ 佣金（通常 0.025% 起）+ 过户费（0.001%）
交易时段：09:15-11:30/13:00-15:00（集合竞价阶段规则不同）

这些规则要求我们在回测时需定制交易引擎，例如下面这个涨跌停价格计算函数：

def limit_price(prev_close: float, is_st: bool = False) -> tuple[float, float]:
    """计算当日涨跌停价格"""
    ratio = 0.05 if is_st else 0.1
    return (round(prev_close * (1 - ratio), 2),  # 跌停价
        round(prev_close * (1 + ratio), 2)   # 涨停价
    )

工具	免费额度	数据质量	更新频率	特色
Tushare Pro	500 次 / 日	较好	日级	社区活跃
AKShare	无限制	一般	实时	支持期货外汇
Wind	付费	优秀	实时	机构级数据

import tushare as ts

# 初始化接口（需先注册获取 token）pro = ts.pro_api('your_token_here')

# 获取沪深 300 成分股
df = pro.index_weight(
    index_code='000300.SH', 
    start_date='20230101',
    end_date='20231231'
)

# 获取个股日线数据
daily = pro.daily(
    ts_code='600519.SH', 
    start_date='20230101',
    end_date='20230401'
)

注意事项：
– 免费版有调用频率限制，建议缓存数据到本地数据库
– 复权处理推荐使用 adj_factor 字段自行计算

from typing import List
import backtrader as bt
import pandas as pd

class MA_Cross(bt.Strategy):
    params = (('fast_ma', 5),   # 快速均线周期
        ('slow_ma', 20),  # 慢速均线周期
        ('printlog', False)
    )

    def __init__(self):
        # 初始化指标计算
        self.fast_ma = bt.indicators.SMA(
            self.data.close, 
            period=self.p.fast_ma
        )
        self.slow_ma = bt.indicators.SMA(
            self.data.close, 
            period=self.p.slow_ma
        )
        self.crossover = bt.indicators.CrossOver(
            self.fast_ma, 
            self.slow_ma
        )

    def next(self):
        if not self.position:  # 没有持仓
            if self.crossover > 0:  # 金叉信号
                cash = self.broker.getcash()
                size = int(cash * 0.9 / self.data.close[0] / 100) * 100  # 按 90% 资金计算手数
                self.buy(size=size)
        elif self.crossover < 0:  # 死叉信号
            self.close()

    def log(self, txt: str, dt=None):
        dt = dt or self.datas[0].datetime.date(0)
        print(f'{dt.isoformat()}, {txt}')

    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            return
        if order.status == order.Completed:
            if order.isbuy():
                self.log(f'买入执行 {order.executed.price:.2f}')
            elif order.issell():
                self.log(f'卖出执行 {order.executed.price:.2f}')

# 创建回测引擎
cerebro = bt.Cerebro()

# 加载数据
data = bt.feeds.PandasData(
    dataname=df,
    datetime='trade_date',
    open='open',
    high='high',
    low='low',
    close='close',
    volume='vol',
    openinterest=-1
)
cerebro.adddata(data)

# 添加策略
cerebro.addstrategy(MA_Cross)

# 设置初始资金
cerebro.broker.setcash(100000.0)

# 添加分析器
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')

# 运行回测
results = cerebro.run()
strat = results[0]

# 打印结果
print('夏普比率:', strat.analyzers.sharpe.get_analysis())
print('最大回撤:', strat.analyzers.drawdown.get_analysis())

# 绘制结果
cerebro.plot(style='candlestick')

典型场景：
– 使用未来数据（如当日收盘价）计算信号
– 在 next() 中引用 self.data.close[1] 却用 [0] 做交易

解决方法：
– 严格使用 [0] 表示当前 bar，[-1]表示前一个 bar
– 回测时添加 cheat_on_open=True 参数检查

# 在回测引擎中添加滑点模型
cerebro.broker.set_slippage_perc(0.001)  # 0.1% 的滑点

# 设置佣金和印花税
cerebro.broker.setcommission(
    commission=0.00025,  # 佣金 0.025%
    margin=0,           # 现货交易无杠杆
    mult=1,             
    stamp_duty=0.001    # 卖出时收取 0.1%
)

单笔风险：单笔亏损不超过总资金的 2%
总风险：总亏损达到 10% 立即停止交易
持仓分散：单个标的仓位不超过 20%

class RiskControl(bt.Strategy):
    params = (('stop_loss', 0.05),  # 5% 止损
        ('take_profit', 0.1)  # 10% 止盈
    )

    def notify_order(self, order):
        if order.status == order.Completed:
            if order.isbuy():
                self.stop_price = order.executed.price * (1 - self.p.stop_loss)
                self.target_price = order.executed.price * (1 + self.p.take_profit)

    def next(self):
        if self.position:
            if self.data.close[0] <= self.stop_price:
                self.close()
            elif self.data.close[0] >= self.target_price:
                self.close()

重要提醒：
– 个人开发者无法直接接入券商交易接口（需通过合规渠道）
– 推荐使用仿真交易平台测试：
– 掘金量化（MyQuant）
– 同花顺仿真交易
– 券商提供的模拟系统

性能优化技巧：
– 使用 Pandas 向量化操作替代循环
– 预计算指标数据
– 使用 numexpr 加速计算

# 向量化计算示例
def vectorized_backtest(df: pd.DataFrame):
    df['fast_ma'] = df['close'].rolling(5).mean()
    df['slow_ma'] = df['close'].rolling(20).mean()
    df['signal'] = np.where(df['fast_ma'] > df['slow_ma'], 1, 0)
    df['position'] = df['signal'].diff()
    return df