动态再平衡与轮动模块“积木化”

时间周期算子，很适合动态再平衡。比如年度，月度再平衡。

动态再平衡

class RunPeriod:
    def __init__(self):
        self.last_date = None

    def __call__(self, ctxs, *args, **kwargs):
        now = utils.get_current_dt(ctxs)

        last_date = self.last_date

        date_to_compare = last_date
        now = pd.Timestamp(now)
        date_to_compare = pd.Timestamp(date_to_compare)

        result = self.compare_dates(now, date_to_compare)
        self.last_date = now
        return result

    @abc.abstractmethod
    def compare_dates(self, now, date_to_compare):
        raise (NotImplementedError("RunPeriod Algo is an abstract class!"))


# https://github.com/pmorissette/bt/blob/master/bt/algos.py
class RunQuarterly(RunPeriod):

    def compare_dates(self, now, date_to_compare):
        if now.quarter != date_to_compare.quarter:
            pyb.param('is_done', False)  # 需要执行下一步
        else:
            pyb.param('is_done', True)


class RunWeekly(RunPeriod):

    def compare_dates(self, now, date_to_compare):
        if now.week != date_to_compare.week:
            pyb.param('is_done', False)  # 需要执行下一步
        else:
            pyb.param('is_done', True)


class RunMonthly(RunPeriod):

    def compare_dates(self, now, date_to_compare):
        if now.month != date_to_compare.month:
            pyb.param('is_done', False)  # 需要执行下一步
        else:
            pyb.param('is_done', True)


class RunYearly(RunPeriod):

    def compare_dates(self, now, date_to_compare):
        if now.year != date_to_compare.year:
            pyb.param('is_done', False)  # 需要执行下一步
        else:
            pyb.param('is_done', True)

与之前版本的区别在于，之前是通过返回值True或False，下游再决定要不要继续执行。这里是通过设置 全局变量 is_done。如果是True，表示当前不需要继续执行了。好处是比较直观，坏处是与pybroker框架有绑定，这个我觉得还好，本身也需要调用框架本身的一些api。

有了算子模块，写策略就容易很多，关键是代码很好维护，不易出错。

与”买入持有”相比，就是改一行RunQuarterly()

这里有一个投资心得，之前我一直有一个感受，就是动态再平衡未必能带来收益率的提升，但肯定能降低回撤（降低波动）。但在当前这个环境上——未必！

免费的午餐，确实只是——分散可以降低波动，这是投资理论，可能通过数学严格证明的。因为波动率是二次方，收益率降低的速度比收益率快得多，所以降低一点预期收益，可以大幅降低波动（风险）。这就是分散的投资逻辑基础。

轮动算子

轮动算子，计算当前所以symbol的指标值。（这里需要看下，如果在某一天没有指标呢？由于pybroker的symbol是独立管理，独立计算，可能存在缺失的可能性）

逻辑就比较简单了，按因子值排序，得分高的，取K个，把选中的传递传存在selected变量中。

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):
        scores = {
            symbol: ctx.indicator(self.order_by)[-1]
            for symbol, ctx in ctxs.items()
        }  # dict {symbol: 指标值}

        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)