昨天，在搭建回測系統主框架 | 從零搭建自己的AI量化平臺 #1裡，我們搭建好一回測框架，這是一個架子，裡面執行的是策略。
傳統回測系統，就是使用者在上面開發、除錯策略邏輯，工作量不小，而且還有可能出錯，有時候不知道是策略不行，還是程式出bug了。你既得很懂平臺，懂技術，還得非常瞭解自己的策略的細節。這個太麻煩了。
所以，我們在設計策略的時候，把策略Strategy拆成運算元，比如選股、賦權重，排序等，只需要選擇相應的“積木塊”就可以很快搭建出一個策略，基本不需要除錯。
from 。account import Accountclass Strategy： def __init__（self， name， algo_list=None）： self。name = name self。algo_list = algo_list self。acc = Account（） def algo_processor（self）： context = {} for algo in self。algo_list： if algo（context） is True： #如果algo返回True，直接不執行，本次不調倉 return None return context［‘weights’］ def onbar（self， index， date， df_bar）： self。acc。update_bar（date， df_bar） weights = self。algo_processor（） if weights： self。acc。adjust_weights（date， weights）
我們的Strategy無比簡潔！
backtest引擎迴圈呼叫我們的onbar函式。函式內，先呼叫account的update_bar更新資料；然後執行“運算元”——這個積木是外部傳入的；然後如果有weights需要調倉，則呼叫account的adjust_weights調倉。
然後我們來看一下，寫一個策略多麼簡單，根本不用繼承什麼類。
algo_list = ［ RunOnce（）， SelectFix（instruments=［‘sh000300’］）， WeightEqually（）］s = Strategy（name=‘買入並持有’，algo_list=algo_list）
寫策略就是“拼”一個積木塊，而且常用的積木塊我都內建了。
Runonce（）是控制演算法執行頻率，不使用預設就是每天執行。SelectFix就是固定選股；WeightEqually就是平均分配倉位。
所以，三個運算元拼起來，就是“買入並持有——滬深300指數”。
如果我們想“滬深300與中證500，7：3的倉位”呢。
algo_list2 = ［ RunOnce（）， SelectFix（instruments=［‘sh000300’， ‘sh000905’］）， WeightFix（weights=［0。7， 0。3］）］
使用我們的WeightFix運算元即可，根本不用寫多餘的程式碼！
兩個策略同時執行，發現加入中證500之後，長期後，年化收益與波動率都升高了，風險收益並存，符合邏輯。
再來看複雜一點的——動態再平衡，我們的策略2，每月再平衡一次。
algo_list3 = ［ RunPeriod（period=22）， SelectFix（instruments=［‘sh000300’， ‘sh000905’］）， WeightFix（weights=［0。7， 0。3］）］
把Runonce演算法改成RunPeriod（period=22），即每22天調倉執行一次即可。
可以看出，再平衡對於投資組合管理是有好處的，當然這裡暫未考慮到交易成本。
來更復雜且實用的——
動量交易策略
。
策略邏輯：
選擇三支指數：
滬深300，中證500及創業板指，分別代表大、中、小盤三個市場。
計算20日動量值，如果動量為正，取大者（top N）持有；
如果動量均為負，則空倉；
每天週期（如果有變化的話）。
轉成規則會描述成：
買入規則：動量(20)>0；
賣出規則：動量（20)<=0;
排序規則：動量20，topK = 1。
按模組化的思路，由於是每天執行，所以不需要頻率運算元，權重也很簡單，在投資組合裡，一般就是top N平均倉位，所以，仍然是WeightEqually（）就可以了。
焦點就在一個新的運算元：SelectWhere上。
這裡按交易訊號選擇top N，其實類似機器學習模型輸出pred_score，然後按score排序，取topN的邏輯是一樣的。我們多一個根據交易訊號來過濾一下標的。
把TopK作為一個獨立運算元，其實就是qlib裡和TopKStrategy。
class SelectTopK（Algo）： def __init__（self， K=2， col=‘pred_score’）： super（）。__init__（） self。K = K self。col = col def __call__（self， context）： bar = context［‘bar’］ bar。sort_values（by=self。col， ascending=False， inplace=True） #倒序 symbols = bar［‘instrument’］［：self。K］ context［‘selected’］ = symbols
由於我們不考慮buy， sell具體動作，而是以管理組合倉位的視角，反倒讓邏輯更加簡單，不需要考慮當前沒有持倉，反正就是取指定列的值倒排即可。得分高的topK個就平分下一期的倉位，其餘的都清倉。
這個運算元對於機器學習模型驅動的策略已經夠用了，只要提供預測的pred_score，選最高分的即可。
傳統的技術分析量化模組分有買入訊號、賣出訊號。我們要計算將要持倉的是“當前持倉+買入-賣出”的排重。
class SelectBySignal（Algo）： def __init__（self， signal_buy， signal_sell）： super（SelectBySignal， self）。__init__（） self。signal_buy = signal_buy self。signal_sell = signal_sell def __call__（self， context）： bar = context［‘bar’］。copy（） # 已經持倉的 acc = context［‘acc’］ holding = acc。get_holding_instruments（） # 再根據兩個訊號列，buy的選入，sell的去除 print（bar［bar［‘to_buy’］］） to_buy = list（bar［bar［self。signal_buy］］。index） to_sell = list（bar［bar［self。signal_sell］］。index） instruments = to_buy + holding - to_sell context［‘selected’］ = instruments
選擇交易買賣訊號進行選股。
所以，我們的動量策略只需要這樣”拼積木“就好了。
algo_list_rolling = ［ SelectFix（instruments=［‘sh000300’， ‘sh000905’， ‘sz399006’］）， SelectBySignal（signal_buy=‘to_buy’， signal_sell=‘to_sell’）， SelectTopK（K=1，col=‘五日動量’）， WeightEqually（）］
後續可以把這個步驟都做成視覺化，那隻需要點一點滑鼠，就可以進行策略生成和回測了。
明天繼續。
搭建回測系統主框架 | 從零搭建自己的AI量化平臺 #1
深度解讀Qlib的TopK策略 | Qlib從入門到精通 #8
