Overnight Momentum Strategy

1from AlgorithmImports import *
2import numpy as np
3from typing import List, Dict
4from pandas.core.frame import DataFrame
5#endregion
6class OvernightMomentumStrategy(QCAlgorithm):
7def Initialize(self) -> None:
8self.SetStartDate(2015, 1, 1)
9self.SetCash(100_000)
10
11self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']	
12self.period: int = 21 # need n of ovenight returns
13market: Symbol = self.AddEquity('SPY', Resolution.Minute).Symbol
14
15self.data: Dict[Symbol, SymbolData] = {} # storing objects of SymbolData under stocks symbols
16self.quantile: int = 10
17self.leverage: int = 20
18self.min_share_price: int = 5
19
20self.traded_quantity: Dict[Symbol, float] = {}
21
22self.fundamental_count: int = 100
23self.fundamental_sorting_key = lambda x: x.DollarVolume
24self.selection_flag: bool = False
25self.UniverseSettings.Leverage = self.leverage
26self.UniverseSettings.Resolution = Resolution.Minute
27self.AddUniverse(self.FundamentalSelectionFunction)
28self.settings.daily_precise_end_time = False
29self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
30self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.BeforeMarketClose(market, 1), self.Selection)
31self.Schedule.On(self.DateRules.EveryDay(market), self.TimeRules.BeforeMarketClose(market, 20), self.MarketClose)
32def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
33for security in changes.AddedSecurities:
34    security.SetFeeModel(CustomFeeModel())
35def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
36# update overnight prices on daily basis
37for stock in fundamental:
38    symbol: Symbol = stock.Symbol
39    
40    if symbol in self.data:
41        # store current stock price
42        self.data[symbol].current_price = stock.AdjustedPrice
43        
44        # get history prices
45        history: DataFrame = self.History(symbol, 1, Resolution.Daily)
46        # update overnight returns based on history prices
47        self.UpdateOvernightReturns(symbol, history)
48
49# monthly rebalance
50if not self.selection_flag:
51    return Universe.Unchanged
52self.selection_flag = False
53
54selected: List[Fundamental] = [
55    x for x in fundamental 
56    if x.HasFundamentalData 
57    and x.Market == 'usa'
58    and x.MarketCap != 0
59    and x.Price > self.min_share_price
60    and x.SecurityReference.ExchangeId in self.exchange_codes
61]
62
63if len(selected) > self.fundamental_count:
64    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
65
66# warm up overnight returns
67for stock in selected:
68    symbol: Symbol = stock.Symbol
69    if symbol in self.data and self.data[symbol].is_overnight_returns_ready():
70        # get overnight returns from RollingWindow object and reverse it's list for simplier calculation of returns accumulation
71        overnight_returns: List[float] = [x for x in self.data[symbol].overnight_returns]
72        overnight_returns.reverse()
73        # calculate accumulated returns
74        accumulated_returns = np.prod([(1 + x) for x in overnight_returns]) - 1
75        # update returns accumulated for last month
76        self.data[symbol].returns_accumulated_last_month = accumulated_returns
77        
78        # go to next iteration, because there is no need for warm up overnight returns
79        continue
80    
81    # initialize SymbolData object for current symbol
82    self.data[symbol] = SymbolData(self.period)
83    # get history of n + 1 days
84    history: DataFrame = self.History(symbol, self.period + 1, Resolution.Daily)
85    # update overnight returns based on history prices
86    self.UpdateOvernightReturns(symbol, history)
87        
88market_cap: Dict[Symbol, float] = {} # storing stocks market capitalization
89last_accumulated_returns: Dict[Symbol, float] = {} # storing stocks last accumuldated returns
90
91for stock in selected:
92    symbol = stock.Symbol
93    if not self.data[symbol].is_ready():
94        continue
95    # store stock's market capitalization
96    market_cap[symbol] = stock.MarketCap
97    # store stock's last accumulated returns
98    last_accumulated_returns[symbol] = self.data[symbol].returns_accumulated_last_month
99
100# not enough data for decile selection     
101if len(last_accumulated_returns) < self.quantile:
102    return Universe.Unchanged
103
104# overnight returns sorting
105quantile: int = int(len(last_accumulated_returns) / self.quantile)
106sorted_by_last_acc_ret: List[Symbol] = [x[0] for x in sorted(last_accumulated_returns.items(), key=lambda item: item[1])]
107
108# long winners 
109long: List[Symbol] = sorted_by_last_acc_ret[-quantile:]
110# short losers
111short: List[Symbol] = sorted_by_last_acc_ret[:quantile]
112
113# market cap weighting
114for i, portfolio in enumerate([long, short]):
115    mc_sum: float = sum(list(map(lambda x: market_cap[x], portfolio)))
116    for symbol in portfolio:
117        if self.data[symbol].current_price != 0:
118            current_price: float = self.data[symbol].current_price
119            w: float = market_cap[symbol] / mc_sum
120            quantity: int = ((-1)**i) * np.floor((self.Portfolio.TotalPortfolioValue * w) / current_price)
121            self.traded_quantity[symbol] = quantity
122return list(self.traded_quantity.keys())
123
124def MarketClose(self) -> None:
125# send market on open and on close orders before market closes
126for symbol, q in self.traded_quantity.items():
127    self.MarketOnCloseOrder(symbol, q)
128    self.MarketOnOpenOrder(symbol, -q)
129
130def UpdateOvernightReturns(self, symbol: Symbol, history: DataFrame) -> None:
131# calculate overnight returns only if history isn't empty
132if history.empty:
133    return
134
135# get open and close prices
136opens = history.loc[symbol].open
137closes = history.loc[symbol].close
138
139# calculate overnight return for each day
140for (_, close_price), (_, open_price) in zip(closes.items(), opens.items()):
141    # check if previous close price isn't None
142    if self.data[symbol].prev_close_price:
143        # calculate overnight return
144        overnight_return = (open_price / self.data[symbol].prev_close_price) - 1
145        # store overnight return
146        self.data[symbol].update(overnight_return)
147    
148    # change value of prev close price for next calculation
149    self.data[symbol].prev_close_price = close_price
150
151def Selection(self) -> None:
152self.selection_flag = True
153self.traded_quantity.clear()
154
155class SymbolData():
156def __init__(self, period: int) -> None:
157self.overnight_returns: RollingWindow = RollingWindow[float](period)
158self.returns_accumulated_last_month: Union[None, float] = None
159self.prev_close_price: Union[None, float] = None
160self.current_price: float = 0.
161
162def update(self, overnight_return: float) -> None:
163self.overnight_returns.Add(overnight_return)
164
165def is_ready(self) -> bool:
166return self.returns_accumulated_last_month
167
168def is_overnight_returns_ready(self) -> bool:
169return self.overnight_returns.IsReady
170
171# Custom fee model.
172class CustomFeeModel(FeeModel):
173def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
174fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
175return OrderFee(CashAmount(fee, "USD"))