原创文章第230篇,专注“个人成长与财富自由、世界运作的逻辑与投资"。 接上一篇文章:因子表达式,积木式策略开发与pybroker框架整合(附源码)把周期的”算子“补充完整。 时间周期算子,很适合动态再平衡。比如年度,月度再平衡。 动态再平衡 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)
|