1 Month Momentum in Bonds

1from AlgorithmImports import *
2class OneMonthMomentum(QCAlgorithm):
3def Initialize(self):
4 self.SetStartDate(2000, 1, 1)
5 self.SetCash(100000)
6 self.symbols = [
7     "EUREX_FGBL1",    # Euro-Bund (10Y) Futures, Continuous Contract #1 (Germany)
8     "CME_TY1",        # 10 Yr Note Futures, Continuous Contract #1 (USA)
9     "MX_CGB1",        # Ten-Year Government of Canada Bond Futures, Continuous Contract #1 (Canada)
10     "ASX_XT1",        # 10 Year Commonwealth Treasury Bond Futures, Continuous Contract #1 (Australia)
11     "SGX_JB1",        # SGX 10-Year Mini Japanese Government Bond Futures, Continuous Contract #1 (Japan)
12     "LIFFE_R1",       # Long Gilt Futures, Continuous Contract #1 (U.K.)
13     "EUREX_FBTP1"     # Long-Term Euro-BTP Futures, Continuous Contract #1 (Italy)
14 ]
15 
16 self.period = 21
17 self.quantile = 5
18 self.SetWarmUp(self.period)
19 
20 # Daily ROC data.
21 self.data = {}
22 
23 for symbol in self.symbols:
24     data = self.AddData(QuantpediaFutures, symbol, Resolution.Daily)
25     data.SetFeeModel(CustomFeeModel())
26     data.SetLeverage(5)
27     
28     self.data[symbol] = self.ROC(symbol, self.period, Resolution.Daily)
29 
30 self.rebalance_flag: bool = False
31 self.Schedule.On(self.DateRules.MonthStart(self.symbols[0]), self.TimeRules.At(0, 0), self.Rebalance)
32 self.settings.daily_precise_end_time = False
33 self.settings.minimum_order_margin_portfolio_percentage = 0.
34
35def on_data(self, slice: Slice) -> None:
36 if not self.rebalance_flag:
37     return
38 self.rebalance_flag = False
39 # Return sorting.
40 sorted_by_return = sorted([x for x in self.data.items() if x[1].IsReady and self.Securities[x[0]].GetLastData() and self.Time.date() < QuantpediaFutures.get_last_update_date()[x[0]] and slice.contains_key(x[0]) and slice[x[0]]], key = lambda x: x[1].Current.Value, reverse = True)
41 long = []
42 short = []
43 if len(sorted_by_return) >= self.quantile:
44     quintile = int(len(sorted_by_return) / self.quantile)
45     long = [x[0] for x in sorted_by_return[:quintile]]
46     short = [x[0] for x in sorted_by_return[-quintile:]]
47 # Trade execution.
48 targets: List[PortfolioTarget] = []
49 for i, portfolio in enumerate([long, short]):
50     for symbol in portfolio:
51         if slice.contains_key(symbol) and slice[symbol]:
52             targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
53 
54 self.SetHoldings(targets, True)
55def Rebalance(self):
56 self.rebalance_flag = True
57# Quantpedia data.
58# NOTE: IMPORTANT: Data order must be ascending (datewise)
59class QuantpediaFutures(PythonData):
60_last_update_date:Dict[Symbol, datetime.date] = {}
61@staticmethod
62def get_last_update_date() -> Dict[Symbol, datetime.date]:
63return QuantpediaFutures._last_update_date
64def GetSource(self, config, date, isLiveMode):
65 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/futures/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
66def Reader(self, config, line, date, isLiveMode):
67 data = QuantpediaFutures()
68 data.Symbol = config.Symbol
69 
70 if not line[0].isdigit(): return None
71 split = line.split(';')
72 
73 data.Time = datetime.strptime(split[0], "%d.%m.%Y") + timedelta(days=1)
74 data['back_adjusted'] = float(split[1])
75 data['spliced'] = float(split[2])
76 data.Value = float(split[1])
77 if config.Symbol.Value not in QuantpediaFutures._last_update_date:
78     QuantpediaFutures._last_update_date[config.Symbol.Value] = datetime(1,1,1).date()
79 if data.Time.date() > QuantpediaFutures._last_update_date[config.Symbol.Value]:
80     QuantpediaFutures._last_update_date[config.Symbol.Value] = data.Time.date()
81 return data
82# Custom fee model.
83class CustomFeeModel(FeeModel):
84def GetOrderFee(self, parameters):
85 fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
86 return OrderFee(CashAmount(fee, "USD"))