策略案例：横截面与多标的轮动

本案例演示多标的横截面轮动：用回看窗口的动量打分，等权持有得分最高的若干只，并通过 on_cross_section 在当日完整切片后统一调仓。

on_bar 按单事件流逐条触发。多标的轮动、排序、打分不要假设一次 on_bar 就代表全部标的已经同时更新。

推荐：日界钩子统一调仓

• 用 on_before_trading 做“前一交易日信息可见”的盘前横截面准备与调仓。
• 用 on_cross_section 做“看到当日所有标的的当拍 bar 后再同周期调仓”。

以下案例是通过 on_cross_section 执行统一调仓，调仓使用 rebalance_weights。

import logging

import numpy as np

from qmfquant import FloatParam, IntParam, Strategy


class QMFStrategy(Strategy):
    """回看窗口动量打分，等权持有得分最高的若干只。"""

    lookback = IntParam(20, ge=5, le=120, title="回看窗口")
    top_n = IntParam(2, ge=1, le=10, title="持仓只数")
    min_valid = IntParam(2, ge=2, le=10, title="最少有效样本")
    rebalance_every = IntParam(5, ge=1, le=20, title="调仓间隔交易日")
    gross = FloatParam(0.95, ge=0.1, le=0.99, title="总仓位")

    def on_start(self) -> None:
        lookback = int(self.params.lookback)
        self.warmup_period = lookback + 1
        self.set_history_depth(lookback)
        self.lot_size = 100
        self.contracts = []
        # 上一轮横截面时的合约数。和本轮不一致说明池子还在变，先不调。
        self._stable_count = 0
        self._days_since_rebalance = int(self.params.rebalance_every)
        self.log(
            "横截面轮动启动 | 回看=%d | 持有=%d | 间隔=%d | 合约待 on_bar 登记"
            % (lookback, int(self.params.top_n), int(self.params.rebalance_every))
        )

    def on_bar(self, bar) -> None:
        """只负责发现页面实际推过来的合约，不在这里打分。"""
        symbol = bar.symbol
        if symbol not in self.contracts:
            self.contracts.append(symbol)
            self.log(
                f"on_bar 新增合约{symbol}，当前合约数 {len(self.contracts)}",
                logging.ERROR,
            )

    def on_cross_section(self, trading_date, timestamp) -> None:
        """用已登记合约打分。数量还在变化时不调仓。"""
        count = len(self.contracts)
        if count == 0 or count != self._stable_count:
            self._stable_count = count
            self.log(f"{trading_date} 合约池未稳定，当前 {count} 个，本轮不调仓")
            return

        self._days_since_rebalance += 1
        if self._days_since_rebalance < int(self.params.rebalance_every):
            return
        if self.get_open_orders():
            self.log(f"{trading_date} 存在未完成订单，跳过本次调仓")
            return

        scores = self._momentum_scores()
        need = int(self.params.min_valid)
        if len(scores) < need:
            self.log(
                f"{trading_date} 有效样本 {len(scores)}/{count}，不足 {need}，不调仓"
            )
            return

        ranked_pairs = sorted(scores.items(), key=lambda item: item[1], reverse=True)
        chosen = [symbol for symbol, _score in ranked_pairs[: int(self.params.top_n)]]
        weight = float(self.params.gross) / len(chosen)
        target_weights = {symbol: weight for symbol in chosen}
        order_ids = self.rebalance_weights(
            target_weights,
            liquidate_unmentioned=True,
        )
        self._days_since_rebalance = 0
        ranked = ", ".join(f"{symbol}={score:.2%}" for symbol, score in ranked_pairs)
        self.log(
            f"{trading_date} 合约数 {count} 目标 {target_weights} "
            f"订单 {list(order_ids or [])} | {ranked}"
        )

    def _momentum_scores(self) -> dict:
        """只对 on_bar 里见过的合约算区间涨跌幅。"""
        lookback = int(self.params.lookback)
        history_map = self.get_history_map(
            count=lookback,
            symbols=self.contracts,
            field="close",
        )
        scores = {}
        if not history_map:
            return scores

        for symbol, closes in history_map.items():
            arr = np.asarray(closes, dtype=float)
            if arr.size < lookback or not np.isfinite(arr).all():
                continue
            base = float(arr[0])
            if base <= 0:
                continue
            scores[symbol] = (float(arr[-1]) - base) / base
        return scores

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

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

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

备选：收齐同一 timestamp 后再执行

没有固定调仓时点时，可在 on_bar 中先缓存同一时间片的标的，收齐后再执行一次横截面逻辑。停牌或缺失数据时，方案可能不触发；可设置“有效样本数达阈值即执行”。

常见坑

• 停牌/缺失数据：某些标的当日无 Bar 时，收齐方案可能不触发。
• Universe 漂移：成分股调整后若仍用旧列表，会出现权重与真实池不一致。
• 调仓时点与成交策略错配：收盘信号配下一根开盘成交时，不要按当日收盘价假设已经换完仓。
• 历史长度不足：新上市或停牌恢复标的数据窗口不完整，评分前检查长度和 NaN。
• 仓位未收敛：多标的先卖后买若资金未及时释放，可能导致买入不足；目标仓位 API 会先减后加，下一时点可二次收敛。

---

相关阅读

• • 策略基础与生命周期
• • 历史数据获取
• • 目标仓位与组合调仓