Upload 3 files

edgar_client.py

@@ -60,6 +60,12 @@ class EdgarDataClient:
         self._search_cache = {}  # search_key -> result
         self._search_cache_max_size = 1000  # Limit cache size
         
+        # Layer 3: Period data cache (avoid re-parsing XBRL for same period)
+        self._period_cache = {}  # period_key -> financial data
+        self._period_cache_timestamps = {}  # period_key -> timestamp
+        self._period_cache_ttl = 1800  # 30 minutes cache (financial data changes rarely)
+        self._period_cache_max_size = 1000  # Limit cache size
+        
         # Common company aliases for intelligent search
         self._company_aliases = {
             'google': ['GOOGL', 'GOOG'],
@@ -415,6 +421,33 @@ class EdgarDataClient:
         
         self._search_cache[cache_key] = result
     
+    def _get_period_cache(self, cache_key):
+        """Get cached period data if valid (Layer 3)"""
+        if cache_key not in self._period_cache_timestamps:
+            return None
+        
+        age = time.time() - self._period_cache_timestamps[cache_key]
+        if age < self._period_cache_ttl:
+            return self._period_cache.get(cache_key)
+        else:
+            # Expired, remove from cache
+            self._period_cache.pop(cache_key, None)
+            self._period_cache_timestamps.pop(cache_key, None)
+            return None
+    
+    def _set_period_cache(self, cache_key, result):
+        """Cache period data with size limit (Layer 3)"""
+        # LRU-like eviction if cache is full
+        if len(self._period_cache) >= self._period_cache_max_size:
+            # Remove oldest half
+            keys_to_remove = list(self._period_cache.keys())[:self._period_cache_max_size // 2]
+            for key in keys_to_remove:
+                self._period_cache.pop(key, None)
+                self._period_cache_timestamps.pop(key, None)
+        
+        self._period_cache[cache_key] = result
+        self._period_cache_timestamps[cache_key] = time.time()
+    
     def get_company_info(self, cik):
         """
         Get basic company information with caching
@@ -607,6 +640,13 @@ class EdgarDataClient:
         if not self.edgar:
             print("sec_edgar_api library not installed")
             return {}
+        
+        # Check period cache first (Layer 3)
+        cache_key = f"period_{cik}_{period}"
+        cached = self._get_period_cache(cache_key)
+        if cached is not None:
+            print(f"[Cache Hit] get_financial_data_for_period({cik}, {period})")
+            return cached.copy()  # Return copy to avoid mutation
             
         try:
             # Get company financial facts
@@ -844,6 +884,9 @@ class EdgarDataClient:
                         if metric_key in result:
                             break
             
+            # Cache the result (Layer 3)
+            self._set_period_cache(cache_key, result)
+            
             return result
         except Exception as e:
             print(f"Error getting financial data for period {period}: {e}")

financial_analyzer.py

@@ -15,6 +15,41 @@ class FinancialAnalyzer:
         """
         self.edgar_client = EdgarDataClient(user_agent)
         
+        # Layer 2: Method-level cache (avoid duplicate API calls)
+        self._method_cache = {}  # method_key -> result
+        self._method_cache_timestamps = {}  # method_key -> timestamp
+        self._method_cache_ttl = 600  # 10 minutes cache
+        self._method_cache_max_size = 500  # Limit cache size
+        
+    def _get_method_cache(self, cache_key):
+        """Get cached method result if valid"""
+        if cache_key not in self._method_cache_timestamps:
+            return None
+        
+        import time
+        age = time.time() - self._method_cache_timestamps[cache_key]
+        if age < self._method_cache_ttl:
+            return self._method_cache.get(cache_key)
+        else:
+            # Expired, remove from cache
+            self._method_cache.pop(cache_key, None)
+            self._method_cache_timestamps.pop(cache_key, None)
+            return None
+    
+    def _set_method_cache(self, cache_key, result):
+        """Cache method result with size limit"""
+        # LRU-like eviction if cache is full
+        if len(self._method_cache) >= self._method_cache_max_size:
+            # Remove oldest half
+            keys_to_remove = list(self._method_cache.keys())[:self._method_cache_max_size // 2]
+            for key in keys_to_remove:
+                self._method_cache.pop(key, None)
+                self._method_cache_timestamps.pop(key, None)
+        
+        import time
+        self._method_cache[cache_key] = result
+        self._method_cache_timestamps[cache_key] = time.time()
+        
     def search_company(self, company_input):
         """
         Search company information (by name, ticker, or CIK) - Optimized version
@@ -141,6 +176,13 @@ class FinancialAnalyzer:
         Returns:
             list: List of financial data
         """
+        # Check method cache first (Layer 2)
+        cache_key = f"extract_metrics_{cik}_{years}"
+        cached = self._get_method_cache(cache_key)
+        if cached is not None:
+            print(f"[Cache Hit] extract_financial_metrics({cik}, {years})")
+            return cached
+        
         financial_data = []
         
         # Step 1: Get company filings to determine what was actually filed
@@ -258,6 +300,9 @@ class FinancialAnalyzer:
                 
                 financial_data.append(data)
         
+        # Cache the result (Layer 2)
+        self._set_method_cache(cache_key, financial_data)
+        
         return financial_data
     
     def get_latest_financial_data(self, cik):
@@ -270,6 +315,13 @@ class FinancialAnalyzer:
         Returns:
             dict: Latest financial data
         """
+        # Check method cache first (Layer 2)
+        cache_key = f"latest_data_{cik}"
+        cached = self._get_method_cache(cache_key)
+        if cached is not None:
+            print(f"[Cache Hit] get_latest_financial_data({cik})")
+            return cached
+        
         # Get latest filing year (supports 10-K and 20-F)
         filings_10k = self.edgar_client.get_company_filings(cik, ['10-K'])
         filings_20f = self.edgar_client.get_company_filings(cik, ['20-F'])
@@ -293,7 +345,12 @@ class FinancialAnalyzer:
             return {}
         
         # Get financial data for latest year
-        return self.edgar_client.get_financial_data_for_period(cik, str(latest_filing_year))
+        result = self.edgar_client.get_financial_data_for_period(cik, str(latest_filing_year))
+        
+        # Cache the result (Layer 2)
+        self._set_method_cache(cache_key, result)
+        
+        return result
     
     def format_financial_data(self, financial_data):
         """

mcp_server_sse.py

@@ -27,7 +27,7 @@ from contextlib import contextmanager
 app = FastAPI(
     title="SEC Financial Report MCP Server API",
     description="Model Context Protocol Server for SEC EDGAR Financial Data",
-    version="2.3.4"
+    version="2.3.5"
 )
 
 # Server startup time for monitoring
@@ -415,7 +415,7 @@ async def handle_mcp_message(request: MCPRequest):
                 },
                 "serverInfo": {
                     "name": "sec-financial-data",
-                    "version": "2.3.4"
+                    "version": "2.3.5"
                 }
             }
         ).dict()
@@ -547,7 +547,7 @@ async def health_check():
     return {
         "status": "healthy",
         "server": "sec-financial-data",
-        "version": "2.3.4",
+        "version": "2.3.5",
         "protocol": "MCP",
         "transport": "SSE",
         "tools_count": len(MCP_TOOLS),