SQL Table builds
Need openfigi key exported
export OPENFIGI_API_KEY="d2506477-3e19-4d6e-8830-281ae776633"
yfinance install
pip install yfinance
pip install --force-reinstall --no-cache-dir curl_cffi yfinance
Pulls the address fields for NVDA
import yfinance as yf
# Initialize the ticker
ticker = yf.Ticker("NVDA")
# Get the company profile info
info = ticker.info
# List of specific fields you requested
fields = [
"longName", "address1",
"city", "state", "zip","country", "sector", "industry"
]
print("--- Company Classification & Location ---")
for field in fields:
# .get() is safer than info[field] because it won't
# throw an error if the field is missing from the API
value = info.get(field, "N/A")
print(f"{field}: {value}")
$ python3 comp.py
--- Company Classification & Location ---
longName: NVIDIA Corporation
address1: 2788 San Tomas Expressway
city: Santa Clara
state: CA
zip: 95051
country: United States
sector: Technology
industry: Semiconductors
Now we need the ticker information present in the S&P 500.
We use the wikipedia link to get the ticker for S&P 500 members and place them in a text file.
import pandas as pd
import requests
from bs4 import BeautifulSoup
url = "https://en.wikipedia.org/wiki/List_of_S%26P_500_companies"
headers = {"User-Agent": "Mozilla/5.0"}
print("Fetching data from Wikipedia...")
response = requests.get(url, headers=headers)
soup = BeautifulSoup(response.text, 'html.parser')
# Find the table by its ID
table = soup.find('table', {'id': 'constituents'})
data = []
# Iterate through rows, skipping the header
for row in table.find_all('tr')[1:]:
cols = row.find_all('td')
if len(cols) >= 3:
# Ticker (with Yahoo Finance formatting)
symbol = cols[0].text.strip().replace('.', '-')
# Security Name
name = cols[1].text.strip()
# GICS Sector
sector = cols[3].text.strip() # Wikipedia column 3 is usually Sector
data.append([symbol, name, sector])
# Create a DataFrame
df = pd.DataFrame(data, columns=['Ticker', 'Security', 'Sector'])
# Save to CSV in the current directory
csv_file = "sp500_tickers.csv"
df.to_csv(csv_file, index=False)
print(f"Success! {len(df)} tickers saved to {csv_file}")
print("\nFirst 5 rows:")
print(df.head())
$ python sp500.py
Fetching data from Wikipedia...
Success! 503 tickers saved to sp500_tickers.csv
First 5 rows:
Ticker Security Sector
0 MMM 3M Industrial Conglomerates
1 AOS A. O. Smith Building Products
2 ABT Abbott Laboratories Health Care Equipment
3 ABBV AbbVie Biotechnology
4 ACN Accenture IT Consulting & Other Services
$ ls -lt sp500_tickers.csv
-rw-rw-r-- 1 john_iacovacci1 john_iacovacci1 22518 Sep 16 13:56 sp500_tickers.csv
We can now create a program to load all tickers from S&P 500 to the company table.
sp500_to_sql.py
import pandas as pd
import yfinance as yf
import sqlalchemy
from sqlalchemy import text
from google.cloud.sql.connector import Connector
import time
# --- SECTION 1: DATABASE CONNECTION SETUP ---
connector = Connector()
def getconn():
conn = connector.connect(
"sentiment-analysis-379200:us-east1:fintech",
"pymysql",
user="root",
password="Uconnstamford1!",
db="jiacovacci",
enable_iam_auth=False
)
return conn
pool = sqlalchemy.create_engine(
"mysql+pymysql://",
creator=getconn,
pool_pre_ping=True,
)
# --- SECTION 2: DATA FETCHING & CLEANING ---
def get_ticker_data(symbol):
ticker = yf.Ticker(symbol)
info = ticker.info
# Using .get() with defaults to handle missing data gracefully
return {
"Ticker": symbol,
"Comp_Name": str(info.get("longName", "N/A")),
"Comp_Street": str(info.get("address1", "N/A")),
"Comp_City": str(info.get("city", "N/A")),
"Comp_State": str(info.get("state", "N/A")),
"Comp_Zip": str(info.get("zip", "N/A")),
"Comp_Country": str(info.get("country", "N/A")),
"Sector": str(info.get("sector", "N/A")),
"Industry": str(info.get("industry", "N/A"))
}
# --- SECTION 3: EXECUTION ---
try:
# 1. Load Tickers from your CSV
sp500_df = pd.read_csv('sp500_tickers.csv')
ticker_list = sp500_df['Ticker'].tolist()
sql_template = """
INSERT INTO company
(Ticker, Comp_Name, Comp_Street, Comp_City, Comp_State, Comp_Zip, Comp_Country, Sector, Industry)
VALUES
(:Ticker, :Comp_Name, :Comp_Street, :Comp_City, :Comp_State, :Comp_Zip, :Comp_Country, :Sector, :Industry)
ON DUPLICATE KEY UPDATE Comp_Name=VALUES(Comp_Name);
"""
with pool.connect() as db_conn:
for symbol in ticker_list:
try:
print(f"Processing {symbol}...")
# Fetch data from yfinance
company_data = get_ticker_data(symbol)
# Execute Insert
db_conn.execute(text(sql_template), company_data)
db_conn.commit()
# Optional: Small sleep to avoid rate limiting from Yahoo Finance
time.sleep(0.1)
except Exception as inner_e:
print(f"Skipping {symbol} due to error: {inner_e}")
continue
print("All tickers processed successfully.")
except Exception as e:
print(f"A critical error occurred: {e}")
finally:
connector.close()
$ python sp500_to_sql.py
Processing MMM...
Processing AOS...
Processing ABT...
Processing ABBV...
Processing ACN...
Processing ADBE...
Before we get the tickers from openFIGI we can use the security.py file to make sure data will come back.
Using data from the site to build the security table. Looking for data for ADI US. Data retrieved matches the SQL security table fields
#!/usr/bin/env python3.12
import json
import urllib.request
import urllib.parse
import os
# --- Configuration ---
OPENFIGI_API_KEY = os.environ.get("OPENFIGI_API_KEY", None)
OPENFIGI_BASE_URL = "https://api.openfigi.com"
# --- Function 1: The API Handler ---
def api_call(path: str, data: list | None = None, method: str = "POST"):
headers = {"Content-Type": "application/json"}
if OPENFIGI_API_KEY:
headers |= {"X-OPENFIGI-APIKEY": OPENFIGI_API_KEY}
request = urllib.request.Request(
url=urllib.parse.urljoin(OPENFIGI_BASE_URL, path),
data=data and bytes(json.dumps(data), encoding="utf-8"),
headers=headers,
method=method,
)
with urllib.request.urlopen(request) as response:
json_response_as_string = response.read().decode("utf-8")
return json.loads(json_response_as_string)
# --- Function 2: SQL Generator ---
def generate_sql_insert(record):
"""Formats OpenFIGI data into your specific SQL table structure."""
data = {
"figi": record.get("figi", "N/A"),
"name": str(record.get("name", "N/A")).replace("'", "''"),
"ticker": record.get("ticker", "N/A"),
"exchCode": record.get("exchCode", "N/A"),
"securityType": record.get("securityType", "N/A"),
"marketSector": record.get("marketSector", "N/A"),
"shareClassFIGI": record.get("shareClassFIGI", "N/A"),
"securityType2": record.get("securityType2", "N/A"),
"securityDescription": record.get("securityDescription", "N/A")
}
sql = """INSERT INTO security (figi, name, ticker, exchCode, securityType, marketSector, shareClassFIGI, securityType2, securityDescription)
VALUES ('{figi}', '{name}', '{ticker}', '{exchCode}', '{securityType}', '{marketSector}', '{shareClassFIGI}', '{securityType2}', '{securityDescription}');""".format(**data)
return sql
# --- Function 3: Main Execution ---
def main():
# Define the request
mapping_request = [
{
"idType": "TICKER",
"idValue": "XLF",
"exchCode": "US"
},
]
print("-- Fetching Data from OpenFIGI --")
try:
# Call the api_call function defined above
mapping_response = api_call("/v3/mapping", mapping_request)
for response_item in mapping_response:
if "error" in response_item:
print(f"Error: {response_item['error']}")
continue
results = response_item.get("data", [])
for record in results:
print(generate_sql_insert(record))
except Exception as e:
print(f"An error occurred: {e}")
if __name__ == "__main__":
main()
$ python security.py
-- Fetching Data from OpenFIGI --
INSERT INTO security (figi, name, ticker, exchCode, securityType, marketSector, shareClassFIGI, securityType2, securityDescription)
VALUES ('BBG000BJ29X7', 'SS FINANCIAL SELECT SECTOR', 'XLF', 'US', 'ETP', 'Equity', 'BBG001S7T223', 'Mutual Fund', 'XLF');
Build security table
Now we use the tickers from the sp500 wikipedia pull to loop thru and populate the security table with data from openfigi API call
openfigi_sp500.py
#!/usr/bin/env python3.12
import json
import urllib.request
import urllib.parse
import os
import sqlalchemy
import pandas as pd
from google.cloud.sql.connector import Connector
# --- Configuration ---
OPENFIGI_API_KEY = os.environ.get("OPENFIGI_API_KEY", None)
OPENFIGI_BASE_URL = "https://api.openfigi.com"
# --- Database Setup ---
connector = Connector()
def getconn():
conn = connector.connect(
"sentiment-analysis-379200:us-east1:fintech",
"pymysql",
user="root",
password="Uconnstamford1!",
db="jiacovacci",
enable_iam_auth=False
)
return conn
pool = sqlalchemy.create_engine(
"mysql+pymysql://",
creator=getconn,
pool_pre_ping=True,
)
# --- Function 1: The API Handler ---
def api_call(path: str, data: list | None = None, method: str = "POST"):
headers = {"Content-Type": "application/json"}
if OPENFIGI_API_KEY:
headers |= {"X-OPENFIGI-APIKEY": OPENFIGI_API_KEY}
request = urllib.request.Request(
url=urllib.parse.urljoin(OPENFIGI_BASE_URL, path),
data=data and bytes(json.dumps(data), encoding="utf-8"),
headers=headers,
method=method,
)
with urllib.request.urlopen(request) as response:
return json.loads(response.read().decode("utf-8"))
# --- Function 2: Database Loader ---
def load_to_db(records):
if not records:
return
# Added ON DUPLICATE KEY UPDATE to handle re-runs smoothly
insert_stmt = sqlalchemy.text("""
INSERT INTO security (figi, name, ticker, exchCode, securityType, marketSector, shareClassFIGI, securityType2, securityDescription)
VALUES (:figi, :name, :ticker, :exchCode, :securityType, :marketSector, :shareClassFIGI, :securityType2, :securityDescription)
ON DUPLICATE KEY UPDATE NAME=VALUES(NAME), TICKER=VALUES(TICKER)
""")
try:
with pool.connect() as db_conn:
for record in records:
params = {
"figi": record.get("figi", "N/A"),
"name": record.get("name", "N/A"),
"ticker": record.get("ticker", "N/A"),
"exchCode": record.get("exchCode", "N/A"),
"securityType": record.get("securityType", "N/A"),
"marketSector": record.get("marketSector", "N/A"),
"shareClassFIGI": record.get("shareClassFIGI", "N/A"),
"securityType2": record.get("securityType2", "N/A"),
"securityDescription": record.get("securityDescription", "N/A")
}
db_conn.execute(insert_stmt, params)
db_conn.commit()
print(f"Successfully loaded {len(records)} records to the database.")
except Exception as e:
print(f"Database Error: {e}")
# --- Main Execution ---
def main():
try:
# 1. Load tickers from CSV
csv_path = 'sp500_tickers.csv'
if not os.path.exists(csv_path):
print(f"Error: {csv_path} not found.")
return
df_sp500 = pd.read_csv(csv_path)
# Ensure 'Ticker' column exists and create the list
if 'Ticker' not in df_sp500.columns:
print("Error: 'Ticker' column not found in CSV.")
return
tickers = df_sp500['Ticker'].tolist()
# 2. Prepare OpenFIGI mapping requests
# We define this AFTER 'tickers' is created
mapping_request = [{"idType": "TICKER", "idValue": str(t).replace('-', '.'), "exchCode": "US"} for t in tickers]
# 3. Process in batches (Lowered to 25 to avoid 413 error)
all_results = []
batch_size = 10
print(f"-- Fetching Data for {len(tickers)} tickers from OpenFIGI --")
for i in range(0, len(mapping_request), batch_size):
batch = mapping_request[i:i + batch_size]
print(f"Requesting batch {i//batch_size + 1}...")
try:
mapping_response = api_call("/v3/mapping", batch)
for response_item in mapping_response:
if "error" in response_item:
continue
results = response_item.get("data", [])
if results:
# Grab the first valid result for this ticker
all_results.append(results[0])
except Exception as batch_error:
print(f"Error in batch {i}: {batch_error}")
continue
# 4. Load all found records to DB
if all_results:
print(f"-- Loading {len(all_results)} total results to Cloud SQL --")
load_to_db(all_results)
else:
print("No data found to load.")
except Exception as e:
print(f"An unexpected error occurred: {e}")
finally:
connector.close()
if __name__ == "__main__":
main()
$ python openfigi_sp500.py
-- Fetching Data for 503 tickers from OpenFIGI --
Requesting batch 1...
Requesting batch 2...
Requesting batch 3...
Requesting batch 4...
Requesting batch 5...
Requesting batch 6...
Requesting batch 7...
Requesting batch 8...
Requesting batch 9...
Requesting batch 10...
Requesting batch 11...
Requesting batch 12...
Requesting batch 13...
Requesting batch 14...
Requesting batch 15...
Requesting batch 16...
Requesting batch 17...
Requesting batch 18...
Requesting batch 19...
Requesting batch 20...
Requesting batch 21...
Requesting batch 22...
Requesting batch 23...
Requesting batch 24...
Requesting batch 25...
Requesting batch 26...
Requesting batch 27...
Requesting batch 28...
Requesting batch 29...
Requesting batch 30...
Requesting batch 31...
Requesting batch 32...
Requesting batch 33...
Requesting batch 34...
Requesting batch 35...
Requesting batch 36...
Requesting batch 37...
Requesting batch 38...
Requesting batch 39...
Requesting batch 40...
Requesting batch 41...
Requesting batch 42...
Requesting batch 43...
Requesting batch 44...
Requesting batch 45...
Requesting batch 46...
Requesting batch 47...
Requesting batch 48...
Requesting batch 49...
Requesting batch 50...
Requesting batch 51...
-- Loading 501 total results to Cloud SQL --
Successfully loaded 501 records to the database.
john_iacovacci1@cloudshell:~/openFIGI (sentiment-analysis-379200)$
Now we can create a program to list securities
security_list.py
#!/usr/bin/env python3.12
import sqlalchemy
import csv
from google.cloud.sql.connector import Connector
# --- SECTION 1: DATABASE CONNECTION SETUP ---
connector = Connector()
def getconn():
conn = connector.connect(
"sentiment-analysis-379200:us-east1:fintech",
"pymysql",
user="root",
password="Uconnstamford1!",
db="jiacovacci",
enable_iam_auth=False
)
return conn
# Create the SQLAlchemy engine
pool = sqlalchemy.create_engine(
"mysql+pymysql://",
creator=getconn,
pool_pre_ping=True,
)
# --- SECTION 2: DATA FETCHING ---
def list_security_table():
"""Fetches all records from the security table and prints them."""
sql_query = "SELECT * FROM security;"
try:
with pool.connect() as db_conn:
result = db_conn.execute(sqlalchemy.text(sql_query))
# Fetch headers and data
column_names = result.keys()
rows = result.fetchall()
if not rows:
print("The security table is currently empty.")
return
# Print Headers
print(f"\n{' | '.join(column_names)}")
print("-" * 120)
# Print Rows
for row in rows:
print(" | ".join(str(value) for value in row))
print(f"\nTotal records found: {len(rows)}")
# Optional: Export to CSV automatically
export_to_csv(column_names, rows)
except Exception as e:
print(f"Database Error: {e}")
finally:
connector.close()
def export_to_csv(headers, data):
filename = "security_list.csv"
with open(filename, mode='w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(headers)
writer.writerows(data)
print(f"Data also exported to {filename}")
# --- Main Execution ---
if __name__ == "__main__":
list_security_table()
$ python security_list.py
figi | name | ticker | exchCode | securityType | marketSector | shareClassFIGI | securityType2 | securityDescription
------------------------------------------------------------------------------------------------------------------------
BBG000B9X8C0 | AMEREN CORPORATION | AEE | US | Common Stock | Equity | BBG001S5NF24 | Common Stock | AEE
BBG000B9XG87 | AMETEK INC | AME | US | Common Stock | Equity | BBG001S5NN54 | Common Stock | AME
BBG000B9XRY4 | APPLE INC | AAPL | US | Common Stock | Equity | BBG001S5N8V8 | Common Stock | AAPL
BBG000B9XYV2 | AMERICAN TOWER CORP | AMT | US | REIT | Equity | BBG001S5NPQ6 | REIT | AMT
BBG000B9YJ35 | AMPHENOL CORP-CL A | APH | US | Common Stock | Equity | BBG001S5NSK6 | Common Stock | APH
No comments:
Post a Comment