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
|
#!/usr/bin/env python3
from bs4 import BeautifulSoup
from multiprocessing import Pool
import ntpath
import os
import csv
from filing import Filing
# Set directories
root_dir = os.path.dirname(os.path.abspath(__file__))
filings_dir = os.path.join(root_dir, "filings")
clean_filings_dir = os.path.join(root_dir, "cleanedfilings")
print("root_dir:", root_dir)
print("filings_dir:", filings_dir)
print("clean_filings_dir:", clean_filings_dir)
def make_batches(lst, batch_size):
res = []
batch = []
for i in range(len(lst)):
batch.append(lst[i])
if (i + 1) % batch_size == 0:
res.append(batch)
batch = []
if len(batch) > 0:
res.append(batch)
return res
# Read CSV data into a list of dicts
csv_filename = os.path.join(root_dir, "data", "progress.csv")
print("Reading CSV data from ", csv_filename)
csv_data_rows = []
with open(csv_filename, newline="") as csvfile:
try:
companyReader = csv.reader(csvfile, delimiter=";")
for row in companyReader:
csv_data_rows.append({
"ticker": row[0],
"cik": row[1],
})
except Exception as ex:
print("Reading CSV failed!")
raise ex
# Construct a list of data_objects (dicts) that contain all necessary info
# for processing.
data_object_lst = []
print("Writing filings to disk and building data objects")
for row in csv_data_rows:
ticker = row["ticker"]
cik = row["cik"]
filings = Filing(cik=cik, filing_type="10-k", count=45)
input_file_dir = os.path.join(filings_dir, cik, "10-k")
input_filenames = []
for r, d, f in os.walk(input_file_dir):
for file_name in f:
input_filenames.append(os.path.join(r, file_name))
for input_filename in input_filenames:
output_filename = os.path.join(
clean_filings_dir, ticker, ntpath.basename(input_filename))
data_object = {
"ticker": ticker,
"cik": cik,
"filings_obj": filings,
"input_filename": input_filename,
"output_filename": output_filename,
"input_string": None,
"output_string": None,
}
data_object_lst.append(data_object)
# Multiprocessing worker function
# Stores result in data objects
def process(data_object):
input_string = data_object["input_string"]
output_string = BeautifulSoup(input_string, "lxml").text
data_object["output_string"] = output_string
return data_object
def process_data_object_batch(batch):
# Fetch filings over HTTP
for data_object in batch:
filings = data_object["filings_obj"]
filings.save(filings_dir)
del data_object["filings_obj"]
# Read files and store content in data objects
print("Reading filing data into memory")
for data_object in batch:
input_filename = data_object["input_filename"]
with open(input_filename, "r+") as rawFile:
input_file_string = rawFile.read()
data_object["input_string"] = input_file_string
print("Parsing filing data with lxml")
# Multiprocessing, 4 cores
with Pool(4) as pool:
result_batch = pool.map(process, batch)
# No multiprocessing, does the same thing
# result_batch = map(process, batch)
# Write output to disk from data objects
print("Writing filing data to disk")
for data_object in result_batch:
output_filename = data_object["output_filename"]
output_dir = os.path.dirname(output_filename)
os.makedirs(output_dir, exist_ok=True)
with open(output_filename, "w") as newFile:
output_string = data_object["output_string"]
newFile.write(output_string)
for data_object in result_batch:
data_object.clear() # Clean up memory
batches = make_batches(data_object_lst, 5)
for i, batch in enumerate(batches):
print("=== Processing batch #{}".format(i))
process_data_object_batch(batch)
print("Done.")
|
