Spaces:
Runtime error
Runtime error
| """ | |
| 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 | |
| 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]}" | |
| 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() | |