Sales Seasonality Premium

1import numpy as np
2from AlgorithmImports import *
3from collections import deque
4from typing import List, Dict, Tuple, Deque
5from numpy import isnan
6class SalesSeasonalityPremium(QCAlgorithm):
7def Initialize(self):
8self.SetStartDate(2000, 1, 1)
9self.SetCash(100000)
10self.avg_SEA_threshold:int = 100
11self.max_period:int = 1000
12self.period:int = 13
13self.leverage:int = 10
14self.min_share_price:int = 5
15self.consecutive_quarter_count:int = 6
16self.percentiles:List[int] = [10, 90]
17self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']	
18self.sea:Dict[Symbol, Deque[Tuple[float]]] = {} # SEA quarterly data
19self.avgsea_alltime:Deque[float] = deque(maxlen=self.max_period)   # All time SEA values.
20market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
21
22self.weight:Dict[Symbol, float] = {}
23
24self.fundamental_count:int = 3000
25self.fundamental_sorting_key = lambda x: x.MarketCap
26self.selection_flag:bool = True
27self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
28self.UniverseSettings.Resolution = Resolution.Daily
29self.AddUniverse(self.FundamentalSelectionFunction)
30self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
31self.settings.daily_precise_end_time = False
32def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
33for security in changes.AddedSecurities:
34    security.SetFeeModel(CustomFeeModel())
35    security.SetLeverage(self.leverage)
36def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
37if not self.selection_flag:
38    return Universe.Unchanged
39selected:List[Fundamental] = [
40    x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa' and \
41    not isnan(x.FinancialStatements.IncomeStatement.TotalRevenue.ThreeMonths) and x.FinancialStatements.IncomeStatement.TotalRevenue.ThreeMonths != 0 and \
42    not isnan(x.FinancialStatements.IncomeStatement.TotalRevenue.TwelveMonths) and x.FinancialStatements.IncomeStatement.TotalRevenue.TwelveMonths != 0 and \
43    not isnan(x.EarningReports.BasicAverageShares.ThreeMonths) and x.EarningReports.BasicAverageShares.ThreeMonths > 0 and \
44    not isnan(x.EarningReports.BasicEPS.TwelveMonths) and x.EarningReports.BasicEPS.TwelveMonths > 0 and \
45    not isnan(x.ValuationRatios.PERatio) and x.ValuationRatios.PERatio > 0 and \
46    x.SecurityReference.ExchangeId in self.exchange_codes and x.MarketCap != 0
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]]
50    
51# Stock which had earnings last month.
52last_month_date:datetime = self.Time - timedelta(days = self.Time.day)
53fine_selected = [x for x in selected if x.EarningReports.FileDate.Value.month == last_month_date.month and x.EarningReports.FileDate.Value.year == last_month_date.year]
54
55avg_sea:Dict[Fundamental, float] = {}
56for stock in fine_selected:
57    symbol = stock.Symbol
58    
59    # store sea data.
60    if symbol not in self.sea:
61        self.sea[symbol] = deque(maxlen = self.period)
62    # SEA calc.
63    curr_month:float = stock.EarningReports.FileDate.Value.month
64    revenue:float = stock.FinancialStatements.IncomeStatement.TotalRevenue.ThreeMonths
65    # annual_revenue = stock.FinancialStatements.IncomeStatement.TotalRevenue.TwelveMonths
66    annual_revenue:float = (revenue / curr_month) * 12
67    
68    if annual_revenue != 0:
69        self.sea[symbol].append((stock.EarningReports.FileDate.Value, revenue / annual_revenue))
70    
71    if len(self.sea[symbol]) == self.sea[symbol].maxlen:
72        curr_year:int = self.Time.year
73        relevant_quarters:List[Tuple[float]] = [x for x in self.sea[symbol] if x[0].year == curr_year - 2 or x[0].year == curr_year - 3]
74        # Make sure we have a consecutive seasonal data => 2 years by 4 quarters.
75        if len(relevant_quarters) == self.consecutive_quarter_count:
76            avgsea:float = np.mean([x[1] for x in relevant_quarters])
77            avg_sea[stock] = avgsea
78            
79            # All time avg_sea values to calculate deciles.
80            self.avgsea_alltime.append(avgsea)
81
82if len(avg_sea) != 0:
83    # Sort by SEA.
84    long:List[Fundamental] = []
85    short:List[Fundamental] = []
86    
87    # Wait for at least 100 last AVGSEA values to estimate percentile values.
88    if len(self.avgsea_alltime) > self.avg_SEA_threshold: 
89        avgsea_values:List[float] = [x for x in self.avgsea_alltime]
90        
91        top_decile:float = np.percentile(avgsea_values, self.percentiles[1])
92        bottom_decile:float = np.percentile(avgsea_values, self.percentiles[0])
93        
94        for stock, avgsea in avg_sea.items():
95            if avgsea > top_decile:
96                short.append(stock)
97            elif avgsea < bottom_decile:
98                long.append(stock)
99    else:
100        return Universe.Unchanged
101        
102    # Market cap weighting.
103    for i, portfolio in enumerate([long, short]):
104        mc_sum:float = sum(list(map(lambda stock: stock.MarketCap, portfolio)))
105        for stock in portfolio:
106            self.weight[stock.Symbol] = ((-1)**i) * stock.MarketCap / mc_sum
107
108return list(self.weight.keys())
109
110def Selection(self) -> None:
111self.selection_flag = True
112
113def OnData(self, data: Slice) -> None:
114if not self.selection_flag:
115    return
116self.selection_flag = False
117
118# Trade execution.
119portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
120self.SetHoldings(portfolio, True)
121self.weight.clear()
122class CustomFeeModel(FeeModel):
123def GetOrderFee(self, parameters):
124fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
125return OrderFee(CashAmount(fee, "USD"))