""" Model Context Protocol (MCP) Server - Tool integration layer (Free Version) Lightweight server for tool execution without external dependencies """ import asyncio import json import uuid from typing import Dict, List, Any, Optional, Callable from datetime import datetime from dataclasses import dataclass from enum import Enum import numpy as np import math import random from scipy import stats import statistics # Import free tools from tools.cfa_calculators import CFACalculatorTools from tools.data_analyzers import DataAnalyzerTools from tools.learning_tools import LearningTools @dataclass class ToolRequest: """Tool execution request""" tool_name: str parameters: Dict[str, Any] request_id: str = None def __post_init__(self): if not self.request_id: self.request_id = f"req_{uuid.uuid4().hex[:8]}" @dataclass class ToolResponse: """Tool execution response""" success: bool result: Optional[Any] = None error: Optional[str] = None execution_time_ms: float = 0.0 request_id: Optional[str] = None tool_name: Optional[str] = None class MCPServer: """Lightweight MCP Server for tool integration (Free Version)""" def __init__(self): self.tools = self._initialize_tools() self.request_history = [] self.active_connections = set() # Performance metrics self.metrics = { 'total_requests': 0, 'successful_requests': 0, 'average_response_time': 0.0, 'tools_available': len(self._get_all_tools()) } def _initialize_tools(self) -> Dict[str, Any]: """Initialize all available tools""" return { 'calculators': CFACalculatorTools(), 'analyzers': DataAnalyzerTools(), 'learning': LearningTools() } def _get_all_tools(self) -> Dict[str, List[str]]: """Get all available tools""" all_tools = {} for category, tool_instance in self.tools.items(): if hasattr(tool_instance, 'list_tools'): all_tools[category] = tool_instance.list_tools() else: # Get public methods methods = [m for m in dir(tool_instance) if not m.startswith('_') and callable(getattr(tool_instance, m))] all_tools[category] = methods return all_tools async def execute_tool(self, tool_request: ToolRequest) -> ToolResponse: """Execute a tool and return response""" start_time = datetime.now() try: # Parse tool name (format: category.method or just method) if '.' in tool_request.tool_name: category, method = tool_request.tool_name.split('.', 1) else: # Try to find the tool in any category category, method = self._find_tool(tool_request.tool_name) if category not in self.tools: return ToolResponse( success=False, error=f"Tool category '{category}' not found", request_id=tool_request.request_id, tool_name=tool_request.tool_name ) tool_instance = self.tools[category] if not hasattr(tool_instance, method): return ToolResponse( success=False, error=f"Tool '{method}' not found in category '{category}'", request_id=tool_request.request_id, tool_name=tool_request.tool_name ) # Get tool method tool_method = getattr(tool_instance, method) # Execute tool if asyncio.iscoroutinefunction(tool_method): result = await tool_method(**tool_request.parameters) else: # Run sync method in thread pool result = await asyncio.get_event_loop().run_in_executor( None, lambda: tool_method(**tool_request.parameters) ) # Calculate execution time exec_time = (datetime.now() - start_time).total_seconds() * 1000 # Update metrics self._update_metrics(True, exec_time) return ToolResponse( success=True, result=result, execution_time_ms=exec_time, request_id=tool_request.request_id, tool_name=tool_request.tool_name ) except Exception as e: exec_time = (datetime.now() - start_time).total_seconds() * 1000 # Update metrics self._update_metrics(False, exec_time) return ToolResponse( success=False, error=f"Tool execution error: {str(e)}", execution_time_ms=exec_time, request_id=tool_request.request_id, tool_name=tool_request.tool_name ) def _find_tool(self, tool_name: str) -> tuple: """Find which category contains a tool""" for category, tool_instance in self.tools.items(): if hasattr(tool_instance, tool_name): return category, tool_name # Default to calculators return 'calculators', tool_name def _update_metrics(self, success: bool, exec_time: float): """Update performance metrics""" self.metrics['total_requests'] += 1 if success: self.metrics['successful_requests'] += 1 # Update average response time current_avg = self.metrics['average_response_time'] n = self.metrics['total_requests'] self.metrics['average_response_time'] = ( (current_avg * (n - 1) + exec_time) / n if n > 1 else exec_time ) def get_metrics(self) -> Dict[str, Any]: """Get server metrics""" return { **self.metrics, 'success_rate': ( self.metrics['successful_requests'] / self.metrics['total_requests'] if self.metrics['total_requests'] > 0 else 0 ), 'uptime': datetime.now().isoformat() # Simple uptime indicator } def list_tools(self) -> Dict[str, Any]: """List all available tools""" return { 'categories': self._get_all_tools(), 'total_tools': sum(len(tools) for tools in self._get_all_tools().values()), 'server_status': 'running', 'version': '1.0.0' } # Tool implementations (Free versions) class CFACalculatorTools: """CFA-specific calculation tools (Free Version)""" def list_tools(self) -> List[str]: return [ 'calculate_npv', 'calculate_irr', 'calculate_sharpe_ratio', 'calculate_sortino_ratio', 'calculate_beta', 'calculate_alpha', 'calculate_treynor_ratio', 'calculate_jensens_alpha', 'calculate_modigliani_ratio', 'calculate_information_ratio', 'calculate_present_value', 'calculate_future_value', 'calculate_annuity_pv', 'calculate_annuity_fv', 'calculate_perpetuity_pv' ] def calculate_npv(self, cash_flows: List[float], discount_rate: float) -> Dict[str, Any]: """Calculate Net Present Value""" try: npv = 0 for t, cf in enumerate(cash_flows): npv += cf / ((1 + discount_rate) ** t) return { 'npv': float(npv), 'cash_flows': cash_flows, 'discount_rate': discount_rate, 'periods': len(cash_flows), 'interpretation': 'Positive NPV indicates value creation' if npv > 0 else 'Negative NPV indicates value destruction' } except Exception as e: return {'error': str(e)} def calculate_irr(self, cash_flows: List[float], max_iterations: int = 100) -> Dict[str, Any]: """Calculate Internal Rate of Return using iterative approximation""" try: if len(cash_flows) < 2: return {'error': 'Need at least two cash flows'} def npv_func(rate): return sum(cf / ((1 + rate) ** i) for i, cf in enumerate(cash_flows)) # Simple iterative approach lower, upper = -0.99, 10.0 tolerance = 0.0001 for _ in range(max_iterations): mid = (lower + upper) / 2 npv_mid = npv_func(mid) if abs(npv_mid) < tolerance: irr = mid break if npv_func(lower) * npv_mid < 0: upper = mid else: lower = mid else: irr = (lower + upper) / 2 return { 'irr': float(irr), 'cash_flows': cash_flows, 'interpretation': f'IRR of {irr:.1%} is the discount rate that makes NPV zero' } except Exception as e: return {'error': str(e)} def calculate_sharpe_ratio(self, returns: List[float], risk_free_rate: float) -> Dict[str, Any]: """Calculate Sharpe Ratio""" try: if len(returns) < 2: return {'error': 'Need at least 2 returns'} excess_returns = [r - risk_free_rate for r in returns] avg_excess = statistics.mean(excess_returns) std_dev = statistics.stdev(returns) if len(returns) > 1 else 0 if std_dev == 0: sharpe = float('inf') if avg_excess > 0 else float('-inf') else: sharpe = avg_excess / std_dev return { 'sharpe_ratio': float(sharpe), 'avg_return': statistics.mean(returns), 'std_dev': std_dev, 'risk_free_rate': risk_free_rate, 'interpretation': 'Higher Sharpe ratio indicates better risk-adjusted returns' } except Exception as e: return {'error': str(e)} def calculate_sortino_ratio(self, returns: List[float], risk_free_rate: float, target_return: float = 0) -> Dict[str, Any]: """Calculate Sortino Ratio""" try: if len(returns) < 2: return {'error': 'Need at least 2 returns'} # Calculate downside returns downside_returns = [r for r in returns if r < target_return] if not downside_returns: sortino = float('inf') downside_deviation = 0 elif len(downside_returns) == 1: sortino = float('inf') downside_deviation = 0 else: downside_deviation = statistics.stdev(downside_returns) avg_return = statistics.mean(returns) sortino = (avg_return - risk_free_rate) / downside_deviation if downside_deviation != 0 else float('inf') return { 'sortino_ratio': float(sortino), 'avg_return': statistics.mean(returns), 'downside_deviation': downside_deviation, 'risk_free_rate': risk_free_rate, 'target_return': target_return, 'interpretation': 'Higher Sortino ratio indicates better downside risk-adjusted returns' } except Exception as e: return {'error': str(e)} def calculate_beta(self, asset_returns: List[float], market_returns: List[float]) -> Dict[str, Any]: """Calculate Beta coefficient""" try: if len(asset_returns) != len(market_returns): return {'error': 'Asset and market returns must have same length'} if len(asset_returns) < 2: return {'error': 'Need at least 2 data points'} # Calculate covariance and variance cov_matrix = np.cov(asset_returns, market_returns) covariance = cov_matrix[0, 1] market_variance = np.var(market_returns) beta = covariance / market_variance if market_variance != 0 else 0 return { 'beta': float(beta), 'covariance': float(covariance), 'market_variance': float(market_variance), 'interpretation': f'Beta of {beta:.2f}: Stock is {"more" if beta > 1 else "less"} volatile than market' } except Exception as e: return {'error': str(e)} def calculate_alpha(self, asset_returns: List[float], market_returns: List[float], risk_free_rate: float) -> Dict[str, Any]: """Calculate Alpha""" try: beta_result = self.calculate_beta(asset_returns, market_returns) if 'error' in beta_result: return beta_result beta = beta_result['beta'] avg_asset_return = statistics.mean(asset_returns) avg_market_return = statistics.mean(market_returns) expected_return = risk_free_rate + beta * (avg_market_return - risk_free_rate) alpha = avg_asset_return - expected_return return { 'alpha': float(alpha), 'beta': beta, 'expected_return': float(expected_return), 'actual_return': float(avg_asset_return), 'interpretation': f'Alpha of {alpha:.2%}: {"Outperformed" if alpha > 0 else "Underperformed"} expectations by {abs(alpha):.2%}' } except Exception as e: return {'error': str(e)} def calculate_treynor_ratio(self, returns: List[float], beta: float, risk_free_rate: float) -> Dict[str, Any]: """Calculate Treynor Ratio""" try: if len(returns) < 1: return {'error': 'Need at least 1 return'} avg_return = statistics.mean(returns) if beta == 0: treynor = float('inf') if avg_return > risk_free_rate else float('-inf') else: treynor = (avg_return - risk_free_rate) / beta return { 'treynor_ratio': float(treynor), 'avg_return': float(avg_return), 'beta': beta, 'risk_free_rate': risk_free_rate, 'interpretation': 'Higher Treynor ratio indicates better systematic risk-adjusted returns' } except Exception as e: return {'error': str(e)} def calculate_jensens_alpha(self, returns: List[float], market_returns: List[float], risk_free_rate: float) -> Dict[str, Any]: """Calculate Jensen's Alpha (same as alpha)""" return self.calculate_alpha(returns, market_returns, risk_free_rate) def calculate_modigliani_ratio(self, returns: List[float], market_returns: List[float], risk_free_rate: float) -> Dict[str, Any]: """Calculate Modigliani Ratio (M2)""" try: sharpe_result = self.calculate_sharpe_ratio(returns, risk_free_rate) if 'error' in sharpe_result: return sharpe_result market_sharpe_result = self.calculate_sharpe_ratio(market_returns, risk_free_rate) if 'error' in market_sharpe_result: return market_sharpe_result sharpe_ratio = sharpe_result['sharpe_ratio'] market_std = np.std(market_returns) if len(market_returns) > 1 else 0 m2 = risk_free_rate + sharpe_ratio * market_std return { 'm2_ratio': float(m2), 'sharpe_ratio': float(sharpe_ratio), 'market_std': float(market_std), 'risk_free_rate': risk_free_rate, 'interpretation': 'M2 measures risk-adjusted returns relative to market risk' } except Exception as e: return {'error': str(e)} def calculate_information_ratio(self, portfolio_returns: List[float], benchmark_returns: List[float]) -> Dict[str, Any]: """Calculate Information Ratio""" try: if len(portfolio_returns) != len(benchmark_returns): return {'error': 'Portfolio and benchmark returns must have same length'} if len(portfolio_returns) < 2: return {'error': 'Need at least 2 data points'} active_returns = [p - b for p, b in zip(portfolio_returns, benchmark_returns)] avg_active_return = statistics.mean(active_returns) tracking_error = statistics.stdev(active_returns) if len(active_returns) > 1 else 0 if tracking_error == 0: info_ratio = float('inf') if avg_active_return > 0 else float('-inf') else: info_ratio = avg_active_return / tracking_error return { 'information_ratio': float(info_ratio), 'avg_active_return': float(avg_active_return), 'tracking_error': float(tracking_error), 'interpretation': 'Higher information ratio indicates better benchmark-relative performance' } except Exception as e: return {'error': str(e)} def calculate_present_value(self, future_value: float, rate: float, periods: int) -> Dict[str, Any]: """Calculate Present Value""" try: pv = future_value / ((1 + rate) ** periods) return { 'present_value': float(pv), 'future_value': future_value, 'rate': rate, 'periods': periods, 'formula': 'PV = FV / (1 + r)^n' } except Exception as e: return {'error': str(e)} def calculate_future_value(self, present_value: float, rate: float, periods: int) -> Dict[str, Any]: """Calculate Future Value""" try: fv = present_value * ((1 + rate) ** periods) return { 'future_value': float(fv), 'present_value': present_value, 'rate': rate, 'periods': periods, 'formula': 'FV = PV × (1 + r)^n' } except Exception as e: return {'error': str(e)} def calculate_annuity_pv(self, payment: float, rate: float, periods: int) -> Dict[str, Any]: """Calculate Present Value of Ordinary Annuity""" try: if rate == 0: pv = payment * periods else: pv = payment * (1 - (1 + rate) ** -periods) / rate return { 'present_value': float(pv), 'payment': payment, 'rate': rate, 'periods': periods, 'formula': 'PV = PMT × [1 - (1 + r)^-n] / r' } except Exception as e: return {'error': str(e)} def calculate_annuity_fv(self, payment: float, rate: float, periods: int) -> Dict[str, Any]: """Calculate Future Value of Ordinary Annuity""" try: if rate == 0: fv = payment * periods else: fv = payment * ((1 + rate) ** periods - 1) / rate return { 'future_value': float(fv), 'payment': payment, 'rate': rate, 'periods': periods, 'formula': 'FV = PMT × [(1 + r)^n - 1] / r' } except Exception as e: return {'error': str(e)} def calculate_perpetuity_pv(self, payment: float, rate: float) -> Dict[str, Any]: """Calculate Present Value of Perpetuity""" try: if rate == 0: return {'error': 'Rate cannot be zero for perpetuity'} pv = payment / rate return { 'present_value': float(pv), 'payment': payment, 'rate': rate, 'formula': 'PV = PMT / r' } except Exception as e: return {'error': str(e)} class DataAnalyzerTools: """Data analysis tools (Free Version)""" def list_tools(self) -> List[str]: return [ 'analyze_time_series', 'calculate_correlation_matrix', 'perform_regression_analysis', 'calculate_rolling_statistics', 'detect_outliers', 'calculate_volatility', 'perform_portfolio_optimization', 'calculate_var', 'calculate_covariance', 'calculate_correlation' ] def analyze_time_series(self, data: List[float]) -> Dict[str, Any]: """Analyze time series data""" try: if len(data) < 2: return {'error': 'Need at least 2 data points'} returns = np.diff(data) / data[:-1] return { 'summary_stats': { 'mean': float(np.mean(data)), 'median': float(np.median(data)), 'std': float(np.std(data)), 'min': float(np.min(data)), 'max': float(np.max(data)), 'range': float(np.max(data) - np.min(data)) }, 'return_stats': { 'mean_return': float(np.mean(returns)) if len(returns) > 0 else 0, 'std_return': float(np.std(returns)) if len(returns) > 0 else 0, 'min_return': float(np.min(returns)) if len(returns) > 0 else 0, 'max_return': float(np.max(returns)) if len(returns) > 0 else 0 }, 'autocorrelation': float(np.corrcoef(data[:-1], data[1:])[0, 1]) if len(data) > 2 else 0 } except Exception as e: return {'error': str(e)} def calculate_correlation_matrix(self, data: Dict[str, List[float]]) -> Dict[str, Any]: """Calculate correlation matrix for multiple time series""" try: # Check all series have same length lengths = [len(series) for series in data.values()] if len(set(lengths)) > 1: return {'error': 'All time series must have same length'} if lengths[0] < 2: return {'error': 'Need at least 2 data points per series'} # Create numpy array series_names = list(data.keys()) series_data = np.array([data[name] for name in series_names]) # Calculate correlation matrix corr_matrix = np.corrcoef(series_data) # Format as dictionary corr_dict = {} for i, name1 in enumerate(series_names): corr_dict[name1] = {} for j, name2 in enumerate(series_names): corr_dict[name1][name2] = float(corr_matrix[i, j]) return { 'correlation_matrix': corr_dict, 'series_names': series_names, 'data_points': lengths[0] } except Exception as e: return {'error': str(e)} def perform_regression_analysis(self, x: List[float], y: List[float]) -> Dict[str, Any]: """Perform linear regression analysis""" try: if len(x) != len(y): return {'error': 'X and Y must have same length'} if len(x) < 2: return {'error': 'Need at least 2 data points'} # Calculate regression using numpy slope, intercept = np.polyfit(x, y, 1) # Calculate R-squared y_pred = slope * np.array(x) + intercept ss_res = np.sum((np.array(y) - y_pred) ** 2) ss_tot = np.sum((np.array(y) - np.mean(y)) ** 2) r_squared = 1 - (ss_res / ss_tot) if ss_tot != 0 else 0 # Calculate standard error n = len(x) std_err = np.sqrt(ss_res / (n - 2)) if n > 2 else 0 return { 'slope': float(slope), 'intercept': float(intercept), 'r_squared': float(r_squared), 'std_error': float(std_err), 'equation': f'y = {slope:.4f}x + {intercept:.4f}', 'data_points': n } except Exception as e: return {'error': str(e)} def calculate_rolling_statistics(self, data: List[float], window: int = 20) -> Dict[str, Any]: """Calculate rolling statistics""" try: if len(data) < window: return {'error': f'Need at least {window} data points'} # Convert to numpy array arr = np.array(data) # Calculate rolling statistics rolling_mean = np.convolve(arr, np.ones(window)/window, mode='valid') rolling_std = np.array([ np.std(arr[i:i+window]) for i in range(len(arr) - window + 1) ]) return { 'rolling_mean': rolling_mean.tolist(), 'rolling_std': rolling_std.tolist(), 'window': window, 'data_points': len(data), 'rolling_points': len(rolling_mean) } except Exception as e: return {'error': str(e)} def detect_outliers(self, data: List[float], method: str = 'iqr') -> Dict[str, Any]: """Detect outliers in data""" try: if len(data) < 3: return {'error': 'Need at least 3 data points'} if method == 'iqr': q1 = np.percentile(data, 25) q3 = np.percentile(data, 75) iqr = q3 - q1 lower_bound = q1 - 1.5 * iqr upper_bound = q3 + 1.5 * iqr outliers = [float(x) for x in data if x < lower_bound or x > upper_bound] return { 'method': 'iqr', 'q1': float(q1), 'q3': float(q3), 'iqr': float(iqr), 'lower_bound': float(lower_bound), 'upper_bound': float(upper_bound), 'outliers': outliers, 'outlier_count': len(outliers), 'outlier_percentage': len(outliers) / len(data) } elif method == 'zscore': mean = np.mean(data) std = np.std(data) if std == 0: return {'error': 'Standard deviation is zero'} z_scores = [(x - mean) / std for x in data] outliers = [float(data[i]) for i, z in enumerate(z_scores) if abs(z) > 3] return { 'method': 'zscore', 'mean': float(mean), 'std': float(std), 'outliers': outliers, 'outlier_count': len(outliers), 'outlier_percentage': len(outliers) / len(data) } else: return {'error': f'Unknown method: {method}'} except Exception as e: return {'error': str(e)} def calculate_volatility(self, returns: List[float], annualization_factor: int = 252) -> Dict[str, Any]: """Calculate volatility metrics""" try: if len(returns) < 2: return {'error': 'Need at least 2 returns'} daily_vol = np.std(returns) annual_vol = daily_vol * np.sqrt(annualization_factor) return { 'daily_volatility': float(daily_vol), 'annual_volatility': float(annual_vol), 'variance': float(np.var(returns)), 'annualization_factor': annualization_factor, 'data_points': len(returns) } except Exception as e: return {'error': str(e)} def perform_portfolio_optimization(self, returns: Dict[str, List[float]], risk_free_rate: float = 0.02) -> Dict[str, Any]: """Perform basic portfolio optimization""" try: # Check all series have same length lengths = [len(series) for series in returns.values()] if len(set(lengths)) > 1: return {'error': 'All return series must have same length'} if lengths[0] < 2: return {'error': 'Need at least 2 data points per series'} assets = list(returns.keys()) n_assets = len(assets) # Convert to numpy array returns_array = np.array([returns[asset] for asset in assets]) # Calculate mean returns and covariance matrix mean_returns = np.mean(returns_array, axis=1) cov_matrix = np.cov(returns_array) # Generate random portfolios n_portfolios = 10000 results = np.zeros((3, n_portfolios)) for i in range(n_portfolios): # Generate random weights weights = np.random.random(n_assets) weights /= np.sum(weights) # Calculate portfolio statistics port_return = np.sum(mean_returns * weights) port_std = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights))) sharpe_ratio = (port_return - risk_free_rate) / port_std if port_std != 0 else 0 results[0, i] = port_return results[1, i] = port_std results[2, i] = sharpe_ratio # Find optimal portfolios max_sharpe_idx = np.argmax(results[2]) min_vol_idx = np.argmin(results[1]) return { 'max_sharpe': { 'return': float(results[0, max_sharpe_idx]), 'volatility': float(results[1, max_sharpe_idx]), 'sharpe_ratio': float(results[2, max_sharpe_idx]) }, 'min_volatility': { 'return': float(results[0, min_vol_idx]), 'volatility': float(results[1, min_vol_idx]), 'sharpe_ratio': float(results[2, min_vol_idx]) }, 'assets': assets, 'n_portfolios': n_portfolios, 'risk_free_rate': risk_free_rate } except Exception as e: return {'error': str(e)} def calculate_var(self, returns: List[float], confidence_level: float = 0.95, method: str = 'historical') -> Dict[str, Any]: """Calculate Value at Risk""" try: if len(returns) < 2: return {'error': 'Need at least 2 returns'} if method == 'historical': var = np.percentile(returns, (1 - confidence_level) * 100) return { 'var': float(var), 'confidence_level': confidence_level, 'method': 'historical', 'data_points': len(returns) } elif method == 'parametric': mean = np.mean(returns) std = np.std(returns) # Use normal distribution approximation from scipy import stats z_score = stats.norm.ppf(confidence_level) var = mean - z_score * std return { 'var': float(var), 'confidence_level': confidence_level, 'method': 'parametric', 'mean': float(mean), 'std': float(std), 'z_score': float(z_score) } else: return {'error': f'Unknown VaR method: {method}'} except Exception as e: return {'error': str(e)} def calculate_covariance(self, x: List[float], y: List[float]) -> Dict[str, Any]: """Calculate covariance between two series""" try: if len(x) != len(y): return {'error': 'X and Y must have same length'} if len(x) < 2: return {'error': 'Need at least 2 data points'} covariance = np.cov(x, y)[0, 1] return { 'covariance': float(covariance), 'data_points': len(x), 'interpretation': 'Positive covariance indicates variables move together' } except Exception as e: return {'error': str(e)} def calculate_correlation(self, x: List[float], y: List[float]) -> Dict[str, Any]: """Calculate correlation between two series""" try: if len(x) != len(y): return {'error': 'X and Y must have same length'} if len(x) < 2: return {'error': 'Need at least 2 data points'} correlation = np.corrcoef(x, y)[0, 1] return { 'correlation': float(correlation), 'data_points': len(x), 'interpretation': f'Correlation of {correlation:.2f}: {"Strong" if abs(correlation) > 0.7 else "Moderate" if abs(correlation) > 0.3 else "Weak"} {"positive" if correlation > 0 else "negative"} relationship' } except Exception as e: return {'error': str(e)} class LearningTools: """Learning and assessment tools (Free Version)""" def list_tools(self) -> List[str]: return [ 'generate_quiz', 'assess_knowledge_gaps', 'create_study_plan', 'simulate_exam', 'track_progress', 'recommend_resources', 'generate_flashcards', 'calculate_readiness_score', 'generate_practice_problem', 'create_study_schedule' ] def generate_quiz(self, topic: str, difficulty: str = 'medium', num_questions: int = 5) -> Dict[str, Any]: """Generate a quiz on a specific topic""" try: # Quiz question templates question_templates = { 'Time Value of Money': [ { 'question': 'What is the future value of $1,000 invested at 5% annual interest for 3 years?', 'options': ['$1,157.63', '$1,150.00', '$1,200.00', '$1,050.00'], 'correct': 0, 'explanation': 'FV = PV × (1 + r)^n = 1000 × (1.05)^3 = $1,157.63' }, { 'question': 'Which of the following is NOT a TVM concept?', 'options': ['Net Present Value', 'Internal Rate of Return', 'Standard Deviation', 'Annuity'], 'correct': 2, 'explanation': 'Standard Deviation is a statistical concept, not specifically TVM' } ], 'Statistics': [ { 'question': 'What does a standard deviation of 0 indicate?', 'options': ['No variability in data', 'Maximum variability', 'Normal distribution', 'Positive skew'], 'correct': 0, 'explanation': 'Standard deviation of 0 means all values are identical' }, { 'question': 'In a normal distribution, what percentage of data falls within ±1 standard deviation?', 'options': ['68%', '95%', '99.7%', '50%'], 'correct': 0, 'explanation': '68% of data falls within ±1 standard deviation in a normal distribution' } ], 'Portfolio Management': [ { 'question': 'What does Beta measure in CAPM?', 'options': ['Systematic risk', 'Unsystematic risk', 'Total risk', 'Default risk'], 'correct': 0, 'explanation': 'Beta measures systematic (market) risk' }, { 'question': 'Which portfolio lies on the efficient frontier?', 'options': ['Maximum return for given risk', 'Minimum risk regardless of return', 'Maximum Sharpe ratio', 'All of the above'], 'correct': 3, 'explanation': 'All portfolios on the efficient frontier are optimal' } ] } # Get questions for topic or use general questions if topic in question_templates: templates = question_templates[topic] else: templates = [ { 'question': f'What is a key concept in {topic}?', 'options': ['Concept A', 'Concept B', 'Concept C', 'Concept D'], 'correct': 0, 'explanation': f'This tests understanding of {topic} fundamentals' }, { 'question': f'Which formula is commonly used in {topic}?', 'options': ['Formula A', 'Formula B', 'Formula C', 'Formula D'], 'correct': 1, 'explanation': f'This tests knowledge of {topic} calculations' } ] # Generate quiz quiz = [] for i in range(min(num_questions, len(templates))): template = templates[i % len(templates)] quiz.append({ 'id': f'q_{i+1}', 'question': template['question'], 'options': template['options'], 'correct_answer': template['correct'], 'explanation': template['explanation'], 'points': 1, 'difficulty': difficulty }) return { 'quiz': quiz, 'topic': topic, 'difficulty': difficulty, 'total_questions': len(quiz), 'total_points': len(quiz), 'time_limit': len(quiz) * 2 # 2 minutes per question } except Exception as e: return {'error': str(e)} def assess_knowledge_gaps(self, performance_data: Dict[str, float], user_profile: Dict[str, Any]) -> Dict[str, Any]: """Assess knowledge gaps based on performance""" try: weak_areas = [] strong_areas = [] for topic, score in performance_data.items(): if score < 0.7: weak_areas.append({ 'topic': topic, 'score': score, 'priority': 'high' if score < 0.5 else 'medium', 'recommendation': self._get_topic_recommendation(topic, score) }) else: strong_areas.append({ 'topic': topic, 'score': score, 'recommendation': f'Maintain proficiency in {topic}' }) # Calculate overall metrics overall_score = np.mean(list(performance_data.values())) if performance_data else 0 return { 'weak_areas': weak_areas, 'strong_areas': strong_areas, 'overall_score': float(overall_score), 'gap_count': len(weak_areas), 'study_priority': sorted(weak_areas, key=lambda x: x['score'])[:3] if weak_areas else [], 'confidence': min(overall_score * 1.2, 0.95) } except Exception as e: return {'error': str(e)} def _get_topic_recommendation(self, topic: str, score: float) -> str: """Get recommendation for a topic based on score""" recommendations = { 'Time Value of Money': [ 'Practice discounting cash flows with different rates', 'Master annuity and perpetuity calculations', 'Understand the relationship between PV and FV' ], 'Statistics': [ 'Practice calculating and interpreting standard deviation', 'Work on hypothesis testing problems', 'Understand correlation vs causation' ], 'Portfolio Management': [ 'Practice calculating portfolio variance', 'Understand the Capital Asset Pricing Model (CAPM)', 'Work through efficient frontier examples' ] } if topic in recommendations: recs = recommendations[topic] if score < 0.5: return recs[0] # Most basic elif score < 0.7: return recs[1] # Intermediate else: return recs[2] # Advanced return f'Review {topic} fundamentals and practice problems' def create_study_plan(self, topics: List[str], available_hours: float, exam_date: Optional[str] = None) -> Dict[str, Any]: """Create a personalized study plan""" try: from datetime import datetime, timedelta if exam_date: exam_date_obj = datetime.strptime(exam_date, '%Y-%m-%d') days_until_exam = (exam_date_obj - datetime.now()).days if days_until_exam <= 0: days_until_exam = 30 else: days_until_exam = 30 # Topic weights (based on CFA curriculum) topic_weights = { 'Ethics': 0.15, 'Quantitative Methods': 0.12, 'Economics': 0.10, 'Financial Reporting': 0.13, 'Corporate Finance': 0.10, 'Equity': 0.11, 'Fixed Income': 0.11, 'Derivatives': 0.08, 'Alternative Investments': 0.06, 'Portfolio Management': 0.14 } # Create study plan study_plan = [] total_hours = 0 for topic in topics: weight = topic_weights.get(topic, 0.1) topic_hours = available_hours * weight study_plan.append({ 'topic': topic, 'allocated_hours': float(topic_hours), 'priority': 'high' if weight > 0.12 else 'medium' if weight > 0.08 else 'low', 'weekly_target': float(topic_hours / (days_until_exam / 7)), 'sessions': int(np.ceil(topic_hours / 2)) # 2-hour sessions }) total_hours += topic_hours hours_per_day = total_hours / days_until_exam return { 'study_plan': study_plan, 'total_hours': float(total_hours), 'hours_per_day': float(hours_per_day), 'days_until_exam': days_until_exam, 'completion_date': (datetime.now() + timedelta(days=days_until_exam)).strftime('%Y-%m-%d'), 'recommendation': f'Aim for {hours_per_day:.1f} hours per day' } except Exception as e: return {'error': str(e)} def simulate_exam(self, num_questions: int = 120, time_limit_minutes: int = 180) -> Dict[str, Any]: """Simulate a CFA exam""" try: # CFA exam structure topic_distribution = { 'Ethics': int(num_questions * 0.15), 'Quantitative Methods': int(num_questions * 0.12), 'Economics': int(num_questions * 0.10), 'Financial Reporting': int(num_questions * 0.13), 'Corporate Finance': int(num_questions * 0.10), 'Equity': int(num_questions * 0.11), 'Fixed Income': int(num_questions * 0.11), 'Derivatives': int(num_questions * 0.08), 'Alternative Investments': int(num_questions * 0.06), 'Portfolio Management': int(num_questions * 0.14) } # Adjust to ensure total matches total = sum(topic_distribution.values()) if total != num_questions: # Adjust the largest category largest_topic = max(topic_distribution.items(), key=lambda x: x[1]) topic_distribution[largest_topic[0]] += (num_questions - total) return { 'exam_info': { 'total_questions': num_questions, 'time_limit_minutes': time_limit_minutes, 'time_per_question': time_limit_minutes / num_questions, 'passing_score': 0.70, 'format': 'Multiple choice' }, 'topic_distribution': topic_distribution, 'instructions': [ 'Read each question carefully', 'Manage your time effectively', 'Answer every question (no penalty for wrong answers)', 'Review marked questions if time permits' ], 'preparation_tips': [ 'Practice with time limits', 'Review weak areas', 'Get familiar with calculator functions', 'Take full practice exams' ] } except Exception as e: return {'error': str(e)} def track_progress(self, study_logs: List[Dict[str, Any]], performance_data: Dict[str, float]) -> Dict[str, Any]: """Track study progress""" try: if not study_logs: return {'error': 'No study logs provided'} # Calculate study metrics total_hours = sum(log.get('hours', 0) for log in study_logs) # Calculate consistency from datetime import datetime study_dates = [datetime.strptime(log['date'], '%Y-%m-%d') for log in study_logs] unique_days = len(set(date.date() for date in study_dates)) total_days = (max(study_dates) - min(study_dates)).days + 1 consistency = unique_days / total_days if total_days > 0 else 0 # Performance analysis if performance_data: avg_score = np.mean(list(performance_data.values())) if avg_score > 0.8: performance_trend = 'excellent' elif avg_score > 0.7: performance_trend = 'good' elif avg_score > 0.6: performance_trend = 'fair' else: performance_trend = 'needs_improvement' else: avg_score = 0 performance_trend = 'unknown' # Topics covered topics_covered = set(log.get('topic', '') for log in study_logs) return { 'study_metrics': { 'total_hours': float(total_hours), 'avg_daily_hours': float(total_hours / max(unique_days, 1)), 'consistency_score': float(consistency), 'days_studied': unique_days, 'topics_covered': len(topics_covered) }, 'performance_metrics': { 'average_score': float(avg_score), 'trend': performance_trend, 'weak_topics': [t for t, s in performance_data.items() if s < 0.7] if performance_data else [] }, 'recommendations': self._generate_progress_recommendations( total_hours, consistency, avg_score ) } except Exception as e: return {'error': str(e)} def _generate_progress_recommendations(self, total_hours: float, consistency: float, avg_score: float) -> List[str]: """Generate progress recommendations""" recommendations = [] if total_hours < 50: recommendations.append(f'Aim for at least {50 - total_hours:.0f} more study hours') if consistency < 0.7: recommendations.append('Increase study consistency - aim for 5+ days per week') if avg_score < 0.7: recommendations.append('Focus on weak areas and practice more problems') elif avg_score < 0.8: recommendations.append('Maintain current pace and focus on application questions') else: recommendations.append('Excellent progress! Focus on timed practice exams') recommendations.append('Review mistakes and understand why answers were wrong') return recommendations def recommend_resources(self, topic: str, learning_style: str = 'general', experience_level: str = 'beginner') -> Dict[str, Any]: """Recommend learning resources""" try: # Resource database resources_db = { 'Time Value of Money': { 'books': ['CFA Program Curriculum - Quantitative Methods', 'Corporate Finance by Ross et al.'], 'videos': ['TVM Basics (Khan Academy)', 'CFA TVM Tutorial (AnalystPrep)'], 'practice': ['TVM Practice Problems (CFAI)', 'Financial Calculator Exercises'], 'tools': ['TVM Calculator', 'Financial Calculator Simulator'] }, 'Statistics': { 'books': ['Statistics for Business and Economics', 'CFA Quantitative Methods'], 'videos': ['Statistics Basics (StatQuest)', 'CFA Statistics (IFT)'], 'practice': ['Statistics Practice Questions', 'Probability Problems'], 'tools': ['Statistical Calculator', 'Data Analysis Practice'] }, 'Portfolio Management': { 'books': ['Modern Portfolio Theory', 'CFA Portfolio Management'], 'videos': ['Portfolio Theory Explained', 'CAPM Tutorial'], 'practice': ['Portfolio Optimization Problems', 'Risk-Return Calculations'], 'tools': ['Portfolio Simulator', 'Efficient Frontier Calculator'] } } # Get resources for topic or general resources if topic in resources_db: topic_resources = resources_db[topic] else: topic_resources = { 'books': ['CFA Program Curriculum'], 'videos': ['General Finance Tutorials'], 'practice': ['Practice Questions and Mock Exams'], 'tools': ['Financial Calculator'] } # Learning style adaptations style_adaptations = { 'visual': { 'priority': ['videos', 'tools'], 'notes': 'Focus on visual explanations and diagrams' }, 'auditory': { 'priority': ['videos', 'audio_books'], 'notes': 'Listen to explanations and discuss concepts' }, 'reading': { 'priority': ['books', 'practice'], 'notes': 'Read extensively and take detailed notes' }, 'kinesthetic': { 'priority': ['tools', 'practice'], 'notes': 'Hands-on practice and real applications' }, 'general': { 'priority': ['books', 'videos', 'practice'], 'notes': 'Mixed approach for comprehensive learning' } } adaptation = style_adaptations.get(learning_style.lower(), style_adaptations['general']) # Experience level adjustments level_notes = { 'beginner': 'Start with foundational concepts and basic problems', 'intermediate': 'Focus on application and integration of concepts', 'advanced': 'Tackle complex problems and advanced topics' } return { 'resources': topic_resources, 'recommended_order': adaptation['priority'], 'study_notes': adaptation['notes'], 'experience_level': experience_level, 'level_notes': level_notes.get(experience_level, ''), 'topic': topic } except Exception as e: return {'error': str(e)} def generate_flashcards(self, topic: str, num_cards: int = 10) -> Dict[str, Any]: """Generate flashcards for a topic""" try: # Flashcard templates card_templates = { 'Time Value of Money': [ ('What is Present Value?', 'Current worth of future cash flows'), ('What is Future Value?', 'Future worth of current investment'), ('What is NPV?', 'Net Present Value - difference between PV of inflows and outflows'), ('What is IRR?', 'Internal Rate of Return - discount rate that makes NPV zero'), ('What is an annuity?', 'Series of equal payments at regular intervals') ], 'Statistics': [ ('What is mean?', 'Average of a dataset'), ('What is standard deviation?', 'Measure of data dispersion'), ('What is correlation?', 'Measure of linear relationship between variables'), ('What is normal distribution?', 'Bell-shaped probability distribution'), ('What is hypothesis testing?', 'Statistical method to test assumptions') ], 'Portfolio Management': [ ('What is diversification?', 'Reducing risk by holding different assets'), ('What is CAPM?', 'Capital Asset Pricing Model - E(R) = Rf + β(Rm - Rf)'), ('What is beta?', 'Measure of systematic risk'), ('What is efficient frontier?', 'Set of optimal portfolios'), ('What is Sharpe ratio?', 'Risk-adjusted return measure') ] } # Get cards for topic or create generic cards if topic in card_templates: templates = card_templates[topic] else: templates = [ (f'Key concept in {topic}', 'Important principle or definition'), (f'Formula for {topic}', 'Relevant formula or calculation'), (f'Application of {topic}', 'Practical use or example'), (f'Common mistake in {topic}', 'Typical error to avoid'), (f'CFA exam tip for {topic}', 'Exam strategy or focus area') ] # Generate flashcards flashcards = [] for i in range(min(num_cards, len(templates) * 2)): template_idx = i % len(templates) question, answer = templates[template_idx] flashcards.append({ 'id': f'card_{i+1}', 'question': question, 'answer': answer, 'topic': topic, 'difficulty': 'easy' if i < 5 else 'medium' if i < 8 else 'hard' }) return { 'flashcards': flashcards, 'topic': topic, 'total_cards': len(flashcards), 'study_tips': [ 'Review cards regularly using spaced repetition', 'Try to recall answer before flipping card', 'Group related concepts together', 'Create your own cards for difficult concepts' ] } except Exception as e: return {'error': str(e)} def calculate_readiness_score(self, mock_exam_scores: List[float], study_hours: float, consistency: float) -> Dict[str, Any]: """Calculate exam readiness score""" try: if not mock_exam_scores: return { 'readiness_score': 0, 'confidence': 'low', 'recommendations': ['Take mock exams to assess readiness'] } # Calculate components avg_mock_score = np.mean(mock_exam_scores) # Score component (0-70 points) score_component = min(avg_mock_score * 70, 70) # Study hours component (0-20 points, target 300 hours) hours_component = min(study_hours / 300 * 20, 20) # Consistency component (0-10 points) consistency_component = consistency * 10 # Total readiness score (0-100) readiness_score = score_component + hours_component + consistency_component # Determine confidence level if readiness_score >= 80: confidence = 'high' recommendation = 'Ready for exam - maintain current pace' elif readiness_score >= 60: confidence = 'medium' recommendation = 'Close to ready - focus on weak areas' else: confidence = 'low' recommendation = 'Needs significant improvement' return { 'readiness_score': float(readiness_score), 'confidence': confidence, 'score_breakdown': { 'mock_exams': float(score_component), 'study_hours': float(hours_component), 'consistency': float(consistency_component) }, 'current_metrics': { 'avg_mock_score': float(avg_mock_score), 'study_hours': float(study_hours), 'consistency': float(consistency) }, 'targets': { 'mock_score_target': 0.75, 'hours_target': 300, 'consistency_target': 0.8 }, 'recommendations': [ recommendation, f'Aim for mock exam scores above {max(0.75, avg_mock_score + 0.05):.0%}', f'Target {max(0, 300 - study_hours):.0f} more study hours' if study_hours < 300 else 'Maintain study hours', 'Increase study consistency' if consistency < 0.7 else 'Maintain consistency' ] } except Exception as e: return {'error': str(e)} def generate_practice_problem(self, topic: str, difficulty: str = 'medium') -> Dict[str, Any]: """Generate a practice problem""" try: # Problem templates problem_templates = { 'Time Value of Money': { 'easy': { 'question': 'What is the future value of $500 invested at 4% annual interest for 2 years?', 'solution': 'FV = 500 × (1.04)^2 = $540.80', 'steps': ['Identify PV = $500', 'Rate = 4% = 0.04', 'n = 2 years', 'Apply FV formula', 'Calculate: 500 × 1.04^2 = 540.80'] }, 'medium': { 'question': 'Calculate the present value of an annuity that pays $100 annually for 5 years at 6% discount rate.', 'solution': 'PV = 100 × [1 - (1.06)^-5] / 0.06 = $421.24', 'steps': ['Identify PMT = $100', 'Rate = 6% = 0.06', 'n = 5 years', 'Apply annuity PV formula', 'Calculate'] } }, 'Statistics': { 'easy': { 'question': 'Calculate the mean of: 10, 15, 20, 25, 30', 'solution': 'Mean = (10+15+20+25+30)/5 = 20', 'steps': ['Sum all values: 10+15+20+25+30 = 100', 'Count values: 5', 'Divide: 100/5 = 20'] }, 'medium': { 'question': 'Calculate the standard deviation of: 5, 10, 15, 20, 25', 'solution': 'Std Dev = 7.07', 'steps': ['Calculate mean: (5+10+15+20+25)/5 = 15', 'Calculate squared differences', 'Sum squared differences', 'Divide by n-1', 'Take square root'] } } } # Get problem template or create generic if topic in problem_templates and difficulty in problem_templates[topic]: template = problem_templates[topic][difficulty] else: template = { 'question': f'A {difficulty} level problem about {topic}', 'solution': f'Sample solution for {topic} problem', 'steps': ['Step 1: Understand the problem', 'Step 2: Apply relevant formula', 'Step 3: Calculate', 'Step 4: Interpret result'] } # Generate multiple choice options for easy/medium problems options = [] if difficulty in ['easy', 'medium']: # Create plausible wrong answers solution_num = self._extract_number(template['solution']) if solution_num is not None: options = [ str(solution_num), str(solution_num * 0.8), str(solution_num * 1.2), str(solution_num * 1.5) ] random.shuffle(options) return { 'problem': { 'question': template['question'], 'difficulty': difficulty, 'topic': topic }, 'solution': { 'answer': template['solution'], 'steps': template['steps'], 'explanation': f'This tests understanding of {topic} at {difficulty} level' }, 'options': options if options else [], 'hints': [ 'Write down known variables', 'Identify the appropriate formula', 'Check units and calculations' ] } except Exception as e: return {'error': str(e)} def _extract_number(self, text: str) -> Optional[float]: """Extract first number from text""" import re match = re.search(r'\d+\.?\d*', text) return float(match.group()) if match else None def create_study_schedule(self, topics: List[str], weeks: int = 8, hours_per_week: float = 10) -> Dict[str, Any]: """Create a weekly study schedule""" try: if weeks <= 0: return {'error': 'Number of weeks must be positive'} # Distribute topics across weeks topics_per_week = max(1, len(topics) // weeks) schedule = [] topic_idx = 0 for week in range(1, weeks + 1): week_topics = [] for _ in range(topics_per_week): if topic_idx < len(topics): week_topics.append(topics[topic_idx]) topic_idx += 1 if week_topics: schedule.append({ 'week': week, 'topics': week_topics, 'hours': hours_per_week, 'focus': f'Master {", ".join(week_topics[:2])}' if week_topics else 'Review', 'milestone': 'Week 1: Foundations' if week == 1 else 'Midpoint: Integration' if week == weeks // 2 else 'Final: Review & Practice' if week == weeks else f'Week {week}: Continued Learning' }) # Add any remaining topics to last week if topic_idx < len(topics): if schedule: schedule[-1]['topics'].extend(topics[topic_idx:]) return { 'schedule': schedule, 'total_weeks': weeks, 'total_hours': weeks * hours_per_week, 'topics_covered': len(topics), 'weekly_target': hours_per_week, 'recommendation': f'Study {hours_per_week} hours per week, focusing on 1-2 topics each week' } except Exception as e: return {'error': str(e)} # Create server instance mcp_server = MCPServer()