-
Notifications
You must be signed in to change notification settings - Fork 4
/
y_toplosers.py
247 lines (208 loc) · 11.5 KB
/
y_toplosers.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
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
#!/usr/bin/python3
#import urllib
#import urllib.request
import requests
from bs4 import BeautifulSoup
import pandas as pd
import numpy as np
import re
import logging
import argparse
import time
import threading
import types
import inspect
from rich import print
# logging setup
logging.basicConfig(level=logging.INFO)
#####################################################
class y_toplosers:
"""Class to extract Top Losers data set from finance.yahoo.com"""
# global accessors
tg_df0 = "" # DataFrame - Full list of top loserers
tg_df1 = "" # DataFrame - Ephemerial list of top 10 loserers. Allways overwritten
tg_df2 = "" # DataFrame - Top 10 ever 10 secs for 60 secs
all_tag_tr = "" # BS4 handle of the <tr> extracted data
rows_extr = 0 # number of rows of data extracted
yti = 0
cycle = 0 # class thread loop counter
def __init__(self, yti):
cmi_debug = __name__+"::"+self.__init__.__name__
logging.info('%s - INSTANTIATE' % cmi_debug )
# init empty DataFrame with present colum names
self.tg_df0 = pd.DataFrame(columns=[ 'Row', 'Symbol', 'Co_name', 'Cur_price', 'Prc_change', 'Pct_change', 'Mkt_cap', 'M_B', 'Time'] )
self.tg_df1 = pd.DataFrame(columns=[ 'ERank', 'Symbol', 'Co_name', 'Cur_price', 'Prc_change', 'Pct_change' 'Mkt_cap', 'M_B', 'Time'] )
self.tg_df2 = pd.DataFrame(columns=[ 'ERank', 'Symbol', 'Co_name', 'Cur_price', 'Prc_change', 'Pct_change' 'Mkt_cap', 'M_B', 'Time'] )
self.yti = yti
return
# method #1
def get_topg_data(self):
"""Connect to finance.yahoo.com and extract (scrape) the raw sring data out of"""
"""the embedded webpage [Stock:Top loserers] html data table. Returns a BS4 handle."""
cmi_debug = __name__+"::"+self.get_topg_data.__name__+".#"+str(self.yti)
logging.info('%s - IN' % cmi_debug )
r = requests.get("https://finance.yahoo.com/losers" )
logging.info('%s - read html stream' % cmi_debug )
self.soup = BeautifulSoup(r.text, 'html.parser')
# ATTR style search. Results -> Dict
# <tr tag in target merkup line has a very complex 'class=' but the attributes are unique. e.g. 'simpTblRow' is just one unique attribute
logging.info('%s - save data handle' % cmi_debug )
self.tag_tbody = self.soup.find('tbody')
self.all_tag_tr = self.soup.find_all(attrs={"class": "simpTblRow"})
# target markup line I am scanning looks like this...
# soup.find_all( "tr", class_="simpTblRow Bgc($extraLightBlue):h BdB Bdbc($finLightGrayAlt) Bdbc($tableBorderBlue):h H(32px) Bgc($extraLightBlue)" )
# Example CSS Selector
#all_tag_tr1 = soup.select( "tr.simpTblRow.Bgc" )
logging.info('%s - close url handle' % cmi_debug )
return
# method #2
def build_tg_df0(self):
"""Build-out a fully populated Pandas DataFrame containg all the"""
"""extracted/scraped fields from the html/markup table data"""
"""Wrangle, clean/convert/format the data correctly."""
cmi_debug = __name__+"::"+self.build_tg_df0.__name__+".#"+str(self.yti)
logging.info('%s - IN' % cmi_debug )
time_now = time.strftime("%H:%M:%S", time.localtime() )
logging.info('%s - Drop all rows from DF0' % cmi_debug )
#self.tg_df0.drop(self.tg_df0.index, inplace=True) # the df is now 100% empty
self.tg_df0 = pd.DataFrame() # new df, but is NULLed. avoids empty concat errors
x = 1 # row counter Also leveraged for unique dataframe key
self.rows_extr = int( len(self.tag_tbody.find_all('tr')) )
for datarow in self.all_tag_tr:
"""
# >>>DEBUG<< for when yahoo.com changes data model...
y = 1
for i in j.find_all('td'):
print ( f"Data {y}: {i.text}" )
logging.info( f'%s - Data: {j.td.strings}' % cmi_debug )
y += 1
print ( f"==============================================" )
# >>>DEBUG<< for when yahoo.com changes data model...
"""
################################ 1 ####################################
# 1st <td> cell : ticker symbol info & has comment of company name
# 2nd <td> cell : company name
# 3rd <td> cell : price
# 4th <td> cell : $ change
# 5th <td> cell : % change
# more cells in <tr> data row...but I'm not interested in them at moment.
# BS4 generator object comes from "extracted strings" BS4 operation (nice)
################################ 3 ####################################
extr_strings = datarow.stripped_strings
co_sym = next(extr_strings)
co_name = next(extr_strings)
price = next(extr_strings)
change = next(extr_strings)
pct = next(extr_strings)
vol = next(extr_strings)
avg_vol = next(extr_strings)
mktcap = next(extr_strings)
co_sym_lj = np.array2string(np.char.ljust(co_sym, 6) ) # left justify TXT in DF & convert to raw string
co_name_lj = (re.sub('[\'\"]', '', co_name) ) # remove " ' and strip leading/trailing spaces
co_name_lj = np.array2string(np.char.ljust(co_name_lj, 25) ) # left justify TXT in DF & convert to raw string
co_name_lj = (re.sub('[\']', '', co_name_lj) ) # remove " ' and strip leading/trailing spaces
price_cl = (re.sub('\,', '', price)) # remove ,
price_clean = float(price_cl)
################################ 2 ####################################
TRILLIONS = re.search('T', mktcap)
BILLIONS = re.search('B', mktcap)
MILLIONS = re.search('M', mktcap)
if TRILLIONS:
mktcap_clean = float(re.sub('T', '', mktcap))
mb = "ST"
logging.info( f'%s - {x} / {co_sym_lj} Mkt Cap: TRILLIONS - ST' % cmi_debug )
if BILLIONS:
mktcap_clean = float(re.sub('B', '', mktcap))
mb = "SB"
logging.info( f'%s - {x} / {co_sym_lj} Mkt cap: BILLIONS - SB' % cmi_debug )
if MILLIONS:
mktcap_clean = float(re.sub('M', '', mktcap))
mb = "SM"
logging.info( f'%s - {x} / {co_sym_lj} Mkt cap: MILLIONS - SM' % cmi_debug )
if not TRILLIONS and not BILLIONS and not MILLIONS:
mktcap_clean = 0 # error condition - possible bad data
mb = "SZ" # Zillions !!
logging.info( f'%s - {x} / {co_sym_lj} bad mktcap data N/A - SZ' % cmi_debug )
# handle bad data in mktcap html page field
if pct == "N/A": # Bad data. FOund a filed with N/A instead of read num
pct = "1.0"
pct = float(re.sub('[\-+,%]', '', pct))
change = float(re.sub('[\-+,%]', '', change))
#float(re.sub('[\+,]', '', change)), \
################################ 5 ####################################
# now construct our list for concatinating to the dataframe
logging.info( f"%s ============= Data prepared for DF =============" % cmi_debug )
self.list_data = [[ \
x, \
re.sub('\'', '', co_sym_lj), \
co_name_lj, \
price_clean, \
change, \
pct, \
mktcap_clean, \
mb, \
time_now ]]
self.df_1_row = pd.DataFrame(self.list_data, columns=[ 'Row', 'Symbol', 'Co_name', 'Cur_price', 'Prc_change', 'Pct_change', 'Mkt_cap', 'M_B', 'Time' ], index=[x] )
self.tg_df0 = pd.concat([self.tg_df0, self.df_1_row])
x+=1
logging.info('%s - populated new DF0 dataset' % cmi_debug )
return x # number of rows inserted into DataFrame (0 = some kind of #FAIL)
# sucess = lobal class accessor (y_toplosers.tg_df0) populated & updated
####################################################################################
# method #3
# Hacking function - keep me arround for a while
def prog_bar(self, x, y):
"""simple progress dialogue function"""
if x % y == 0:
print ( " " )
else:
print ( ".", end="" )
return
####################################################################################
# method #4
def topg_listall(self):
"""Print the full DataFrame table list of Yahoo Finance Top loserers"""
"""Sorted by % Change"""
cmi_debug = __name__+"::"+self.topg_listall.__name__+".#"+str(self.yti)
logging.info('%s - IN' % cmi_debug )
pd.set_option('display.max_rows', None)
pd.set_option('max_colwidth', 30)
print ( self.tg_df0.sort_values(by='Pct_change', ascending=False ) ) # only do after fixtures datascience dataframe has been built
return
####################################################################################
# method #5
def build_top10(self):
"""Get top 10 loserers from main DF (df0) -> temp DF (df1)"""
"""Number of rows to grab is now set from num of rows that BS4 actually extracted (rows_extr)"""
"""df1 is ephemerial. Is allways overwritten on each run"""
cmi_debug = __name__+"::"+self.build_top10.__name__+".#"+str(self.yti)
logging.info('%s - IN' % cmi_debug )
logging.info('%s - Drop all rows from DF1' % cmi_debug )
self.tg_df1.drop(self.tg_df1.index, inplace=True)
logging.info('%s - Copy DF0 -> ephemerial DF1' % cmi_debug )
#self.tg_df1 = self.tg_df0.sort_values(by='Pct_change', ascending=False ).copy(deep=True) # create new DF via copy of top 10 entries
self.tg_df1 = self.tg_df0.sort_values(by='Pct_change', ascending=False ).head(self.rows_extr).copy(deep=True) # create new DF via copy of top 10 entries
self.tg_df1.rename(columns = {'Row':'ERank'}, inplace = True) # Rank is more accurate for this Ephemerial DF
self.tg_df1.reset_index(inplace=True, drop=True) # reset index each time so its guaranteed sequential
return
####################################################################################
# method #7
def print_top10(self):
"""Prints the Top 10 Dataframe"""
"""Number of rows to grab is now set from num of rows that BS4 actually extracted (rows_extr)"""
cmi_debug = __name__+"::"+self.print_top10.__name__+".#"+str(self.yti)
logging.info('%s - IN' % cmi_debug )
pd.set_option('display.max_rows', None)
pd.set_option('max_colwidth', 30)
print ( self.tg_df1.sort_values(by='Pct_change', ascending=False ).head(self.rows_extr) )
return
####################################################################################
# method #6
def build_tenten60(self, cycle):
"""Build-up 10x10x060 historical DataFrame (df2) from source df1"""
"""Generally called on some kind of cycle"""
cmi_debug = __name__+"::"+self.build_tenten60.__name__+".#"+str(self.yti)
logging.info('%s - IN' % cmi_debug )
self.tg_df2 = self.tg_df2.append(self.tg_df1, ignore_index=False) # merge top 10 into
self.tg_df2.reset_index(inplace=True, drop=True) # ensure index is allways unique + sequential
return