策略案例：双均线 T+1

本案例用快慢两条均线判断趋势：空仓且快线在慢线上方时买入，持仓且快线在慢线下方时清仓可卖部分。同时按买入日期记录持仓，在 T+1 约束下不允许当日卖出。

完整代码

import logging

import numpy as np

from qmfquant import (
    Bar,
    BoolParam,
    IntParam,
    Order,
    OrderSide,
    OrderStatus,
    Strategy,
    Trade,
)


class DoubleMovingAverageStrategy(Strategy):
    """快线在慢线上方买入，下方卖出可卖仓位。多标的，可按买入日限制 T+1。"""

    fast_window = IntParam(5, ge=2, le=100, title="短均线周期")
    slow_window = IntParam(20, ge=3, le=250, title="长均线周期")
    quantity = IntParam(100, ge=100, title="每次交易股数")
    t1 = BoolParam(True, title="是否T+1(日线及以上周期可关闭)")

    def on_start(self) -> None:
        fast = int(self.params.fast_window)
        slow = int(self.params.slow_window)
        quantity = int(self.params.quantity)
        if fast >= slow:
            raise ValueError("fast_window 必须小于 slow_window")
        if quantity % 100 != 0:
            raise ValueError("quantity 必须是 100 的整数倍")

        self.warmup_period = slow
        self.set_history_depth(slow)
        self.lot_size = 100
        # symbol -> {buy_date, volume, buy_price}
        self.positions_dict = {}
        self.pending_symbols = set()
        self.schedule_daily("14:50:00", "daily-risk-check")
        self.log("双均线策略启动")

    def _trade_date(self, timestamp_iso) -> str:
        text = str(timestamp_iso)
        if "T" in text:
            return text.split("T", 1)[0]
        return text[:10]

    def on_bar(self, bar: Bar) -> None:
        symbol = bar.symbol
        if self.get_open_orders(symbol):
            return
        # 订单已结束但终态回调没对上时，不要让挂单标记把后面的信号挡住
        self.pending_symbols.discard(symbol)

        slow = int(self.params.slow_window)
        closes = self.get_history(slow, symbol, field="close")
        if closes is None or closes.size < slow or np.isnan(closes).any():
            return

        fast = int(self.params.fast_window)
        fast_ma = float(np.mean(closes[-fast:]))
        slow_ma = float(np.mean(closes[-slow:]))
        position = self.get_position(symbol)
        available = self.get_available_position(symbol)

        if position == 0:
            self.positions_dict.pop(symbol, None)
            if fast_ma > slow_ma:
                quantity = int(self.params.quantity)
                receipt = self.buy(symbol, quantity=quantity, tag="ma-long")
                self.pending_symbols.add(symbol)
                self.log(
                    f"[{bar.timestamp_iso}] {symbol} 快线在慢线上方，"
                    f"买入 {quantity}：{receipt.primary}"
                )
            return

        if position > 0 and fast_ma < slow_ma:
            held = self.positions_dict.get(symbol)
            if self.params.t1 and held is not None:
                if self._trade_date(bar.timestamp_iso) <= held["buy_date"]:
                    return
            if available <= 0:
                return
            self.close_position(symbol)
            self.pending_symbols.add(symbol)
            self.log(f"[{bar.timestamp_iso}] {symbol} 快线在慢线下方，提交清仓")

    def on_timer(self, payload: str) -> None:
        if payload == "daily-risk-check":
            account = self.get_account()
            self.log(
                f"每日检查：权益={account.get('equity', 0):.2f} "
                f"现金={account.get('cash', 0):.2f}"
            )

    def _release_pending(self, symbol: str) -> None:
        self.pending_symbols.discard(symbol)

    def on_order(self, order: Order) -> None:
        self.log(
            f"订单状态：{order.id} {order.status} "
            f"{order.filled_quantity}/{order.quantity}"
        )
        if order.status in {
            OrderStatus.Filled,
            OrderStatus.Cancelled,
            OrderStatus.Rejected,
            OrderStatus.Expired,
        }:
            self._release_pending(order.symbol)

    def on_trade(self, trade: Trade) -> None:
        symbol = trade.symbol
        trade_date = self._trade_date(trade.timestamp_iso)
        quantity = float(trade.quantity)
        self._release_pending(symbol)
        self.log(
            f"成交回报：{symbol} {trade.side} {trade.quantity}@{trade.price}"
        )

        if trade.side == OrderSide.Buy:
            held = self.positions_dict.get(symbol)
            if held is None:
                self.positions_dict[symbol] = {
                    "buy_date": trade_date,
                    "volume": quantity,
                    "buy_price": float(trade.price),
                }
            else:
                held["volume"] = float(held["volume"]) + quantity
            return

        if trade.side == OrderSide.Sell:
            held = self.positions_dict.get(symbol)
            if held is None:
                return
            held["volume"] = float(held["volume"]) - quantity
            if held["volume"] <= 0:
                self.positions_dict.pop(symbol, None)

    def on_reject(self, order: Order) -> None:
        self._release_pending(order.symbol)
        self.log(
            f"拒单：{order.id}，原因：{order.reject_reason}",
            logging.ERROR,
        )

    def on_stop(self) -> None:
        self.log(f"策略停止：权益={self.equity:.2f} 持仓={self.positions}")

要点说明

• on_start 里先校验参数，并把 slow_window 同时设为预热长度和历史缓存深度，保证第一次信号计算时数据够用。
• pending_symbols 用来避免同一标的在订单未结束时重复触发信号；on_order 进入终态、on_trade 成交、on_reject 拒单时都会释放标记。
• T+1 判断不是读框架的持仓天数，而是自己在 on_trade 里记录买入日期，在 on_bar 里与当前 K 线日期比较。
• 卖出前必须检查 get_available_position()，当天买入的部分不在可卖数量里。

相关阅读

• • 外置参数与数据对象
• • 历史数据获取
• • 报单与成交