Momentum Effect in Stocks Combined with Stop-Losses

1import numpy as np
2from AlgorithmImports import *
3class MomentumEffectStocksCombinedStopLosses(QCAlgorithm):
4def Initialize(self):
5self.SetStartDate(2000, 1, 1)
6self.SetCash(100000)
7self.period:int = 7
8self.leverage:int = 10
9self.quantile:int = 10
10self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
11market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
12
13self.fundamental_count:int = 500
14self.fundamental_sorting_key = lambda x: x.DollarVolume
15# Symbol data.
16self.data:Dict[Symbol, SymbolData] = {}
17self.weight:Dict[Symbol, float] = {}
18
19# Opened stop-orders.
20self.opened_orders:List[OrderTicket] = []
21
22self.last_month:int = -1
23self.selection_flag:bool = False
24self.UniverseSettings.Resolution = Resolution.Daily
25self.AddUniverse(self.FundamentalSelectionFunction)
26self.settings.daily_precise_end_time = False
27self.settings.minimum_order_margin_portfolio_percentage = 0.
28self.schedule.on(self.date_rules.month_start(market),
29                self.time_rules.after_market_open(market),
30                self.selection)
31def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
32for security in changes.AddedSecurities:
33    symbol:Symbol = security.Symbol
34    security.SetFeeModel(CustomFeeModel())
35    security.SetLeverage(self.leverage)
36    if symbol not in self.data:
37        self.data[symbol] = SymbolData(self.period)
38        
39def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
40if not self.selection_flag:
41    return Universe.Unchanged
42# Update the rolling window every month.
43for stock in fundamental:
44    symbol:Symbol = stock.Symbol
45    # Store monthly price.
46    if symbol in self.data:
47        self.data[symbol].update(stock.AdjustedPrice)
48selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.MarketCap != 0 and \
49    x.SecurityReference.ExchangeId in self.exchange_codes and x.CompanyReference.IsREIT == 0]
50if len(selected) > self.fundamental_count:
51	selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]] 
52
53performance:Dict[Fundamental, float] = {}
54# Warmup price rolling windows.
55for stock in fundamental:
56    symbol:Symbol = stock.Symbol
57    
58    if symbol not in self.data:
59        self.data[symbol] = SymbolData(self.period)
60        history = self.History(symbol, self.period*30, Resolution.Daily)
61        if history.empty:
62            self.Log(f"Not enough data for {symbol} yet.")
63            continue
64        closes = history.loc[symbol].close
65        
66        closes_len = len(closes.keys())
67        # Find monthly closes.
68        for index, time_close in enumerate(closes.items()):
69            # index out of bounds check.
70            if index + 1 < closes_len:
71                date_month = time_close[0].date().month
72                next_date_month = closes.keys()[index + 1].month
73            
74                # Found last day of month.
75                if date_month != next_date_month:
76                    self.data[symbol].update(time_close[1])
77    
78    if self.data[symbol].is_ready():
79        performance[stock] = self.data[symbol].performance()
80# Performance sorting. 
81if len(performance) >= self.quantile:
82    sorted_by_return:List = sorted(performance.items(), key = lambda x: x[1], reverse = True)
83    quantile:int = int(len(sorted_by_return) / self.quantile)
84    long:List[Fundamental] = [x[0] for x in sorted_by_return[:quantile]]
85
86    # Market cap weighting.
87    mc_sum:float = sum([x.MarketCap for x in long])
88    for stock in long:
89        self.weight[stock.Symbol] = stock.MarketCap / mc_sum
90return list(self.weight.keys())
91
92def OnData(self, data: Slice) -> None:
93if not self.selection_flag:
94    return
95self.selection_flag = False
96# Trade execution.
97for symbol, w in self.weight.items():
98    if symbol in data.Keys and data[symbol]:
99        curr_price:float = data[symbol].Value
100        if curr_price != 0:
101            # Unit size calc.
102            unit_size:float = self.CalculateOrderQuantity(symbol, w)
103    
104            # Buy order.
105            if unit_size != 0:
106                self.MarketOrder(symbol, unit_size)
107    
108                # SL setting.
109                sl_price:float = curr_price * 0.95    # 5% SL.
110                ticket:OrderTicket = self.StopMarketOrder(symbol, -unit_size, sl_price, 'SL')
111                self.opened_orders.append(ticket)
112
113self.weight.clear()
114def selection(self) -> None:
115self.selection_flag = True
116# Liquidate and cancel pending orders.
117self.Liquidate()
118for ticket_index in range(len(self.opened_orders)-1, 0, -1):
119    response = self.opened_orders[ticket_index].Cancel("Canceled Trade")
120    self.opened_orders.remove(self.opened_orders[ticket_index])
121class SymbolData():
122def __init__(self, period: int):
123self._price:RollingWindow = RollingWindow[float](period)
124
125def update(self, price: float) -> None:
126self._price.Add(price)
127
128def is_ready(self) -> bool:
129return self._price.IsReady
130
131# 12 month momentum, one month skipped.
132def performance(self) -> float:
133return self._price[1] / self._price[self._price.Count - 1] - 1
134# Custom fee model.
135class CustomFeeModel(FeeModel):
136def GetOrderFee(self, parameters):
137fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
138return OrderFee(CashAmount(fee, "USD"))