Financial Daily Loads
Ask Gemini
I want to create a python program that runs once a day to retrieve prices from yfinance api and put them in daily_price table - daily_price ( ticker VARCHAR(10) NOT NULL, date DATE NOT NULL, open DECIMAL(19, 4), high DECIMAL(19, 4), low DECIMAL(19, 4), close DECIMAL(19, 4), volume BIGINT, PRIMARY KEY (ticker, date));
update_daily_prices.py
from datetime import date
from google.cloud.sql.connector import Connector
import sqlalchemy
import yfinance as yf
# --- 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
pool = sqlalchemy.create_engine(
"mysql+pymysql://",
creator=getconn,
pool_pre_ping=True,
)
def setup_table(db_conn):
"""Ensures the daily_price table exists matching your exact schema."""
db_conn.execute(
sqlalchemy.text("""
CREATE TABLE IF NOT EXISTS daily_price (
ticker VARCHAR(10) NOT NULL,
date DATE NOT NULL,
open DECIMAL(19, 4),
high DECIMAL(19, 4),
low DECIMAL(19, 4),
close DECIMAL(19, 4),
volume BIGINT,
PRIMARY KEY (ticker, date)
);
""")
)
db_conn.commit()
# --- SECTION 2: DATA FETCHING & PROCESSING ---
def get_tickers_from_db(db_conn):
"""Fetches the list of active tickers from the company table."""
sql_query = sqlalchemy.text(
"SELECT ticker FROM company WHERE ticker IS NOT NULL;"
)
result = db_conn.execute(sql_query)
# Flatten the result rows into a clean list of uppercase strings
tickers = [row[0].strip().upper() for row in result.fetchall()]
return tickers
def fetch_yfinance_price(symbol):
"""Pulls historical price data via yfinance for the latest trading session."""
try:
print(f"Fetching daily price data for {symbol}...")
tk = yf.Ticker(symbol)
df = tk.history(period="2d")
if df.empty:
print(f" -> Warning: No history data returned for {symbol}.")
return None
# Get the latest available trading day
latest = df.dropna().iloc[-1]
row_date = latest.name.date().strftime("%Y-%m-%d")
return {
"ticker": symbol,
"date": row_date,
"open": float(latest["Open"]),
"high": float(latest["High"]),
"low": float(latest["Low"]),
"close": float(latest["Close"]),
"volume": int(latest["Volume"]),
}
except Exception as ex:
print(f" -> Error fetching price data for {symbol}: {ex}")
return None
# --- SECTION 3: EXECUTION ---
try:
with pool.connect() as db_conn:
# Step 0: Ensure table exists
setup_table(db_conn)
# Step 1: Read target tickers from company table
tickers = get_tickers_from_db(db_conn)
print(f"Found {len(tickers)} tickers to process from the company table.")
if not tickers:
print("No tickers found in 'company' table. Exiting.")
else:
# Step 2: Loop through tickers and fetch yfinance daily prices
price_records = []
for ticker in tickers:
record = fetch_yfinance_price(ticker)
if record:
price_records.append(record)
# Step 3: Batch insert/upsert records into daily_price table
if not price_records:
print("No valid price records compiled to insert.")
else:
sql_template = """
INSERT INTO daily_price (
ticker, date, open, high, low, close, volume
) VALUES (
:ticker, :date, :open, :high, :low, :close, :volume
)
ON DUPLICATE KEY UPDATE
open = VALUES(open),
high = VALUES(high),
low = VALUES(low),
close = VALUES(close),
volume = VALUES(volume);
"""
print(
f"Uploading {len(price_records)} records to 'daily_price' table..."
)
db_conn.execute(sqlalchemy.text(sql_template), price_records)
db_conn.commit()
print("Daily price database upload successful!")
for rec in price_records:
print(
f" -> Upserted {rec['ticker']} for {rec['date']} | Close:"
f" ${rec['close']:.2f}"
)
except Exception as e:
print(f"An error occurred: {e}")
finally:
connector.close()
print("Cloud SQL connection closed.")
$ python update_daily_prices.py
Found 506 tickers to process from the company table.
Fetching daily price data for A...
Fetching daily price data for AAPL...
Fetching daily price data for ABBV...
Fetching daily price data for ABNB
Uploading 496 records to 'daily_price' table...
Daily price database upload successful!
-> Upserted A for 2026-09-23 | Close: $165.93
-> Upserted AAPL for 2026-09-23 | Close: $338.06
How to Automate It to Run Once a Day
Because this script runs locally or on a persistent instance rather than a restricted Cloud Function gateway, you can schedule it cleanly without timing out
Linux / Cloud Shell (via Crontab): Open your crontab editor (crontab -e) and add a rule to run the script every weekday at 6:00 PM (after market close):
Bash
0 18 * * 1-5 /usr/bin/python3 /path/to/update_daily_prices.py >> /path/to/cron.log 2>&1
Gemini
create a python program to list the daily prices file
list_prices.py
from google.cloud.sql.connector import Connector
from sqlalchemy import create_engine, text
# --- 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 SQLAlchemy engine using the Cloud SQL connector
engine = create_engine("mysql+pymysql://", creator=getconn)
def list_daily_prices():
"""Queries and displays all rows from the daily_price table."""
query = text("""
SELECT ticker, date, open, high, low, close, volume
FROM daily_price
ORDER BY date DESC, ticker ASC;
""")
print("--- Daily Price Database Log ---")
print(
f"{'Ticker':<8} | {'Date':<12} | {'Open':<10} | {'High':<10} |"
f" {'Low':<10} | {'Close':<10} | {'Volume':<12}"
)
print("-" * 88)
with engine.connect() as connection:
result = connection.execute(query)
rows = result.fetchall()
if not rows:
print("No records found in the daily_price table.")
return
for row in rows:
ticker, p_date, open_p, high_p, low_p, close_p, volume = row
# Format price and volume values nicely
open_str = f"${open_p:,.2f}" if open_p is not None else "N/A"
high_str = f"${high_p:,.2f}" if high_p is not None else "N/A"
low_str = f"${low_p:,.2f}" if low_p is not None else "N/A"
close_str = f"${close_p:,.2f}" if close_p is not None else "N/A"
vol_str = f"{volume:,}" if volume is not None else "N/A"
print(
f"{ticker:<8} | {str(p_date):<12} | {open_str:<10} | {high_str:<10}"
f" | {low_str:<10} | {close_str:<10} | {vol_str:<12}"
)
print("-" * 88)
print(f"Total price records listed: {len(rows)}")
if __name__ == "__main__":
list_daily_prices()
$ python list_prices.py
--- Daily Price Database Log ---
Ticker | Date | Open | High | Low | Close | Volume
----------------------------------------------------------------------------------------
A | 2026-09-23 | $166.54 | $168.94 | $164.64 | $165.93 | 1,034,439
AAPL | 2026-09-23 | $341.08 | $341.80 | $336.41 | $338.06 | 9,663,509
ABBV | 2026-09-23 | $266.00 | $269.39 | $265.05 | $266.06 | 696,629
ABNB | 2026-09-23 | $159.52 | $160.20 | $153.56 | $153.61 | 2,305,781
Lets look at daily metrics
daily_metric_loads.py
from datetime import date
import sqlalchemy
from google.cloud.sql.connector import Connector
import yfinance as yf
# --- 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
pool = sqlalchemy.create_engine(
"mysql+pymysql://",
creator=getconn,
pool_pre_ping=True,
)
# --- SECTION 2: DATA FETCHING & PROCESSING ---
def get_tickers_from_db(db_conn):
"""Fetches the list of active tickers from the company table."""
# Assumes your company table has a column named 'ticker'
sql_query = sqlalchemy.text("SELECT ticker FROM company WHERE ticker IS NOT NULL;")
result = db_conn.execute(sql_query)
# Flatten the result rows into a clean list of uppercase strings
tickers = [row[0].strip().upper() for row in result.fetchall()]
return tickers
def fetch_yfinance_metrics(symbol):
"""Pulls the info dict from yfinance and cleans/maps fields to match SQL schema."""
try:
print(f"Fetching metrics data for {symbol}...")
ticker = yf.Ticker(symbol)
info = ticker.info
# If info dict is empty or invalid, skip it
if not info or 'currentPrice' not in info:
print(f"Skipping {symbol}: No valid metric data found.")
return None
# Map fields to match table structure, using .get() to safely fall back to None
return {
"ticker": symbol,
"date": date.today().strftime('%Y-%m-%d'),
"target_high_price": info.get('targetHighPrice'),
"target_low_price": info.get('targetLowPrice'),
"target_mean_price": info.get('targetMeanPrice'),
"target_median_price": info.get('targetMedianPrice'),
"current_price": info.get('currentPrice'),
"recommendation_key": info.get('recommendationKey'),
"recommendation_mean": info.get('recommendationMean'),
"number_of_analyst_opinions": info.get('numberOfAnalystOpinions'),
"trailing_pe": info.get('trailingPE'),
"forward_pe": info.get('forwardPE'),
"price_to_book": info.get('priceToBook'),
"peg_ratio": info.get('pegRatio'),
"ebitda": info.get('ebitda'),
"revenue_growth": info.get('revenueGrowth'),
"return_on_equity": info.get('returnOnEquity'),
"fifty_two_week_high": info.get('fiftyTwoWeekHigh'),
"fifty_two_week_low": info.get('fiftyTwoWeekLow'),
"fifty_day_average": info.get('fiftyDayAverage'),
"two_hundred_day_average": info.get('twoHundredDayAverage'),
"beta": info.get('beta')
}
except Exception as e:
print(f"Error fetching metrics data for {symbol}: {e}")
return None
# --- SECTION 3: EXECUTION ---
try:
with pool.connect() as db_conn:
# Step 1: Read target tickers from company table
tickers = get_tickers_from_db(db_conn)
print(f"Found {len(tickers)} tickers to process from the database.")
if not tickers:
print("No tickers found in 'company' table. Exiting.")
else:
# Step 2: Loop through tickers and fetch yfinance data
metrics_records = []
for ticker in tickers:
record = fetch_yfinance_metrics(ticker)
if record:
metrics_records.append(record)
# Step 3: Batch insert/upsert records into daily_metrics table
if not metrics_records:
print("No valid metric records compiled to insert.")
else:
# SQL Template structured for MySQL's duplicate handling
sql_template = """
INSERT INTO daily_metrics (
ticker, date, target_high_price, target_low_price, target_mean_price,
target_median_price, current_price, recommendation_key, recommendation_mean,
number_of_analyst_opinions, trailing_pe, forward_pe, price_to_book,
peg_ratio, ebitda, revenue_growth, return_on_equity, fifty_two_week_high,
fifty_two_week_low, fifty_day_average, two_hundred_day_average, beta
) VALUES (
:ticker, :date, :target_high_price, :target_low_price, :target_mean_price,
:target_median_price, :current_price, :recommendation_key, :recommendation_mean,
:number_of_analyst_opinions, :trailing_pe, :forward_pe, :price_to_book,
:peg_ratio, :ebitda, :revenue_growth, :return_on_equity, :fifty_two_week_high,
:fifty_two_week_low, :fifty_day_average, :two_hundred_day_average, :beta
)
ON DUPLICATE KEY UPDATE
target_high_price = VALUES(target_high_price),
target_low_price = VALUES(target_low_price),
target_mean_price = VALUES(target_mean_price),
target_median_price = VALUES(target_median_price),
current_price = VALUES(current_price),
recommendation_key = VALUES(recommendation_key),
recommendation_mean = VALUES(recommendation_mean),
number_of_analyst_opinions = VALUES(number_of_analyst_opinions),
trailing_pe = VALUES(trailing_pe),
forward_pe = VALUES(forward_pe),
price_to_book = VALUES(price_to_book),
peg_ratio = VALUES(peg_ratio),
ebitda = VALUES(ebitda),
revenue_growth = VALUES(revenue_growth),
return_on_equity = VALUES(return_on_equity),
fifty_two_week_high = VALUES(fifty_two_week_high),
fifty_two_week_low = VALUES(fifty_two_week_low),
fifty_day_average = VALUES(fifty_day_average),
two_hundred_day_average = VALUES(two_hundred_day_average),
beta = VALUES(beta);
"""
print(f"Uploading {len(metrics_records)} records to 'daily_metrics' table...")
# SQLAlchemy handles execution mapping automatically for lists of dicts
db_conn.execute(sqlalchemy.text(sql_template), metrics_records)
db_conn.commit()
print("Database upload successful!")
for rec in metrics_records:
print(f"Processed/Upserted metrics for: {rec['ticker']}")
except Exception as e:
print(f"An error occurred: {e}")
finally:
# Always cleanly close the Google Cloud SQL Connector
connector.close()
print("Cloud SQL connection closed.")
list_metrics.py
from google.cloud.sql.connector import Connector
import sqlalchemy
# --- 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
pool = sqlalchemy.create_engine("mysql+pymysql://", creator=getconn, pool_pre_ping=True)
def list_daily_metrics():
"""Queries and displays all rows from the daily_metrics table."""
query = sqlalchemy.text("""
SELECT ticker, date, current_price, target_mean_price, recommendation_key,
trailing_pe, forward_pe, price_to_book, revenue_growth, beta
FROM daily_metrics
ORDER BY date DESC, ticker ASC;
""")
print("--- Daily Metrics Database Log ---")
print(
f"{'Ticker':<8} | {'Date':<12} | {'Current':<9} | {'Target Mean':<11} |"
f" {'Rec':<10} | {'Trailing P/E':<12} | {'Forward P/E':<11} | {'Beta':<6}"
)
print("-" * 92)
try:
with pool.connect() as db_conn:
result = db_conn.execute(query)
rows = result.fetchall()
if not rows:
print("No records found in the daily_metrics table.")
return
for row in rows:
(
ticker,
m_date,
cur_p,
tgt_m,
rec,
t_pe,
f_pe,
p_book,
rev_g,
beta,
) = row
# Format values cleanly with fallbacks for N/A
cur_str = f"${cur_p:,.2f}" if cur_p is not None else "N/A"
tgt_str = f"${tgt_m:,.2f}" if tgt_m is not None else "N/A"
rec_str = str(rec) if rec else "N/A"
t_pe_str = f"{t_pe:.2f}" if t_pe is not None else "N/A"
f_pe_str = f"{f_pe:.2f}" if f_pe is not None else "N/A"
beta_str = f"{beta:.2f}" if beta is not None else "N/A"
print(
f"{ticker:<8} | {str(m_date):<12} | {cur_str:<9} | {tgt_str:<11} |"
f" {rec_str:<10} | {t_pe_str:<12} | {f_pe_str:<11} | {beta_str:<6}"
)
print("-" * 92)
print(f"Total metrics records listed: {len(rows)}")
except Exception as e:
print(f"Error querying daily_metrics table: {e}")
finally:
connector.close()
if __name__ == "__main__":
list_daily_metrics()
No comments:
Post a Comment