Advertising Effect within Stocks

1from AlgorithmImports import *
2from typing import Dict, List
3import numpy as np
4class AdvertisingEffect(QCAlgorithm):
5def Initialize(self) -> None:
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100_000)
8self.UniverseSettings.Leverage = 10
9self.UniverseSettings.Resolution = Resolution.Daily
10self.AddUniverse(self.FundamentalSelectionFunction)
11self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
12self.fundamental_count: int = 3_000
13self.fundamental_sorting_key = lambda x: x.MarketCap
14self.min_market_cap: int = 20_000_000
15self.quantile: int = 10
16self.buy_month: int = 6
17self.sell_month: int = 12
18self.selection_flag: bool = False
19self.adv_expenses: Dict[Symbol, float] = {}
20self.long_symbols: List[Symbol] = []
21self.short_symbols: List[Symbol] = []
22self.settings.daily_precise_end_time = False
23self.settings.minimum_order_margin_portfolio_percentage = 0.
24market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
25self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
26
27def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
28for security in changes.AddedSecurities:
29    security.SetFeeModel(CustomFeeModel())
30
31def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
32if not self.selection_flag:
33    return Universe.Unchanged
34
35filtered: List[Fundamental] = [
36    f for f in fundamental if f.HasFundamentalData
37    and f.SecurityReference.ExchangeId in self.exchange_codes
38    and f.MarketCap > self.min_market_cap
39    and not np.isnan(f.FinancialStatements.IncomeStatement.SellingAndMarketingExpense.ThreeMonths)
40    and f.FinancialStatements.IncomeStatement.SellingAndMarketingExpense.ThreeMonths > 0
41]
42sorted_filter: List[Fundamental] = sorted(filtered,
43                                        key=self.fundamental_sorting_key,
44                                        reverse=True)[:self.fundamental_count]
45d_adv: Dict[Symbol, float] = {}
46for f in sorted_filter:        
47    if f.Symbol not in self.adv_expenses:
48        self.adv_expenses[f.Symbol] = -1
49    
50    adv_expenses: float = f.FinancialStatements.IncomeStatement.SellingAndMarketingExpense.ThreeMonths
51    if f.Symbol in self.adv_expenses and self.adv_expenses[f.Symbol] != -1:
52        d_adv[f.Symbol] = adv_expenses / self.adv_expenses[f.Symbol] - 1
53    # Update adv expense value
54    self.adv_expenses[f.Symbol] = adv_expenses
55    
56# NOTE: Get rid of old advertisment records so we work with latest values
57for symbol in self.adv_expenses:
58    if symbol not in [x.Symbol for x in sorted_filter]:
59        self.adv_expenses[symbol] = -1
60
61if len(d_adv) >= self.quantile:
62    sorted_by_adv: list = sorted(d_adv.items(), key=lambda x: x[1], reverse=True)
63    decile: int = int(len(sorted_by_adv) / self.quantile)
64    self.long_symbols = [x[0] for x in sorted_by_adv[-decile:]]
65    self.short_symbols = [x[0] for x in sorted_by_adv[:decile]]
66
67return self.long_symbols + self.short_symbols
68
69def OnData(self, slice: Slice) -> None:
70if not self.selection_flag:
71    return
72self.selection_flag = False
73
74# Trade execution
75targets: List[PortfolioTarget] = []
76for i, portfolio in enumerate([self.long_symbols, self.short_symbols]):
77    for symbol in portfolio:
78        if slice.ContainsKey(symbol) and slice[symbol] is not None:
79            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
80self.SetHoldings(targets, True)
81self.long_symbols.clear()
82self.short_symbols.clear()
83
84def Selection(self) -> None:
85if self.Time.month == self.buy_month:
86    self.selection_flag = True
87elif self.Time.month == self.sell_month:
88    self.Liquidate()
89class CustomFeeModel(FeeModel):
90def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
91fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
92return OrderFee(CashAmount(fee, "USD"))