Weighted Frequency of Losses

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3from typing import List, Dict
4import numpy as np
5from collections import deque
6# endregion
7
8class WeightedFrequencyofLosses(QCAlgorithm):
9
10def Initialize(self) -> None:
11self.SetStartDate(2000, 1, 1)
12self.SetCash(100000)  
13
14self.market:Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
15self.tickers_to_ignore:List[str] = ['CDLB']
16
17self.weight:Dict[Symbol, float] = {}
18self.data:Dict[Symbol, deque] = {}
19
20self.r:float = 0.13
21self.quantile:int = 10
22self.leverage:int = 5
23self.period = 30
24
25self.fundamental_count:int = 3000
26self.selection_flag:bool = False
27self.UniverseSettings.Resolution = Resolution.Daily
28self.Settings.MinimumOrderMarginPortfolioPercentage = 0
29self.AddUniverse(self.FundamentalSelectionFunction)
30self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
31
32def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
33for security in changes.AddedSecurities:
34    security.SetLeverage(self.leverage)
35    security.SetFeeModel(CustomFeeModel())
36
37def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
38# update the rolling window every day
39for stock in fundamental:
40    symbol:Symbol = stock.Symbol
41
42    # store monthly price
43    if symbol in self.data:
44        self.data[symbol].append((self.Time, stock.AdjustedPrice))
45
46if not self.selection_flag:
47    return Universe.Unchanged
48
49selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Symbol.Value not in self.tickers_to_ignore and x.MarketCap != 0 and \
50        (x.SecurityReference.ExchangeId == 'NYS') or (x.SecurityReference.ExchangeId == 'NAS') or (x.SecurityReference.ExchangeId == 'ASE')]
51# selected:List[Fundamental] = [x
52#     for x in sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Symbol.Value not in self.tickers_to_ignore],
53#         key = lambda x: x.DollarVolume, reverse = True)[:self.fundamental_count]]
54
55# warmup price rolling windows
56for stock in selected:
57    symbol:Symbol = stock.Symbol
58    if symbol in self.data:
59        continue
60    
61    self.data[symbol] = deque(maxlen=self.period)
62    history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
63    if history.empty:
64        self.Log(f"Not enough data for {symbol} yet.")
65        continue
66    closes:pd.Series = history.loc[symbol].close
67    for time, close in closes.items():
68        self.data[symbol].append((time, close))
69
70# selected = [x for x in selected if x.MarketCap != 0 and \
71#         (x.SecurityReference.ExchangeId == 'NYS') or (x.SecurityReference.ExchangeId == 'NAS') or (x.SecurityReference.ExchangeId == 'ASE')]
72
73if len(selected) > self.fundamental_count:
74    selected = sorted(selected, key=lambda x: x.MarketCap, reverse=True)[:self.fundamental_count]
75
76selected:Dict[Symbol, Fundamental] = {x.Symbol: x for x in selected if len(self.data[x.Symbol]) == self.data[x.Symbol].maxlen}
77
78if len(selected) != 0:
79    # create dataframe from saved prices
80    selected_stocks:Dict[Symbol, List[float]] = {symbol: [i[1] for i in value] for symbol, value in self.data.items() if symbol in selected.keys()}
81    df_stocks:DataFrame = pd.DataFrame(selected_stocks, index=[i[0] for i in list(self.data.values())[0]])
82    
83    # trim dataframe to most recent month period
84    last_month_start:datetime.date = (self.Time.date() - timedelta(self.Time.day + 1)).replace(day=1)
85    df_stocks = df_stocks.pct_change()
86    df_stocks = df_stocks[df_stocks.index.date >= last_month_start]
87    
88    # indicator function
89    df_stocks[df_stocks >= 0.] = 0.
90    df_stocks[df_stocks < 0.] = 1.
91    
92    # weighted frequency
93    weights:np.ndarray = np.array([np.exp(self.r * (j - 1)) for j in range(1, len(df_stocks) + 1)])
94    weights = weights / sum(weights)
95
96    df_stocks = df_stocks.mul(weights, axis=0)
97    df_stocks = df_stocks.sum(axis=0)
98
99    VFL:Dict[Symbol, float] = df_stocks.to_dict()
100
101    # sort and divide to upper decile and lower decile
102    if len(VFL) >= self.quantile:
103        sorted_VFL:List[Symbol] = sorted(VFL, key=VFL.get, reverse=True)
104        quantile:int = int(len(sorted_VFL) / self.quantile)
105        long:List[Symbol] = sorted_VFL[:quantile]
106        short:List[Symbol] = sorted_VFL[-quantile:]
107
108        # calculate weights based on marketcap
109        for i, portfolio in enumerate([long, short]):
110            mc_sum:float = sum([selected[x].MarketCap for x in portfolio])
111            for symbol in portfolio:
112                self.weight[symbol] = ((-1) ** i) * (selected[symbol].MarketCap / mc_sum)
113
114return list(self.weight.keys())
115
116def OnData(self, data: Slice) -> None:
117# monthly rebalance
118if not self.selection_flag:
119    return
120self.selection_flag = False
121
122# trade execution
123portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
124self.SetHoldings(portfolio, True)
125
126self.weight.clear()
127
128def Selection(self) -> None:
129self.selection_flag = True
130
131# custom fee model
132class CustomFeeModel(FeeModel):
133def GetOrderFee(self, parameters):
134fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
135return OrderFee(CashAmount(fee, "USD"))