import os from datetime import datetime import re from dotenv import load_dotenv from sendgrid import SendGridAPIClient from sendgrid.helpers.mail import Mail load_dotenv() def build_html_report ( portfolio: dict, metrics: dict, forecasts: dict, sentiment: dict, explanation: str )->str : """Build a rich HTML email report.""" stock_rows = "" for stock, weight in sorted(portfolio.items(), key=lambda x:x[1] , reverse=True): forecast = forecasts.get(stock, {}) ret = forecast.get("return", 0) vol = forecast.get("volatility", 0) sig = sentiment.get(stock, {}).get("signal", "neutral") color = "#00b894" if ret > 0 else "#d63031" sig_icon = "🟢" if sig == "bullish" else \ "🔴" if sig == "bearish" else "⚪" stock_rows += f""" {stock} {weight:.1%} {ret :.2%} ±{vol:.2%} {sig_icon} {sig} """ return f"""

📈 Daily Portfolio Report

{datetime.now().strftime("%A, %B %d, %Y")}

Expected Return

{metrics.get('expected_return', 0):+.2%}

Sharpe Ratio

{metrics.get('sharpe_ratio', 0):.2f}

Volatility

{metrics.get('volatility', 0):.2%}

📊 Portfolio Allocation

{stock_rows}

Stock	Weight	Forecast	Volatility	Sentiment

🤖 AI Portfolio Analysis

{explanation}

⚠️ This report is for informational purposes only. Not financial advice.
Generated by Portfolio Optimizer · LangChain · Groq · GitHub Actions

""" def send_daily_report( results :dict)->bool : """ Send the dailyportfolio report via email Args: results:Pipeline results dict containing portfolio, mertics,etc Returns : True if sent successfully , False otherwise """ print("\n Sending daily portfolio report email ...") html_content = build_html_report( portfolio = results.get("portfolio", {}), metrics = results.get("metrics", {}), forecasts = results.get("forecasts", {}), sentiment = results.get("sentiment", {}), explanation = results.get("explanation", "No explanation available.") ) sender_email = os.getenv("SENDER_EMAIL") recipient_email = os.getenv("RECIPIENT_EMAIL") api_key = os.getenv("SENDGRID_API_KEY") if not all([sender_email, recipient_email, api_key]): print(" ⚠️ Missing email configuration in .env — skipping email") return False message = Mail( from_email = sender_email, to_emails = recipient_email, subject = f"📈 Daily Portfolio Report — " f"{datetime.now().strftime('%B %d, %Y')}", html_content = html_content ) try: sg = SendGridAPIClient(api_key) response = sg.send(message) print(f" ✅ Email sent successfully! Status: {response.status_code}") return True except Exception as e: if hasattr(e, 'body'): print(f" ❌ Email failed: {e.body}") return False if __name__ == "__main__": # Test with mock data mock_results = { "portfolio": {"AAPL": 0.285, "MSFT": 0.321, "GOOGL": 0.154}, "metrics": { "expected_return": 0.184, "volatility": 0.092, "sharpe_ratio": 1.82 }, "forecasts": { "AAPL": {"return": 0.082, "volatility": 0.021}, "MSFT": {"return": 0.051, "volatility": 0.018}, "GOOGL": {"return": -0.023, "volatility": 0.032}, }, "sentiment": { "AAPL": {"signal": "bullish"}, "MSFT": {"signal": "bullish"}, "GOOGL": {"signal": "bearish"}, }, "explanation": "This is a test explanation of the portfolio strategy." } send_daily_report(mock_results)