Cross-sectional momentum

• ✓Shines when the market has a clear direction and breakouts follow through
• ⚠Gets chopped up by whipsaws and false breakouts in sideways markets

Cross-sectional selection over a synthetic 8-symbol universe — not a promise of future returns.

"""横截面动量选股（Cross-sectional momentum）—— 嵌入式筛选器。

思路：在一个标的池里按 lookback 区间收益做横截面排名，做多最强的 top_n 个（等名义额），每 rebalance
步重排一次、退出跌出榜单的标的。经典 cross-sectional momentum 因子——不预测单一标的方向，而是"买强汰弱"。
本 demo 直接在 on_tick 里持有一个 Screener 并调用 select()，演示"把选股嵌入策略代码"的用法。

适用性：instrument="perp"、venues=["HYPERLIQUID"]。回测需多标的 universe feed（source="universe"）。
"""
from __future__ import annotations

from ..base import StrategyBase
from ..context import StrategyContext
from ..registry import register
from ..screener import Screener


@register("横截面动量")
class XSectMomentum(StrategyBase):
    description = "横截面动量选股：按区间收益排名做多最强 top_n（等权），定期换仓、退出跌出榜的标的。"
    params = {"lookback": 30, "top_n": 2, "rebalance": 20, "slice": 8000.0}
    venues = ["HYPERLIQUID"]
    symbols: list = []
    instrument = "perp"

    def __init__(self, lookback: int = 30, top_n: int = 2, rebalance: int = 20,
                 slice: float = 8000.0) -> None:
        self.screener = Screener("momentum", lookback=int(lookback), top_n=int(top_n), direction="top")
        self.rebalance = int(rebalance)
        self.slice = float(slice)
        self._t = 0
        self._active: set[str] = set()

    async def on_tick(self, ctx: StrategyContext) -> None:
        if self._t % max(1, self.rebalance) == 0:
            new = set(await self.screener.select(ctx))
            for s in self._active - new:
                await ctx.target(s, 0.0)  # 退出跌出榜单的标的
            self._active = new
        for s in self._active:
            px = await ctx.price(s)
            if px > 0:
                qty = self.slice / px
                await ctx.target(s, qty, band=qty * 0.1)
        self._t += 1

The full SDK and all strategies are MIT-licensed open source — backtest, paper-trade, or fork them directly.