Notes![what is notes.io? What is notes.io?](/theme/images/whatisnotesio.png)
![]() ![]() Notes - notes.io |
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.action_chains import ActionChains
import pandas as pd
from pandas import DataFrame
from datetime import datetime
import mysql.connector
from urllib.parse import quote_plus
from sqlalchemy.engine import create_engine
chrome_path = r"C:UsersJawaharDesktopchromedriver_win32chromedriver.exe"
s = Service(chrome_path)
driver = webdriver.Chrome(service=s)
# Creating connection string
#credentials = "mysql://wadmin@uaa-db:%[email protected]:3306"
engine = create_engine("mysql://%s:Tcs#[email protected]:3306" % quote_plus("wadmin@uaa-db"))
# Creating connection object
mydb = mysql.connector.connect(
host = "uaa-db.mysql.database.azure.com",
user = "wadmin@uaa-db",
password = ("Tcs#1234")
)
mydbcursor = mydb.cursor()
mydbcursor.execute("show databases")
# for databases in mydbcursor:
# # print(databases)
##mycccursor = mydb.cursor()
##ccode_data = mycccursor.execute("SELECT company_code, company_name FROM dt_retail.company_tbl ORDER BY company_code")
##comcodedf = DataFrame(ccode_data.fetchall())
#my_data = pd.read_sql("SELECT company_code, company_name FROM dt_retail.company_tbl ORDER BY company_code", mydb)
#print(my_data)
def inserting_Data():
pass
def getCompanyCodes():
# read in your SQL query results using pandas
comcodedf = pd.read_sql("""
SELECT company_code, company_name
FROM dt_retail.company_tbl
ORDER BY company_code
""", engine)
return comcodedf
def dataCounting(selectQry):
#print(selectQry)
#cnts = pd.read_sql('''SELECT COUNT(1) FROM dt_retail.income_statement_tbl''',engine)
cnts = pd.read_sql(selectQry, engine)
count = cnts.values[0]
return count
def insertISMetricData(insertQry, selectQry):
# insert data into the Income Statement table
#try:
count = dataCounting(selectQry)
print("count ::: " + str(count) + str(type(count)) + str(type(int(count))) + str(int(count)) )
if(int(count) == 0):
print(selectQry)
engine.execute(insertQry)
#except:
# print("insert error !!")
# print(ccdf)
def getQuarter():
# read in your SQL query results using pandas
quarterdf = pd.read_sql("""
SELECT quarter_code, quarter_name
FROM dt_retail.quarter_tbl
ORDER BY quarter_code
""", engine)
return quarterdf
qtrdf = getQuarter()
# print(qtrdf)
def getMetrics():
# read in your SQL query results using pandas
metricsdf = pd.read_sql("""
SELECT metric_id, metric_name, level, parent, metric_type
FROM dt_retail.metrics_tbl
ORDER BY metric_id
""", engine)
return metricsdf
mtricsdf = getMetrics()
print(mtricsdf)
# Select rows where company_code is AAP
### print(ccdf.loc[ccdf['company_code'] == 'AAP'])
def conStrToDateTime(datetime_str):
try:
datetime_object = datetime.strptime(datetime_str, '%m/%d/%Y')
print(datetime_object)
return datetime_object
except ValueError as ve:
print('ValueError Raised:', ve)
def getQtrFromDt(datetimeObj):
qtrOfDate = f'Q{(datetimeObj.month-1)//3+1}'
## qtrAppendQ = "Q" + qtrOfDate
return qtrOfDate
def getYearFromDt(datetimeObj):
yearOfDate = datetimeObj.year
return yearOfDate
def get_data_from_yahooFinance(url_link,ticker):
driver.get(url_link)
time.sleep(2)
#For hiding the popup coming to the screen
if(len(driver.find_elements(By.XPATH,"//button[text()='Maybe later']")) > 0):
hide_popup=driver.find_element(By.XPATH,"//button[text()='Maybe later']")
ActionChains(driver).click(hide_popup).perform()
#Quarterly
quarter=driver.find_element(By.XPATH,"//span[text()='Quarterly']")
ActionChains(driver).click(quarter).perform()
#For expanding all columns
expand=driver.find_element(By.XPATH,"//span[text()='Expand All']")
ActionChains(driver).click(expand).perform()
WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.XPATH, '//div[@class="D(tbhg)"]')))
headers_elem = driver.find_elements(By.XPATH, '//div[@class="D(tbhg)"]/div/div')
col_headers = [header.text for header in headers_elem]
col_headers_copy = []
for i in range(len(col_headers)):
##print(type(col_headers[i]))
if(i <= 1):
col_headers_copy.append(col_headers[i])
if(i > 1):
dttime = conStrToDateTime(col_headers[i])
headerDate = str(getYearFromDt(dttime)) + " " + getQtrFromDt(dttime)
col_headers_copy.append(headerDate)
df = pd.DataFrame(columns = col_headers_copy)
rows = driver.find_element(By.XPATH, '//div[@class="D(tbrg)"]')
rows_text=[row_value for row_value in rows.text.split("n")]
for i in range(len(rows_text)):
if(i%2 != 0):
row_data=[]
row_data.append(rows_text[i-1])
row_data+=rows_text[i].split(" ")
new_row_data = []
for i in row_data:
data = i.replace(',',"")
data = i.replace('-',"0")
data1 = string_to_int(data)
#print(str(data1) + str(type(data1)))
new_row_data.append(data1)
df.loc[len(df)] = new_row_data
df['Ticker'] = ticker
df.set_index("Breakdown")
first_column = df.pop('Ticker')
df.insert(0, 'Ticker', first_column)
# Drop last column of a dataframe
#df = df.iloc[: , :-1]
# print("dasffffffffffff")
# print(df)
return df
def string_to_int(s):
if not s:
return s
try:
f = float(s)
i = int(f)
return i if f == i else f
except ValueError:
return s
def get_tickers_lists():
fh = open("tickers.txt")
tickers_string = fh.read()
tickers_list = tickers_string.split("n")
tickers_list = list(filter(None,tickers_list))
return tickers_list
def build_income_statement_url(ticker):
url=f"https://finance.yahoo.com/quote/{ticker}/financials?p={ticker}"
return url
def build_balance_sheet_url(ticker):
url=f"https://finance.yahoo.com/quote/{ticker}/balance-sheet?p={ticker}"
return url
def build_cash_flow_url(ticker):
url=f"https://finance.yahoo.com/quote/{ticker}/cash-flow?p={ticker}"
return url
def up_da_ter3(df):
columns = df.columns.tolist()
dateData = []
tickerid = ""
for _, i in df.iterrows():
for c in columns:
##print(i[c])
## print("column: " + str(c))
if(c == "Breakdown"):
metricName = i[c]
## get the metric data metric df
if(c == "Ticker"):
tickerid = i[c]
print("ticker : " + tickerid)
if(False == ((c == "Breakdown") or (c == "TTM") or (c == "Ticker"))):
print("column: " + str(c))
##dateData.append(c.split(" "))
## print("dateData : " )
##print(dateData)
yearDate = c.split(" ")[0]
quarter = c.split(" ")[1]
# print("yeardate :" + yearDate)
# print("qter : " + quarter)
# print("met: " + metricName)
# insert into dt_retail.income_statement_tbl (metric_id, metric_name, metric_year, metric_quarter, metric_value, company_code, parent_metric_id, metric_level, created_by, updated_by)
# values (1, 'Total Revenue', 2022, 'Q4', 123455, 'AAP', 0, 1, 'JD', 'JD');
is_insert_1stprt = "insert into dt_retail.income_statement_tbl (metric_id, metric_name, metric_year, metric_quarter, metric_value, company_code, parent_metric_id, metric_level, created_by, updated_by) values ("
# print("############")
print("metricValue :" + str(i[c]).replace(',',""))
print(mtricsdf.loc[mtricsdf['metric_name'] == metricName])
metricDtlsrow = mtricsdf.loc[mtricsdf['metric_name'] == metricName]
# metric_id metric_name level parent metric_type
if (metricDtlsrow.empty):
continue
m_id = metricDtlsrow['metric_id'].values[0]
print("m_id : " + str(m_id))
m_name = metricDtlsrow['metric_name'].values[0]
m_level = metricDtlsrow['level'].values[0]
m_parent = metricDtlsrow['parent'].values[0]
print("metric level : " + str(m_level))
m_type = metricDtlsrow['metric_type'].values[0]
is_insert_2ndprt = str(m_id) + ', "' + metricName + '", ' + yearDate + ', "' + quarter + '", ' + str(i[c]).replace(',',"") + ', "' + tickerid + '", ' + str(m_parent) + ', ' + str(m_level) + ', '
print('"user1"' + ', ' + '"user1"')
is_insert_3rdprt = '"user1"' + ', ' + '"user1"' + ' );'
is_insert_query = is_insert_1stprt + is_insert_2ndprt + is_insert_3rdprt
# print(is_insert_query)
isSelectQry_prt1 = "select count(1) from dt_retail.income_statement_tbl"
isSelectQry_prt2 = " where metric_year = " + yearDate
isSelectQry_prt3 = " and metric_quarter = '" + quarter + "'"
isSelectQry_prt4 = " and metric_id = " + str(m_id)
isSelectQry_prt5 = " and company_code = '" + tickerid + "';"
isSelectQry = isSelectQry_prt1 + isSelectQry_prt2 + isSelectQry_prt3 + isSelectQry_prt4 + isSelectQry_prt5
# count = dataCounting(isSelectQry)
# print(" count : " + str(count))
insertISMetricData(is_insert_query, isSelectQry)
def write_to_excel(df,filename):
df.to_excel(filename)
def get_income_statement_data():
list_of_tickers=get_tickers_lists()
df=pd.DataFrame()
for ticker in list_of_tickers:
income_statement_url=build_income_statement_url(ticker)
df1=get_data_from_yahooFinance(income_statement_url,ticker)
df=pd.concat([df,df1],axis=0,ignore_index=True)
print(df)
write_to_excel(df,'income_statement_sample.xlsx')
up_da_ter3(df)
def get_balance_sheet_data():
list_of_tickers=get_tickers_lists()
df=pd.DataFrame()
for ticker in list_of_tickers:
balance_sheet_url=build_balance_sheet_url(ticker)
df1=get_data_from_yahooFinance(balance_sheet_url,ticker)
df=pd.concat([df,df1],axis=0,ignore_index=True)
# print(df)
write_to_excel(df,'balance_sheet.xlsx')
def get_cash_flow_data():
list_of_tickers=get_tickers_lists()
df=pd.DataFrame()
for ticker in list_of_tickers:
cash_flow_url=build_cash_flow_url(ticker)
df1=get_data_from_yahooFinance(cash_flow_url,ticker)
df=pd.concat([df,df1],axis=0,ignore_index=True)
# print(df)
write_to_excel(df,'cash_flow.xlsx')
def main():
get_income_statement_data()
#get_balance_sheet_data()
#get_cash_flow_data()
print("--------Completed--------")
main()
#bb = dataCounting()
#print("#############")
#print(type(bb))
# print(bb)
#to replace (-) with blank
#string to datetime &
#date time conversion to year & quarter
![]() |
Notes.io is a web-based application for taking notes. You can take your notes and share with others people. If you like taking long notes, notes.io is designed for you. To date, over 8,000,000,000 notes created and continuing...
With notes.io;
- * You can take a note from anywhere and any device with internet connection.
- * You can share the notes in social platforms (YouTube, Facebook, Twitter, instagram etc.).
- * You can quickly share your contents without website, blog and e-mail.
- * You don't need to create any Account to share a note. As you wish you can use quick, easy and best shortened notes with sms, websites, e-mail, or messaging services (WhatsApp, iMessage, Telegram, Signal).
- * Notes.io has fabulous infrastructure design for a short link and allows you to share the note as an easy and understandable link.
Fast: Notes.io is built for speed and performance. You can take a notes quickly and browse your archive.
Easy: Notes.io doesn’t require installation. Just write and share note!
Short: Notes.io’s url just 8 character. You’ll get shorten link of your note when you want to share. (Ex: notes.io/q )
Free: Notes.io works for 12 years and has been free since the day it was started.
You immediately create your first note and start sharing with the ones you wish. If you want to contact us, you can use the following communication channels;
Email: [email protected]
Twitter: http://twitter.com/notesio
Instagram: http://instagram.com/notes.io
Facebook: http://facebook.com/notesio
Regards;
Notes.io Team