A股量化交易系统开发入门指南：从零搭建高可靠策略引擎

1次阅读

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

A 股市场与海外市场存在显著差异：T+ 1 交易制度限制当日买入的股票无法卖出，10% 的涨跌停板机制导致价格波动受限，且交易所对 API 调用有严格的频率限制（如每秒最多 3 笔订单）。这些特性要求量化系统必须具备特殊的容错设计和延迟敏感性。

优点：开发速度快，无需处理底层协议
缺点：功能受限（如不支持 Level2 行情），无法自定义订单路由逻辑
典型场景 ：低频策略验证阶段

优点：内置 CTP 接口封装，支持快速实盘对接
缺点：事件驱动架构学习曲线陡峭，策略隔离性差
典型场景 ：中小型私募的标准化策略部署

行情网关 ：负责多源数据聚合与标准化
策略容器 ：隔离策略运行环境
风控总线 ：实时监控保证金占用
交易路由 ：智能拆单与最优执行

import asyncio
from websockets import connect

class MarketFeed:
    def __init__(self, uri):
        self._uri = uri
        self._ws = None

    async def _reconnect(self):
        while True:
            try:
                self._ws = await connect(self._uri)
                await self._subscribe()
                return
            except Exception as e:
                print(f"Connection failed: {e}, retrying...")
                await asyncio.sleep(5)

    async def _subscribe(self):
        await self._ws.send('{"action":"subscribe","codes":["600519.SH"]}')

    async def run(self):
        await self._reconnect()
        async for msg in self._ws:
            print(f"Tick: {msg}")  # 实际处理应放入异步队列

class DualMAStrategy:
    def __init__(self, fast_period=5, slow_period=20):
        self.fast_ma = []
        self.slow_ma = []
        self.position = 0  # 当前持仓
        self.max_pos = 0.3  # 单策略最大仓位

    def on_tick(self, price):
        self.fast_ma.append(price)
        self.slow_ma.append(price)

        if len(self.fast_ma) > fast_period:
            self.fast_ma.pop(0)
        if len(self.slow_ma) > slow_period:
            self.slow_ma.pop(0)

        # 交易信号生成
        if len(self.fast_ma) == fast_period and len(self.slow_ma) == slow_period:
            fast_val = sum(self.fast_ma)/fast_period
            slow_val = sum(self.slow_ma)/slow_period

            if fast_val > slow_val and self.position < self.max_pos:
                return "BUY"  # 实际应返回订单对象
            elif fast_val < slow_val and self.position > 0:
                return "SELL"

import pandas as pd

def backtest(df: pd.DataFrame):
    # 计算收益率
    df['return'] = df['close'].pct_change()

    # 生成信号 (例：简单突破策略)
    df['signal'] = (df['close'] > df['high'].rolling(20).mean()).astype(int)

    # 计算策略收益 (未考虑滑点)
    df['strategy_return'] = df['signal'].shift(1) * df['return']

    # 年化波动率计算
    ann_vol = df['strategy_return'].std() * np.sqrt(240)