基于有限投资者关注度的谷歌搜索策略

1from AlgorithmImports import *
2import numpy as np
3from typing import List, Dict
4#endregion
5class GoogleSearchStrategyBasedOnLimitedInvestorAttention(QCAlgorithm):
6def Initialize(self) -> None:
7 self.SetStartDate(2004, 1, 1)               # Google search data are since 2004
8 self.SetCash(100000)
9 
10 self.data: Dict[Symbol, List[float]] = {}   # Storing search values about each stock in list
11 self.selected: Dict[str, Symbol] = {}       # Storing stocks, which will be traded
12 self.tickers: List[str] = []                # Storing S&P100 tickers
13 self.symbols: List[Symbol] = []             # Storing symbols, to get search values about stocks
14 self.last_ASV: Dict[Symbol, float] = {}     # Storing last Abnormal Search Volume for each stock
15 
16 self.period: int = 8                        # 8 months of search values
17 self.leverage: int = 5
18 
19 # Load csv with S&P100 tickers
20 csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/sp100.csv')
21 line: List[str] = csv_string_file.split('\r\n')
22 line_split: List[str] = line[0].split(';')
23 
24 for ticker in line_split:
25     self.tickers.append(ticker)
26     
27     ticker: str = ticker
28     # Subscribe to QuantpediaGoogleSearch with csv name
29     symbol: Symbol = self.AddData(QuantpediaGoogleSearch, ticker, Resolution.Daily).Symbol
30     
31     # Add subscribed symbol to self.symbols
32     self.symbols.append(symbol)
33     # Create list for each subcribed symbol
34     self.data[symbol] = []
35 
36 symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
37 
38 self.selection_flag: bool = False
39 self.settings.daily_precise_end_time = False
40 self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
41 self.UniverseSettings.Resolution = Resolution.Daily
42 self.AddUniverse(self.FundamentalSelectionFunction)
43 self.Schedule.On(self.DateRules.MonthStart(symbol), self.TimeRules.AfterMarketOpen(symbol), self.Selection)
44 
45def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
46 for security in changes.AddedSecurities:
47     security.SetFeeModel(CustomFeeModel())
48     security.SetLeverage(self.leverage)
49 
50def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
51 # Rebalance monthly
52 if not self.selection_flag:
53     return Universe.Unchanged
54         
55 # Select S&P 100 stocks
56 self.selected = {x.Symbol.Value: x.Symbol for x in fundamental if x.Symbol.Value in self.tickers}
57 return list(self.selected.values())
58def OnData(self, data: Slice) -> None:
59 # Store search data for each S&P100 stock
60 for symbol in self.symbols:
61     if symbol in data and data[symbol]:
62         value: float = data[symbol].Value
63         self.data[symbol].append(value)
64 
65 # Rebalance monthly
66 if not self.selection_flag:
67     return
68 self.selection_flag = False
69 
70 long: List[Symbol] = []
71 
72 GSV_symbols_last_update_date: datetime.date = QuantpediaGoogleSearch.get_last_update_date()
73 for symbol in self.symbols:
74     if self.Securities[symbol].GetLastData() and self.Time.date() <= GSV_symbols_last_update_date[symbol]:
75         if len(self.data[symbol]) > self.period: # Wait until we have more than self.period months of data
76             search_volumes: List[int] = self.data[symbol]
77             # Get largest search value
78             max_value: int = max(search_volumes)
79             
80             current_asv: float = 0.
81             
82             if max_value > 0:
83                 # Calculate last search volume
84                 last_search_volume: int = search_volumes[-1] / max_value
85                 
86                 # Get and calculate last self.period search volumes
87                 last_n_search_volume: int = np.median([x / max_value for x in search_volumes[-self.period + 1:][:-1]])
88                 
89                 # NOTE: We can't work with infinite numbers,
90                 # which will be created by log of last_search_volume and log of last_n_search_volume
91                 if last_search_volume != 0 and last_n_search_volume != 0:
92                     
93                     # Need to make log from last_search_volume and last_n_search_volume for accurate mathematical form
94                     last_search_volume: float = np.log(last_search_volume)
95                     last_n_search_volume: float = np.log(last_n_search_volume)
96                 
97                     # In mathematical form, it is: ASVt = ln [GSVt] – ln [Med (GSVt-1, …,GSVt-8)]
98                     current_asv: float = last_search_volume - last_n_search_volume
99             
100             # Check if symbol has last ASV
101             if symbol in self.last_ASV:
102                 # Go long if last week’s attention for the stock was abnormally low or remains unchanged, i.e. ASVt-1 ≤ 0.
103                 if current_asv == self.last_ASV[symbol] or self.last_ASV[symbol] <= 0:
104                     long.append(symbol)
105             
106             # Change last ASV           
107             self.last_ASV[symbol] = current_asv
108             
109 # If long list is empty liquidate all stocks
110 if len(long) == 0:
111     self.Liquidate()
112     return
113 # Trade execution.
114 targets: List[PortfolioTarget] = []
115 for symbol in long:
116     if symbol.Value in self.selected:
117         # Get QC symbol for trade
118         symbol: Symbol = self.selected[symbol.Value]
119         if symbol in data and data[symbol]:
120             targets.append(PortfolioTarget(symbol, 1 / len(long)))
121 
122 self.SetHoldings(targets, True)
123def Selection(self) -> None:
124 self.selection_flag = True
125# Quantpedia data.
126# NOTE: IMPORTANT: Data order must be ascending (datewise)
127# NOTE: IMPORTANT: Name of the csv file has to be in upper case
128class QuantpediaGoogleSearch(PythonData):
129_last_update_date:Dict[Symbol, datetime.date] = {}
130@staticmethod
131def get_last_update_date() -> Dict[Symbol, datetime.date]:
132return QuantpediaGoogleSearch._last_update_date
133def GetSource(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
134 return SubscriptionDataSource("data.quantpedia.com/backtesting_data/google_search/{0}_STOCK.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
135 
136def Reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData:
137 data = QuantpediaGoogleSearch()
138 data.Symbol = config.Symbol
139 
140 if not line[0].isdigit(): return None
141 split: List[str] = line.split(';')
142 
143 data.Time = datetime.strptime(split[0], "%d.%m.%Y") + timedelta(days=1)
144 data['value'] = float(split[1])
145 data.Value = float(split[1])
146 if config.Symbol not in QuantpediaGoogleSearch._last_update_date:
147     QuantpediaGoogleSearch._last_update_date[config.Symbol] = datetime(1,1,1).date()
148 if data.Time.date() > QuantpediaGoogleSearch._last_update_date[config.Symbol]:
149     QuantpediaGoogleSearch._last_update_date[config.Symbol] = data.Time.date()
150 return data
151# Custom fee model
152class CustomFeeModel(FeeModel):
153def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
154 fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
155 return OrderFee(CashAmount(fee, "USD"))