日内配对交易策略

1import itertools as it
2from AlgorithmImports import *
3from typing import List, Dict
4import numpy as np
5class PairsTradingIntradayBasis(QCAlgorithm):
6
7def Initialize(self) -> None:
8self.SetStartDate(2010, 1, 1)
9self.SetCash(100000)
10   
11self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
12# Daily price data.
13self.history_price:Dict[Symbol, RollingWindow] = {}
14self.formation_period:int = 264
15self.trading_period:int = 132
16self.period:int = 22
17self.leverage:int = 5
18self.min_share_price:int = 5
19
20self.overlapping_portfolios_count:int = 6  # 132 / 22 = 6
21self.overlapping_portfolios:List[Symbol] = []
22
23# Equally weighted brackets.
24self.max_traded_pairs:int = 5
25
26self.fundamental_sorting_key = lambda x: x.DollarVolume
27self.fundamental_count:int = 100
28self.selection_hours:int = 1
29self.selection_flag:bool = True
30self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
31self.settings.daily_precise_end_time = False
32self.UniverseSettings.Resolution = Resolution.Hour
33self.AddUniverse(self.FundamentalSelectionFunction)
34def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
35for security in changes.AddedSecurities:
36    security.SetFeeModel(CustomFeeModel())
37    security.SetLeverage(self.leverage)
38    
39for security in changes.RemovedSecurities:
40    symbol:Symbol = security.Symbol
41    if symbol in self.history_price:
42        del self.history_price[symbol]
43    
44symbols:List[Symbol] = [x for x in self.history_price.keys() if x != self.symbol]
45self.symbol_pairs:List[Tuple[Symbol]] = list(it.combinations(symbols, 2))
46
47# minimize the sum of squared deviations
48distances:Dict[Tuple[Symbol], float] = {}
49for pair in self.symbol_pairs:
50    if self.history_price[pair[0]].IsReady and self.history_price[pair[1]].IsReady:
51        distances[pair] = self.Distance(self.history_price[pair[0]], self.history_price[pair[1]])
52
53new_portfolio = TradedPortfolio(self)
54
55# sorting 5 pairs with no assets repetition
56if len(distances) != 0:
57    sorted_pairs:List[Tuple[Tuple[Symbol], float]] = sorted(distances.items(), key = lambda x: x[1])
58    sorted_pairs:List[Tuple[Symbol]] = [x[0] for x in sorted_pairs]
59    
60    # self.sorted_pairs.append(sorted_pairs[0])
61    new_portfolio.add_pair(sorted_pairs[0])
62    
63    selected_assets:List[List[Symbol]] = [sorted_pairs[0][0], sorted_pairs[0][1]]
64    
65    for pair in sorted_pairs:
66        symbol1:Symbol = pair[0]
67        symbol2:Symbol = pair[1]
68        
69        # symbols which has not been selected yet
70        if symbol1 not in selected_assets and \
71            symbol2 not in selected_assets:
72                # self.sorted_pairs.append(pair)
73                new_portfolio.add_pair(pair)
74                selected_assets.extend([symbol1, symbol2])
75                
76                if len(new_portfolio.sorted_pairs) == self.max_traded_pairs:
77                    break
78
79# liquidate firstly added portfolio
80if len(self.overlapping_portfolios) == self.overlapping_portfolios_count:
81    item_to_remove:Symbol = self.overlapping_portfolios[0]
82    result = item_to_remove.liquidate()
83    if not result:
84        self.Log('Nothing to liquidate or liquidation error')
85    self.overlapping_portfolios.remove(item_to_remove)
86# append new portfolio
87self.overlapping_portfolios.append(new_portfolio)
88
89def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
90if not self.selection_flag:
91    return Universe.Unchanged
92self.selection_flag = False
93
94selected:List[Fundamental] = [
95    x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa'
96]
97    
98if len(selected) > self.fundamental_count:
99    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
100
101# Warmup price rolling windows.
102for stock in selected:
103    symbol:Symbol = stock.Symbol
104    
105    if symbol in self.history_price:
106        continue
107    
108    self.history_price[symbol] = RollingWindow[float](self.formation_period)
109    history:DataFrame = self.History(symbol, self.formation_period, Resolution.Hour)
110    if history.empty:
111        self.Log(f"Not enough data for {symbol} yet")
112        continue
113    closes:Series = history.loc[symbol].close
114    for time, close in closes.items():
115        self.history_price[symbol].Add(close)
116        
117return [x.Symbol for x in selected if self.history_price[x.Symbol].IsReady]
118
119def OnData(self, data: Slice) -> None:
120for symbol in self.history_price:
121    symbol_obj:Symbol = self.Symbol(symbol)
122    if symbol_obj in data and data[symbol_obj]:
123        price:float = data[symbol_obj].Value
124        self.history_price[symbol].Add(price)
125                
126if self.selection_hours == self.period:
127    self.selection_flag = True
128    self.selection_hours = 1
129else:
130    self.selection_hours += 1
131
132# manage overlapping portfolios
133for portfolio in self.overlapping_portfolios:
134    pairs_to_remove:List[Tuple[Symbol]] = []
135    
136    for pair in portfolio.sorted_pairs:
137        if pair[0] not in self.history_price or pair[1] not in self.history_price:
138            continue
139        
140        if not self.history_price[pair[0]].IsReady or not self.history_price[pair[1]].IsReady:
141            continue
142        
143        # Calculate the spread of two price series.
144        price_a:List[float] = [x for x in self.history_price[pair[0]]]
145        price_b:List[float] = [x for x in self.history_price[pair[1]]]
146        
147        norm_a:np.ndarray = np.array(price_a) / price_a[-1]
148        norm_b:np.ndarray = np.array(price_b) / price_b[-1]
149        
150        spread:np.ndarray = norm_a - norm_b
151        mean:float = np.mean(spread)
152        std:float = np.std(spread)
153        actual_spread:float = spread[0]
154        
155        # Long-short position is opened when pair prices have diverged by two standard deviations.
156        traded_portfolio_value = self.Portfolio.TotalPortfolioValue / self.overlapping_portfolios_count / self.max_traded_pairs
157        if actual_spread > mean + 2*std or actual_spread < mean - 2*std:
158            if pair not in portfolio.traded_pairs:
159                # open new position for pair, if there's place for it.
160                if len(portfolio.traded_pairs) < self.max_traded_pairs:
161                    symbol_a:Symbol = pair[0]
162                    symbol_b:Symbol = pair[1]
163                    a_price_norm:float = norm_a[0]
164                    b_price_norm:float = norm_b[0]
165                    a_price:float = price_a[0]
166                    b_price:float = price_b[0]
167                    
168                    if symbol_a in data and data[symbol_a] and symbol_b in data and data[symbol_b]: 
169                        # a stock's price > b stock's price
170                        if a_price_norm > b_price_norm:
171                            long_q:float = traded_portfolio_value / b_price    # long b stock
172                            short_q:float = -traded_portfolio_value / a_price  # short a stock
173                            if self.Securities.ContainsKey(symbol_a) and self.Securities.ContainsKey(symbol_b) and \
174                                self.Securities[symbol_a].Price != 0 and self.Securities[symbol_a].IsTradable and \
175                                self.Securities[symbol_b].Price != 0 and self.Securities[symbol_b].IsTradable:
176                                self.MarketOrder(symbol_a, short_q)
177                                self.MarketOrder(symbol_b, long_q)
178                                
179                                portfolio.traded_quantity[pair] = (short_q, long_q)
180                                portfolio.traded_pairs.append(pair)
181                        # b stock's price > a stock's price
182                        else:
183                            long_q = traded_portfolio_value / a_price
184                            short_q = -traded_portfolio_value / b_price
185                            if self.Securities.ContainsKey(symbol_a) and self.Securities.ContainsKey(symbol_b) and \
186                                self.Securities[symbol_a].Price != 0 and self.Securities[symbol_a].IsTradable and \
187                                self.Securities[symbol_b].Price != 0 and self.Securities[symbol_b].IsTradable:
188                                self.MarketOrder(symbol_a, long_q)
189                                self.MarketOrder(symbol_b, short_q)
190
191                                portfolio.traded_quantity[pair] = (long_q, short_q)
192                                portfolio.traded_pairs.append(pair)
193        # The position is closed when prices revert back.
194        else:
195            portfolio.liquidate_pair(pair)
196def Distance(self, price_a:List[float], price_b:List[float]):
197# Calculate the sum of squared deviations between two normalized price series.
198price_a:List[float] = [x for x in price_a]
199price_b:List[float] = [x for x in price_b]
200
201norm_a:np.ndarray = np.array(price_a) / price_a[-1]
202norm_b:np.ndarray = np.array(price_b) / price_b[-1]
203return sum((norm_a - norm_b)**2)
204
205# Custom fee model.
206class CustomFeeModel(FeeModel):
207def GetOrderFee(self, parameters):
208fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
209return OrderFee(CashAmount(fee, "USD"))
210class TradedPortfolio():
211def __init__(self, algorithm):
212self.traded_pairs = []
213self.traded_quantity = {}
214self.sorted_pairs = []
215
216self.algorithm = algorithm
217
218def add_pair(self, pair) -> None:
219self.sorted_pairs.append(pair)
220
221def liquidate(self) -> bool:
222result = False
223for pair in self.traded_pairs:
224    if pair in self.traded_quantity:
225        self.algorithm.MarketOrder(pair[0], -self.traded_quantity[pair][0])
226        self.algorithm.MarketOrder(pair[1], -self.traded_quantity[pair][1])
227        result = True
228return result
229def liquidate_pair(self, pair) -> None:
230if pair in self.traded_pairs and pair in self.traded_quantity:
231    self.algorithm.MarketOrder(pair[0], -self.traded_quantity[pair][0])
232    self.algorithm.MarketOrder(pair[1], -self.traded_quantity[pair][1])
233    self.traded_pairs.remove(pair)
234    del self.traded_quantity[pair]