隔夜股票交易策略

1import numpy as np
2from AlgorithmImports import *
3from typing import Dict, List
4class OvernightStockTrading(QCAlgorithm):
5def Initialize(self):
6self.SetStartDate(2010, 1, 1)
7self.SetCash(100000)
8self.leverage:int = 10
9self.quantile:int = 10
10self.period:int = 12 * 21
11self.min_share_price:int = 5
12self.symbol:Symbol = self.AddEquity('SPY', Resolution.Minute).Symbol
13
14self.fundamental_sorting_key = lambda x: x.DollarVolume
15self.fundamental_count:int = 100
16
17self.selected:List[Symbol] = [] # symbols of selected stocks from fundamentalSelectionFunction
18
19self.long:List[Symbol] = []
20self.short:List[Symbol] = []
21self.data:Dict[Symbol, SymbolData] = {}
22
23self.months_counter:int = 1
24self.selection_flag:bool = True
25self.UniverseSettings.Resolution = Resolution.Minute
26self.AddUniverse(self.FundamentalSelectionFunction)
27self.settings.daily_precise_end_time = False
28self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
29self.Schedule.On(self.DateRules.EveryDay(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol, 20), self.MarketClose)
30self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol), self.Selection)
31def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
32for security in changes.AddedSecurities:
33    security.SetFeeModel(CustomFeeModel())
34    security.SetLeverage(self.leverage)
35    
36def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
37# updating overnight data and stock price every day
38for stock in fundamental:
39    symbol:Symbol = stock.Symbol
40    
41    if symbol in self.data:
42        # update stock price
43        self.data[symbol].price = stock.AdjustedPrice
44        
45        # get history data
46        history:DataFrame = self.History(symbol, 1, Resolution.Daily)
47        # update overnight return and change prev_close_price
48        self.UpdateOvernightReturns(history, symbol)
49        
50# one year rebalance
51if not self.selection_flag:
52    return Universe.Unchanged
53self.selection_flag = False
54
55# sort stocks by dollar volume
56selected:List[Fundamental] = [
57    x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa'
58]
59
60if len(selected) > self.fundamental_count:
61    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
62
63for stock in selected:
64    symbol:Symbol = stock.Symbol
65    if symbol in self.data:
66        continue
67    
68    # create object of SymbolData class for current stock
69    self.data[symbol] = SymbolData(self.period)
70    
71    # get history data
72    history:DataFrame = self.History(symbol, self.period + 1, Resolution.Daily)
73    # update overnight return and change prev_close_price
74    self.UpdateOvernightReturns(history, symbol)
75    
76# change self.selected list on rebalance
77self.selected = [x.Symbol for x in selected]
78    
79return self.selected
80    
81def MarketClose(self) -> None:
82total_performance:Dict[Symbol, float] = {} # storing total overnight returns performance for self.period overnight returns
83
84# calculate total overnight performances
85for symbol in self.selected:
86    if not self.data[symbol].is_overnight_returns_ready():
87        continue
88    
89    # calculate and store total overnight performance
90    total_performance[symbol] = self.data[symbol].total_overnight_performance()
91    
92if len(total_performance) >= self.quantile:
93    # quantile selection
94    quantile:int = int(len(total_performance) / self.quantile)    
95    sorted_by_total_perf:List[Symbol] = [x[0] for x in sorted(total_performance.items(), key=lambda item: item[1])]
96    
97    # long top quantile stocks and short bottom quantile stocks
98    self.long = sorted_by_total_perf[-quantile:]
99    self.short = sorted_by_total_perf[:quantile]
100    
101    long_length:int = len(self.long)
102    short_length:int = len(self.short)
103    
104    # trade execution
105    for symbol in self.long:
106        current_price = self.data[symbol].price
107        if current_price != 0 and self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable: 
108            quantity = np.floor((self.Portfolio.TotalPortfolioValue / long_length) / current_price)
109            self.MarketOnCloseOrder(symbol, quantity)
110            self.MarketOnOpenOrder(symbol, -quantity)
111        
112    for symbol in self.short:
113        current_price = self.data[symbol].price
114        if current_price != 0 and self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable: 
115            quantity = np.floor((self.Portfolio.TotalPortfolioValue / short_length) / current_price)
116            self.MarketOnCloseOrder(symbol, -quantity)
117            self.MarketOnOpenOrder(symbol, quantity)
118def UpdateOvernightReturns(self, history, symbol: Symbol) -> None:
119''' update overnight returns for specific stock according to history data '''
120
121# check if history isn't empty and history dataframe has required attributes
122if not history.empty and hasattr(history, 'close') and hasattr(history, 'open'):
123    # get open and close prices from dataframe
124    opens:Series = history['open']
125    closes:Series = history['close']
126    
127    # update overnight return 
128    for (_, open_price), (_, close_price) in zip(opens.items(), closes.items()):
129        # update overnight return and change prev_close_price
130        self.data[symbol].update(open_price, close_price)
131
132def Selection(self) -> None:
133if self.months_counter % 12 == 0:
134    self.selection_flag = True
135self.months_counter += 1
136
137class SymbolData():
138def __init__(self, period: int):
139self.overnight_returns:RollingWindow = RollingWindow[float](period)
140self.prev_close_price:Union[None, float] = None
141self.price:int = 0
142
143def update(self, open_price: float, close_price: float) -> None:
144# update overnight returns only if prev_close_price isn't None
145if self.prev_close_price:
146    overnight_return = open_price / self.prev_close_price - 1
147    self.overnight_returns.Add(overnight_return)
148# change previous close price to current close price
149self.prev_close_price = close_price
150
151def total_overnight_performance(self) -> float:
152return sum(list(self.overnight_returns))
153
154def is_overnight_returns_ready(self) -> bool:
155return self.overnight_returns.IsReady
156# Custom fee model
157class CustomFeeModel(FeeModel):
158def GetOrderFee(self, parameters):
159fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
160return OrderFee(CashAmount(fee, "USD"))