-
Notifications
You must be signed in to change notification settings - Fork 9
/
holdings_dl.py
212 lines (196 loc) · 9.55 KB
/
holdings_dl.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
#!/usr/bin/env python3
import argparse
import math
import time
import pandas as pd
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as ec
from selenium.webdriver.firefox.options import Options
from selenium.webdriver.support.ui import WebDriverWait
'''
File name: holdings_dl.py
Author: Piper Batey
Date created: 7/13/2021
Date last modified: 9/3/2021
Python Version: 3.8
Description: A simple Python script that downloads
the holdings of one or more ETFs into .csv files.
'''
class HoldingsDownloader:
def __init__(self):
# variables
self.firefox_options = Options() # default: headless
self.etf_symbols = []
self.valid_etfs = []
self.log_entries = []
self.file_name = ""
self.num_files = 0
self.wait_time = 15
self.log_mode = False
self.quiet_mode = False
self.sort_mode = False
self.raw_mode = False
# init
self._parse_command_args()
if self.file_name:
self._read_input_file()
if self.sort_mode:
self.etf_symbols.sort()
def _parse_command_args(self):
parser = argparse.ArgumentParser(
formatter_class=lambda prog: argparse.HelpFormatter(prog, max_help_position=27)) # type error is ok
title_group = parser.add_argument_group(title="required arguments") # required arguments header in output
input_type_group = title_group.add_mutually_exclusive_group(required=True)
input_type_group.add_argument("--symbol", nargs='+', metavar="SYM", help="specify one or more ETF symbols")
input_type_group.add_argument("--file", metavar="FILE", help="specify a file containing a list of ETF symbols")
parser.add_argument("-r", "--raw", action="store_true", help="save raw data without symbols or units")
parser.add_argument("-l", "--log", action="store_true",
help="create a log of the downloaded ETFs in etf-log.csv")
parser.add_argument("-a", "--alpha", action="store_true",
help="sort ETF symbols into alphabetical order for output")
parser.add_argument("-w", "--window", action="store_false",
help="run web driver with firefox window visible")
parser.add_argument("-q", "--quiet", action="store_true", help="suppress verbose terminal output")
parser.add_argument("-t", "--time", default=15, type=int,
help="set the maximum time in seconds the program will wait for web pages to load "
"(default: 15)")
args = parser.parse_args()
self.firefox_options.headless = args.window
self.quiet_mode = args.quiet
self.log_mode = args.log
self.sort_mode = args.alpha
self.wait_time = args.time
self.raw_mode = args.raw
if args.file:
self.file_name = args.file
if args.symbol:
self.etf_symbols = args.symbol
def _read_input_file(self):
if not self.quiet_mode:
print("Reading symbols from {} ...".format(self.file_name), end=" ")
with open(self.file_name, 'r') as input_file:
for name in input_file:
self.etf_symbols.append(name[:-1]) # avoid the newline at the end
if not self.quiet_mode:
print("complete")
def _convert_units_to_float(self, x): # gets raw data for dataframe .apply()
start = 0
if type(x) == float:
return x
if x[0] == '-': # set negative portfolio weights to 0
return 0
if x[0] == '$':
start = 1
if x[-1] == '%':
return float(x[:-1]) / 100
if x[-1] == 'K':
return float(x[start:-1]) * 10e3
if x[-1] == 'M':
return float(x[start:-1]) * 10e6
if x[-1] == 'B':
return float(x[start:-1]) * 10e9
else:
return float(x[start:])
def _get_etf_from_schwab(self, etf_symbol):
if not self.quiet_mode:
print("Opening {} database".format(etf_symbol))
driver = webdriver.Firefox(options=self.firefox_options)
driver.implicitly_wait(self.wait_time)
wait = WebDriverWait(driver, 30, poll_frequency=1)
url = "https://www.schwab.wallst.com/schwab/Prospect/research/etfs/schwabETF" \
"/index.asp?type=holdings&symbol={}".format(etf_symbol)
try:
driver.get(url)
show_sixty_items = driver.find_element_by_xpath("//a[@perpage='60']")
show_sixty_items.click()
except ec.NoSuchElementException:
if not self.quiet_mode:
print("{} is not a valid ETF (not found in schwab database)\n".format(etf_symbol))
driver.close()
return False
except ec.WebDriverException:
if not self.quiet_mode:
print("{} not retrieved (web driver error)\n".format(etf_symbol))
driver.close()
return False
page_elt = wait.until(ec.visibility_of_element_located((By.CLASS_NAME, "paginationContainer"))) # REQUIRED!!!
try:
num_pages = math.ceil(float(page_elt.text.split(" ")[4]) / 60)
except ec.StaleElementReferenceException: # fixes stale reference bug
page_elt = wait.until(ec.visibility_of_element_located((By.CLASS_NAME, "paginationContainer")))
num_pages = math.ceil(float(page_elt.text.split(" ")[4]) / 60)
# Read holdings data
if not self.quiet_mode:
print("{}: page 1 of {} ...".format(etf_symbol, num_pages), end=" ")
time.sleep(.5) # force wait needed for reliability
dataframe_list = [pd.read_html(driver.page_source)[1]]
if not self.quiet_mode:
print("complete")
current_page = 2
while current_page <= num_pages:
if not self.quiet_mode:
print("{}: page {} of {} ...".format(etf_symbol, current_page, num_pages), end=" ")
driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
next_button = driver.find_element_by_xpath("//li[@pagenumber='{}']".format(current_page))
driver.execute_script("arguments[0].click();", next_button)
while True: # wait until the new data has loaded (read_html() is from pandas so can't use selenium wait)
time.sleep(.25)
df = pd.read_html(driver.page_source, match="Symbol")[0]
if not df.equals(dataframe_list[-1]):
break
dataframe_list.append(df)
current_page += 1
if not self.quiet_mode:
print("complete")
# end while
concat_result = pd.concat(dataframe_list) # merge into a single dataframe
result_df = concat_result.drop_duplicates()
result_df.columns = ['Symbol', 'Description', 'Portfolio Weight', 'Shares Held', 'Market Value']
if self.raw_mode: # strip symbols and units
result_df['Portfolio Weight'] = result_df['Portfolio Weight'].apply(self._convert_units_to_float)
result_df['Shares Held'] = result_df['Shares Held'].apply(self._convert_units_to_float)
result_df['Market Value'] = result_df['Market Value'].apply(self._convert_units_to_float)
result_df.to_csv("{}-holdings.csv".format(etf_symbol), index=False) # create the csv
if self.log_mode:
driver.execute_script("window.scrollTo(0, -document.body.scrollHeight);") # info is at top of page
header_elt = driver.find_element_by_xpath("//div[@modulename='FirstGlance']")
header_text = header_elt.text.split("\n")
full_name = header_text[0].split(" {}:".format(etf_symbol))[0].encode("ascii", "ignore").decode()
last_price = header_text[2].split(" ")[0]
self.log_entries.append([etf_symbol, full_name, last_price, result_df.shape[0]])
driver.close()
if not self.quiet_mode:
print("{}: {} holdings retrieved\n".format(etf_symbol, result_df.shape[0]))
return True
# _get_etf_from_schwab()
def run_schwab_download(self):
for symbol in self.etf_symbols:
if symbol in self.valid_etfs: # skip duplicates
continue
if self._get_etf_from_schwab(symbol):
self.num_files += 1
self.valid_etfs.append(symbol)
def generate_log_file(self):
if not self.quiet_mode:
print("Generating log file...", end=' ')
log_dataframe = pd.DataFrame(self.log_entries, columns=['Symbol', 'Name', 'Last Price', 'Number of Holdings'])
log_dataframe.to_csv("etf-log.csv", index=False)
self.num_files += 1
if not self.quiet_mode:
print("complete")
def print_end_summary(self):
print("\n{} file(s) have been generated for {} ETF(s):".format(self.num_files, len(self.valid_etfs)))
if self.log_mode:
print("etf-log.csv")
for symbol in self.valid_etfs:
print("{}-holdings.csv".format(symbol))
def main():
downloader = HoldingsDownloader()
downloader.run_schwab_download()
if downloader.log_mode:
downloader.generate_log_file()
if not downloader.quiet_mode:
downloader.print_end_summary()
if __name__ == "__main__":
main()