-
Notifications
You must be signed in to change notification settings - Fork 4
/
y_techevents.py
401 lines (363 loc) · 21.3 KB
/
y_techevents.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
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
#!/usr/bin/python3
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 json
# logging setup
logging.basicConfig(level=logging.INFO)
#####################################################
class y_techevents:
"""
Class to extract Simple Today/Short/Medium/Long term indicator data set from finance.yahoo.com
"""
# global accessors
symbol = "" # class global
te_sentiment = {} # Dict contains Tech Events elements...e.g.
# 0: ("today_only", "1D", Bullish/Bearish/Neutral or N/A )
# 1: ("short_term", "2W - 6W", Bullish/Bearish/Neutral or N/A )
# 2: ("med_term", "6W - 9M", Bullish/Bearish/Neutral or N/A )
# 3: ("long_term", "9M+", Bullish/Bearish/Neutral or N/A )
# 4: count_of_Bullish
# 5: Tech sentiment computed ranking
te_df0 = ""
te_resp0 = ""
te_jsondata0 = ""
te_zone = ""
te_short = ""
te_mid = ""
te_long = ""
te_srs_zone = ""
te_all_url = ""
yti = 0
cycle = 0 # class thread loop counter
def __init__(self, yti):
cmi_debug = __name__+"::"+self.__init__.__name__
logging.info( f"{cmi_debug} - Instantiate.#{yti}" )
self.te_df0 = pd.DataFrame(columns=[ 'Symbol', 'Today', 'Short', 'Mid', 'Long', 'Bullcount', 'Senti', 'Time' ] ) # init empty DF with preset columns
self.yti = yti
return
#######################################################################
# method #1
def form_api_endpoints(self, symbol):
"""
For Technical event Indicators endpoint for the req get()
This page is where the Free view into some technical idicators are show with full TEXT strings descriptions.
NOTE: This is a teaser page. Tech Events are only available to paid-for subscrption users. But the teaser page shows
a clutch of free indicators...So we'll take the free Tech Event indicators, for now.
1. Today, Short, intermediate, Long Term technical analysis
2. Support, Resistance, Stop loss levels
"""
cmi_debug = __name__+"::"+self.form_api_endpoints.__name__+".#"+str(self.yti)
logging.info( f"{cmi_debug} - form API endpoint URL(s)" )
self.symbol = symbol.upper()
self.te_url = "https://finance.yahoo.com/chart/" + self.symbol + "?Technical="
#self.te_all_url = "https://finance.yahoo.com/quote/" + self.symbol + "?p=" + self.symbol + "&.tsrc=fin-srch"
self.te_all_url = "https://finance.yahoo.com/quote/" + self.symbol
self.te_short_url = "https://finance.yahoo.com/chart/" + self.symbol + "?Technical=short"
self.te_mid_url = "https://finance.yahoo.com/chart/" + self.symbol + "?Technical=intermediate"
self.te_long_url = "https://finance.yahoo.com/chart/" + self.symbol + "?Technical=long"
#
# 2024 new URL's
# https://finance.yahoo.com/chart/T
logging.info( f"================================ Tech Events API endpoints ================================" )
logging.info( f"{cmi_debug} - API endpoint #0: [ {self.te_url} ]" )
logging.info( f"{cmi_debug} - API endpoint #1: [ {self.te_all_url} ]" )
logging.info( f"{cmi_debug} - API endpoint #2: [ {self.te_short_url} ]" )
logging.info( f"{cmi_debug} - API endpoint #3: [ {self.te_mid_url} ]" )
logging.info( f"{cmi_debug} - API endpoint #4: [ {self.te_long_url} ]" )
return
###########################################################################
# method #2
def get_te_zones(self, me):
"""
Connect to finance.yahoo.com and extract (scrape) the raw JSON data out of
the embedded webpage [finance.yahoo.com/chart/GOL?technical=short] html data table.
Sabe JSON to class global attribute: self.te_resp0.text
"""
cmi_debug = __name__+"::"+self.get_te_zones.__name__+".#"+str(self.yti)+"."+str(me)
logging.info( f"{cmi_debug} - IN : {self.te_all_url}" )
with requests.get( self.te_all_url, stream=True, timeout=5 ) as self.te_resp0:
logging.info( f"{cmi_debug} - get() data / storing..." )
self.soup = BeautifulSoup(self.te_resp0.text, 'html.parser')
logging.info( f"{cmi_debug} - Zone #1 / [Entire page] {len(self.soup)} lines extracted / Done" )
logging.info( f"{cmi_debug} - Zone #2 / search [ul zones]..." )
#self.te_zone = self.soup.find_all(attrs={"id": "dropdown-menu"} )
self.te_zone = self.soup.find_all("ul")
logging.info( f"{cmi_debug} - zone #2 / [ul zones] found: {len(self.te_zone)}" )
#logging.info( f"{cmi_debug} - Zone #2\n{self.te_zone}" )
#logging.info( f"{cmi_debug} - Zone #3 / Tech Events zone [tag: <li>]..." )
self.te_lizones = self.te_zone.li
logging.info( f"{cmi_debug} - >>>DEBUG<<< Zone #3 / [li zones]: {len(self.te_lizones)}" )
print ( f">>>DEBUG:<<< {self.te_lizones}" )
try:
self.te_lizones = self.te_zone.find_all("li")
logging.info( f"{cmi_debug} - Zone #3 / [li zones]: {len(self.te_lizones)}" )
except AttributeError as ae_inst: # ae_inst = my custom AE var
if ae_inst.__cause__ is None: # interrogate error raised - was it for [NoneType]
logging.info( f"{cmi_debug} - Zone #3 / [li tag] BS4 search failed: None" )
return -1
else:
logging.info( f"{cmi_debug} - Zone #3 / [li tag] Attribute ERROR: {ae_inst.__cause__}" )
return -2
else:
logging.info( f"{cmi_debug} - Zone #4: Embeded Data block [label tag]" )
self.bb_datablock = self.te_zone.find_all('label')
#self.bb_today = self.te_zone.find(attrs={"class": "W(1/4)--mobp W(1/2) IbBox"} )
return 0 # success
###########################################################################
# method #3
def build_te_data(self, me):
"""
Build out BULLISH/BEARISH/NEUTRAL Perfromance Outlook Technical Events ranking for each symbol
Rankings are weighted, depending on which timeframe this Bull/Bear/Neutral indicator relates to (i.e. today/short/med/long term)
- Weightings can get complicated: e.g. ...
: Short term Bullish -> more weighting than Long term Bullish
: Bullish today, _> more weighting than Bullish short term
: Neutral short term -> more weighting than Neutral Long term
: etc, etc... (see bb_weights DICT)
Dict structure: { key: (embeded 5 element tuple) }
e.g. {0: (te_sml, te_timeframe, "Grey", "Sideways", "Neutral") }
1. tme_sml : (Short/Medium/Long)
2. tm_timeframe : Time frame that this Tech Event covers (days/weeks/months)
3. Tech Event Indicator: Red/Grey/Green * not included in final dict
4. Tech Event Indicator: Down/Sideways/Up * not included in final dict
5. Tech Event Indicator: Bearish/Neutral/Bullish
"""
#logging.disable(0) # ENABLE Log level = INFO
#logging.disable(20) # DISABLE Logging
cmi_debug = __name__+"::"+self.build_te_data.__name__+".#"+str(self.yti)+"."+str(me)
logging.info( f"{cmi_debug} - CALLED" )
time_now = time.strftime("%H:%M:%S", time.localtime() )
logging.info( f"{cmi_debug} - Scan Tech Event indicators" )
bullcount = 0 # counter / how many BULLISH rankings for this symbol
rankalgo = 0 # FINAL Bull/Bear ranking value assigned to this symbol
y = 0 # current dict key/index (i.e. column we are working on..today/short/med/long)
########################################################################################
# Internal helper method
def algo_te_autorank(bull_bear, y, te_sml, rankalgo, te_timeframe, bullcount):
# PRIVATE: Helper method, since we do this 4 x ~50 ticker_sybols
# figurre out which Technical indiciator we are looking at, which column we
# are in, and what the current Bill_Bear code is...: then track some stats & store
# this data in some important places
# bull_bear:
# Bullish, Bearish, Neutral, N/A
te_bb_state = bull_bear
y = y
te_sml = te_sml
rankalgo = rankalgo
bullcount = bullcount
te_timeframe = te_timeframe
logging.info( f"{cmi_debug} - #1 - y_col:{y} / looking for term: {te_sml} / decoded into:{bb_term.get(te_sml)}" )
te_term = bb_term.get(te_sml) # decode yahoo time period -> Short_Med_Long_N/A
bb_getrank = bb_weights.get(te_bb_state) # select DICT index that matches timeframe : result -> DICT
z = bb_getrank.get(te_term) # get algo ranking weight for this col/term timeframe
logging.info( f"{cmi_debug} - #1 - y_col:{y} / te_term:{te_term}: / BB_state:{te_bb_state} / algo rank:{z}" )
rankalgo += z # set ranking
self.te_sentiment.update({y: (te_sml, te_timeframe, te_bb_state)} )
if bull_bear is 'Bullish': bullcount += 1
y += 1 # incr dict/index (timeframe column)
return (y, rankalgo, bullcount)
####################################################
# algo hinter DICT to decode Techncial weightings
# We can customize these as we see fit
bb_weights = { 'Bullish': {'Today': 5, 'Short': 4, 'Med': 4, 'Long': 4},
'Neutral': {'Today': 2, 'Short': 1, 'Med': 1, 'Long': 1},
'N/A': {'Today': 0, 'Short': 0, 'Med': 0, 'Long': 0},
'Bearish': {'Today': -5, 'Short': -4, 'Med': -3, 'Long': -2}
}
####################################################
# algo hinter DICT to decode data range terminology
bb_term = { 'Today': 'Today',
'Short Term (2-3 weeks)': 'Short',
'Mid Term (6 weeks - 9 months)': 'Med',
'Long Term (9+ months)': 'Long',
'N/A': 'N/A'
}
######################################################################################
# BEGIN : Auto ranking algo
# TODAY is treated slightly differnt, becasue its the FIRST element & we know this...no need to guess col/pos
""" bb_today = self.bb_today.next_element.next_element.string # get the Bull/Bear sentimewnt
self.te_sentiment.update({y: ("Today", "1D", bb_today)} )
te_term = bb_term.get('Today') # decode yahoo time periods -> Short_Med_Lon_N/A
bb_getrank = bb_weights.get(bb_today) # select DICT index that matches timeframe : result -> DICT
z = bb_getrank.get(te_term) # get algo ranking weight for this col/term tag (which is TODAY)
logging.info( f"{cmi_debug} - #0 - y_col:{y} / te_term:{te_term}: / BB_state:{bb_today} / algo rank:{z}" )
timeframe_window = bb_weights.get(bb_today) # get tupple @ index = Bull-Bear State
rankalgo += z # set rank weighting for @pos TODAY
logging.info( f"{cmi_debug} - #0 - y_col:{y} / bb_today:{bb_today} / rank:{rankalgo}" )
if bb_today == "Bullish": bullcount += 1 # keep count of BULLISH
"""
# CONTINUE : ranking/weighting algo
# figurre out what COL we are in, what the BB status is and apply a weighting from HINTER table
y += 1 # done with TODAY . incr dict key/index to next timeframe
z = 0 # reset ranking
#for j in self.te_lizones: # eval historical sentiments across all timeframes
for j in self.bb_datablock: # eval historical sentiments across all timeframes
for i in j:
te_strings = i.strings
te_timeframe = next(te_strings) # Timeframe string... "Short Term (2-3 weeks)", "Mid Term (6 weeks - 9 months)", "Long Term (9+ months)"
te_sml = next(te_strings) # Tech sentiment string... Bullish, Neutral, Bearish
print ( f">>>>>DEBUG: sentiment: {te_sml} <<<<<" )
algo_te_autorank(te_sml, y, te_sml, rankalgo, te_timeframe, bullcount)
"""
if i.svg is not None:
red = i.svg.parent.contents
red_down = re.search('180deg', str(red) )
grey_neutral = re.search('90deg', str(red) )
if red_down: # Red = Bearish
y, rankalgo, bullcount = algo_te_autorank('Bearish', y, te_sml, rankalgo, te_timeframe, bullcount)
elif grey_neutral: # Grey = Neutral
y, rankalgo, bullcount = algo_te_autorank('Neutral', y, te_sml, rankalgo, te_timeframe, bullcount)
else: # Green = Bullish
y, rankalgo, bullcount = algo_te_autorank('Bullish', y, te_sml, rankalgo, te_timeframe, bullcount)
else:
pass
y, rankalgo, bullcount = algo_te_autorank('N/A', y, te_sml, rankalgo, te_timeframe, bullcount)
z = 0
"""
self.te_sentiment.update({y: bullcount} )
y += 1 # advance dict pointer
self.te_sentiment.update({y: rankalgo} )
logging.info('%s - populated new Tech Event dict' % cmi_debug )
#logging.disable(0) # ENABLE Logging
#logging.disable(20) # DISABLE Logging
return y # number of rows inserted into Tech events dict
###########################################################################
# method #4
def build_te_summary(self, te_combo_df, me):
"""
Build a Performance Outlook Technical Events DataFrame that is...
A nice easy to read summary table
With Quick to identify stats on BUllish/Bearish Outlook
And can be quickly visually correlated to the Master Summary DataFrame
"""
cmi_debug = __name__+"::"+self.build_te_summary.__name__+".#"+str(self.yti)+"."+str(me)
logging.info( f"{cmi_debug} - get combo_df uniques" )
time_now = time.strftime("%H:%M:%S", time.localtime() )
#DEBUG
print ( f"COMBO_DF: {te_combo_df}" )
te_source = te_combo_df.list_uniques() # work on the combo DataFrame (unique only, no DUPES)
cols = 1
print ( f"\n===== Build Bullish/Bearish outlook summary ==============================" )
for this_sym in te_source['Symbol'].tolist(): # list of symbols to work on
nq_symbol = this_sym.strip().upper() # clearn each symbol (DF pads out with spaces)
print ( f"{this_sym}...", end="", flush=True )
self.form_api_endpoints(nq_symbol)
te_status = self.get_te_zones(me)
if te_status != 0: # FAIL : cant get te_zone data
self.te_is_bad() # FAIL : build a FAILURE dict
self.build_te_df(me) # FAIL: insert failure status into DataFrame for this symbol
print ( f"!", end="", flush=True )
logging.info( f"{cmi_debug} - FAILED to get Tech Event data: Clear all dicts" )
self.te_sentiment.clear()
cols += 1
else:
print ( f"+", end="", flush=True ) # GOOD : suceeded to get TE indicators
self.build_te_data(me) # extract Tech Events data elements
self.build_te_df(me+1) # debug helper, since we call method multiple times
cols += 1
if cols == 8:
print ( f" " ) # only print 8 symbols per row
cols = 1
else:
print ( f" / ", end="", flush=True )
self.te_sentiment.clear()
return
###########################################################################
# method #5
def te_is_bad(self):
"""
Build a Technical Events dict showing all [ BAD / N/A ] indicators
This method is leveraged if we experince issues scraping the TE indicators from the
the Tech performance page/zones becasue they are flakey & unreliable. (yahoo wants you
to pay for premium serivce to get access to them). Also, some tickers dont contain TE data
(for example: some ETF's, Funds, Trusts)
Dict structure: { key: (embeded 5 element tuple) }
e.g. {0: (te_sml, te_timeframe, "Grey", "Sideways", "Neutral") }
1. tme_sml : (Short/Medium/Long)
2. tm_timeframe : Time frame that this Tech Event covers (days/weeks/months)
3. Tech Event Indicator: - set to N/A
4. Tech Event Indicator: - set to N/A
5. Tech Event Indicator: - set to N/A
"""
cmi_debug = __name__+"::"+self.te_is_bad.__name__+".#"+str(self.yti)
logging.info( f"{cmi_debug} - CALLED" )
time_now = time.strftime("%H:%M:%S", time.localtime() )
logging.info( f"{cmi_debug} - Set ALL Tech Event indicators to BAD: N/A and 0" )
bb_today = "N/A"
self.te_sentiment.update({0: ("today_only", "1D", "N/A")} )
self.te_sentiment.update({1: ("short_term", "2W - 6W", "N/A")} )
self.te_sentiment.update({2: ("med_term", "6W - 9M", "N/A")} )
self.te_sentiment.update({3: ("long_term", "9M+", "N/A")} )
self.te_sentiment.update({4: 0})
self.te_sentiment.update({5: 0})
logging.info( f"{cmi_debug} - populated dict as BAD data: All values set to N/A" )
return 4 # number of rows inserted into Tech events dict
###########################################################################
# method #6
def build_te_df(self, me):
"""
Add a ROW into the sentiment DataFrame
ROW is for current symbol last worked on, so method must *only* be called after
you have sucessfullu built & populated the Tech Events sentiment dict.
"""
cmi_debug = __name__+"::"+self.build_te_df.__name__+".#"+str(self.yti)+"."+str(me)
logging.info( f"{cmi_debug} - CALLED" )
time_now = time.strftime("%H:%M:%S", time.localtime() )
logging.info( f"{cmi_debug} - Create Tech Event Perf DataFrame" )
####################################################################
# craft final data structure.
# NOTE: globally accessible and used by quote DF and quote DICT
logging.info( f"{cmi_debug} - Build Dataframe dataset: {self.symbol}" ) # so we can access it natively if needed, without using pandas
data0 = [[ \
self.symbol, \
self.te_sentiment[0][2], \
self.te_sentiment[1][2], \
self.te_sentiment[2][2], \
self.te_sentiment[3][2], \
self.te_sentiment[4], \
self.te_sentiment[5], \
time_now ]]
# self.te_df0.drop(self.te_df0.index, inplace=True) # ensure the DF is empty
logging.info( f"{cmi_debug} - Populate DF with Tech Events emphemerial dict data" )
self.df_1_row = pd.DataFrame(data0, columns=[ 'Symbol', 'Today', 'Short', 'Mid', 'Long', 'Bullcount', 'Senti', 'Time' ] )
self.te_df0 = pd.concat([self.te_df0, self.df_1_row])
#self.te_df0 = self.te_df0.append(te_temp_df0, ignore_index=True)
logging.info( f"{cmi_debug} - Tech Event DF created" )
return
###########################################################################
# method #7
def reset_te_df0(self):
"""
Reset DataFrame index to be sequential, sarting from 0
"""
cmi_debug = __name__+"::"+self.reset_te_df0.__name__+".#"+str(self.yti)
logging.info( f"{cmi_debug} - CALLED" )
self.te_df0.reset_index(inplace=True, drop=True)
logging.info( f"{cmi_debug} - completed" )
return
###########################################################################
# method #8
def te_into_nquote(self, nqinst):
"""
Push the Tech Event Bull/Bear Indicators into their location within the nasdaq quote dict
which is now owned by the nasdaq_wrangler::nq_wrangler class in the accessor dict {} - qd_quote
NOTE: This is .update is BAD form b/c we reshape the dict struct of qd_quote{} so that it no longer
matches the original defiend shape in the class where its managed (nasdaq_wrangler::nq_wrangler)
"""
cmi_debug = __name__+"::"+self.te_into_nquote.__name__+".#"+str(self.yti)
logging.info( f"{cmi_debug} - CALLED" )
nqinst.update({"today_only": self.te_sentiment[0][2]} )
nqinst.update({"short_term": self.te_sentiment[1][2]} )
nqinst.update({"med_term": self.te_sentiment[2][2]} )
nqinst.update({"long_term": self.te_sentiment[3][2]} )
nqinst.update({"Bull_count": self.te_sentiment[4]} )
nqinst.update({"Senti_algo": self.te_sentiment[5]} )
logging.info( f"{cmi_debug} - completed" )
return