策略案例：多周期均线趋势

本案例用大周期过滤趋势、基础周期决定进出场：大周期窗口收盘价在均线上方才允许开仓，基础周期金叉买入、死叉清仓。

基础周期就是页面上选的 K 线，在 on_bar 里做金叉、死叉。更粗的周期在 __init__ 里 subscribe_bars，窗口闭合时进 on_window_bar，这里只用来判断趋势。on_window_bar 不是每次 on_bar 都触发，只在大周期窗口闭合时触发，并且发生在闭合它的那根基础 K 线的 on_bar 之后。窗口 K 线不参与撮合，下单仍写在 on_bar。

多个标的不用再单独订阅。subscribe_bars 省略 symbols 表示聚合页面上的全部标的。on_bar 和 on_window_bar 都是每个标的单独回调，用 bar.symbol 区分。

大周期要外置时，页面参数会作为构造参数传入，在 __init__ 里读取后再订阅。订阅表在 __init__ 之后冻结，不能写到 on_start。大周期必须粗于页面基础周期，相同或更细会报 ValueError。

两个周期的分工

两个周期分工不同，每个标的单独计算。

• 大周期只看能不能做。 窗口收盘价高于该周期均线，这个标的才允许开仓。价格落到均线下方后不再开新仓，已经持有的仓位不因此平掉。
• 基础周期决定什么时候做。 快均线上穿慢均线为金叉：当前空仓，并且大周期仍在均线上方，才买入。快均线下穿慢均线为死叉：有可卖数量就平仓。只出现金叉、大周期不在均线上方时，这次不开仓。
• 死叉当天卖不掉就记下。 卖出仍用 get_available_position()。T+1 导致当天可卖数量为 0 时，先把代码记下来，之后每根基础 K 线只要可卖数量大于 0 就继续平仓。

完整代码

import logging

import numpy as np

from qmfquant import Bar, ChoiceParam, IntParam, Order, Strategy, Trade


SESSION_WINDOWS = [("09:30", "11:30"), ("13:00", "15:00")]
LARGE_FREQ_CHOICES = ("30min", "1h", "1d")


class MultiPeriodMaStrategy(Strategy):
    """大周期过滤趋势，基础周期金叉开仓、死叉平仓。"""

    fast_window = IntParam(5, ge=2, le=60, title="小周期快线")
    slow_window = IntParam(20, ge=3, le=120, title="小周期慢线")
    trend_window = IntParam(20, ge=2, le=120, title="大周期均线")
    large_freq = ChoiceParam("1h", choices=list(LARGE_FREQ_CHOICES), title="大周期")
    quantity = IntParam(100, ge=100, title="每次交易股数")

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        freq = kwargs.get("large_freq")
        if not freq:
            freq = getattr(self.params, "large_freq", "1h")
        freq = str(freq)
        if freq not in LARGE_FREQ_CHOICES:
            raise ValueError(
                f"large_freq 只能是 {list(LARGE_FREQ_CHOICES)}，收到 {freq}"
            )
        self.large_freq_used = freq
        self.subscribe_bars(freq, session_windows=SESSION_WINDOWS)
        self.above_ma = {}
        self.to_close = set()

    def on_start(self) -> None:
        fast = int(self.params.fast_window)
        slow = int(self.params.slow_window)
        trend = int(self.params.trend_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 = 0
        self.set_history_depth(max(slow + 1, trend))
        self.lot_size = 100
        self.log(
            f"多周期双均线启动：基础周期快{fast}/慢{slow}，"
            f"大周期={self.large_freq_used}，均线{trend}，每次 {quantity} 股"
        )

    def on_window_bar(self, bar: Bar) -> None:
        if bar.freq != self.large_freq_used:
            return
        trend = int(self.params.trend_window)
        closes = self.get_history(trend, bar.symbol, "close")
        if closes is None or closes.size < trend or np.isnan(closes).any():
            return

        ma = float(np.mean(closes))
        above = float(closes[-1]) > ma
        previous = self.above_ma.get(bar.symbol)
        self.above_ma[bar.symbol] = above
        if previous != above:
            state = "均线上方" if above else "均线下方"
            self.log(
                f"[{bar.timestamp_iso}] {bar.symbol} {self.large_freq_used} "
                f"收盘 {closes[-1]:.3f} 均线 {ma:.3f}，进入{state}"
            )

    def on_bar(self, bar: Bar) -> None:
        symbol = bar.symbol
        if self.get_open_orders(symbol):
            return

        fast = int(self.params.fast_window)
        slow = int(self.params.slow_window)
        closes = self.get_history(slow + 1, symbol, "close")
        golden = False
        death = False
        if closes is not None and closes.size >= slow + 1 and not np.isnan(closes).any():
            fast_prev = float(np.mean(closes[-fast - 1:-1]))
            fast_now = float(np.mean(closes[-fast:]))
            slow_prev = float(np.mean(closes[-slow - 1:-1]))
            slow_now = float(np.mean(closes[-slow:]))
            golden = fast_prev <= slow_prev and fast_now > slow_now
            death = fast_prev >= slow_prev and fast_now < slow_now

        position = self.get_position(symbol)
        if golden and position == 0:
            if self.above_ma.get(symbol) is True:
                quantity = int(self.params.quantity)
                receipt = self.buy(symbol, quantity=quantity, tag="mtf-golden")
                self.log(
                    f"[{bar.timestamp_iso}] {symbol} 金叉且大周期在均线上方，"
                    f"买入 {quantity}：{receipt.primary}"
                )
            else:
                self.log(f"[{bar.timestamp_iso}] {symbol} 金叉，但大周期不在均线上方")

        if death:
            self.to_close.add(symbol)
        if symbol not in self.to_close:
            return
        if position <= 0:
            self.to_close.discard(symbol)
            return
        if self.get_available_position(symbol) <= 0:
            return
        self.close_position(symbol)
        self.to_close.discard(symbol)
        self.log(f"[{bar.timestamp_iso}] {symbol} 死叉，提交清仓")

    def on_order(self, order: Order) -> None:
        self.log(
            f"订单状态：{order.id} {order.status} "
            f"{order.filled_quantity}/{order.quantity}"
        )

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

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

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

要点说明

• subscribe_bars 只能写在 __init__，且要在 super().__init__() 之后调用；写在 on_start 会因订阅表已冻结而无效。
• large_freq 既可能由页面参数传入，也可能走内联参数默认值，所以 __init__ 里先取 kwargs 再回落到 self.params。
• above_ma 只保存最近一次大周期状态，on_bar 用它判断是否允许开仓；它不会因为价格回落而自动平掉已有持仓。
• to_close 记录死叉但当天卖不掉的标的，后续每根基础 K 线只要可卖数量大于 0 就继续平仓。

---

相关阅读

• • 历史数据获取
• • 报单与成交
• • 多周期