Fibonacci Supports and Resistances in Cross-Sectional Stock Trading

1from AlgorithmImports import *
2from data_tools import CustomFeeModel, SymbolData
3from datetime import date
4from pandas.core.frame import DataFrame
5# endregion
6
7class FibonacciSupportsAndResistancesInCrossSectionalStockTrading(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2000, 1, 1)
11self.SetCash(100000)
12
13self.history_start:datetime.date = date(1999, 1, 1)
14
15self.leverage:int = 5
16self.quantile:int = 5
17self.total_portfolio_parts:int = 2  # long + short
18
19# fibinacci levels: 0, 0.381, 0.5, 0.612, 1
20self.fibonacci_levels:List[float] = [0.5]
21
22self.data:Dict[Symbol, SymbolData] = {}
23self.managed_queue:List[List[Symbol, float]] = []
24self.prev_managed_queue:List[List[Symbol, float]] = []
25
26self.market_symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
27
28self.coarse_count:int = 500
29self.selection_flag:bool = False
30self.UniverseSettings.Resolution = Resolution.Daily
31self.AddUniverse(self.FundamentalSelectionFunction)
32self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
33self.Schedule.On(self.DateRules.WeekStart(self.market_symbol), self.TimeRules.BeforeMarketClose(self.market_symbol, 0), self.Selection)
34
35def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
36for security in changes.AddedSecurities:
37    security.SetFeeModel(CustomFeeModel())
38    security.SetLeverage(self.leverage)
39
40def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
41for equity in fundamental:
42    symbol:Symbol = equity.Symbol
43
44    if symbol in self.data:
45        self.data[symbol].update(equity.AdjustedPrice)
46
47if not self.selection_flag:
48    return Universe.Unchanged
49
50selected:List[Fundamental] = sorted([x for x in fundamental if x.HasFundamentalData and \
51    x.SecurityReference.ExchangeId == 'NYS' and x.MarketCap != 0],
52        key=lambda x: x.DollarVolume, reverse=True)[:self.coarse_count]
53
54warm_up_period:int = (self.Time.date() - self.history_start).days
55approach_values:Dict[float, Dict[Symbol, float]] = { fibonacci_level: {} for fibonacci_level in self.fibonacci_levels }
56
57for stock in selected:
58    symbol:Symbol = stock.Symbol
59
60    if symbol not in self.data:
61        self.data[symbol] = SymbolData()
62
63        history:DataFrame = self.History(symbol, warm_up_period, Resolution.Daily)
64        if history.empty:
65            continue
66        
67        closes:pd.Series = history.loc[symbol].close
68
69        for _, close in closes.items():
70            self.data[symbol].update(close)
71
72    if self.data[symbol].ath_atl_ready():
73        for fibonacci_level in self.fibonacci_levels:
74            fib_level_value:float = self.data[symbol].get_fibonacci_level_value(fibonacci_level)
75            approach_value:float = self.data[symbol].get_approach_value(fib_level_value)
76
77            approach_values[fibonacci_level][symbol] = approach_value
78
79if len(list(approach_values.values())[0]) < self.quantile:
80    return Universe.Unchanged
81
82selected_symbols:Set(Symbol) = set()
83
84total_fibonacci_levels:int = len(self.fibonacci_levels)
85for fibonacci_level, approach_value_by_symbol in approach_values.items():
86    quantile:int = int(len(approach_value_by_symbol) / self.quantile)
87    sorted_by_approach:List[Symbol] = [x[0] for x in sorted(approach_value_by_symbol.items(), key=lambda item: abs(item[1]))]
88    lowest_quantile:List[Symbol] = sorted_by_approach[:quantile]
89
90    long:List[Symbol] = list(filter(lambda symbol: approach_value_by_symbol[symbol] > 0, lowest_quantile))
91    short:List[Symbol] = list(filter(lambda symbol: approach_value_by_symbol[symbol] < 0, lowest_quantile))
92
93    if len(long) > 0 and len(short) > 0:
94        for i, portfolio in enumerate([long, short]):
95            w:float = self.Portfolio.TotalPortfolioValue / total_fibonacci_levels / self.total_portfolio_parts / len(portfolio)
96            for symbol in portfolio:
97                selected_symbols.add(symbol)
98                quantity:float = ((-1) ** i) * np.floor(w / self.data[symbol].get_latest_price())
99                self.managed_queue.append([symbol, quantity])
100
101return list(selected_symbols)
102
103def OnData(self, data: Slice) -> None:
104# rebalance monthly
105if not self.selection_flag:
106    return
107self.selection_flag = False
108
109# liquidate prev month trades
110for symbol, quantity in self.prev_managed_queue:
111    if self.Securities[symbol].Invested:
112        self.MarketOrder(symbol, -quantity)
113
114for symbol, quantity in self.managed_queue:
115    if symbol in data and data[symbol]:
116        self.MarketOrder(symbol, quantity)
117
118self.prev_managed_queue = self.managed_queue
119self.managed_queue = []
120
121def Selection(self) -> None:
122self.selection_flag = True