Momentum Based on Fractional-Difference Filter

1from AlgorithmImports import *
2import data_tools
3import numpy as np
4from dateutil.relativedelta import relativedelta
5from typing import List, Dict, Tuple
6from math import factorial, prod
7from decimal import Decimal
8# endregion
9
10class MomentumBasedonFractionalDifferenceFilter(QCAlgorithm):
11
12def Initialize(self):
13self.SetStartDate(2000, 1, 1)
14self.SetCash(100000)
15
16self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
17
18self.data:Dict[Symbol, float] = {}
19
20self.stocks_to_liquidate:List[data_tools.HoldingItem] = []
21self.traded_portfolio_portion:Dict[Symbol, float] = {}
22
23self.t:int = 252
24self.period:int = self.t * 5
25self.quantile:int = 5
26self.leverage:int = 25
27self.holding_period:int = 5
28self.fundamental_count:int = 50
29self.d:Decimal = Decimal(0.9)
30
31# pi and weights calculations
32self.pi_list:List[Decimal] = [ self.pi(i, self.d) for i in range(self.period) ]
33self.weights:List[Decimal] = [ self.w(np.array(list(range(self.period))), self.pi_list[i], i) for i in range(self.period) ]
34
35self.rebalance_flag:bool = False
36self.Settings.MinimumOrderMarginPortfolioPercentage = 0
37self.UniverseSettings.Resolution = Resolution.Daily
38self.AddUniverse(self.FundamentalSelectionFunction)
39self.Schedule.On(self.DateRules.EveryDay(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
40
41def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
42for security in changes.AddedSecurities:
43    security.SetFeeModel(data_tools.CustomFeeModel())
44    security.SetLeverage(self.leverage)
45
46def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
47# store daily stock prices
48for stock in fundamental:
49    symbol:Symbol = stock.Symbol
50
51    if symbol in self.data:
52        self.data[symbol].update_daily_return(stock.AdjustedPrice)
53
54selected:List[Symbol] = [x.Symbol
55    for x in sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.MarketCap != 0 and \
56        (x.SecurityReference.ExchangeId == 'NYS') or (x.SecurityReference.ExchangeId == 'NAS') or (x.SecurityReference.ExchangeId == 'ASE')],
57        key = lambda x: x.DollarVolume, reverse = True)[:self.fundamental_count]]
58
59# price warmup and prediction
60predict_by_symbol:Dict[Symbol, float] = {}
61
62for symbol in selected:
63    if symbol not in self.data:
64        self.data[symbol] = data_tools.SymbolData(self.period)
65
66        # price warmup
67        history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
68        if history.empty:
69            self.Log(f"Not enough data for {symbol} yet.")
70            continue
71        closes:pd.Series = history.loc[symbol].close
72        for time, close in closes.items():
73            self.data[symbol].update_daily_return(close)
74    
75    if self.data[symbol].is_ready():
76        # prediction
77        prices:List[float] = np.log(self.data[symbol].get_daily_prices())
78        predict:float = - (self.weights[len(self.weights) - self.t] / self.t - 1) * Decimal(sum([ prices[i] - prices[len(self.weights) - self.t] for i in range(len(self.weights) - self.t + 1, len(self.weights) - 1) ])) \
79                        + Decimal(sum([ (self.weights[i] + (self.weights[len(self.weights) - self.t] / (self.t - 1))) * Decimal(prices[i] - prices[len(self.weights) - 1]) for i in range(len(self.weights) - self.t + 1, len(self.weights) - 1) ]))
80        predict_by_symbol[symbol] = predict
81
82# sort and divide into quantiles
83if len(predict_by_symbol) >= self.quantile:
84    sorted_predicts:List[Symbol] = sorted(predict_by_symbol, key=predict_by_symbol.get, reverse=True)
85    quantile:int = len(sorted_predicts) // self.quantile
86    long = sorted_predicts[:quantile]
87    short = sorted_predicts[-quantile:]
88
89    for i, portfolio in enumerate([long, short]):
90        for symbol in portfolio:
91            self.traded_portfolio_portion[symbol] = ((-1) ** i) / len(portfolio) * (self.Portfolio.TotalPortfolioValue / self.holding_period)
92                
93    self.rebalance_flag = True
94
95return selected
96
97def OnData(self, data: Slice) -> None:
98if not self.rebalance_flag:
99    return
100self.rebalance_flag = False
101
102items_to_remove:List[data_tools.HoldingItem] = []
103
104# execute order and hold for holding period
105for item in self.stocks_to_liquidate:
106    item._holding_period += 1
107    if item._holding_period >= self.holding_period:
108        self.MarketOrder(item._symbol, -item._quantity)
109        items_to_remove.append(item)    
110
111# remove from collection
112for item in items_to_remove:
113    self.stocks_to_liquidate.remove(item)    
114
115# execute order
116for price_symbol, portfolio_portion in self.traded_portfolio_portion.items():
117    if price_symbol in data and data[price_symbol]:
118        final_quantity:int = portfolio_portion // data[price_symbol].Price
119        if portfolio_portion != 0:
120            self.MarketOrder(price_symbol, final_quantity)
121            self.stocks_to_liquidate.append(data_tools.HoldingItem(price_symbol, final_quantity))
122
123self.traded_portfolio_portion.clear()
124
125def Selection(self) -> None:
126self.selection_flag = True
127
128def w(self, price_dataset:np.ndarray, pi:Decimal, u:int) -> Decimal:
129result:Decimal = (np.sum((np.arange((len(price_dataset) - u - self.t + 1), len(price_dataset) - u)) * pi) / self.t) - self.pi((len(price_dataset) - u + 1), self.d) if u < len(price_dataset) - self.t \
130            else ((np.sum((np.arange(len(price_dataset) - u)) * pi) / self.t) - self.pi((len(price_dataset) - u + 1), self.d) if u in range((len(price_dataset) - self.t + 1) , len(price_dataset) - 1) \
131            else (self.pi(0, self.d) / self.t) - self.pi(1, self.d) - 1)
132return result
133
134def pi(self, s:int, d:Decimal) -> Decimal:
135s_factorial:Decimal = Decimal(factorial(s))
136result:Decimal = Decimal(((-1) ** s) * (prod([ Decimal(d - i) / s_factorial for i in range(s-1) ])))
137return result