第 45 篇：实盘接口全景 QMT、PTrade 与开源框架

# ---------- 1. 交易接口抽象层：定义「渠道无关」的下单协议 ----------
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class Order:
    symbol: str
    side: str            # "buy" / "sell"
    price: float
    volume: int
    status: str = "pending"          # pending/filled/partial/rejected
    filled_volume: int = 0
    msgs: list = field(default_factory=list)

class BrokerBase(ABC):
    """所有实盘/模拟通道都要实现的接口"""
    @abstractmethod
    def buy(self, symbol, price, volume) -> Order: ...
    @abstractmethod
    def sell(self, symbol, price, volume) -> Order: ...
    @abstractmethod
    def position(self, symbol) -> int: ...
    @abstractmethod
    def cash(self) -> float: ...

class PaperBroker(BrokerBase):
    """纸面交易通道（第46篇主角）：本地记账，验证流程用"""
    def __init__(self, init_cash=100000.0, fee=0.001, slip=0.001):
        self._cash = init_cash
        self._pos = {}
        self.fee, self.slip = fee, slip

    def _fill(self, symbol, side, price, volume):
        # 模拟滑点：买入向上滑、卖取向下滑
        fill_px = price * (1 + self.slip if side == "buy" else -self.slip)
        cost = volume * fill_px * (1 + self.fee if side == "buy"
                                   else -self.fee)
        self._cash -= cost
        self._pos[symbol] = self._pos.get(symbol, 0) + \
            (volume if side == "buy" else -volume)
        o = Order(symbol, side, price, volume, "filled", volume,
                  [f"{datetime.now():%H:%M:%S} 成交价{fill_px:.2f}"])
        return o

    def buy(self, symbol, price, volume):
        return self._fill(symbol, "buy", price, volume)

    def sell(self, symbol, price, volume):
        return self._fill(symbol, "sell", price, volume)

    def position(self, symbol):
        return self._pos.get(symbol, 0)

    def cash(self):
        return self._cash

# ---------- 2. 模拟「断线重连」的健壮下单封装 ----------
import time

def robust_order(broker, fn, symbol, price, volume,
                 retries=3, wait=2.0):
    """带重试的下单：网络类异常退避重试，业务类异常立刻报告"""
    for attempt in range(1, retries + 1):
        try:
            return fn(symbol, price, volume)
        except ConnectionError as e:        # 瞬时：重试
            if attempt == retries:
                raise RuntimeError(f"下单失败(重试耗尽): {e}")
            time.sleep(wait * attempt)
        except ValueError as e:             # 业务：如资金不足
            raise RuntimeError(f"订单被拒: {e}")
    return None

# ---------- 3. 演示：通道切换不改策略代码 ----------
def strategy_entry(broker: BrokerBase, symbol, price, volume):
    """策略只依赖抽象接口——明天把 PaperBroker 换成 QmtBroker，
    这行代码一个字都不用改"""
    if broker.position(symbol) == 0:
        order = robust_order(broker, broker.buy, symbol, price, volume)
        print(f"买入 {symbol} {volume}股 @ {price}, "
              f"剩余现金 {broker.cash():,.0f}")
        return order
    return None

paper = PaperBroker()
strategy_entry(paper, "510300", 4.020, 10000)
print(f"持仓: {paper.position('510300')} 股")

# ---------- 4. QMT 侧的对接示意（伪代码，需开通后按官方文档实现） ----------
class QmtBroker(BrokerBase):
    """示意结构：实际方法名/参数以所开通 QMT 的官方文档为准"""
    def __init__(self, path, session_id):
        # from xtquant import xttrader ...
        # self.xt = xttrader.XtQuantTrader(path, session_id)
        # self.acc = 资金账号对象
        raise NotImplementedError("需开通QMT后按官方SDK实现")

    def buy(self, symbol, price, volume):
        # 委托: self.xt.order_stock(self.acc, symbol, 买,
        #        volume, 限价, price...)
        ...
    def sell(self, symbol, price, volume): ...
    def position(self, symbol):
        # 查询: self.xt.query_stock_positions(self.acc) ...
        ...
    def cash(self): ...

示例输出（示意）:
买入 510300 10000股 @ 4.02, 剩余现金 59,548
持仓: 10000 股