import os
import smtplib
import yfinance as yf
import requests
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime
import pandas as pd
import schedule
import time
import logging
import socket
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)
# Set user agent to avoid Yahoo Finance blocking
yf.set_tz_cache_location("/tmp/yfinance_cache")
def check_network():
"""Verify network connectivity to key services"""
try:
host = "query2.finance.yahoo.com"
ip = socket.gethostbyname(host)
logger.info(f"Successfully resolved {host} to {ip}")
# Simple socket connection check
with socket.create_connection((host, 443), timeout=5):
logger.info(f"Successfully connected to {host}:443")
except Exception as e:
logger.error(f"Network check failed for {host}: {e}")
class FinanceEmailer:
def __init__(self):
self.gmail_user = os.environ.get('GMAIL_USER')
self.gmail_app_password = os.environ.get('GMAIL_APP_PASSWORD')
self.recipient_email = os.environ.get('RECIPIENT_EMAIL')
if not all([self.gmail_user, self.gmail_app_password, self.recipient_email]):
raise ValueError("Missing required environment variables")
def get_exchange_rate(self):
"""Fetch AUD to USD exchange rate using Frankfurter API"""
try:
url = "https://api.frankfurter.app/latest?from=AUD&to=USD"
response = requests.get(url, timeout=10)
response.raise_for_status()
data = response.json()
return data['rates']['USD']
except Exception as e:
logger.error(f"Error fetching exchange rate: {e}")
return None
def get_hpe_stock_price(self):
"""Fetch HPE stock price using yfinance"""
try:
logger.info("Attempting to fetch HPE stock data...")
# Let yfinance handle the session
stock = yf.Ticker("HPE")
logger.info(f"Created ticker object for HPE")
# Try to get data from the last 5 days to handle weekends/holidays
logger.info("Fetching 5-day history...")
try:
data = stock.history(period="5d")
except Exception as history_err:
logger.error(f"stock.history() raised an exception: {history_err}", exc_info=True)
data = pd.DataFrame() # Treat as empty
logger.info(f"Data received - Empty: {data.empty}, Shape: {data.shape if not data.empty else 'N/A'}")
if not data.empty:
logger.debug(f"Data preview:\n{data}")
latest_price = data['Close'].iloc[-1]
logger.info(f"Successfully fetched HPE price: ${latest_price}")
return round(latest_price, 2)
else:
logger.warning("DataFrame is empty - no data returned")
# Try alternative method
logger.info("Trying alternative: stock.info...")
try:
# Force info fetch with error catching
info = stock.info
logger.debug(f"Info keys available: {list(info.keys())[:10]}...")
price = info.get('currentPrice') or info.get('regularMarketPrice') or info.get('previousClose')
if price:
logger.info(f"Got price from info: ${price}")
return round(price, 2)
else:
logger.warning(f"No valid price in info. currentPrice={info.get('currentPrice')}, regularMarketPrice={info.get('regularMarketPrice')}")
except Exception as info_err:
logger.error(f"stock.info failed with error: {info_err}", exc_info=True)
logger.error("All methods failed - No stock data available")
return None
except Exception as e:
logger.error(f"Error fetching HPE stock price: {e}", exc_info=True)
return None
def send_email(self, aud_usd_rate, hpe_price):
"""Send email with financial data"""
try:
# Format subject with data
aud_str = f"{aud_usd_rate:.4f}" if aud_usd_rate else "N/A"
hpe_str = f"${hpe_price}" if hpe_price else "N/A"
msg = MIMEMultipart('alternative')
msg['Subject'] = f"AUD/USD: {aud_str} | HPE: {hpe_str} - Daily Update"
msg['From'] = self.gmail_user
msg['To'] = self.recipient_email
# Create email body
text = f"""
Daily Financial Update
=======================
Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
AUD to USD Exchange Rate: {aud_usd_rate if aud_usd_rate else 'N/A'}
HPE Stock Price: ${hpe_price if hpe_price else 'N/A'}
---
Automated daily update
"""
html = f"""
Daily Financial Update
Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
| Metric |
Value |
| AUD to USD |
{aud_usd_rate if aud_usd_rate else 'N/A'} |
| HPE Stock Price |
${hpe_price if hpe_price else 'N/A'} |
Automated daily update
"""
part1 = MIMEText(text, 'plain')
part2 = MIMEText(html, 'html')
msg.attach(part1)
msg.attach(part2)
# Send email via Gmail SMTP
with smtplib.SMTP_SSL('smtp.gmail.com', 465) as server:
server.login(self.gmail_user, self.gmail_app_password)
server.send_message(msg)
logger.info(f"Email sent successfully")
return True
except Exception as e:
logger.error(f"Error sending email: {e}")
return False
def run_daily_update(self):
"""Fetch data and send email"""
logger.info(f"Running daily update")
check_network()
aud_usd = self.get_exchange_rate()
hpe_price = self.get_hpe_stock_price()
self.send_email(aud_usd, hpe_price)
def main():
emailer = FinanceEmailer()
# Schedule daily at 9:00 AM
schedule.every().day.at("09:00").do(emailer.run_daily_update)
# Run once immediately on startup
logger.info("Starting Finance Emailer...")
logger.info(f"System time: {datetime.now()}")
emailer.run_daily_update()
# Keep running
while True:
schedule.run_pending()
time.sleep(60)
if __name__ == "__main__":
main()