Financial Data MCP Server

#!/usr/bin/env python3 """ Daily Recommendation Tracking Script Monitors performance of buy recommendations and generates tracking reports """ import json import logging from pathlib import Path from datetime import datetime, timedelta from typing import List, Dict, Any import yfinance as yf import pandas as pd import matplotlib.pyplot as plt import matplotlib.dates as mdates # Configure matplotlib for headless operation import matplotlib matplotlib.use('Agg') # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class RecommendationTracker: def __init__(self): """Initialize the recommendation tracker""" self.project_dir = Path(__file__).parent self.recommendations_file = self.project_dir / "buy_recommendations.json" self.tracking_history_file = self.project_dir / "tracking_history.json" self.tracking_charts_dir = self.project_dir / "tracking_charts" self.tracking_charts_dir.mkdir(exist_ok=True) logger.info("Recommendation Tracker initialized") logger.info(f"Recommendations file: {self.recommendations_file}") logger.info(f"Tracking charts: {self.tracking_charts_dir}") def load_recommendations(self) -> List[Dict[str, Any]]: """Load current buy recommendations""" try: if self.recommendations_file.exists(): with open(self.recommendations_file, 'r') as f: recommendations = json.load(f) logger.info(f"Loaded {len(recommendations)} recommendations") return recommendations else: logger.warning("No recommendations file found") return [] except Exception as e: logger.error(f"Error loading recommendations: {e}") return [] def load_tracking_history(self) -> Dict[str, List[Dict[str, Any]]]: """Load historical tracking data""" try: if self.tracking_history_file.exists(): with open(self.tracking_history_file, 'r') as f: return json.load(f) else: return {} except Exception as e: logger.error(f"Error loading tracking history: {e}") return {} def save_tracking_history(self, history: Dict[str, List[Dict[str, Any]]]): """Save tracking history to file""" try: with open(self.tracking_history_file, 'w') as f: json.dump(history, f, indent=2) except Exception as e: logger.error(f"Error saving tracking history: {e}") def get_current_price(self, symbol: str) -> float: """Get current price for a symbol""" try: ticker = yf.Ticker(symbol) data = ticker.history(period="1d") if not data.empty: return float(data['Close'].iloc[-1]) else: logger.warning(f"No price data for {symbol}") return None except Exception as e: logger.error(f"Error fetching price for {symbol}: {e}") return None def get_price_history(self, symbol: str, days: int = 30) -> pd.DataFrame: """Get price history for a symbol""" try: ticker = yf.Ticker(symbol) data = ticker.history(period=f"{days}d") return data if not data.empty else None except Exception as e: logger.error(f"Error fetching history for {symbol}: {e}") return None def calculate_performance(self, initial_price: float, current_price: float) -> Dict[str, Any]: """Calculate performance metrics""" if not initial_price or not current_price: return {"change_pct": 0, "change_abs": 0, "status": "unknown"} change_abs = current_price - initial_price change_pct = (change_abs / initial_price) * 100 # Determine status based on performance if change_pct >= 5: status = "strong_gain" elif change_pct >= 2: status = "moderate_gain" elif change_pct >= -2: status = "neutral" elif change_pct >= -5: status = "moderate_loss" else: status = "strong_loss" return { "change_pct": round(change_pct, 2), "change_abs": round(change_abs, 2), "status": status } def create_tracking_chart(self, symbol: str, recommendation_data: Dict[str, Any], price_history: pd.DataFrame) -> str: """Create tracking chart showing performance since recommendation""" try: fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10)) fig.suptitle(f'{symbol} - Performance Tracking Since Recommendation', fontsize=16, fontweight='bold') # Chart 1: Price with recommendation date marked ax1.plot(price_history.index, price_history['Close'], linewidth=2, color='blue', label='Price') # Mark recommendation date if available rec_date = recommendation_data.get('recommendation_date') if rec_date and recommendation_data.get('initial_price'): try: rec_datetime = datetime.strptime(rec_date, '%Y-%m-%d') initial_price = recommendation_data['initial_price'] ax1.axvline(x=rec_datetime, color='green', linestyle='--', linewidth=2, label=f'Recommendation Date') ax1.axhline(y=initial_price, color='orange', linestyle=':', alpha=0.7, label=f'Entry Price: ${initial_price:.2f}') except: pass # Current price line current_price = price_history['Close'].iloc[-1] ax1.axhline(y=current_price, color='red', linestyle=':', alpha=0.7, label=f'Current: ${current_price:.2f}') ax1.set_title(f'Price Movement - {symbol}') ax1.set_ylabel('Price ($)') ax1.legend() ax1.grid(True, alpha=0.3) # Format x-axis ax1.xaxis.set_major_formatter(mdates.DateFormatter('%m/%d')) ax1.xaxis.set_major_locator(mdates.WeekdayLocator(interval=1)) # Chart 2: Volume ax2.bar(price_history.index, price_history['Volume'], alpha=0.6, color='gray') ax2.set_title('Volume') ax2.set_ylabel('Volume') ax2.set_xlabel('Date') # Add performance text box performance = recommendation_data.get('performance', {}) perf_text = f""" PERFORMANCE SUMMARY: • Recommendation: {recommendation_data.get('recommendation', 'N/A')} • Confidence: {recommendation_data.get('confidence', 'N/A')} • Entry Price: ${recommendation_data.get('initial_price', 0):.2f} • Current Price: ${recommendation_data.get('current_price', 0):.2f} • Change: {performance.get('change_pct', 0):+.2f}% • Status: {performance.get('status', 'unknown').replace('_', ' ').title()} """.strip() ax1.text(0.02, 0.98, perf_text, transform=ax1.transAxes, verticalalignment='top', fontsize=9, bbox=dict(boxstyle='round,pad=0.5', facecolor='lightblue', alpha=0.8)) plt.tight_layout() # Save chart timestamp = datetime.now().strftime("%Y%m%d") chart_filename = f"{symbol}_tracking_{timestamp}.png" chart_path = self.tracking_charts_dir / chart_filename plt.savefig(chart_path, dpi=300, bbox_inches='tight') plt.close() logger.info(f"Tracking chart saved: {chart_path}") return str(chart_path) except Exception as e: logger.error(f"Error creating tracking chart for {symbol}: {e}") plt.close() return "" def update_recommendations(self, recommendations: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Update recommendations with current data""" updated_recommendations = [] for rec in recommendations: symbol = rec['symbol'] logger.info(f"Updating {symbol}") # Get current price current_price = self.get_current_price(symbol) if current_price is None: logger.warning(f"Could not get current price for {symbol}") updated_recommendations.append(rec) continue # Set initial price if not set if rec.get('initial_price') is None: rec['initial_price'] = current_price logger.info(f"Set initial price for {symbol}: ${current_price:.2f}") # Update current price rec['current_price'] = current_price rec['last_updated'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # Calculate performance performance = self.calculate_performance(rec.get('initial_price'), current_price) rec['performance'] = performance # Get price history and create chart price_history = self.get_price_history(symbol, days=30) if price_history is not None: chart_path = self.create_tracking_chart(symbol, rec, price_history) rec['latest_chart'] = chart_path updated_recommendations.append(rec) logger.info(f"{symbol}: ${rec.get('initial_price', 0):.2f} → ${current_price:.2f} " f"({performance.get('change_pct', 0):+.2f}%)") return updated_recommendations def generate_tracking_report(self, recommendations: List[Dict[str, Any]]) -> str: """Generate tracking report""" report_filename = f"tracking_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" report_path = self.tracking_charts_dir / report_filename try: with open(report_path, 'w') as f: f.write("="*80 + "\n") f.write("DAILY RECOMMENDATION TRACKING REPORT\n") f.write("="*80 + "\n") f.write(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"Total recommendations tracked: {len(recommendations)}\n\n") # Performance summary total_performance = 0 active_count = 0 status_counts = {} for rec in recommendations: if rec.get('status') == 'active' and rec.get('performance'): perf = rec['performance'].get('change_pct', 0) total_performance += perf active_count += 1 status = rec['performance'].get('status', 'unknown') status_counts[status] = status_counts.get(status, 0) + 1 avg_performance = total_performance / active_count if active_count > 0 else 0 f.write("PERFORMANCE SUMMARY:\n") f.write(f"• Average Performance: {avg_performance:.2f}%\n") f.write(f"• Active Recommendations: {active_count}\n") f.write("\nStatus Breakdown:\n") for status, count in status_counts.items(): f.write(f"• {status.replace('_', ' ').title()}: {count}\n") f.write("\n") # Detailed tracking f.write("DETAILED TRACKING:\n") f.write("-"*80 + "\n") # Sort by performance sorted_recs = sorted(recommendations, key=lambda x: x.get('performance', {}).get('change_pct', 0), reverse=True) for rec in sorted_recs: f.write(f"Symbol: {rec['symbol']}\n") f.write(f"Recommendation: {rec.get('recommendation', 'N/A')}\n") f.write(f"Confidence: {rec.get('confidence', 'N/A')}\n") f.write(f"Recommendation Date: {rec.get('recommendation_date', 'N/A')}\n") f.write(f"Entry Price: ${rec.get('initial_price', 0):.2f}\n") f.write(f"Current Price: ${rec.get('current_price', 0):.2f}\n") if rec.get('performance'): perf = rec['performance'] f.write(f"Performance: {perf.get('change_pct', 0):+.2f}% " f"(${perf.get('change_abs', 0):+.2f})\n") f.write(f"Status: {perf.get('status', 'unknown').replace('_', ' ').title()}\n") f.write(f"Last Updated: {rec.get('last_updated', 'N/A')}\n") f.write("-"*40 + "\n") # Recommendations for action f.write("\nACTION RECOMMENDATIONS:\n") f.write("="*50 + "\n") strong_gains = [r for r in