基于pybroker的动量轮动+排序模型，年化11%（附代码）

除了少数的投资策略，比如套利之外，多数投资组合的构建思路都可归为“多因子模型”。因子可以是基本面，技术面，另类数据等。

单个标的的操作，比如期货、BTC等，重要技术指标加上仓位策略。仓位策略，止盈损之类的非常关键。

而组合型的投资，如etf，转债，股票等，一般不会重仓单支，而是投资组合的形式，这里就涉及一个优先的问题，优选的本质就是“多因子模型”。当然也可以附加大盘择时。

多因子模型一般是轮动模型，就是定期评估组合内各标的的得分，持有得分高的，卖出得分低的。

轮动模型，核心是对某因子进行排序，选得分高的TOP K个。

轮动算子

class SelectTopK:
    def __init__(self, K=1, order_by='order_by', b_ascending=False):
        self.K = K
        self.order_by = order_by
        self.b_ascending = b_ascending

    def __call__(self, ctxs: dict[str, ExecContext], *args, **kwargs):
        selected = None
        if pyb.param('selected'):
            selected = pyb.param('selected')

        scores = {}
        for symbol, ctx in ctxs.items():
            if selected and symbol not in selected:
                continue
            scores[symbol] = ctx.indicator(self.order_by)[-1]

        sorted_scores = sorted(
            scores.items(),
            key=lambda score: score[1],
            reverse=True
        )

        threshold = self.K
        top_scores = sorted_scores[:threshold]
        top_symbols = [score[0] for score in top_scores]
        pyb.param('selected', top_symbols)

规则选股

这里就是传统规则型多因子策略的逻辑，比如roc_20>0.08 且均线20突破收盘价，甚至可以加上3个条件满足2个及以上即可。

正则表达式加上eval，相当的强大！！！

class SelectBySignal:
    def __init__(self, buy_rules=[], buy_at_least_count=1, sell_rules=[], sell_rules_at_least_count=1):
        self.buy_rules = buy_rules
        self.buy_at_least_count = buy_at_least_count
        self.sell_rules = sell_rules
        self.sell_rules_at_least_count = sell_rules_at_least_count


    def _check_if_matched(self, ctxs, rules, at_least_count):
        matched_items = []
        for symbol, ctx in ctxs.items():
            match = 0
            for i, rule in enumerate(rules):
                expr = re.sub('ind\((.*)\)', 'ctx.indicator("\\1")[-1]', rule)
                if eval(expr):
                    match += 1
            if match >= at_least_count:
                matched_items.append(symbol)
        return matched_items

    def __call__(self, ctxs):
        to_buy = self._check_if_matched(ctxs, self.buy_rules, self.buy_at_least_count)
        to_sell = self._check_if_matched(ctxs, self.sell_rules, self.sell_rules_at_least_count)
        holding = utils.get_current_holdings(ctxs)

        new_hold = list(set(to_buy + holding))
        for s in to_sell:
            if s in new_hold:
                new_hold.remove(s)

        pyb.param('selected', new_hold)