Monthly Reversal and/or Momentum Based on Intraday Returns

1from AlgorithmImports import *
2from typing import List, Dict
3class MonthlyReversalMomentumBasedIntradayReturns(QCAlgorithm):
4def Initialize(self):
5self.SetStartDate(2010, 1, 1)
6self.SetCash(100000)
7self.min_share_price:int = 5
8self.leverage:int = 5
9self.quantile:int = 10
10self.days_to_lookup:int = 21
11self.period:int = self.days_to_lookup * 7
12
13market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
14
15self.fundamental_sorting_key = lambda x: x.DollarVolume
16self.fundamental_count:int = 300
17self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
18
19self.last_course:List[Symbol] = []
20
21# Relevant hourly closes (at 14 and 16)
22self.daily_return:Dict[Symbol, float] = {}
23self.price_14:Dict[Symbol, float] = {}
24
25self.settings.daily_precise_end_time = False
26self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
27self.UniverseSettings.Resolution = Resolution.Hour
28self.AddUniverse(self.FundamentalSelectionFunction)
29self.selection_flag:bool = False
30
31self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
32
33def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
34for security in changes.AddedSecurities:
35    security.SetFeeModel(CustomFeeModel())
36    security.SetLeverage(self.leverage)
37    
38def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
39if not self.selection_flag:
40    return Universe.Unchanged
41
42self.selection_flag = False
43selected:List[Fundamental] = [
44    x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and \
45    x.Market == 'usa' and x.SecurityReference.ExchangeId in self.exchange_codes\
46]
47    
48if len(selected) > self.fundamental_count:
49    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
50self.last_course = [x.Symbol for x in selected[:self.fundamental_count]]
51
52return self.last_course
53
54def OnData(self, data: Slice):
55# Store 14h price to calculate daily return.
56if self.Time.hour == 14:
57    for symbol in self.last_course:
58        if symbol in data and data[symbol]:
59            price_14:float = data[symbol].Value
60            if price_14 != 0:
61                # 14h price value.
62                self.price_14[symbol] = price_14
63
64# Calculate daily return.
65if self.Time.hour == 16:
66    
67    accumulated_returns:Dict[Symbol, float] = {}
68    for symbol in self.last_course:
69        if symbol in self.price_14 and symbol in data and data[symbol]:
70            price_16:float = data[symbol].Value
71            if price_16 != 0:
72                # Calculate intraday return.
73                price_14 = self.price_14[symbol] # 14h price
74                ret:float = price_16 / price_14 - 1
75                    
76                if symbol not in self.daily_return:
77                    self.daily_return[symbol] = RollingWindow[float](21) # One month of daily returns
78                self.daily_return[symbol].Add(ret)
79                    
80                # Month worth of daily return is ready.
81                if self.daily_return[symbol].IsReady:
82                    acc_ret:float = sum([x for x in self.daily_return[symbol]])
83                    accumulated_returns[symbol] = acc_ret
84    if len(accumulated_returns) == 0:
85        return
86    # Sort by daily accumulated returns.
87    sorted_by_return:List[Tuple[Symbol, float]] = sorted(accumulated_returns.items(), key = lambda x: x[1], reverse = True)
88    quantile:int = int(len(sorted_by_return) / self.quantile)
89    long:List[Symbol] = [x[0] for x in sorted_by_return[-quantile:]]
90    short:List[Symbol] = [x[0] for x in sorted_by_return[:quantile]]
91    # Trade execution
92    targets:List[PortfolioTarget] = []
93    for i, portfolio in enumerate([long, short]):
94        for symbol in portfolio:
95            if symbol in data and data[symbol]:
96                targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
97    
98    self.SetHoldings(targets, True)
99    self.daily_return.clear()
100    
101def Selection(self):
102self.selection_flag = True
103
104# Custom fee model
105class CustomFeeModel(FeeModel):
106def GetOrderFee(self, parameters):
107fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
108return OrderFee(CashAmount(fee, "USD"))