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

1次阅读

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

与海外市场相比，A 股量化交易存在几个关键差异点：

T+ 1 交易制度 ：当日买入的股票需次交易日才能卖出，影响高频策略设计
涨跌停限制 ：普通股票每日波动幅度为 10%（ST 股为 5%），需特别处理边界情况
交易时段分段 ：早盘集合竞价（9:15-9:25），连续竞价（9:30-11:30/13:00-15:00），收盘集合竞价（14:57-15:00）
手续费结构 ：包含印花税（卖出 0.1%）、过户费（0.001%）和券商佣金（通常 0.025% 起）

接口类型	适用场景	优点	缺点
CTP 官方接口	期货公司柜台	低延迟（<1ms）	需要期货公司资质
券商原生 API	股票交易	支持 Level2 行情	各家协议不统一
第三方封装 SDK	快速开发	跨券商兼容	存在中间层延迟
网页协议模拟	小型策略	无需申请权限	稳定性差

# 安装依赖库
pip install requests pandas websocket-client

import hashlib
import time

class TradeAPI:
    def __init__(self):
        self.base_url = "https://simtrade.example.com/api"
        self.ws_url = "wss://simtrade.example.com/stream"

    def _gen_sign(self, params):
        """生成 API 签名"""
        param_str = '&'.join([f'{k}={v}' for k,v in sorted(params.items())])
        return hashlib.md5((param_str + "API_SECRET_KEY").encode()).hexdigest()

    def connect(self, user_id, password):
        """建立会话连接"""
        auth_params = {
            'user_id': user_id,
            'timestamp': int(time.time() * 1000),
            'version': '1.0'
        }
        auth_params['sign'] = self._gen_sign(auth_params)

        resp = requests.post(f"{self.base_url}/auth", json=auth_params)
        self.session_id = resp.json()['session_id']

        # 启动心跳线程
        self._start_heartbeat()

import pandas as pd

class MarketData:
    def parse_l2_snapshot(self, data):
        """解析 Level2 快照数据"""
        df = pd.DataFrame(data['asks'], columns=['price', 'volume'])
        df['type'] = 'ask'
        df_bid = pd.DataFrame(data['bids'], columns=['price', 'volume'])
        df_bid['type'] = 'bid'

        return pd.concat([df, df_bid]).sort_values('price', ascending=False)

典型状态转换流程：

stateDiagram
    [*] --> PENDING
    PENDING --> PARTIAL_FILLED: 部分成交
    PENDING --> FILLED: 全部成交
    PENDING --> CANCELLED: 已撤单
    PARTIAL_FILLED --> FILLED
    PARTIAL_FILLED --> CANCELLED
    PENDING --> REJECTED: 拒单

异常处理要点：

使用指数退避重试（建议初始间隔 500ms）
记录原始订单 ID 防止重复提交
对「已报待撤」状态做特殊处理

控制报单频率：单账户每秒不超过 3 笔委托
避免价格突破：下单价格距离最新价不超过±2%（防止乌龙指）
使用冰山订单：大单拆分为小单分批发送

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), 
       wait=wait_exponential(multiplier=1, max=10))
def place_order(self, symbol, price, quantity):
    try:
        return self._raw_place_order(symbol, price, quantity)
    except ConnectionError as e:
        self.logger.error(f"Network error: {str(e)}")
        raise