原创文章第315篇，专注“个人成长与财富自由、世界运作的逻辑与投资"。
日拱一卒，功不唐捐
1、duckdb loader优化为二级目录。
2、行业轮动子策略。动量，rsrs，多指标符合买入卖出。
3、增加波动率，风险资产标识，系数乘以波动率倒数和之比例。
我们的回测系统基本稳定。
内核是支持传统规则量化和机器学习AI量化的回测引擎，外围的算子，基本覆盖选股，择时，权重分配，市场择时逻辑。辅助模块包括数据获取、数据加载，因子表达式与计算等。
使用方式开始在星球日更了，请同学们关注下面这个专栏，快速熟悉平台的使用与代码结构，尽快开发出适合自己的策略。
后面会围绕： 系统、算子、数据、因子、模型（规则），策略的维度逐步迭代。本月的重点目标是ETF的绝对收益策略交付实盘，敬请期待。
从大的策略模板来看，大致分为：大类资产配置、择时、轮动、策略组合。提炼出共同点就是选股、买卖规则，排序规则（可以是因子排序，合成因子排序，机器模型排序），策略组合就是多个策略组合在一起，绝对型收益很有用。
主要就是如下这个算子：可以给定买入规则，卖出规则，至少要满足几件规则等：
class SelectBySignal(AlgoExtend):
def __init__(self, data: pd.DataFrame, buy_rules=[], buy_at_least_count=1, sell_rules=[], sell_rules_at_least_count=1):
super(SelectBySignal, self).__init__(data)
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, df_bar, rules, at_least_count):
matched_items = []
for symbol in list(df_bar.index):
bar = df_bar.loc[symbol]
match = 0
for i, rule in enumerate(rules):
# expr = re.sub('ind\((.*?)\)', 'bar["\\1"]', rule)
#if eval(expr):
if rule in list(bar.index):
if bar[rule]:
match += 1
if match >= at_least_count:
matched_items.append(symbol)
return matched_items
def __call__(self, target):
df_bar = self.get_bar_df(target)
if self.buy_rules and len(self.buy_rules):
to_buy = self._check_if_matched(df_bar, self.buy_rules, self.buy_at_least_count)
else:
to_buy = list(df_bar.index)
to_sell = self._check_if_matched(df_bar, self.sell_rules, self.sell_rules_at_least_count)
holding = self.get_long_position_symbols(target)
#print(holding)
#holding = target.position
#holding = context.get_long_symbols()
new_hold = list(set(to_buy + holding))
for s in to_sell:
if s in new_hold:
new_hold.remove(s)
#if len(new_hold) > 0:
#print(new_hold)
target.temp['selected'] = new_hold
return True
然后配合SelectTopK轮动使用：
class SelectTopK(AlgoExtend):
def __init__(self, df_factor: pd.DataFrame, factor_name='order_by', K=1, drop_top_n=0, b_ascending=False):
self.K = K
self.drop_top_n = drop_top_n # 这算是一个魔改，就是把最强的N个弃掉，尤其动量指标，过尤不及。
self.df_factor = df_factor
self.factor_name = factor_name
self.b_ascending = b_ascending
def __call__(self, target):
selected = None
key = 'selected'
# print(target.now)
df_bar = self.df_factor.loc[target.now]
if type(df_bar) is pd.Series:
df_bar = df_bar.to_frame().T
df_bar.set_index('symbol', inplace=True)
# factor = factor[factor['']]
factor_sorted = df_bar.sort_values(by=self.factor_name, ascending=self.b_ascending)
symbols = factor_sorted.index
ture_symbols = []
for s in symbols:
price = target.universe.loc[target.now][s]
if price < 0:
continue
ture_symbols.append(s)
# bar_df = bar_df.sort_values(self.order_by, ascending=self.b_ascending)
if not selected:
start = 0
if self.drop_top_n <= len(ture_symbols):
start = self.drop_top_n
ordered = ture_symbols[start: start + self.K]
else:
ordered = []
else:
ordered = []
count = 0
for s in ture_symbols: # 一定是当天有记录的
if s in selected:
count += 1
if count > self.drop_top_n:
ordered.append(s)
if len(ordered) >= self.K:
break
target.temp[key] = ordered
return True
最近几年，ETF行业指数动量轮动的效果一般，主要是风格切换太快。
代码已经提交至星球：
联系客服
