从零搭建自己的A股量化交易系统：一个Python新手的实战记录

“本文记录了一个Python小白从零开始，一步步搭建A股量化交易系统的完整过程。从Tushare数据获取、双均线策略编写，到Backtrader回测和PyQt5图形界面开发，所有代码均可直接运行。希望能帮到同样想入门量化的朋友。

一、前言

作为一个Python新手，我一直对量化交易感兴趣。经过一个月的摸索，我终于搭建了一套能实盘运行的双均线策略系统。本文将完整记录从环境搭建、数据接入到策略回测的全过程。

二、环境准备

• Python 3.9安装与配置
• 解决C++编译工具问题（新手最容易卡住的地方）
• 安装Tushare、pandas等核心库
  ┌─────────────────┐
  │ 数据层 │
  │ Tushare获取数据 │
  └────────┬────────┘
  ↓
  ┌─────────────────┐
  │ 策略层 │
  │ 双均线策略实现 │
  └────────┬────────┘
  ↓
  ┌─────────────────┐
  │ 回测层 │
  │ Backtrader回测 │
  └────────┬────────┘
  ↓
  ┌─────────────────┐
  │ 可视化层 │
  │ PyQt5图形界面 │
  └─────────────────┘

三、数据获取（Tushare Pro实战）

• 注册账号与积分获取（注册100分+修改资料20分）
• 日线行情接口调用示例
• 核心股票池构建（30只主流股覆盖主要板块）

代码示例：

import tushare as ts
pro = ts.pro_api('你的token')
df = pro.daily(ts_code='600519.SH', start_date='20260101', end_date='20260227')
print(df.head())
# 安装Tushare
!pip install tushare

import tushare as ts
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta

# 设置Tushare token（注册后获取）
# 我的Tushare ID：983160
ts.set_token('你的token')
pro = ts.pro_api()
def get_stock_data(ts_code='600519.SH', start_date='20240101', end_date='20241231'):
    """
    获取股票日线数据
    :param ts_code: 股票代码，默认贵州茅台
    :param start_date: 开始日期
    :param end_date: 结束日期
    :return: DataFrame
    """
    # 获取日线数据
    df = pro.daily(ts_code=ts_code, 
                   start_date=start_date, 
                   end_date=end_date)
    
    # 按交易日期排序
    df = df.sort_values('trade_date')
    
    # 将trade_date转为datetime类型
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    
    # 设置日期为索引
    df.set_index('trade_date', inplace=True)
    
    # 添加常用技术指标
    df['ma5'] = df['close'].rolling(window=5).mean()
    df['ma10'] = df['close'].rolling(window=10).mean()
    df['ma20'] = df['close'].rolling(window=20).mean()
    df['ma60'] = df['close'].rolling(window=60).mean()
    
    # 计算涨跌幅
    df['pct_change'] = df['close'].pct_change() * 100
    
    # 计算成交量均线
    df['vol_ma5'] = df['vol'].rolling(window=5).mean()
    
    return df

# 获取贵州茅台数据
df_maotai = get_stock_data()
print(df_maotai.tail())
            open    high     low   close     vol  ...  ma20  ma60  pct_change  vol_ma5
trade_date                                         ...                               
2024-12-20  1650.0  1675.0  1645.0  1668.0  32145  ...  1650  1620        1.52    30123
2024-12-23  1670.0  1685.0  1665.0  1680.0  35678  ...  1655  1625        0.72    31234
2024-12-24  1682.0  1700.0  1678.0  1695.0  40123  ...  1660  1630        0.89    33456
2024-12-25  1695.0  1710.0  1690.0  1705.0  38901  ...  1665  1635        0.59    34567
2024-12-26  1705.0  1725.0  1700.0  1720.0  42345  ...  1670  1640        0.88    35678
# 定义股票池
stock_pool = {
    '600519.SH': '贵州茅台',
    '000858.SZ': '五粮液',
    '000333.SZ': '美的集团',
    '600036.SH': '招商银行',
    '300750.SZ': '宁德时代'
}

def get_multi_stocks(stock_dict, start_date='20240101', end_date='20241231'):
    """
    获取多只股票数据
    """
    data_dict = {}
    for ts_code, name in stock_dict.items():
        try:
            df = get_stock_data(ts_code, start_date, end_date)
            data_dict[name] = df
            print(f"✅ 成功获取 {name} 数据，共 {len(df)} 条记录")
        except Exception as e:
            print(f"❌ 获取 {name} 失败：{e}")
    
    return data_dict

# 获取股票池数据
stock_data = get_multi_stocks(stock_pool)
四、策略实现模块（双均线策略）
4.1 策略核心逻辑
def generate_signals(df, short_window=5, long_window=20):
    """
    生成交易信号
    策略：短期均线上穿长期均线买入，下穿卖出
    """
    # 创建信号DataFrame
    signals = pd.DataFrame(index=df.index)
    signals['signal'] = 0.0
    
    # 计算短期和长期均线
    signals['short_ma'] = df['close'].rolling(window=short_window, min_periods=1).mean()
    signals['long_ma'] = df['close'].rolling(window=long_window, min_periods=1).mean()
    
    # 生成信号
    # 当短期均线 > 长期均线时，信号为1（买入/持有）
    # 当短期均线 < 长期均线时，信号为-1（卖出/空仓）
    signals['signal'][short_window:] = np.where(
        signals['short_ma'][short_window:] > signals['long_ma'][short_window:], 1.0, 0.0
    )
    
    # 生成交易指令（信号变化点）
    signals['position'] = signals['signal'].diff()
    
    return signals

# 贵州茅台信号生成
signals_maotai = generate_signals(df_maotai)
print(signals_maotai[signals_maotai['position'] != 0].head(10))
def plot_signals(df, signals, stock_name='贵州茅台'):
    """
    绘制价格曲线和交易信号
    """
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(15, 10), 
                                    gridspec_kw={'height_ratios': [3, 1]})
    
    # 绘制价格和均线
    ax1.plot(df.index, df['close'], label='收盘价', color='black', alpha=0.7)
    ax1.plot(signals.index, signals['short_ma'], label='5日均线', color='blue', linestyle='--')
    ax1.plot(signals.index, signals['long_ma'], label='20日均线', color='red', linestyle='--')
    
    # 标记买入点
    buy_signals = signals[signals['position'] == 1]
    ax1.plot(buy_signals.index, 
             df.loc[buy_signals.index]['close'], 
             '^', markersize=12, color='green', label='买入信号')
    
    # 标记卖出点
    sell_signals = signals[signals['position'] == -1]
    ax1.plot(sell_signals.index, 
             df.loc[sell_signals.index]['close'], 
             'v', markersize=12, color='red', label='卖出信号')
    
    ax1.set_title(f'{stock_name} - 双均线策略交易信号')
    ax1.set_ylabel('价格')
    ax1.legend()
    ax1.grid(True, alpha=0.3)
    
    # 绘制持仓信号
    ax2.fill_between(signals.index, 0, signals['signal'], 
                     color='green', alpha=0.3, label='持仓区域')
    ax2.set_ylabel('持仓信号')
    ax2.set_xlabel('日期')
    ax2.set_ylim(-0.1, 1.1)
    ax2.legend()
    ax2.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()

# 绘制信号图
plot_signals(df_maotai, signals_maotai)
五、回测验证模块
5.1 手动回测函数
def backtest_strategy(df, signals, initial_capital=100000.0):
    """
    回测策略表现
    """
    # 创建回测DataFrame
    portfolio = pd.DataFrame(index=signals.index)
    portfolio['holdings'] = 0.0  # 持仓市值
    portfolio['cash'] = 0.0      # 现金
    portfolio['total'] = 0.0     # 总资产
    portfolio['returns'] = 0.0    # 收益率
    
    # 初始化
    cash = initial_capital
    position = 0  # 0表示空仓，1表示持仓
    buy_price = 0
    
    # 逐日回测
    for i, date in enumerate(portfolio.index):
        price = df.loc[date, 'close']
        
        # 处理交易信号
        if i > 0:
            signal = signals.loc[date, 'position']
            
            # 买入信号且空仓
            if signal == 1 and position == 0:
                position = 1
                buy_price = price
                shares = cash // price
                cash -= shares * price
                print(f"📈 买入：{date.date()}，价格：{price:.2f}，股数：{shares}")
            
            # 卖出信号且持仓
            elif signal == -1 and position == 1:
                position = 0
                cash += shares * price
                profit = (price - buy_price) * shares
                print(f"📉 卖出：{date.date()}，价格：{price:.2f}，盈利：{profit:.2f}")
        
        # 计算当前持仓市值
        if position == 1:
            holdings = shares * price
        else:
            holdings = 0
        
        total = cash + holdings
        
        # 记录数据
        portfolio.loc[date, 'holdings'] = holdings
        portfolio.loc[date, 'cash'] = cash
        portfolio.loc[date, 'total'] = total
        portfolio.loc[date, 'returns'] = (total / initial_capital - 1) * 100
    
    return portfolio

# 执行回测
portfolio = backtest_strategy(df_maotai, signals_maotai)

# 回测结果统计
final_return = portfolio['returns'].iloc[-1]
print(f"\n📊 回测结果（贵州茅台）")
print(f"初始资金：100,000元")
print(f"最终资金：{portfolio['total'].iloc[-1]:.2f}元")
print(f"总收益率：{final_return:.2f}%")

# 计算最大回撤
rolling_max = portfolio['total'].expanding().max()
drawdown = (portfolio['total'] - rolling_max) / rolling_max * 100
max_drawdown = drawdown.min()
print(f"最大回撤：{max_drawdown:.2f}%")
# 安装backtrader
!pip install backtrader

import backtrader as bt
import backtrader.analyzers as btanalyzers

class DualMAStrategy(bt.Strategy):
    """双均线策略"""
    
    params = (
        ('short_period', 5),
        ('long_period', 20),
    )
    
    def __init__(self):
        # 计算均线
        self.short_ma = bt.indicators.SMA(self.data.close, period=self.params.short_period)
        self.long_ma = bt.indicators.SMA(self.data.close, period=self.params.long_period)
        
        # 金叉死叉信号
        self.crossover = bt.indicators.CrossOver(self.short_ma, self.long_ma)
        
        # 记录订单
        self.order = None
        
    def next(self):
        # 如果有未完成订单，跳过
        if self.order:
            return
            
        # 没有持仓且金叉
        if not self.position:
            if self.crossover > 0:
                self.order = self.buy()
                
        # 有持仓且死叉
        elif self.crossover < 0:
            self.order = self.sell()
    
    def notify_order(self, order):
        if order.status in [order.Completed]:
            if order.isbuy():
                print(f'买入：{self.data.datetime.date(0)}，价格：{order.executed.price:.2f}')
            else:
                print(f'卖出：{self.data.datetime.date(0)}，价格：{order.executed.price:.2f}')
            
            self.order = None

# 执行Backtrader回测
def run_backtrader(df, stock_name='贵州茅台'):
    """
    使用Backtrader进行回测
    """
    # 创建cerebro引擎
    cerebro = bt.Cerebro()
    
    # 添加策略
    cerebro.addstrategy(DualMAStrategy)
    
    # 准备数据
    data = bt.feeds.PandasData(
        dataname=df[['open', 'high', 'low', 'close', 'vol']],
        datetime=None,  # 使用index作为日期
        open='open',
        high='high',
        low='low',
        close='close',
        volume='vol'
    )
    
    # 添加数据
    cerebro.adddata(data)
    
    # 设置初始资金
    cerebro.broker.setcash(100000.0)
    
    # 设置手续费
    cerebro.broker.setcommission(commission=0.0005)  # 万五
    
    # 添加分析器
    cerebro.addanalyzer(btanalyzers.SharpeRatio, _name='sharpe')
    cerebro.addanalyzer(btanalyzers.Returns, _name='returns')
    cerebro.addanalyzer(btanalyzers.DrawDown, _name='drawdown')
    
    # 运行回测
    print(f'初始资金：{cerebro.broker.getvalue():.2f}')
    results = cerebro.run()
    print(f'最终资金：{cerebro.broker.getvalue():.2f}')
    
    # 获取分析结果
    strat = results[0]
    sharpe = strat.analyzers.sharpe.get_analysis()
    returns = strat.analyzers.returns.get_analysis()
    drawdown = strat.analyzers.drawdown.get_analysis()
    
    print(f"\n📊 {stock_name} 策略表现：")
    print(f"夏普比率：{sharpe.get('sharperatio', 0):.2f}")
    print(f"年化收益率：{returns.get('rnorm100', 0):.2f}%")
    print(f"最大回撤：{drawdown.get('max', {}).get('drawdown', 0):.2f}%")
    
    # 绘制结果
    cerebro.plot(style='candlestick')
    
    return results

# 运行专业回测
results = run_backtrader(df_maotai)
# 安装PyQt5
!pip install PyQt5

import sys
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
from PyQt5.QtGui import *
import matplotlib
matplotlib.use('Qt5Agg')
from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas
from matplotlib.figure import Figure

class QuantTradingGUI(QMainWindow):
    """量化交易图形界面"""
    
    def __init__(self):
        super().__init__()
        self.initUI()
        self.data = None
        self.signals = None
        
    def initUI(self):
        """初始化界面"""
        self.setWindowTitle('A股量化交易系统 v1.0')
        self.setGeometry(100, 100, 1200, 800)
        
        # 创建中央控件
        central_widget = QWidget()
        self.setCentralWidget(central_widget)
        
        # 创建主布局
        main_layout = QHBoxLayout()
        central_widget.setLayout(main_layout)
        
        # ===== 左侧控制面板 =====
        left_panel = QWidget()
        left_layout = QVBoxLayout()
        left_panel.setLayout(left_layout)
        left_panel.setMaximumWidth(300)
        
        # 标题
        title = QLabel('量化策略回测系统')
        title.setStyleSheet('font-size: 16px; font-weight: bold; padding: 10px;')
        title.setAlignment(Qt.AlignCenter)
        left_layout.addWidget(title)
        
        # 股票选择
        stock_group = QGroupBox('股票选择')
        stock_layout = QFormLayout()
        
        self.stock_combo = QComboBox()
        stocks = ['600519.SH (贵州茅台)', '000858.SZ (五粮液)', 
                  '000333.SZ (美的集团)', '600036.SH (招商银行)',
                  '300750.SZ (宁德时代)']
        self.stock_combo.addItems(stocks)
        stock_layout.addRow('股票代码：', self.stock_combo)
        
        stock_group.setLayout(stock_layout)
        left_layout.addWidget(stock_group)
        
        # 日期范围
        date_group = QGroupBox('日期范围')
        date_layout = QFormLayout()
        
        self.start_date = QDateEdit()
        self.start_date.setDate(QDate(2024, 1, 1))
        self.start_date.setCalendarPopup(True)
        date_layout.addRow('开始日期：', self.start_date)
        
        self.end_date = QDateEdit()
        self.end_date.setDate(QDate(2024, 12, 31))
        self.end_date.setCalendarPopup(True)
        date_layout.addRow('结束日期：', self.end_date)
        
        date_group.setLayout(date_layout)
        left_layout.addWidget(date_group)
        
        # 策略参数
        param_group = QGroupBox('策略参数')
        param_layout = QFormLayout()
        
        self.short_ma = QSpinBox()
        self.short_ma.setRange(3, 30)
        self.short_ma.setValue(5)
        param_layout.addRow('短期均线：', self.short_ma)
        
        self.long_ma = QSpinBox()
        self.long_ma.setRange(10, 60)
        self.long_ma.setValue(20)
        param_layout.addRow('长期均线：', self.long_ma)
        
        self.init_capital = QDoubleSpinBox()
        self.init_capital.setRange(10000, 10000000)
        self.init_capital.setValue(100000)
        self.init_capital.setSingleStep(10000)
        param_layout.addRow('初始资金：', self.init_capital)
        
        param_group.setLayout(param_layout)
        left_layout.addWidget(param_group)
        
        # 按钮
        self.load_btn = QPushButton('1. 加载数据')
        self.load_btn.clicked.connect(self.load_data)
        left_layout.addWidget(self.load_btn)
        
        self.run_btn = QPushButton('2. 运行回测')
        self.run_btn.clicked.connect(self.run_backtest)
        self.run_btn.setEnabled(False)
        left_layout.addWidget(self.run_btn)
        
        # 结果显示
        result_group = QGroupBox('回测结果')
        result_layout = QVBoxLayout()
        
        self.result_text = QTextEdit()
        self.result_text.setReadOnly(True)
        self.result_text.setMaximumHeight(150)
        result_layout.addWidget(self.result_text)
        
        result_group.setLayout(result_layout)
        left_layout.addWidget(result_group)
        
        # 退出按钮
        exit_btn = QPushButton('退出系统')
        exit_btn.clicked.connect(self.close)
        left_layout.addWidget(exit_btn)
        
        left_layout.addStretch()
        
        # ===== 右侧图表面板 =====
        right_panel = QWidget()
        right_layout = QVBoxLayout()
        right_panel.setLayout(right_layout)
        
        # 创建matplotlib画布
        self.figure = Figure(figsize=(8, 6))
        self.canvas = FigureCanvas(self.figure)
        right_layout.addWidget(self.canvas)
        
        # 添加状态栏
        self.statusBar().showMessage('准备就绪')
        
        # 添加到主布局
        main_layout.addWidget(left_panel)
        main_layout.addWidget(right_panel, 1)
    
    def load_data(self):
        """加载数据"""
        try:
            self.statusBar().showMessage('正在加载数据...')
            
            # 获取选择的股票
            stock_text = self.stock_combo.currentText()
            ts_code = stock_text.split(' ')[0]
            
            # 获取日期
            start = self.start_date.date().toString('yyyyMMdd')
            end = self.end_date.date().toString('yyyyMMdd')
            
            # 调用Tushare获取数据
            # 这里使用之前定义的get_stock_data函数
            self.data = get_stock_data(ts_code, start, end)
            
            self.statusBar().showMessage(f'✅ 数据加载完成，共{len(self.data)}条记录')
            self.run_btn.setEnabled(True)
            
            # 显示数据预览
            preview = f"数据预览：\n"
            preview += f"日期范围：{self.data.index[0].strftime('%Y-%m-%d')} 至 {self.data.index[-1].strftime('%Y-%m-%d')}\n"
            preview += f"最新收盘价：{self.data['close'].iloc[-1]:.2f}\n"
            preview += f"期间涨幅：{(self.data['close'].iloc[-1]/self.data['close'].iloc[0]-1)*100:.2f}%"
            
            self.result_text.setText(preview)
            
        except Exception as e:
            QMessageBox.critical(self, '错误', f'数据加载失败：{str(e)}')
    
    def run_backtest(self):
        """运行回测"""
        try:
            self.statusBar().showMessage('正在运行回测...')
            
            # 生成信号
            short = self.short_ma.value()
            long = self.long_ma.value()
            self.signals = generate_signals(self.data, short, long)
            
            # 执行回测
            capital = self.init_capital.value()
            portfolio = backtest_strategy(self.data, self.signals, capital)
            
            # 计算指标
            final_value = portfolio['total'].iloc[-1]
            total_return = (final_value / capital - 1) * 100
            
            # 计算年化收益率
            days = len(portfolio)
            years = days / 245  # A股年交易天数约245天
            annual_return = ((1 + total_return/100) ** (1/years) - 1) * 100
            
            # 计算夏普比率（简化版）
            daily_returns = portfolio['total'].pct_change().dropna()
            sharpe = np.sqrt(245) * daily_returns.mean() / daily_returns.std() if daily_returns.std() != 0 else 0
            
            # 显示结果
            result = f"📊 回测结果\n"
            result += f"{'='*30}\n"
            result += f"初始资金：{capital:,.2f}\n"
            result += f"最终资金：{final_value:,.2f}\n"
            result += f"总收益率：{total_return:.2f}%\n"
            result += f"年化收益率：{annual_return:.2f}%\n"
            result += f"夏普比率：{sharpe:.2f}\n"
            result += f"交易次数：{len(self.signals[self.signals['position']!=0])}\n"
            
            self.result_text.setText(result)
            
            # 绘制图表
            self.figure.clear()
            ax = self.figure.add_subplot(111)
            
            # 绘制价格和均线
            ax.plot(self.data.index, self.data['close'], label='收盘价', alpha=0.7)
            ax.plot(self.signals.index, self.signals['short_ma'], 
                   label=f'{short}日均线', linestyle='--')
            ax.plot(self.signals.index, self.signals['long_ma'], 
                   label=f'{long}日均线', linestyle='--')
            
            # 标记买卖点
            buy_signals = self.signals[self.signals['position'] == 1]
            sell_signals = self.signals[self.signals['position'] == -1]
            
            ax.scatter(buy_signals.index, 
                      self.data.loc[buy_signals.index, 'close'],
                      color='green', marker='^', s=100, label='买入', zorder=5)
            ax.scatter(sell_signals.index,
                      self.data.loc[sell_signals.index, 'close'],
                      color='red', marker='v', s=100, label='卖出', zorder=5)
            
            ax.set_title(f'双均线策略回测结果（{short}日/ {long}日）')
            ax.set_xlabel('日期')
            ax.set_ylabel('价格')
            ax.legend()
            ax.grid(True, alpha=0.3)
            
            self.figure.autofmt_xdate()
            self.canvas.draw()
            
            self.statusBar().showMessage('✅ 回测完成')
            
        except Exception as e:
            QMessageBox.critical(self, '错误', f'回测失败：{str(e)}')

# 运行GUI
if __name__ == '__main__':
    app = QApplication(sys.argv)
    gui = QuantTradingGUI()
    gui.show()
    sys.exit(app.exec_())
    七、实盘经验分享
    7.1 策略优化思路
经过几个月的实盘测试，我发现几个优化方向：
# 只在放量时确认信号
df['vol_ratio'] = df['vol'] / df['vol_ma5']
# 只有成交量放大1.2倍以上才确认信号
# 只在放量时确认信号
df['vol_ratio'] = df['vol'] / df['vol_ma5']
# 只有成交量放大1.2倍以上才确认信号
# 只在上升趋势中做多
df['trend'] = np.where(df['close'] > df['ma60'], 1, 0)
# 根据均线多头排列程度动态调整仓位
ma_alignment = (df['ma5'] > df['ma10']).astype(int) + \
               (df['ma10'] > df['ma20']).astype(int) + \
               (df['ma20'] > df['ma60']).astype(int)
position_size = ma_alignment / 3  # 0-100%仓位
7.2 踩过的坑
1. 未来函数：回测时不小心用了未来数据，导致回测收益率超高，实盘一塌糊涂
2. 过拟合：参数优化得太精细，换段时间就失效
3. 滑点成本：实盘时发现成交价总比信号价差一点，来回几次就把利润吃掉了
7.3 实盘建议
1. 先用模拟盘跑3个月
2. 每次只修改一个参数
3. 留足安全边际（按80%的仓位执行）
4. 做好资金管理（单票不超过20%）

八、总结与展望
这套系统我从零开始搭建，虽然还有很多不完善的地方，但已经能稳定运行了。回顾这一年的学习历程，最大的感悟是：
量化交易不是印钞机，而是一个帮你执行纪律的工具。
它不会让你一夜暴富，但能帮你：
• ✅ 摆脱情绪化交易
• ✅ 系统化复盘优化
• ✅ 节省盯盘时间
下一步计划：
1. 加入更多因子（动量、估值、情绪等）
2. 开发多策略组合
3. 接入实盘交易API

附：完整代码获取
本文所有代码都可以在Tushare社区下载。如果你也想开始量化学习，可以先注册Tushare获取数据：
• Tushare ID：983160
• 注册送100积分，发文章还能换更多积分
• 有问题欢迎在评论区交流