EasyReportDataMCP/edgar_client.py

"""EDGAR API Client Module with Performance Optimization"""

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import urllib3
try:
    from sec_edgar_api.EdgarClient import EdgarClient
except ImportError:
    EdgarClient = None
import json
import time
import threading
from functools import lru_cache
from datetime import datetime, timedelta
import re
import difflib

# Disable SSL warnings for better compatibility
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class EdgarDataClient:
    # Class-level cache for company_tickers.json (shared across instances)
    _company_tickers_cache = None
    _company_tickers_cache_time = None
    _company_tickers_cache_ttl = 3600  # 1 hour TTL
    _cache_lock = threading.Lock()
    
    # Class-level rate limiter (SEC requires max 10 requests per second)
    _last_request_time = 0
    _rate_limit_lock = threading.Lock()
    _min_request_interval = 0.11  # 110ms between requests (9 req/sec, safe margin)
    
    # 新增：公司索引（加速搜索，避免每次遍历全量数据）
    _by_ticker = None  # ticker -> company info
    _by_title = None  # title (lowercase) -> company info
    _by_title_norm = None  # normalized title -> company info
    _all_keys = None  # 用于模糊匹配的所有key列表
    _index_built_time = None
    _index_ttl = 3600  # 1 hour
    
    # 新增：常见别名映射（提升搜索智能性）
    _alias_map = {
        "google": "alphabet inc",
        "alphabet": "alphabet inc",
        "facebook": "meta platforms, inc.",
        "meta": "meta platforms, inc.",
        "amazon": "amazon.com, inc.",
        "apple": "apple inc.",
        "microsoft": "microsoft corporation",
        "netflix": "netflix, inc.",
        "nvidia": "nvidia corporation",
        "tesla": "tesla, inc.",
        "adobe": "adobe inc.",
        "oracle": "oracle corporation",
        "ibm": "international business machines corporation",
        "paypal": "paypal holdings, inc.",
        "shopify": "shopify inc.",
    }
    
    def __init__(self, user_agent="Juntao Peng Financial Report Metrics App (jtyxabc@gmail.com)"):
        """Initialize EDGAR client with connection pooling and timeout"""
        self.user_agent = user_agent
        
        # 新增：实例级搜索缓存（进一步减少重复搜索开销）
        self._search_cache = {}
        
        # Configure requests session with connection pooling
        self.session = requests.Session()
        
        # Configure retry strategy with enhanced retries for stability
        retry_strategy = Retry(
            total=5,  # Increased from 3 to 5 for better reliability
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["HEAD", "GET", "OPTIONS"]
        )
        
        adapter = HTTPAdapter(
            pool_connections=10,
            pool_maxsize=20,
            max_retries=retry_strategy,
            pool_block=False
        )
        
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
        
        # Set default timeout with connection and read timeouts
        self.timeout = (10, 30)  # (connect timeout, read timeout)
        
        # Initialize sec_edgar_api client with timeout wrapper
        if EdgarClient:
            self.edgar = EdgarClient(user_agent=user_agent)
            # Monkey patch to add timeout
            self._patch_edgar_client_timeout()
        else:
            self.edgar = None
    
    def _patch_edgar_client_timeout(self):
        """Monkey patch sec_edgar_api to add timeout support"""
        if not self.edgar:
            return
        
        # Wrap get_submissions and get_company_facts with timeout (thread-based, Gradio compatible)
        original_get_submissions = self.edgar.get_submissions
        original_get_company_facts = self.edgar.get_company_facts
        
        def get_submissions_with_timeout(cik):
            """Thread-based timeout wrapper for get_submissions (Gradio compatible)"""
            result = [None]
            exception = [None]
            
            def wrapper():
                try:
                    result[0] = original_get_submissions(cik)
                except Exception as e:
                    exception[0] = e
            
            thread = threading.Thread(target=wrapper, daemon=True)
            thread.start()
            # Use read timeout value (second element of timeout tuple)
            timeout_seconds = self.timeout[1] if isinstance(self.timeout, tuple) else self.timeout
            thread.join(timeout=timeout_seconds)
            
            if thread.is_alive():
                raise TimeoutError(f"SEC API request timeout ({timeout_seconds}s)")
            
            if exception[0]:
                raise exception[0]
            
            return result[0]
        
        def get_company_facts_with_timeout(cik):
            """Thread-based timeout wrapper for get_company_facts (Gradio compatible)"""
            result = [None]
            exception = [None]
            
            def wrapper():
                try:
                    result[0] = original_get_company_facts(cik)
                except Exception as e:
                    exception[0] = e
            
            thread = threading.Thread(target=wrapper, daemon=True)
            thread.start()
            # Use read timeout value (second element of timeout tuple)
            timeout_seconds = self.timeout[1] if isinstance(self.timeout, tuple) else self.timeout
            thread.join(timeout=timeout_seconds)
            
            if thread.is_alive():
                raise TimeoutError(f"SEC API request timeout ({timeout_seconds}s)")
            
            if exception[0]:
                raise exception[0]
            
            return result[0]
        
        self.edgar.get_submissions = get_submissions_with_timeout
        self.edgar.get_company_facts = get_company_facts_with_timeout
    
    def _rate_limit(self):
        """Thread-safe rate limiting to comply with SEC requirements"""
        with self._rate_limit_lock:
            current_time = time.time()
            time_since_last = current_time - EdgarDataClient._last_request_time
            
            if time_since_last < self._min_request_interval:
                sleep_time = self._min_request_interval - time_since_last
                time.sleep(sleep_time)
            
            EdgarDataClient._last_request_time = time.time()
    
    def _normalize_text(self, s: str) -> str:
        """规范化文本：用于提升匹配准确度"""
        if not s:
            return ""
        s = s.lower().strip()
        s = s.replace("&", " and ")
        s = re.sub(r"[.,()\-_/]", " ", s)
        s = re.sub(r"\s+", " ", s)
        # 移除常见后缀词
        stopwords = {"inc", "inc.", "incorporated", "corp", "corporation", "co", "company", "plc", "ltd", "llc", "the"}
        tokens = [t for t in s.split() if t not in stopwords]
        return " ".join(tokens).strip()
    
    def _ensure_company_index(self):
        """确保公司索引已构建（按需构建或过期重建）"""
        with self._cache_lock:
            current_time = time.time()
            
            # 若 company_tickers 缓存不存在或已过期，先刷新
            if (EdgarDataClient._company_tickers_cache is None or
                EdgarDataClient._company_tickers_cache_time is None or
                current_time - EdgarDataClient._company_tickers_cache_time >= self._company_tickers_cache_ttl):
                # 拉取并更新 company_tickers 缓存
                self._rate_limit()
                url = "https://www.sec.gov/files/company_tickers.json"
                headers = {"User-Agent": self.user_agent}
                response = self.session.get(url, headers=headers, timeout=self.timeout)
                response.raise_for_status()
                companies = response.json()
                EdgarDataClient._company_tickers_cache = companies
                EdgarDataClient._company_tickers_cache_time = current_time
            else:
                companies = EdgarDataClient._company_tickers_cache
            
            # 若索引不存在或已过期，则重建索引
            if (EdgarDataClient._by_ticker is None or
                EdgarDataClient._by_title is None or
                EdgarDataClient._by_title_norm is None or
                EdgarDataClient._all_keys is None or
                EdgarDataClient._index_built_time is None or
                current_time - EdgarDataClient._index_built_time >= EdgarDataClient._index_ttl):
                
                by_ticker = {}
                by_title = {}
                by_title_norm = {}
                all_keys = []
                
                for _, company in companies.items():
                    title = company.get("title", "")
                    ticker = company.get("ticker", "")
                    cik_str = str(company.get("cik_str", "")).zfill(10)
                    
                    title_lower = title.lower()
                    ticker_lower = ticker.lower()
                    title_norm = self._normalize_text(title)
                    
                    # 构建索引：ticker、title、normalized title
                    if ticker_lower:
                        by_ticker[ticker_lower] = {"cik": cik_str, "name": title, "ticker": ticker}
                        all_keys.append(ticker_lower)
                    if title_lower:
                        by_title[title_lower] = {"cik": cik_str, "name": title, "ticker": ticker}
                    if title_norm:
                        by_title_norm[title_norm] = {"cik": cik_str, "name": title, "ticker": ticker}
                        all_keys.append(title_norm)
                
                EdgarDataClient._by_ticker = by_ticker
                EdgarDataClient._by_title = by_title
                EdgarDataClient._by_title_norm = by_title_norm
                EdgarDataClient._all_keys = all_keys
                EdgarDataClient._index_built_time = current_time
        
    def search_company_by_name(self, company_name):
        """Search company CIK by company name with caching and optimized ticker matching"""
        try:
            # 实例级缓存命中检查（按规范化后的query）
            norm_query = self._normalize_text(company_name)
            cache_hit = self._search_cache.get(norm_query)
            if cache_hit:
                return cache_hit
            
            # 确保索引已构建（首次或过期后会重建）
            self._ensure_company_index()
            
            # 获取索引引用（已在锁内构建完成）
            by_ticker = EdgarDataClient._by_ticker
            by_title = EdgarDataClient._by_title
            by_title_norm = EdgarDataClient._by_title_norm
            all_keys = EdgarDataClient._all_keys
            
            # ✅ OPTIMIZATION 1: Ticker 优先匹配（遵循项目规范）
            raw = company_name.strip().lower()
            raw_compact = re.sub(r"[^a-z0-9]", "", raw)
            is_ticker_like = len(raw_compact) <= 5 and len(raw_compact) >= 1
            
            if is_ticker_like and raw_compact in by_ticker:
                result = by_ticker[raw_compact]
                self._search_cache[norm_query] = result
                return result
            
            # ✅ OPTIMIZATION 2: 别名映射（如 'google' -> 'alphabet inc'）
            alias_target = EdgarDataClient._alias_map.get(norm_query)
            if alias_target:
                alias_norm = self._normalize_text(alias_target)
                # 先尝试规范化标题
                if alias_norm in by_title_norm:
                    result = by_title_norm[alias_norm]
                    self._search_cache[norm_query] = result
                    return result
                # 再尝试原始标题
                alias_lower = alias_target.lower()
                if alias_lower in by_title:
                    result = by_title[alias_lower]
                    self._search_cache[norm_query] = result
                    return result
                # 最后尝试 ticker（有些别名可能实际上是ticker）
                alias_ticker = re.sub(r"[^a-z0-9]", "", alias_lower)
                if alias_ticker in by_ticker:
                    result = by_ticker[alias_ticker]
                    self._search_cache[norm_query] = result
                    return result
            
            # ✅ OPTIMIZATION 3: 精确匹配（原始标题）
            title_lower = company_name.lower().strip()
            if title_lower in by_title:
                result = by_title[title_lower]
                self._search_cache[norm_query] = result
                return result
            
            # ✅ OPTIMIZATION 4: 精确匹配（规范化标题）
            if norm_query in by_title_norm:
                result = by_title_norm[norm_query]
                self._search_cache[norm_query] = result
                return result
            
            # ✅ OPTIMIZATION 5: 精确匹配（ticker，再次尝试原始输入）
            if raw_compact in by_ticker:
                result = by_ticker[raw_compact]
                self._search_cache[norm_query] = result
                return result
            
            # ✅ OPTIMIZATION 6: 部分包含匹配
            partial_matches = []
            for key in by_title_norm.keys():
                if norm_query in key:
                    partial_matches.append(key)
            if not partial_matches:
                for t in by_ticker.keys():
                    if norm_query in t:
                        partial_matches.append(t)
            if partial_matches:
                best_key = max(
                    partial_matches,
                    key=lambda k: difflib.SequenceMatcher(None, norm_query, k).ratio()
                )
                result = by_title_norm.get(best_key) or by_ticker.get(best_key)
                if result:
                    self._search_cache[norm_query] = result
                    return result
            
            # ✅ OPTIMIZATION 7: 模糊匹配（difflib，用于拼写近似的情况）
            close = difflib.get_close_matches(norm_query, all_keys, n=1, cutoff=0.78)
            if close:
                best = close[0]
                result = by_title_norm.get(best) or by_ticker.get(best)
                if result:
                    self._search_cache[norm_query] = result
                    return result
            
            # 未找到
            return None
            
        except TimeoutError as e:
            print(f"Timeout searching company: {e}")
            return None
        except Exception as e:
            print(f"Error searching company: {e}")
            return None
    
    @lru_cache(maxsize=128)
    def get_company_info(self, cik):
        """
        Get basic company information (cached)
        
        Args:
            cik (str): Company CIK code
            
        Returns:
            dict: Dictionary containing company information
        """
        if not self.edgar:
            print("sec_edgar_api library not installed")
            return None
            
        try:
            self._rate_limit()
            # Get company submissions (now has timeout protection)
            submissions = self.edgar.get_submissions(cik=cik)
            
            return {
                "cik": cik,
                "name": submissions.get("name", ""),
                "tickers": submissions.get("tickers", []),
                "sic": submissions.get("sic", ""),
                "sic_description": submissions.get("sicDescription", "")
            }
        except TimeoutError as e:
            print(f"Timeout getting company info for CIK {cik}: {e}")
            return None
        except Exception as e:
            print(f"Error getting company info: {e}")
            return None
    
    @lru_cache(maxsize=128)
    def get_company_filings(self, cik, form_types=None):
        """
        Get all company filing documents (cached)
        
        Args:
            cik (str): Company CIK code
            form_types (tuple): Tuple of form types, e.g., ('10-K', '10-Q'), None for all types
            
        Returns:
            list: List of filing documents
        """
        if not self.edgar:
            print("sec_edgar_api library not installed")
            return []
        
        # Convert list to tuple for caching (lists are not hashable)
        if form_types and isinstance(form_types, list):
            form_types = tuple(form_types)
            
        try:
            self._rate_limit()
            # Get company submissions (now has timeout protection)
            submissions = self.edgar.get_submissions(cik=cik)
            
            # Extract filing information
            filings = []
            recent = submissions.get("filings", {}).get("recent", {})
            
            # Get data from each field
            form_types_list = recent.get("form", [])
            filing_dates = recent.get("filingDate", [])
            accession_numbers = recent.get("accessionNumber", [])
            primary_documents = recent.get("primaryDocument", [])
            
            # Iterate through all filings
            for i in range(len(form_types_list)):
                form_type = form_types_list[i]
                
                # ✅ 归一化表单类型: "10-K/A" -> "10-K", "20-F/A" -> "20-F"
                # 这样修订版年报也能被正确识别和使用
                normalized_form_type = form_type.split('/')[0]
                
                # Filter by form type if specified (使用归一化后的类型)
                if form_types and normalized_form_type not in form_types:
                    continue
                
                filing_date = filing_dates[i] if i < len(filing_dates) else ""
                accession_number = accession_numbers[i] if i < len(accession_numbers) else ""
                primary_document = primary_documents[i] if i < len(primary_documents) else ""
                
                filing = {
                    "form_type": form_type,  # 保留原始form_type供参考
                    "filing_date": filing_date,
                    "accession_number": accession_number,
                    "primary_document": primary_document
                }
                
                filings.append(filing)
            
            return filings
        except TimeoutError as e:
            print(f"Timeout getting company filings for CIK {cik}: {e}")
            return []
        except Exception as e:
            print(f"Error getting company filings: {e}")
            return []
    
    @lru_cache(maxsize=128)
    def get_company_facts(self, cik):
        """
        Get all company financial facts data (cached)
        
        Args:
            cik (str): Company CIK code
            
        Returns:
            dict: Company financial facts data
        """
        if not self.edgar:
            print("sec_edgar_api library not installed")
            return {}
            
        try:
            self._rate_limit()
            # Now has timeout protection via monkey patch
            facts = self.edgar.get_company_facts(cik=cik)
            return facts
        except TimeoutError as e:
            print(f"Timeout getting company facts for CIK {cik}: {e}")
            return {}
        except Exception as e:
            print(f"Error getting company facts: {e}")
            return {}
    
    def get_financial_data_for_period(self, cik, period):
        """
        Get financial data for a specific period (supports annual and quarterly) - Cached
        
        Args:
            cik (str): Company CIK code
            period (str): Period in format 'YYYY' or 'YYYYQX' (e.g., '2025' or '2025Q3')
            
        Returns:
            dict: Financial data dictionary
        """
        if not self.edgar:
            print("sec_edgar_api library not installed")
            return {}
        
        # 实例级缓存（避免重复计算）
        cache_key = f"period_{cik}_{period}"
        if hasattr(self, '_period_cache') and cache_key in self._period_cache:
            return self._period_cache[cache_key]
        
        if not hasattr(self, '_period_cache'):
            self._period_cache = {}
            
        try:
            # Get company financial facts
            facts = self.get_company_facts(cik)
            
            if not facts:
                return {}
            
            # Extract us-gaap and ifrs-full financial data (20-F may use IFRS)
            us_gaap = facts.get("facts", {}).get("us-gaap", {})
            ifrs_full = facts.get("facts", {}).get("ifrs-full", {})
            
            # Define financial metrics and their XBRL tags
            # Include multiple possible tags to improve match rate (including US-GAAP and IFRS tags)
            financial_metrics = {
                "total_revenue": ["Revenues", "RevenueFromContractWithCustomerExcludingAssessedTax", "RevenueFromContractWithCustomerIncludingAssessedTax", "SalesRevenueNet", "RevenueFromContractWithCustomer", "Revenue"],
                "net_income": ["NetIncomeLoss", "ProfitLoss", "NetIncome", "ProfitLossAttributableToOwnersOfParent"],
                "earnings_per_share": ["EarningsPerShareBasic", "EarningsPerShare", "BasicEarningsPerShare", "BasicEarningsLossPerShare"],
                "operating_expenses": ["OperatingExpenses", "OperatingCostsAndExpenses", "OperatingExpensesExcludingDepreciationAndAmortization", "CostsAndExpenses", "GeneralAndAdministrativeExpense", "CostOfRevenue", "ResearchAndDevelopmentExpense", "SellingAndMarketingExpense"],
                "operating_cash_flow": ["NetCashProvidedByUsedInOperatingActivities", "NetCashProvidedUsedInOperatingActivities", "NetCashFlowsFromUsedInOperatingActivities", "CashFlowsFromUsedInOperatingActivities"],
            }
            
            # Store result with new optimized structure
            result = {
                "period": period,
                "_metadata": {},
                "metrics": {}
            }
            
            # Determine target form types to search
            if 'Q' in period:
                # Quarterly data, mainly search 10-Q (20-F usually doesn't have quarterly reports)
                target_forms = ("10-Q",)  # Use tuple for caching
                target_forms_annual = ("10-K", "20-F")  # for fallback
                year = int(period.split('Q')[0])
                quarter = period.split('Q')[1]
            else:
                # Annual data, search 10-K and 20-F annual forms
                target_forms = ("10-K", "20-F")  # Use tuple for caching
                target_forms_annual = target_forms
                year = int(period)
                quarter = None
            
            # Get company filings to find accession number and primary document
            filings = self.get_company_filings(cik, form_types=target_forms)
            filings_map = {}  # Map: form -> {accession_number, primary_document, filing_date}
            
            # Build filing map for quick lookup
            for filing in filings:
                form_type = filing.get("form_type", "")
                filing_date = filing.get("filing_date", "")
                accession_number = filing.get("accession_number", "")
                primary_document = filing.get("primary_document", "")
                
                if filing_date and accession_number:
                    # Extract year from filing_date (format: YYYY-MM-DD)
                    file_year = int(filing_date[:4]) if len(filing_date) >= 4 else 0
                    
                    # ✅ 归一化表单类型: "10-K/A" -> "10-K", "20-F/A" -> "20-F"
                    # 使用归一化后的类型构建 key,这样 facts 中的 "10-K" 能命中 "10-K/A"
                    normalized_form_type = form_type.split('/')[0]
                    
                    # ✅ FIXED: Remove year filter to keep all filings
                    # 20-F forms are often filed in the year after the fiscal year
                    # We'll match them later using fiscal year (fy) and filed date
                    key = f"{normalized_form_type}_{file_year}"  # 使用归一化后的类型
                    if key not in filings_map:
                        filings_map[key] = {
                            "accession_number": accession_number,
                            "primary_document": primary_document,
                            "form_type": form_type,  # 保留原始 form_type
                            "filing_date": filing_date
                        }
            
            # Iterate through each financial metric
            for metric_key, metric_tags in financial_metrics.items():
                # Support multiple possible tags
                for metric_tag in metric_tags:
                    # Search both US-GAAP and IFRS tags
                    metric_data = None
                    data_source = None
                    
                    if metric_tag in us_gaap:
                        metric_data = us_gaap[metric_tag]
                        data_source = "us-gaap"
                    elif metric_tag in ifrs_full:
                        metric_data = ifrs_full[metric_tag]
                        data_source = "ifrs-full"
                    
                    if metric_data:
                        units = metric_data.get("units", {})
                        
                        # Find USD unit data (supports USD and USD/shares)
                        usd_data = None
                        if "USD" in units:
                            usd_data = units["USD"]
                        elif "USD/shares" in units and metric_key == "earnings_per_share":
                            # EPS uses USD/shares unit
                            usd_data = units["USD/shares"]
                        
                        if usd_data:
                            # Try exact match first, then loose match
                            matched_entry = None
                            
                            # Search for data in the specified period
                            for entry in usd_data:
                                form = entry.get("form", "")
                                fy = entry.get("fy", 0)
                                fp = entry.get("fp", "")
                                end_date = entry.get("end", "")
                                
                                if not end_date or len(end_date) < 4:
                                    continue
                                    
                                entry_year = int(end_date[:4])
                                
                                # Check if form type matches
                                if form in target_forms:
                                    if quarter:
                                        # Quarterly data match
                                        if entry_year == year and fp == f"Q{quarter}":
                                            # If already matched, compare end date, choose the latest
                                            if matched_entry:
                                                if entry.get("end", "") > matched_entry.get("end", ""):
                                                    matched_entry = entry
                                            else:
                                                matched_entry = entry
                                    else:
                                        # Annual data match - prioritize fiscal year (fy) field
                                        # Strategy 1: Exact match by fiscal year
                                        if fy == year and (fp == "FY" or fp == "" or not fp):
                                            # If already matched, compare end date, choose the latest
                                            if matched_entry:
                                                if entry.get("end", "") > matched_entry.get("end", ""):
                                                    matched_entry = entry
                                            else:
                                                matched_entry = entry
                                        # Strategy 2: Match by end date year (when fy not available or doesn't match)
                                        elif not matched_entry and entry_year == year and (fp == "FY" or fp == "" or not fp):
                                            matched_entry = entry
                                        # Strategy 3: Allow fy to differ by 1 year (fiscal year vs calendar year mismatch)
                                        elif not matched_entry and fy > 0 and abs(fy - year) <= 1 and (fp == "FY" or fp == "" or not fp):
                                            matched_entry = entry
                                        # Strategy 4: Match by frame field for 20-F
                                        elif not matched_entry and form == "20-F" and "frame" in entry:
                                            frame = entry.get("frame", "")
                                            if f"CY{year}" in frame or str(year) in end_date:
                                                matched_entry = entry
                            
                            # If quarterly data not found, try finding from annual report (fallback strategy)
                            if not matched_entry and quarter and target_forms_annual:
                                for entry in usd_data:
                                    form = entry.get("form", "")
                                    end_date = entry.get("end", "")
                                    fp = entry.get("fp", "")
                                    
                                    if form in target_forms_annual and end_date:
                                        # Check if end date is within this quarter range
                                        if str(year) in end_date and f"Q{quarter}" in fp:
                                            matched_entry = entry
                                            break
                            
                            # Apply matched data
                            if matched_entry:
                                # Store metric value and tag
                                result["metrics"][metric_key] = {
                                    "value": matched_entry.get("val", 0),
                                    "tag": metric_tag
                                }
                                
                                # Get form and accession info - only populate metadata once
                                if not result["_metadata"]:
                                    form_type = matched_entry.get("form", "")
                                    accn_from_facts = matched_entry.get('accn', '').replace('-', '')
                                    filed_date = matched_entry.get('filed', '')
                                    
                                    # Multi-strategy filing lookup for 20-F and cross-year submissions
                                    filing_info = None
                                    
                                    # Strategy 1: Try matching by fiscal year
                                    filing_key = f"{form_type}_{year}"
                                    filing_info = filings_map.get(filing_key)
                                    
                                    # Strategy 2: Try matching by filed year (for 20-F filed in next year)
                                    if not filing_info and filed_date:
                                        filed_year = int(filed_date[:4]) if len(filed_date) >= 4 else 0
                                        if filed_year > 0:
                                            filing_key = f"{form_type}_{filed_year}"
                                            filing_info = filings_map.get(filing_key)
                                    
                                    # Strategy 3: Try fiscal year + 1 (common for 20-F)
                                    if not filing_info:
                                        filing_key = f"{form_type}_{year + 1}"
                                        filing_info = filings_map.get(filing_key)
                                    
                                    # Strategy 4: Search all filings with matching form type and accession
                                    if not filing_info and accn_from_facts:
                                        for key, finfo in filings_map.items():
                                            if finfo["form_type"] == form_type:
                                                filing_accn = finfo["accession_number"].replace('-', '')
                                                if filing_accn == accn_from_facts:
                                                    filing_info = finfo
                                                    break
                                    
                                    # Generate source URL
                                    source_url = ""
                                    if filing_info:
                                        accession_number = filing_info["accession_number"].replace('-', '')
                                        primary_document = filing_info["primary_document"]
                                        if primary_document:
                                            source_url = f"https://www.sec.gov/Archives/edgar/data/{cik}/{accession_number}/{primary_document}"
                                        else:
                                            source_url = f"https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany&CIK={cik}&type={form_type}&dateb=&owner=exclude&count=100"
                                    else:
                                        source_url = f"https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany&CIK={cik}&type={form_type}&dateb=&owner=exclude&count=100"
                                    
                                    # Populate metadata (only once per period)
                                    result["_metadata"] = {
                                        "form": matched_entry.get("form", ""),
                                        "fiscal_year": matched_entry.get("fy", 0),
                                        "fiscal_period": matched_entry.get("fp", ""),
                                        "start_date": matched_entry.get("start", ""),
                                        "end_date": matched_entry.get("end", ""),
                                        "filed_date": matched_entry.get("filed", ""),
                                        "source_url": source_url,
                                        "data_source": data_source
                                    }
                        
                        # If data is found, break out of tag loop
                        if metric_key in result["metrics"]:
                            break
            
            # 缓存结果
            if result and "period" in result:
                self._period_cache[cache_key] = result
            
            return result
        except Exception as e:
            print(f"Error getting financial data for period {period}: {e}")
            return {}