-
Notifications
You must be signed in to change notification settings - Fork 0
/
scrapper_app.py
112 lines (87 loc) · 4.82 KB
/
scrapper_app.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
from scrappers.core.yf_statement import YFStatement
from scrappers.core.yf_summary import YFSummary
from scrappers.core.yf_statistics import YFStatistics
from scrappers.core.analysis.growth_analysis import CAGR
from scrappers.utils.multi_threader import MultiThreader
from scrappers.utils.config_reader import ConfigReader
from scrappers.utils.data_writer import DataWriter
from scrappers.utils.logger import Logger
from scrappers.core.analysis.proportion import Proportion
from collections import OrderedDict
from time import strftime
from pandas import DataFrame
import sys
from pdb import set_trace
class ScrapperApp:
"""
Central control of Scrapper application
:param args: list[String] => list of ticker names
:param store_location: String => root directory on hard drive to save output
:param folder_name: String => folder name to be created in the directory of store_location
:param file_save: Boolean => whether to save the output
:param comprehensive: Boolean => whether to perform statement analysis
"""
def __init__(self, tickers, store_location, file_save, comprehensive):
self.tickers = tickers
self.store_location = store_location
self.file_save = file_save
self.comprehensive = comprehensive
# #self.start_time =
self.logger = Logger("global").create_logger()
self.config = ConfigReader("application_configurations.json").get_configurations()
@staticmethod
def parallel_runner(class__):
class__.run()
def scrapper_start(self):
implied_peg = OrderedDict()
statistics = YFStatistics(self.tickers, self.store_location, self.config.get("statistics").get("folder_name"), self.file_save)
summary = YFSummary(self.tickers, self.store_location, self.config.get("summary").get("folder_name"), self.file_save)
MultiThreader.run_thread_pool([statistics, summary], self.parallel_runner, 2)
ranking = statistics.get_ranking()
score = statistics.get_score()
ignored_stats = statistics.get_ignored_stats()
summary_data = summary.get_summary()
trailing_pe_list = statistics.get_downsized_df().loc["Trailing P/E"]
pb_list = statistics.get_downsized_df().loc["Price/Book"]
if self.comprehensive:
statement = YFStatement(self.tickers, self.store_location, self.config.get("statement").get("IS").get("folder_name"), self.file_save, statement_type="IS")
statement.run()
cagr, cagr_compare = CAGR(statements=statement.get_statement("growth"), statement_type="IS").run()
statements = statement.get_statement("raw")
#Proportion(statements, "IS").run()
for ticker in self.tickers:
if trailing_pe_list[ticker] == "N/A" or cagr.get(ticker) == "N/A":
implied_peg[ticker] = 0
else:
implied_peg[ticker] = float(trailing_pe_list[ticker]) / cagr.get(ticker) / 100
if self.file_save:
DataWriter().csv_writer(self.store_location, "", "CAGR_COMPARE.csv", cagr_compare)
else:
for ticker in ranking.columns:
implied_peg[ticker] = 0
print("$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$\n")
summary_dict = OrderedDict()
summary_columns = ["Score", "Trailing PE", "Price/Book", "Implied PEG", "Potential", "Price Percentile"]
for ticker, value in zip(score.keys(), score.values()):
ticker_data = [round(value, 2), trailing_pe_list[ticker], pb_list[ticker],
implied_peg[ticker], str(summary_data.loc["Growth Potential", ticker]),
str(summary_data.loc["52 Week Percentile", ticker])]
summary_dict[ticker] = ticker_data
summary_df = DataFrame(summary_dict, index=summary_columns).transpose()
print(summary_df)
if ignored_stats:
print("ignored items:")
for column in ignored_stats.keys():
print("{}: {}".format(column, ignored_stats.get(column)))
print("\n$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$")
if self.file_save:
DataWriter().csv_writer(self.store_location, "", "SCORE_COMPARE.csv", summary_df)
if __name__ == "__main__":
user_input = input("Please select the ticker you wish you analyze: ")
user_input = user_input.replace(' ', '').split(sep=",")
config = ConfigReader(file="application_configurations.json").get_configurations()
general_conf = config.get("general")
if user_input == [""]:
user_input = general_conf.get("symbols")["industry"]
ScrapperApp(user_input, store_location=general_conf.get("store_location"),
file_save=general_conf.get("file_save"), comprehensive=False).scrapper_start()