A股量化实盘入门指南：从零搭建你的第一个策略交易系统

1次阅读

没有评论

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

A 股量化交易相比国际市场有其独特特点：

T+ 1 交易制度限制日内交易频率
涨跌停板机制影响流动性
市场参与者结构特殊（散户占比高）
政策因素影响显著

对新手而言，主要面临三大难点：

数据获取：需要稳定可靠的行情和基本面数据源
回测陷阱：容易忽略滑点、手续费等实际交易成本
实盘对接：券商接口复杂且风控要求严格

主流 Python 量化框架对比：

vn.py
优势：专注实盘交易，支持多家券商接口
不足：学习曲线较陡，文档不够完善
Backtrader
优势：回测功能强大，社区资源丰富
不足：实盘支持较弱，性能一般
PyAlgoTrade
优势：简单易用，适合教学
不足：功能较少，已停止维护

推荐选择：vn.py（实盘）+ Backtrader（回测）组合方案

import tushare as ts

# 设置 token（需先在 Tushare 官网注册）ts.set_token('你的 token')
pro = ts.pro_api()

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

# 数据预处理
df = df.sort_values('trade_date')
df['trade_date'] = pd.to_datetime(df['trade_date'])
df.set_index('trade_date', inplace=True)

import numpy as np
import pandas as pd

class DualMAStrategy:
    def __init__(self, short_window=5, long_window=20):
        self.short_window = short_window
        self.long_window = long_window

    def generate_signals(self, data):
        signals = pd.DataFrame(index=data.index)
        signals['price'] = data['close']
        signals['short_ma'] = data['close'].rolling(self.short_window).mean()
        signals['long_ma'] = data['close'].rolling(self.long_window).mean()

        # 产生交易信号
        signals['signal'] = 0
        signals['signal'][self.short_window:] = np.where(signals['short_ma'][self.short_window:] > signals['long_ma'][self.short_window:], 
            1, 0)

        # 信号差分得到实际买卖点
        signals['positions'] = signals['signal'].diff()
        return signals

from vnpy.app.cta_strategy import CtaTemplate
from vnpy.trader.constant import Interval

class DualMAStrategy(CtaTemplate):
    """实盘双均线策略"""

    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
        super().__init__(cta_engine, strategy_name, vt_symbol, setting)

        # 参数设置
        self.short_window = 5
        self.long_window = 20

        # 变量初始化
        self.bg = BarGenerator(self.on_bar)
        self.am = ArrayManager()

    def on_tick(self, tick: TickData):
        """行情推送处理"""
        self.bg.update_tick(tick)

    def on_bar(self, bar: BarData):
        """K 线处理"""
        self.am.update_bar(bar)
        if not self.am.inited:
            return

        # 计算均线
        short_ma = self.am.sma(self.short_window)
        long_ma = self.am.sma(self.long_window)

        # 交易逻辑
        if short_ma > long_ma and self.pos == 0:
            self.buy(bar.close_price, 1)
        elif short_ma < long_ma and self.pos > 0:
            self.sell(bar.close_price, abs(self.pos))