📚 课程目录第3章实盘账户接入与下单模块

3. 实盘账户接入与下单模块

任务目标

本案例通过券商官方API（以华泰证券MATIC为例）连接实盘账户，实现三个核心功能：下单、撤单、查询持仓。代码采用同步阻塞方式，便于理解API调用链路；实际生产环境可替换为异步或事件驱动模式。

环境依赖

Python 3.8+
华泰证券MATIC交易API（需向券商申请开通权限，获取 client_id 和 client_secret）
安装依赖：pip install pymatic

关键完整代码

import time
import json
from pymatic import MaticClient

class LiveTrading:
    def __init__(self, client_id: str, client_secret: str):
        self.client = MaticClient(client_id, client_secret)
        self.client.login()

    def place_order(self, symbol: str, side: str, order_type: str, quantity: int, price: float = None):
        """下单。side: 'BUY'/'SELL'; order_type: 'LIMIT'/'MARKET'"""
        order_params = {
            "symbol": symbol,
            "side": side,
            "order_type": order_type,
            "quantity": quantity,
        }
        if order_type == "LIMIT":
            if price is None:
                raise ValueError("LIMIT订单必须指定价格")
            order_params["price"] = price

        response = self.client.place_order(**order_params)
        if response.get("status") == "success":
            return response["order_id"]
        else:
            raise RuntimeError(f"下单失败: {response.get('message')}")

    def cancel_order(self, order_id: str):
        """撤单"""
        response = self.client.cancel_order(order_id)
        if response.get("status") != "success":
            raise RuntimeError(f"撤单失败: {response.get('message')}")
        return True

    def get_positions(self):
        """查询持仓"""
        response = self.client.get_positions()
        if response.get("status") != "success":
            raise RuntimeError(f"查询持仓失败: {response.get('message')}")
        return response["positions"]

    def run_demo(self):
        """演示完整流程：下单→查持仓→撤单"""
        # 1. 下单（以买入100股平安银行为例）
        order_id = self.place_order(
            symbol="000001.SZ",
            side="BUY",
            order_type="LIMIT",
            quantity=100,
            price=10.50
        )
        print(f"订单已提交，order_id: {order_id}")

        # 2. 查询持仓（验证订单是否成交）
        time.sleep(2)  # 等待交易所回报
        positions = self.get_positions()
        print(f"当前持仓: {json.dumps(positions, ensure_ascii=False)}")

        # 3. 撤单（若订单未成交则撤销）
        self.cancel_order(order_id)
        print(f"订单 {order_id} 已撤销")

if __name__ == "__main__":
    # 请替换为实际申请的凭证
    trader = LiveTrading(
        client_id="your_client_id",
        client_secret="your_client_secret"
    )
    trader.run_demo()

运行方式

将上述代码保存为 live_trading.py。
替换 your_client_id 和 your_client_secret 为券商实际分配的凭证。
执行 python live_trading.py。

验证方法与判定标准

下单验证：程序输出包含 order_id，且券商交易终端（或MATIC控制台）能看到对应委托记录。
持仓查询验证：若订单已成交，get_positions() 返回的持仓列表中应包含 000001.SZ 及对应数量；若未成交，持仓列表不变。
撤单验证：撤单成功后，券商终端中该订单状态变为“已撤”，且 cancel_order 返回 True。

常见问题

登录失败：确认 client_id/client_secret 正确，且IP已在券商白名单中。
下单被拒：检查是否为交易时段、股票是否停牌、价格是否超出涨跌停范围。
撤单失败：订单若已成交则无法撤销，需先查询订单状态确认“已报”或“部分成交”时才可撤单。