MACD-V: Volatility Normalised Momentum

1from AlgorithmImports import *
2# endregion
3
4class MACDVVolatilityNormalizedMomentum(QCAlgorithm):
5
6def Initialize(self):
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000)
9
10self.short_period:int = 12
11self.long_period:int = 26
12self.short_hold_period:int = 15
13self.long_hold_period:int = 77
14
15self.target_limit:float = 1.0285
16self.macd_entry_threshold:int = 70
17
18self.price_at_trade:float = 0.
19self.holding_counter:int = 0
20
21future = self.AddData(QuantpediaFutures, 'EUREX_FDAX1', Resolution.Daily)
22future.SetFeeModel(CustomFeeModel())
23self.future = future.Symbol
24
25self.short_period_EMA:ExponentialMovingAverage = self.EMA(self.future, self.short_period)
26self.long_period_EMA:ExponentialMovingAverage = self.EMA(self.future, self.long_period)
27self.ATR:AverageTrueRange = self.ATR(self.future, self.long_period)
28
29self.SetWarmup(self.long_period, Resolution.Daily)
30
31def OnData(self, data: Slice) -> None:
32if self.IsWarmingUp:
33    return
34
35future_last_update_data:datetime.date = QuantpediaFutures._last_update_date
36
37# check if data is still coming
38if self.Securities[self.future].GetLastData() and self.Time.date() >= future_last_update_data:
39    self.Liquidate()
40    return
41
42if self.future in data and data[self.future]:
43    if self.Portfolio.Invested:
44        self.holding_counter += 1
45        if (data[self.future].Value > (self.price_at_trade * self.target_limit)) and (self.holding_counter == 1):
46            self.Liquidate()
47        
48        elif (data[self.future].Value > self.price_at_trade) and (self.holding_counter == self.short_hold_period):
49            self.Liquidate()
50
51        elif self.holding_counter == self.long_hold_period:
52            self.Liquidate()
53
54    if all(x.IsReady for x in [self.short_period_EMA, self.long_period_EMA, self.ATR]):
55        MACD_V:float = ((self.short_period_EMA.Current.Value - self.long_period_EMA.Current.Value) / self.ATR.Current.Value) * 100
56
57        if MACD_V > self.macd_entry_threshold and (not self.Portfolio.Invested):
58            # buy future
59            self.SetHoldings(self.future, 1)
60            self.price_at_trade = data[self.future].Value
61            self.holding_counter = 0
62
63# Quantpedia data.
64# NOTE: IMPORTANT: Data order must be ascending (datewise)
65class QuantpediaFutures(PythonData):
66_last_update_date:datetime.date = datetime(1,1,1).date()
67
68@staticmethod
69def get_last_update_date() -> Dict[Symbol, datetime.date]:
70return QuantpediaFutures._last_update_date
71
72def GetSource(self, config, date, isLiveMode):
73return SubscriptionDataSource("data.quantpedia.com/backtesting_data/futures/EUREX_FDAX1.csv", SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
74
75def Reader(self, config, line, date, isLiveMode):
76data = QuantpediaFutures()
77data.Symbol = config.Symbol
78
79if not line[0].isdigit(): return None
80split = line.split(';')
81
82data.Time = datetime.strptime(split[0], "%d.%m.%Y") + timedelta(days=1)
83data['back_adjusted'] = float(split[1])
84data['spliced'] = float(split[2])
85data.Value = float(split[1])
86
87if data.Time.date() > QuantpediaFutures._last_update_date:
88    QuantpediaFutures._last_update_date = data.Time.date()
89
90return data
91
92# Custom fee model
93class CustomFeeModel(FeeModel):
94def GetOrderFee(self, parameters):
95fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
96return OrderFee(CashAmount(fee, "USD"))