Create agents / calculator_agent.py

agents / calculator_agent.py

@@ -0,0 +1,539 @@
+"""
+Calculator Agent - Performs complex financial calculations
+"""
+
+import numpy as np
+import pandas as pd
+from typing import Dict, List, Any, Optional, Union
+from datetime import datetime
+import re
+import math
+
+
+class CalculatorAgent:
+    """Performs complex financial and quantitative calculations"""
+    
+    def __init__(self):
+        self.calculation_history = []
+        self.formulas = self._load_formulas()
+        
+    def _load_formulas(self) -> Dict[str, Dict]:
+        """Load financial formulas and their implementations"""
+        return {
+            'future_value': {
+                'description': 'Future value of a single cash flow',
+                'formula': 'FV = PV × (1 + r)^n',
+                'parameters': ['PV', 'r', 'n'],
+                'function': self._calculate_fv
+            },
+            'present_value': {
+                'description': 'Present value of a single cash flow',
+                'formula': 'PV = FV / (1 + r)^n',
+                'parameters': ['FV', 'r', 'n'],
+                'function': self._calculate_pv
+            },
+            'annuity_pv': {
+                'description': 'Present value of an ordinary annuity',
+                'formula': 'PV = PMT × [1 - (1 + r)^-n] / r',
+                'parameters': ['PMT', 'r', 'n'],
+                'function': self._calculate_annuity_pv
+            },
+            'annuity_fv': {
+                'description': 'Future value of an ordinary annuity',
+                'formula': 'FV = PMT × [(1 + r)^n - 1] / r',
+                'parameters': ['PMT', 'r', 'n'],
+                'function': self._calculate_annuity_fv
+            },
+            'perpetuity_pv': {
+                'description': 'Present value of a perpetuity',
+                'formula': 'PV = PMT / r',
+                'parameters': ['PMT', 'r'],
+                'function': self._calculate_perpetuity_pv
+            },
+            'npv': {
+                'description': 'Net present value of cash flows',
+                'formula': 'NPV = Σ [CF_t / (1 + r)^t]',
+                'parameters': ['cash_flows', 'r'],
+                'function': self._calculate_npv
+            },
+            'irr': {
+                'description': 'Internal rate of return',
+                'formula': '0 = Σ [CF_t / (1 + IRR)^t]',
+                'parameters': ['cash_flows'],
+                'function': self._calculate_irr
+            },
+            'mean': {
+                'description': 'Arithmetic mean',
+                'formula': 'μ = Σ x_i / n',
+                'parameters': ['data'],
+                'function': self._calculate_mean
+            },
+            'variance': {
+                'description': 'Variance',
+                'formula': 'σ² = Σ (x_i - μ)² / (n-1)',
+                'parameters': ['data'],
+                'function': self._calculate_variance
+            },
+            'standard_deviation': {
+                'description': 'Standard deviation',
+                'formula': 'σ = √Variance',
+                'parameters': ['data'],
+                'function': self._calculate_std_dev
+            },
+            'correlation': {
+                'description': 'Correlation coefficient',
+                'formula': 'ρ = Cov(X,Y) / (σ_X × σ_Y)',
+                'parameters': ['x', 'y'],
+                'function': self._calculate_correlation
+            },
+            'covariance': {
+                'description': 'Covariance',
+                'formula': 'Cov(X,Y) = Σ [(x_i - μ_x)(y_i - μ_y)] / (n-1)',
+                'parameters': ['x', 'y'],
+                'function': self._calculate_covariance
+            }
+        }
+    
+    def process(self,
+                query: str,
+                user_profile: Dict[str, Any],
+                conversation_history: List[Dict] = None) -> Dict[str, Any]:
+        """
+        Parse and execute calculation requests
+        
+        Args:
+            query: Calculation request
+            user_profile: User's learning profile
+            conversation_history: Previous conversation
+        
+        Returns:
+            Calculation results
+        """
+        # Extract calculation type and parameters
+        calc_type, params = self._parse_calculation_request(query)
+        
+        if not calc_type:
+            return {
+                'response': "I couldn't identify a specific calculation in your query. "
+                          "Please specify what you want to calculate (e.g., 'Calculate NPV', "
+                          "'What is the future value?').",
+                'confidence': 0.3,
+                'metadata': {'parsed': False}
+            }
+        
+        try:
+            # Perform calculation
+            result = self._perform_calculation(calc_type, params)
+            
+            # Format response
+            response = self._format_calculation_response(calc_type, params, result)
+            
+            # Log calculation
+            self._log_calculation(calc_type, params, result, user_profile)
+            
+            return {
+                'response': response,
+                'result': result,
+                'calculation_type': calc_type,
+                'confidence': 0.95,
+                'metadata': {
+                    'formula': self.formulas[calc_type]['formula'],
+                    'parameters_used': params
+                }
+            }
+            
+        except Exception as e:
+            error_response = f"Error in calculation: {str(e)}\n\n"
+            error_response += f"Formula: {self.formulas[calc_type]['formula']}\n"
+            error_response += f"Required parameters: {', '.join(self.formulas[calc_type]['parameters'])}"
+            
+            return {
+                'response': error_response,
+                'error': str(e),
+                'calculation_type': calc_type,
+                'confidence': 0.1,
+                'metadata': {'error': True}
+            }
+    
+    def _parse_calculation_request(self, query: str) -> tuple:
+        """Parse calculation request to identify type and parameters"""
+        query_lower = query.lower()
+        
+        # Map keywords to calculation types
+        keyword_mapping = {
+            'future value': 'future_value',
+            'fv': 'future_value',
+            'present value': 'present_value',
+            'pv': 'present_value',
+            'annuity': 'annuity_pv',
+            'perpetuity': 'perpetuity_pv',
+            'npv': 'npv',
+            'net present value': 'npv',
+            'irr': 'irr',
+            'internal rate of return': 'irr',
+            'mean': 'mean',
+            'average': 'mean',
+            'variance': 'variance',
+            'standard deviation': 'standard_deviation',
+            'std dev': 'standard_deviation',
+            'correlation': 'correlation',
+            'covariance': 'covariance',
+            'calculate': None,  # Generic - need to infer
+            'compute': None,
+            'solve': None
+        }
+        
+        # Identify calculation type
+        calc_type = None
+        for keyword, ctype in keyword_mapping.items():
+            if keyword in query_lower:
+                calc_type = ctype
+                break
+        
+        # If generic calculation term, try to infer from context
+        if not calc_type:
+            # Look for formula components
+            if any(term in query_lower for term in ['(1+r)', '^n', 'discount']):
+                calc_type = 'future_value'
+            elif 'cash flow' in query_lower:
+                calc_type = 'npv'
+        
+        # Extract parameters
+        params = self._extract_parameters(query, calc_type)
+        
+        return calc_type, params
+    
+    def _extract_parameters(self, query: str, calc_type: str) -> Dict[str, Any]:
+        """Extract parameters from query text"""
+        params = {}
+        
+        # Extract numbers with context
+        number_pattern = r'(\d+\.?\d*)'
+        numbers = re.findall(number_pattern, query)
+        
+        # Extract percentages
+        percent_pattern = r'(\d+\.?\d*)\s*%'
+        percents = re.findall(percent_pattern, query)
+        
+        # Extract monetary values
+        money_pattern = r'\$(\d+\.?\d*)'
+        money = re.findall(money_pattern, query)
+        
+        # Map based on calculation type
+        if calc_type == 'future_value':
+            if money:
+                params['PV'] = float(money[0])
+            if percents:
+                params['r'] = float(percents[0]) / 100
+            if numbers and 'n' not in params:
+                # Assume last number is n if not already assigned
+                for num in reversed(numbers):
+                    try:
+                        if float(num) != params.get('PV', 0) and float(num) != params.get('r', 0)*100:
+                            params['n'] = float(num)
+                            break
+                    except:
+                        pass
+        
+        elif calc_type == 'present_value':
+            if money:
+                params['FV'] = float(money[0])
+            if percents:
+                params['r'] = float(percents[0]) / 100
+            if numbers and 'n' not in params:
+                for num in reversed(numbers):
+                    try:
+                        if float(num) != params.get('FV', 0) and float(num) != params.get('r', 0)*100:
+                            params['n'] = float(num)
+                            break
+                    except:
+                        pass
+        
+        elif calc_type in ['annuity_pv', 'annuity_fv']:
+            if money:
+                params['PMT'] = float(money[0])
+            if percents:
+                params['r'] = float(percents[0]) / 100
+            if numbers and 'n' not in params:
+                for num in reversed(numbers):
+                    try:
+                        if float(num) != params.get('PMT', 0) and float(num) != params.get('r', 0)*100:
+                            params['n'] = float(num)
+                            break
+                    except:
+                        pass
+        
+        elif calc_type == 'npv':
+            # Try to extract cash flow array
+            cf_pattern = r'cash flows?\s*[:=]\s*\[([-\d\s,\.]+)\]'
+            cf_match = re.search(cf_pattern, query, re.IGNORECASE)
+            if cf_match:
+                cash_flows_str = cf_match.group(1)
+                params['cash_flows'] = [float(x.strip()) for x in cash_flows_str.split(',')]
+            
+            if percents:
+                params['r'] = float(percents[0]) / 100
+        
+        elif calc_type in ['mean', 'variance', 'standard_deviation']:
+            # Try to extract data array
+            data_pattern = r'data\s*[:=]\s*\[([\d\s,\.]+)\]'
+            data_match = re.search(data_pattern, query, re.IGNORECASE)
+            if data_match:
+                data_str = data_match.group(1)
+                params['data'] = [float(x.strip()) for x in data_str.split(',')]
+        
+        elif calc_type in ['correlation', 'covariance']:
+            # Try to extract two data arrays
+            x_pattern = r'x\s*[:=]\s*\[([\d\s,\.]+)\]'
+            y_pattern = r'y\s*[:=]\s*\[([\d\s,\.]+)\]'
+            
+            x_match = re.search(x_pattern, query, re.IGNORECASE)
+            y_match = re.search(y_pattern, query, re.IGNORECASE)
+            
+            if x_match and y_match:
+                x_str = x_match.group(1)
+                y_str = y_match.group(1)
+                params['x'] = [float(x.strip()) for x in x_str.split(',')]
+                params['y'] = [float(y.strip()) for y in y_str.split(',')]
+        
+        return params
+    
+    def _perform_calculation(self, calc_type: str, params: Dict[str, Any]) -> Dict[str, Any]:
+        """Perform the actual calculation"""
+        formula_info = self.formulas[calc_type]
+        calculation_func = formula_info['function']
+        
+        result = calculation_func(params)
+        
+        return {
+            'value': result,
+            'formula': formula_info['formula'],
+            'description': formula_info['description'],
+            'parameters': params
+        }
+    
+    # Calculation implementations
+    def _calculate_fv(self, params: Dict) -> float:
+        """Calculate future value: FV = PV × (1 + r)^n"""
+        pv = params.get('PV', 0)
+        r = params.get('r', 0)
+        n = params.get('n', 0)
+        return pv * (1 + r) ** n
+    
+    def _calculate_pv(self, params: Dict) -> float:
+        """Calculate present value: PV = FV / (1 + r)^n"""
+        fv = params.get('FV', 0)
+        r = params.get('r', 0)
+        n = params.get('n', 0)
+        return fv / (1 + r) ** n
+    
+    def _calculate_annuity_pv(self, params: Dict) -> float:
+        """Calculate present value of annuity: PV = PMT × [1 - (1 + r)^-n] / r"""
+        pmt = params.get('PMT', 0)
+        r = params.get('r', 0)
+        n = params.get('n', 0)
+        
+        if r == 0:
+            return pmt * n
+        return pmt * (1 - (1 + r) ** -n) / r
+    
+    def _calculate_annuity_fv(self, params: Dict) -> float:
+        """Calculate future value of annuity: FV = PMT × [(1 + r)^n - 1] / r"""
+        pmt = params.get('PMT', 0)
+        r = params.get('r', 0)
+        n = params.get('n', 0)
+        
+        if r == 0:
+            return pmt * n
+        return pmt * ((1 + r) ** n - 1) / r
+    
+    def _calculate_perpetuity_pv(self, params: Dict) -> float:
+        """Calculate present value of perpetuity: PV = PMT / r"""
+        pmt = params.get('PMT', 0)
+        r = params.get('r', 0)
+        
+        if r == 0:
+            raise ValueError("Discount rate cannot be zero for perpetuity")
+        return pmt / r
+    
+    def _calculate_npv(self, params: Dict) -> float:
+        """Calculate net present value: NPV = Σ [CF_t / (1 + r)^t]"""
+        cash_flows = params.get('cash_flows', [])
+        r = params.get('r', 0.1)
+        
+        npv = 0
+        for t, cf in enumerate(cash_flows):
+            npv += cf / (1 + r) ** t
+        return npv
+    
+    def _calculate_irr(self, params: Dict) -> float:
+        """Calculate internal rate of return"""
+        cash_flows = params.get('cash_flows', [])
+        
+        # Simple IRR calculation using numpy's IRR function
+        # In production, use numpy.irr or a more robust implementation
+        if len(cash_flows) < 2:
+            raise ValueError("Need at least two cash flows for IRR calculation")
+        
+        # Simple implementation for demonstration
+        # Actual implementation would use numerical methods
+        def npv_func(rate):
+            return sum(cf / (1 + rate) ** i for i, cf in enumerate(cash_flows))
+        
+        # Use binary search to find IRR
+        lower, upper = -0.99, 10.0
+        for _ in range(100):
+            mid = (lower + upper) / 2
+            if npv_func(mid) > 0:
+                lower = mid
+            else:
+                upper = mid
+        
+        return (lower + upper) / 2
+    
+    def _calculate_mean(self, params: Dict) -> float:
+        """Calculate mean: μ = Σ x_i / n"""
+        data = params.get('data', [])
+        if not data:
+            raise ValueError("No data provided")
+        return sum(data) / len(data)
+    
+    def _calculate_variance(self, params: Dict) -> float:
+        """Calculate sample variance: σ² = Σ (x_i - μ)² / (n-1)"""
+        data = params.get('data', [])
+        if len(data) < 2:
+            raise ValueError("Need at least 2 data points for variance")
+        
+        mean = self._calculate_mean({'data': data})
+        squared_diffs = [(x - mean) ** 2 for x in data]
+        return sum(squared_diffs) / (len(data) - 1)
+    
+    def _calculate_std_dev(self, params: Dict) -> float:
+        """Calculate standard deviation: σ = √Variance"""
+        variance = self._calculate_variance(params)
+        return math.sqrt(variance)
+    
+    def _calculate_covariance(self, params: Dict) -> float:
+        """Calculate covariance: Cov(X,Y) = Σ [(x_i - μ_x)(y_i - μ_y)] / (n-1)"""
+        x = params.get('x', [])
+        y = params.get('y', [])
+        
+        if len(x) != len(y):
+            raise ValueError("X and Y must have same length")
+        if len(x) < 2:
+            raise ValueError("Need at least 2 data points")
+        
+        mean_x = self._calculate_mean({'data': x})
+        mean_y = self._calculate_mean({'data': y})
+        
+        cov_sum = sum((x[i] - mean_x) * (y[i] - mean_y) for i in range(len(x)))
+        return cov_sum / (len(x) - 1)
+    
+    def _calculate_correlation(self, params: Dict) -> float:
+        """Calculate correlation: ρ = Cov(X,Y) / (σ_X × σ_Y)"""
+        cov = self._calculate_covariance(params)
+        std_x = self._calculate_std_dev({'data': params.get('x', [])})
+        std_y = self._calculate_std_dev({'data': params.get('y', [])})
+        
+        if std_x == 0 or std_y == 0:
+            return 0
+        return cov / (std_x * std_y)
+    
+    def _format_calculation_response(self, calc_type: str, params: Dict, result: Dict) -> str:
+        """Format calculation results for display"""
+        formula_info = self.formulas[calc_type]
+        
+        response = f"## 🧮 Calculation Result: {formula_info['description']}\n\n"
+        response += f"**Formula:** `{formula_info['formula']}`\n\n"
+        
+        response += "**Parameters:**\n"
+        for param, value in params.items():
+            if isinstance(value, list):
+                value_str = f"[{', '.join(str(v) for v in value[:5])}" + ('...' if len(value) > 5 else ']')
+            elif param == 'r' and 'r' in params:
+                response += f"- {param} = {value:.1%}\n"
+                continue
+            else:
+                value_str = str(value)
+            response += f"- {param} = {value_str}\n"
+        
+        response += f"\n**Result:** "
+        
+        # Format result based on calculation type
+        result_value = result['value']
+        if calc_type in ['future_value', 'present_value', 'annuity_pv', 'annuity_fv', 'perpetuity_pv', 'npv']:
+            response += f"**${result_value:,.2f}**\n"
+        elif calc_type == 'irr':
+            response += f"**{result_value:.1%}**\n"
+        elif calc_type in ['correlation', 'covariance']:
+            response += f"**{result_value:.4f}**\n"
+        else:
+            response += f"**{result_value:.4f}**\n"
+        
+        response += "\n**Calculation Steps:**\n"
+        
+        # Add calculation steps based on type
+        if calc_type == 'future_value':
+            pv = params.get('PV', 0)
+            r = params.get('r', 0)
+            n = params.get('n', 0)
+            response += f"1. FV = PV × (1 + r)^n\n"
+            response += f"2. FV = ${pv:,.2f} × (1 + {r:.3f})^{n}\n"
+            response += f"3. FV = ${pv:,.2f} × {((1 + r) ** n):.4f}\n"
+            response += f"4. FV = **${result_value:,.2f}**\n"
+        
+        elif calc_type == 'mean':
+            data = params.get('data', [])
+            response += f"1. Sum of values: {sum(data)}\n"
+            response += f"2. Number of values: {len(data)}\n"
+            response += f"3. Mean = {sum(data)} / {len(data)}\n"
+            response += f"4. Mean = **{result_value:.4f}**\n"
+        
+        response += "\n💡 *Tip: Try varying the parameters to see how they affect the result.*"
+        
+        return response
+    
+    def _log_calculation(self, calc_type: str, params: Dict, result: Dict, user_profile: Dict):
+        """Log calculation for monitoring and analytics"""
+        log_entry = {
+            'timestamp': datetime.now().isoformat(),
+            'calculation_type': calc_type,
+            'parameters': params,
+            'result': result['value'],
+            'user_level': user_profile.get('cfa_level', 'I'),
+            'topics': user_profile.get('study_focus', [])
+        }
+        self.calculation_history.append(log_entry)
+        
+        # Keep only last 1000 calculations
+        if len(self.calculation_history) > 1000:
+            self.calculation_history = self.calculation_history[-1000:]
+    
+    def get_calculation_history(self, limit: int = 50) -> List[Dict]:
+        """Get recent calculation history"""
+        return self.calculation_history[-limit:]
+    
+    def get_metrics(self) -> Dict[str, Any]:
+        """Get calculator performance metrics"""
+        if not self.calculation_history:
+            return {
+                'total_calculations': 0,
+                'most_common_type': None,
+                'recent_accuracy': 1.0
+            }
+        
+        # Count calculation types
+        type_counts = {}
+        for entry in self.calculation_history:
+            ctype = entry['calculation_type']
+            type_counts[ctype] = type_counts.get(ctype, 0) + 1
+        
+        most_common = max(type_counts.items(), key=lambda x: x[1])[0] if type_counts else None
+        
+        return {
+            'total_calculations': len(self.calculation_history),
+            'most_common_type': most_common,
+            'type_distribution': type_counts,
+            'recent_activity': len([e for e in self.calculation_history 
+                                  if datetime.fromisoformat(e['timestamp']).date() == datetime.now().date()])
+        }