test

app.py

@@ -5,9 +5,13 @@ import re
 import pandas as pd
 import globals as g
 from service.mysql_service import get_companys, insert_company, get_company_by_name
-from service.chat_service import get_analysis_report, search_company, search_news, get_invest_suggest, chat_bot
+from service.chat_service import get_analysis_report, get_stock_price_from_bailian, search_company, search_news, get_invest_suggest, chat_bot
 from service.company import check_company_exists
 from service.hf_upload import get_hf_files_with_links
+from service.report_mcp import query_financial_data
+from service.report_tools import build_financial_metrics_three_year_data, calculate_yoy_comparison, extract_financial_table, extract_last_three_with_fallback, get_yearly_data
+from service.three_year_table_tool import build_table_format
+from service.three_year_tool import  process_financial_data_with_metadata
 from service.tool_processor import get_stock_price
 
 custom_css = """
@@ -620,6 +624,104 @@ def create_report_section():
     report_display = gr.HTML(initial_content)
     return report_display
 
+def format_financial_metrics(data: dict, prev_data: dict = None) -> list:  # pyright: ignore[reportArgumentType]
+    """
+    将原始财务数据转换为 financial_metrics 格式。
+    
+    Args:
+        data (dict): 当前财年数据（必须包含 total_revenue, net_income 等字段）
+        prev_data (dict, optional): 上一财年数据，用于计算 change。若未提供，change 设为 "--"
+    
+    Returns:
+        list[dict]: 符合 financial_metrics 格式的列表
+    """
+    
+    def format_currency(value: float) -> str:
+        """将数字格式化为 $XB / $XM / $XK"""
+        if value >= 1e9:
+            return f"${value / 1e9:.2f}B"
+        elif value >= 1e6:
+            return f"${value / 1e6:.2f}M"
+        elif value >= 1e3:
+            return f"${value / 1e3:.2f}K"
+        else:
+            return f"${value:.2f}"
+
+    def calculate_change(current: float, previous: float) -> tuple:
+        """计算变化百分比和颜色"""
+        if previous == 0:
+            return "--", "gray"
+        change_pct = (current - previous) / abs(previous) * 100
+        sign = "+" if change_pct >= 0 else ""
+        color = "green" if change_pct >= 0 else "red"
+        return f"{sign}{change_pct:.1f}%", color
+
+    # 定义指标映射
+    metrics_config = [
+        {
+            "key": "total_revenue",
+            "label": "Total Revenue",
+            "is_currency": True,
+            "eps_like": False
+        },
+        {
+            "key": "net_income",
+            "label": "Net Income",
+            "is_currency": True,
+            "eps_like": False
+        },
+        {
+            "key": "earnings_per_share",
+            "label": "Earnings Per Share",
+            "is_currency": False,  # EPS 不用 B/M 单位
+            "eps_like": True
+        },
+        {
+            "key": "operating_expenses",
+            "label": "Operating Expenses",
+            "is_currency": True,
+            "eps_like": False
+        },
+        {
+            "key": "operating_cash_flow",
+            "label": "Cash Flow",
+            "is_currency": True,
+            "eps_like": False
+        }
+    ]
+
+    result = []
+    for item in metrics_config:
+        key = item["key"]
+        current_val = data.get(key)
+        if current_val is None:
+            continue
+
+        # 格式化 value
+        if item["eps_like"]:
+            value_str = f"${current_val:.2f}"
+        elif item["is_currency"]:
+            value_str = format_currency(current_val)
+        else:
+            value_str = str(current_val)
+
+        # 计算 change（如果有上期数据）
+        if prev_data and key in prev_data:
+            prev_val = prev_data[key]
+            change_str, color = calculate_change(current_val, prev_val)
+        else:
+            change_str = "--"
+            color = "gray"
+
+        result.append({
+            "label": item["label"],
+            "value": value_str,
+            "change": change_str,
+            "color": color
+        })
+
+    return result
+
 def create_sidebar():
     """创建侧边栏组件"""
     # 初始化 companies_map
@@ -710,7 +812,83 @@ def create_sidebar():
 
     # 返回公司列表组件和报告部分组件
     return company_list, report_section_group, report_display
-
+    
+def build_income_table(table_data):
+            # 兼容两种数据结构：
+            # 1. 新结构：包含 list_data 和 yoy_rates 的字典
+            # 2. 旧结构：直接是二维数组
+            if isinstance(table_data, dict) and "list_data" in table_data:
+                # 新结构
+                income_statement = table_data["list_data"]
+                yoy_rates = table_data["yoy_rates"] or []
+            else:
+                # 旧结构，直接使用传入的数据
+                income_statement = table_data
+                yoy_rates = []
+            
+            # 创建一个映射，将年份列索引映射到增长率
+            yoy_map = {}
+            if len(yoy_rates) > 1 and len(yoy_rates[0]) > 1:
+                # 获取增长率表头（跳过第一列"Category"）
+                yoy_headers = yoy_rates[0][1:]
+                
+                # 为每个指标行创建增长率映射
+                for i, yoy_row in enumerate(yoy_rates[1:], 1):  # 跳过标题行
+                    category = yoy_row[0]
+                    yoy_map[category] = {}
+                    for j, rate in enumerate(yoy_row[1:]):
+                        if j < len(yoy_headers):
+                            yoy_map[category][yoy_headers[j]] = rate
+            
+            table_rows = ""
+            header_row = income_statement[0]
+            
+            for i, row in enumerate(income_statement):
+                if i == 0:
+                    row_style = "background-color: #f5f5f5; font-weight: 500;"
+                else:
+                    row_style = "background-color: #f9f9f9;"
+                cells = ""
+                
+                for j, cell in enumerate(row):
+                    if j == 0:
+                        cells += f"<td style='padding: 8px; border: 1px solid #ddd; text-align: center; font-size: 14px;'>{cell}</td>"
+                    else:
+                        # 添加增长率箭头（如果有的话）
+                        growth = None
+                        category = row[0]
+                        # j是当前单元格索引，0是类别列，1,2,3...是数据列
+                        # yoy_map的键是年份，例如"2024/FY"
+                        if i > 0 and category in yoy_map and j > 0 and j < len(header_row):
+                            year_header = header_row[j]
+                            if year_header in yoy_map[category]:
+                                growth = yoy_map[category][year_header]
+                        
+                        if growth and growth != "N/A":
+                            arrow = "▲" if growth.startswith("+") else "▼"
+                            color = "green" if growth.startswith("+") else "red"
+                            cells += f"""<td style='padding: 8px; border: 1px solid #ddd; text-align: center; font-size: 14px; position: relative;'>
+                            <div>{cell}</div>
+                            <div style='position: absolute; bottom: -5px; right: 5px; font-size: 10px; color: {color};'>{arrow}{growth}</div>
+                            </td>"""
+                        else:
+                            cells += f"<td style='padding: 8px; border: 1px solid #ddd; text-align: center; font-size: 14px;'>{cell}</td>"
+                table_rows += f"<tr style='{row_style}'>{cells}</tr>"
+            
+            html = f"""
+            <div style="min-width: 400px;max-width: 600px;height: 300px !important;border: 1px solid #e0e0e0; border-radius: 8px; padding: 16px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); font-family: 'Segoe UI', sans-serif;">
+                <div style="display: flex; align-items: center; gap: 8px; margin-bottom: 16px;">
+                    <svg width="16" height="16" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+                        <path d="M12 2L15.09 8.26L19 9.07L16 14L16 19L12 19L8 14L8 9.07L4.91 8.26L8 2L12 2Z" fill="#0066cc"/>
+                    </svg>
+                    <div style="font-size: 18px; font-weight: 600;">Income Statement and Cash Flow</div>
+                </div>
+                <table style="width: 100%; border-collapse: collapse; font-size: 14px;">
+                    {table_rows}
+                </table>
+            </div>
+            """
+            return html
 def create_metrics_dashboard():
     """创建指标仪表板组件"""
     with gr.Row(elem_classes=["metrics-dashboard"]):
@@ -737,23 +915,43 @@ def create_metrics_dashboard():
             "volume": "27.10M"
         }
 
+        # financial_metrics = query_financial_data("NVDA", "最新财务数据")
+        # print(f"最新财务数据: {financial_metrics}")
         financial_metrics = [
-            {"label": "Total Revenue", "value": "$2.84B", "change": "+12.4%", "color": "green"},
-            {"label": "Net Income", "value": "$685M", "change": "-3.2%", "color": "red"},
-            {"label": "Earnings Per Share", "value": "$2.15", "change": "-3.2%", "color": "red"},
-            {"label": "Operating Expenses", "value": "$1.2B", "change": "+5.1%", "color": "green"},
-            {"label": "Cash Flow", "value": "$982M", "change": "+8.7%", "color": "green"}
-        ]
-
-        income_statement = [
-            ["Category", "2024/FY", "2023/FY", "2022/FY"],
-            ["Total", "130350M", "126491M", "134567M"],
-            ["Net Income", "11081", "10598M", "9818.4M"],
-            ["Earnings Per Share", "4.38", "4.03", "3.62"],
-            ["Operating Expenses", "31990.9M", "31439.6M", "34516.2M"],
-            ["Cash Flow", "25289.9M", "29086M", "22517.2M"]
+            {"label": "Total Revenue", "value": "N/A", "change": "N/A", "color": "grey"},
+            {"label": "Net Income", "value": "N/A", "change": "N/A", "color": "grey"},
+            {"label": "Earnings Per Share", "value": "N/A", "change": "N/A", "color": "grey"},
+            {"label": "Operating Expenses", "value": "N/A", "change": "N/A", "color": "grey"},
+            {"label": "Cash Flow", "value": "N/A", "change": "N/A", "color": "grey"}
         ]
-
+        # income_statement = [
+        #     ["Category", "2024/FY", "2023/FY", "2022/FY"],
+        #     ["Total", "130350M", "126491M", "134567M"],
+        #     ["Net Income", "11081", "10598M", "9818.4M"],
+        #     ["Earnings Per Share", "4.38", "4.03", "3.62"],
+        #     ["Operating Expenses", "31990.9M", "31439.6M", "34516.2M"],
+        #     ["Cash Flow", "25289.9M", "29086M", "22517.2M"]
+        # ]
+        income_statement = {
+        "list_data": [
+        ["Category", "N/A/FY", "N/A/FY", "N/A/FY"],
+        ["Total", "N/A", "N/A", "N/A"],
+        ["Net Income", "N/A", "N/A", "N/A.4M"],
+        ["Earnings Per Share", "N/A", "N/A", "N/A"],
+        ["Operating Expenses", "N/A", "N/A", "N/A"],
+        ["Cash Flow", "N/A", "N/A", "N/A"]
+     ],
+     "yoy_rates": []
+    #  "yoy_rates": [
+    #     ["Category", "N/A/FY", "N/A/FY"],
+    #     ["Total", "N/A", "N/A"],
+    #     ["Net Income", "+3.05%", "-6.00%"],
+    #     ["Earnings Per Share", "+3.05%", "-6.00%"],
+    #     ["Operating Expenses", "+29.17%", "-6.00%"],
+    #     ["Cash Flow", "-13.05%", "-6.00%"]
+    #  ]
+    }
+        yearly_data = 'N/A'
         # 增长变化的 HTML 字符（箭头+百分比）
         def render_change(change: str, color: str):
             if change.startswith("+"):
@@ -798,62 +996,15 @@ def create_metrics_dashboard():
                     <svg width="16" height="16" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
                         <path d="M12 2L15.09 8.26L19 9.07L16 14L16 19L12 19L8 14L8 9.07L4.91 8.26L8 2L12 2Z" fill="#0066cc"/>
                     </svg>
-                    <div style="font-size: 18px; font-weight: 600;">2025 Q3 Financial Metrics</div>
+                    <div style="font-size: 18px; font-weight: 600;">{yearly_data} Financial Metrics</div>
                 </div>
                 {metrics_html}
             </div>
             """
             return html
 
-        # 构建右侧表格
-        def build_income_table():
-            table_rows = ""
-            for i, row in enumerate(income_statement):
-                if i == 0:
-                    row_style = "background-color: #f5f5f5; font-weight: 500;"
-                else:
-                    row_style = "background-color: #f9f9f9;"
-                cells = ""
-                for j, cell in enumerate(row):
-                    if j == 0:
-                        cells += f"<td style='padding: 8px; border: 1px solid #ddd; text-align: left; font-size: 14px;'>{cell}</td>"
-                    else:
-                        # 添加增长箭头（模拟数据）
-                        growth = None
-                        if i == 1 and j == 1: growth = "+3.05%"
-                        elif i == 1 and j == 2: growth = "-6.00%"
-                        elif i == 2 and j == 1: growth = "+3.05%"
-                        elif i == 2 and j == 2: growth = "-6.00%"
-                        elif i == 3 and j == 1: growth = "+3.05%"
-                        elif i == 3 and j == 2: growth = "-6.00%"
-                        elif i == 4 and j == 1: growth = "+29.17%"
-                        elif i == 4 and j == 2: growth = "+29.17%"
-                        elif i == 5 and j == 1: growth = "-13.05%"
-                        elif i == 5 and j == 2: growth = "+29.17%"
-
-                        if growth:
-                            arrow = "▲" if growth.startswith("+") else "▼"
-                            color = "green" if growth.startswith("+") else "red"
-                            cells += f"<td style='padding: 8px; border: 1px solid #ddd; text-align: right; font-size: 14px; position: relative;'><div>{cell}</div><div style='position: absolute; bottom: -5px; right: 5px; font-size: 10px; color: {color};'>{arrow}{growth}</div></td>"
-                        else:
-                            cells += f"<td style='padding: 8px; border: 1px solid #ddd; text-align: right; font-size: 14px;'>{cell}</td>"
-                table_rows += f"<tr style='{row_style}'>{cells}</tr>"
-            
-            html = f"""
-            <div style="min-width: 400px;max-width: 600px;height: 300px !important;border: 1px solid #e0e0e0; border-radius: 8px; padding: 16px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); font-family: 'Segoe UI', sans-serif;">
-                <div style="display: flex; align-items: center; gap: 8px; margin-bottom: 16px;">
-                    <svg width="16" height="16" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
-                        <path d="M12 2L15.09 8.26L19 9.07L16 14L16 19L12 19L8 14L8 9.07L4.91 8.26L8 2L12 2Z" fill="#0066cc"/>
-                    </svg>
-                    <div style="font-size: 18px; font-weight: 600;">Income Statement and Cash Flow</div>
-                </div>
-                <table style="width: 100%; border-collapse: collapse; font-size: 14px;">
-                    {table_rows}
-                </table>
-            </div>
-            """
-            return html
-       # 主函数：返回所有 HTML 片段
+
+        # 主函数：返回所有 HTML 片段
         def get_dashboard():
             with gr.Row():
                 with gr.Column(scale=1, min_width=250, elem_classes=["metric-card-col-left"]):
@@ -861,7 +1012,8 @@ def create_metrics_dashboard():
                 with gr.Column(scale=1, min_width=300, elem_classes=["metric-card-col-middle"]):
                     financial_metrics_html = gr.HTML(build_financial_metrics(), elem_classes=["metric-card-middle"])
                 with gr.Column(scale=1, min_width=450, elem_classes=["metric-card-col-right"]):
-                    income_table_html = gr.HTML(build_income_table(), elem_classes=["metric-card-right"])
+                    # 传递income_statement参数
+                    income_table_html = gr.HTML(build_income_table(income_statement), elem_classes=["metric-card-right"])
             return stock_card_html, financial_metrics_html, income_table_html
         
         # 创建指标仪表板并保存引用
@@ -875,77 +1027,160 @@ def create_metrics_dashboard():
 def update_metrics_dashboard(company_name):
     """根据选择的公司更新指标仪表板"""
     # 模拟数据
-    company_info = {
-        "name": company_name,
-        "symbol": "NYSE:BABA",
-        "price": 157.65,
-        "change": 0.64,
-        "change_percent": 0.41,
-        "open": 165.20,
-        "high": 166.37,
-        "low": 156.15,
-        "prev_close": 157.01,
-        "volume": "27.10M"
-    }
-    
+    # company_info = {
+    #     "name": company_name,
+    #     "symbol": "NYSE:BABA",
+    #     "price": 157.65,
+    #     "change": 0.64,
+    #     "change_percent": 0.41,
+    #     "open": 165.20,
+    #     "high": 166.37,
+    #     "low": 156.15,
+    #     "prev_close": 157.01,
+    #     "volume": "27.10M"
+    # }
+    company_info = {}
     # 尝试获取股票价格数据，但不中断程序执行
     try:
         # 根据选择的公司获取股票代码
         stock_code = get_stock_code_by_company_name(company_name)
-        company_info2 = get_stock_price(stock_code)
-        print(f"股票价格数据: {company_info2}")
-        
+        # company_info2 = get_stock_price(stock_code)
+        company_info2 = get_stock_price_from_bailian(stock_code)
+        # print(f"股票价格数据: {company_info2}")
+        company_info = company_info2
         # 如果成功获取数据，则用实际数据替换模拟数据
-        if company_info2 and "content" in company_info2 and len(company_info2["content"]) > 0:
-            import json
-            # 解析返回的JSON数据
-            data_text = company_info2["content"][0]["text"]
-            stock_data = json.loads(data_text)
+        # if company_info2 and "content" in company_info2 and len(company_info2["content"]) > 0:
+        #     import json
+        #     # 解析返回的JSON数据
+        #     data_text = company_info2["content"][0]["text"]
+        #     stock_data = json.loads(data_text)
             
-            # 提取数据
-            quote = stock_data["Global Quote"]
+        #     # 提取数据
+        #     quote = stock_data["Global Quote"]
             
-            # 转换交易量单位
-            volume = int(quote['06. volume'])
-            if volume >= 1000000:
-                volume_str = f"{volume / 1000000:.2f}M"
-            elif volume >= 1000:
-                volume_str = f"{volume / 1000:.2f}K"
-            else:
-                volume_str = str(volume)
+        #     # 转换交易量单位
+        #     volume = int(quote['06. volume'])
+        #     if volume >= 1000000:
+        #         volume_str = f"{volume / 1000000:.2f}M"
+        #     elif volume >= 1000:
+        #         volume_str = f"{volume / 1000:.2f}K"
+        #     else:
+        #         volume_str = str(volume)
             
-            company_info = {
-                "name": company_name,
-                "symbol": f"NYSE：{quote['01. symbol']}",
-                "price": float(quote['05. price']),
-                "change": float(quote['09. change']),
-                "change_percent": float(quote['10. change percent'].rstrip('%')),
-                "open": float(quote['02. open']),
-                "high": float(quote['03. high']),
-                "low": float(quote['04. low']),
-                "prev_close": float(quote['08. previous close']),
-                "volume": volume_str
-            }
+        #     company_info = {
+        #         "name": company_name,
+        #         "symbol": f"NYSE：{quote['01. symbol']}",
+        #         "price": float(quote['05. price']),
+        #         "change": float(quote['09. change']),
+        #         "change_percent": float(quote['10. change percent'].rstrip('%')),
+        #         "open": float(quote['02. open']),
+        #         "high": float(quote['03. high']),
+        #         "low": float(quote['04. low']),
+        #         "prev_close": float(quote['08. previous close']),
+        #         "volume": volume_str
+        #     }
     except Exception as e:
         print(f"获取股票价格数据失败: {e}")
         company_info2 = None
 
-    financial_metrics = [
-        {"label": "Total Revenue", "value": "$2.84B", "change": "+12.4%", "color": "green"},
-        {"label": "Net Income", "value": "$685M", "change": "-3.2%", "color": "red"},
-        {"label": "Earnings Per Share", "value": "$2.15", "change": "-3.2%", "color": "red"},
-        {"label": "Operating Expenses", "value": "$1.2B", "change": "+5.1%", "color": "green"},
-        {"label": "Cash Flow", "value": "$982M", "change": "+8.7%", "color": "green"}
-    ]
-
-    income_statement = [
-        ["Category", "2024/FY", "2023/FY", "2022/FY"],
-        ["Total", "130350M", "126491M", "134567M"],
-        ["Net Income", "11081", "10598M", "9818.4M"],
-        ["Earnings Per Share", "4.38", "4.03", "3.62"],
-        ["Operating Expenses", "31990.9M", "31439.6M", "34516.2M"],
-        ["Cash Flow", "25289.9M", "29086M", "22517.2M"]
-    ]
+    # financial_metrics = [
+    #     {"label": "Total Revenue", "value": "$2.84B", "change": "+12.4%", "color": "green"},
+    #     {"label": "Net Income", "value": "$685M", "change": "-3.2%", "color": "red"},
+    #     {"label": "Earnings Per Share", "value": "$2.15", "change": "-3.2%", "color": "red"},
+    #     {"label": "Operating Expenses", "value": "$1.2B", "change": "+5.1%", "color": "green"},
+    #     {"label": "Cash Flow", "value": "$982M", "change": "+8.7%", "color": "green"}
+    # ]
+    financial_metrics_pre = query_financial_data(company_name, "5-Year")
+    # financial_metrics_pre = query_financial_data(company_name, "5年趋势")
+    # print(f"最新财务数据: {financial_metrics_pre}")
+    # financial_metrics = format_financial_metrics(financial_metrics_pre)
+
+
+    # financial_metrics_pre_2 = extract_last_three_with_fallback(financial_metrics_pre)
+    # print(f"提取的3年数据: {financial_metrics_pre_2}")
+    # financial_metrics_pre = {
+    #     "metrics": financial_metrics_pre_2
+    # }
+    financial_metrics = []
+    # try:
+    #     # financial_metrics = calculate_yoy_comparison(financial_metrics_pre)
+    #     financial_metrics = build_financial_metrics_three_year_data(financial_metrics_pre)
+    #     print(f"格式化后的财务数据: {financial_metrics}")
+    # except Exception as e:
+    #     print(f"Error calculating YOY comparison: {e}")
+    year_data = None
+    three_year_data = None
+    try:
+        # financial_metrics = process_financial_data_with_metadata(financial_metrics_pre)
+        result = process_financial_data_with_metadata(financial_metrics_pre)
+
+        # 按需提取字段
+        financial_metrics = result["financial_metrics"]
+        year_data = result["year_data"]
+        three_year_data = result["three_year_data"]
+        print(f"格式化后的财务数据: {financial_metrics}")
+    except Exception as e:
+        print(f"Error process_financial_data: {e}")
+   
+
+    # income_statement = [
+    #     ["Category", "2024/FY", "2023/FY", "2022/FY"],
+    #     ["Total", "130350M", "126491M", "134567M"],
+    #     ["Net Income", "11081", "10598M", "9818.4M"],
+    #     ["Earnings Per Share", "4.38", "4.03", "3.62"],
+    #     ["Operating Expenses", "31990.9M", "31439.6M", "34516.2M"],
+    #     ["Cash Flow", "25289.9M", "29086M", "22517.2M"]
+    # ]
+
+    # table_data = None
+    # try:
+    #     table_data = extract_financial_table(financial_metrics_pre)
+    #     print(table_data)
+    # except Exception as e:
+    #     print(f"Error extract_financial_table: {e}")
+    # yearly_data = None
+    # try:
+    #     yearly_data = get_yearly_data(financial_metrics_pre)
+    # except Exception as e:
+    #     print(f"Error get_yearly_data: {e}")
+
+    # ======
+    # table_data = [
+    #     ["Category", "2024/FY", "2023/FY", "2022/FY"],
+    #     ["Total", "130350M", "126491M", "134567M"],
+    #     ["Net Income", "11081", "10598M", "9818.4M"],
+    #     ["Earnings Per Share", "4.38", "4.03", "3.62"],
+    #     ["Operating Expenses", "31990.9M", "31439.6M", "34516.2M"],
+    #     ["Cash Flow", "25289.9M", "29086M", "22517.2M"]
+    # ]   
+    yearly_data = year_data
+    table_data = build_table_format(three_year_data)
+    print(f"table_data: {table_data}")
+      # yearly_data = None
+    # try:
+    #     yearly_data = get_yearly_data(financial_metrics_pre)
+    # except Exception as e:
+    #     print(f"Error get_yearly_data: {e}")
+    #=======
+
+    # exp = {
+    #     "list_data": [
+    #     ["Category", "2024/FY", "2023/FY", "2022/FY"],
+    #     ["Total", "130350M", "126491M", "134567M"],
+    #     ["Net Income", "11081", "10598M", "9818.4M"],
+    #     ["Earnings Per Share", "4.38", "4.03", "3.62"],
+    #     ["Operating Expenses", "31990.9M", "31439.6M", "34516.2M"],
+    #     ["Cash Flow", "25289.9M", "29086M", "22517.2M"]
+    #  ],
+    #  "yoy_rates": [
+    #     ["Category", "2024/FY", "2023/FY"],
+    #     ["Total", "+3.05%", "-6.00%"],
+    #     ["Net Income", "+3.05%", "-6.00%"],
+    #     ["Earnings Per Share", "+3.05%", "-6.00%"],
+    #     ["Operating Expenses", "+29.17%", "-6.00%"],
+    #     ["Cash Flow", "-13.05%", "-6.00%"]
+    #  ]
+    # }
 
     # 增长变化的 HTML 字符（箭头+百分比）
     def render_change(change: str, color: str):
@@ -955,45 +1190,128 @@ def update_metrics_dashboard(company_name):
             return f'<span style="color:{color};">▼{change}</span>'
 
     # 构建左侧卡片
-    def build_stock_card():
-        # 检查是否获取到了股票数据，如果没有则显示N/A
-        if company_info2 and "content" in company_info2 and len(company_info2["content"]) > 0:
-            price = company_info["price"]
-            change = company_info["change"]
-            change_percent = company_info["change_percent"]
-
-            # 格式化价格变动
-            change_color = "green" if change > 0 else "red"
-            change_html = f'<span style="color:{change_color};">+{change:.2f}({change_percent:+.2f}%)</span>' if change > 0 else \
-                        f'<span style="color:{change_color};">{change:.2f}({change_percent:+.2f}%)</span>'
-        else:
-            # 如果没有获取到数据，所有数值显示N/A
-            price = "N/A"
-            change = 0
-            change_percent = 0
-            change_html = "<span style=\"color:#888;\">N/A</span>"
+    def format_volume(vol_str):
+        """将成交量字符串转为带 M/B 的简洁格式"""
+        if vol_str == "N/A" or not vol_str:
+            return "N/A"
+        try:
+            vol = int(float(vol_str))  # 兼容 "21453064" 或 "2.145e7"
+            if vol >= 1_000_000_000:
+                val = vol / 1_000_000_000
+                return f"{val:.2f}B".rstrip('0').rstrip('.')
+            elif vol >= 1_000_000:
+                val = vol / 1_000_000
+                return f"{val:.2f}M".rstrip('0').rstrip('.')
+            elif vol >= 1_000:
+                return f"{vol // 1_000}K"
+            else:
+                return str(vol)
+        except (ValueError, TypeError):
+            return "N/A"
+    def build_stock_card(company_info):
+        try:
+            if not company_info or not isinstance(company_info, dict):
+                company_name = "N/A"
+                symbol = "N/A"
+                price = "N/A"
+                change_html = '<span style="color:#888;">N/A</span>'
+                open_val = high_val = low_val = prev_close_val = volume_display = "N/A"
+            else:
+                company_name = company_info.get("company", "N/A")
+                symbol = company_info.get("symbol", "N/A")
+                price = company_info.get("price", "N/A")
+                
+                # 解析 change
+                change_str = company_info.get("change", "0")
+                try:
+                    change = float(change_str)
+                except (ValueError, TypeError):
+                    change = 0.0
 
-        html = f"""
-        <div style="width: 250px;height: 300px !important;border: 1px solid #e0e0e0; border-radius: 8px; padding: 16px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); font-family: 'Segoe UI', sans-serif;">
-            <div style="font-size: 16px; color: #555;font-weight: 500;">{company_info['name']}</div>
-            <div style="font-size: 12px; color: #888;">{company_info['symbol'] if (company_info2 and 'content' in company_info2 and len(company_info2['content']) > 0) else 'NYSE：N/A'}</div>
-            <div style="display: flex; align-items: center; gap: 10px; margin: 8px 0;">
-                <div style="font-size: 32px; font-weight: bold;">{price}</div>
-                <div style="font-size: 14px;">{change_html}</div>
-            </div>
-            <div style="margin-top: 12px; display: grid; grid-template-columns: auto 1fr; gap: 8px;">
-                <div style="font-size: 14px; color: #555;">Open</div><div style="font-size: 14px; font-weight: 500;text-align: center;">{company_info['open'] if (company_info2 and 'content' in company_info2 and len(company_info2['content']) > 0) else 'N/A'}</div>
-                <div style="font-size: 14px; color: #555;">High</div><div style="font-size: 14px; font-weight: 500;text-align: center;">{company_info['high'] if (company_info2 and 'content' in company_info2 and len(company_info2['content']) > 0) else 'N/A'}</div>
-                <div style="font-size: 14px; color: #555;">Low</div><div style="font-size: 14px; font-weight: 500;text-align: center;">{company_info['low'] if (company_info2 and 'content' in company_info2 and len(company_info2['content']) > 0) else 'N/A'}</div>
-                <div style="font-size: 14px; color: #555;">Prev Close</div><div style="font-size: 14px; font-weight: 500;text-align: center;">{company_info['prev_close'] if (company_info2 and 'content' in company_info2 and len(company_info2['content']) > 0) else 'N/A'}</div>
-                <div style="font-size: 14px; color: #555;">Vol</div><div style="font-size: 14px; font-weight: 500;text-align: center;">{company_info['volume'] if (company_info2 and 'content' in company_info2 and len(company_info2['content']) > 0) else 'N/A'}</div>
+                # 解析 change_percent
+                change_percent_str = company_info.get("change_percent", "0%")
+                try:
+                    change_percent = float(change_percent_str.rstrip('%'))
+                except (ValueError, TypeError):
+                    change_percent = 0.0
+
+                change_color = "green" if change >= 0 else "red"
+                sign = "+" if change >= 0 else ""
+                change_html = f'<span style="color:{change_color};">{sign}{change:.2f} ({change_percent:+.2f}%)</span>'
+
+                # 其他价格字段（可选：也可格式化为 2 位小数）
+                open_val = company_info.get("open", "N/A")
+                high_val = company_info.get("high", "N/A")
+                low_val = company_info.get("low", "N/A")
+                prev_close_val = company_info.get("previous close", "N/A")
+                raw_volume = company_info.get("volume", "N/A")
+                volume_display = format_volume(raw_volume)
+
+            html = f"""
+            <div style="width: 250px; height: 300px !important; border: 1px solid #e0e0e0; border-radius: 8px; padding: 16px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); font-family: 'Segoe UI', sans-serif;">
+                <div style="font-size: 16px; color: #555; font-weight: 500;">{company_name}</div>
+                <div style="font-size: 12px; color: #888;">{symbol}</div>
+                <div style="display: flex; align-items: center; gap: 10px; margin: 8px 0;">
+                    <div style="font-size: 32px; font-weight: bold;">{price}</div>
+                    <div style="font-size: 14px;">{change_html}</div>
+                </div>
+                <div style="margin-top: 12px; display: grid; grid-template-columns: auto 1fr; gap: 8px;">
+                    <div style="font-size: 14px; color: #555;">Open</div><div style="font-size: 14px; font-weight: 500; text-align: center;">{open_val}</div>
+                    <div style="font-size: 14px; color: #555;">High</div><div style="font-size: 14px; font-weight: 500; text-align: center;">{high_val}</div>
+                    <div style="font-size: 14px; color: #555;">Low</div><div style="font-size: 14px; font-weight: 500; text-align: center;">{low_val}</div>
+                    <div style="font-size: 14px; color: #555;">Prev Close</div><div style="font-size: 14px; font-weight: 500; text-align: center;">{prev_close_val}</div>
+                    <div style="font-size: 14px; color: #555;">Vol</div><div style="font-size: 14px; font-weight: 500; text-align: center;">{volume_display}</div>
+                </div>
             </div>
-        </div>
-        """
-        return html
+            """
+            return html
+
+        except Exception as e:
+            print(f"Error building stock card: {e}")
+            return '<div style="width:250px; padding:16px; color:red;">Error loading stock data</div>'
+    # def build_stock_card():
+    #     try:
+    #          # 检查是否获取到了股票数据，如果没有则显示N/A
+    #         if company_info:
+    #             price = company_info["price"]
+    #             change = int(company_info["change"])
+    #             change_percent = company_info["change_percent"]
+
+    #             # 格式化价格变动
+    #             change_color = "green" if change > 0 else "red"
+    #             change_html = f'<span style="color:{change_color};">+{change:.2f}({change_percent:+.2f}%)</span>' if change > 0 else \
+    #                         f'<span style="color:{change_color};">{change:.2f}({change_percent:+.2f}%)</span>'
+    #         else:
+    #             # 如果没有获取到数据，所有数值显示N/A
+    #             price = "N/A"
+    #             change = 0
+    #             change_percent = 0
+    #             change_html = "<span style=\"color:#888;\">N/A</span>"
+    #             html = f"""
+    #                 <div style="width: 250px;height: 300px !important;border: 1px solid #e0e0e0; border-radius: 8px; padding: 16px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); font-family: 'Segoe UI', sans-serif;">
+    #                     <div style="font-size: 16px; color: #555;font-weight: 500;">{company_info['name']}</div>
+    #                     <div style="font-size: 12px; color: #888;">{company_info['symbol'] if (company_info and 'symbol' in company_info) else 'NYSE：N/A'}</div>
+    #                     <div style="display: flex; align-items: center; gap: 10px; margin: 8px 0;">
+    #                         <div style="font-size: 32px; font-weight: bold;">{price}</div>
+    #                         <div style="font-size: 14px;">{change_html}</div>
+    #                     </div>
+    #                     <div style="margin-top: 12px; display: grid; grid-template-columns: auto 1fr; gap: 8px;">
+    #                         <div style="font-size: 14px; color: #555;">Open</div><div style="font-size: 14px; font-weight: 500;text-align: center;">{company_info['open'] if (company_info and 'open' in company_info) else 'N/A'}</div>
+    #                         <div style="font-size: 14px; color: #555;">High</div><div style="font-size: 14px; font-weight: 500;text-align: center;">{company_info['high'] if (company_info and 'high' in company_info) else 'N/A'}</div>
+    #                         <div style="font-size: 14px; color: #555;">Low</div><div style="font-size: 14px; font-weight: 500;text-align: center;">{company_info['low'] if (company_info and 'low' in company_info) else 'N/A'}</div>
+    #                         <div style="font-size: 14px; color: #555;">Prev Close</div><div style="font-size: 14px; font-weight: 500;text-align: center;">{company_info['prev_close'] if (company_info and 'prev_close' in company_info) else 'N/A'}</div>
+    #                         <div style="font-size: 14px; color: #555;">Vol</div><div style="font-size: 14px; font-weight: 500;text-align: center;">{company_info['volume'] if (company_info and 'volume' in company_info) else 'N/A'}</div>
+    #                     </div>
+    #                 </div>
+    #                 """
+    #             return html
+    #     except Exception as e:
+    #         print(f"Error building stock card: {e}")
+
+        
 
     # 构建中间卡片
-    def build_financial_metrics():
+    def build_financial_metrics(yearly_data):
         metrics_html = ""
         for item in financial_metrics:
             change_html = render_change(item["change"], item["color"])
@@ -1010,7 +1328,7 @@ def update_metrics_dashboard(company_name):
                 <svg width="16" height="16" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
                     <path d="M12 2L15.09 8.26L19 9.07L16 14L16 19L12 19L8 14L8 9.07L4.91 8.26L8 2L12 2Z" fill="#0066cc"/>
                 </svg>
-                <div style="font-size: 18px; font-weight: 600;">2025 Q3 Financial Metrics</div>
+                <div style="font-size: 18px; font-weight: 600;">{yearly_data} Financial Metrics</div>
             </div>
             {metrics_html}
         </div>
@@ -1018,56 +1336,59 @@ def update_metrics_dashboard(company_name):
         return html
 
     # 构建右侧表格
-    def build_income_table():
-        table_rows = ""
-        for i, row in enumerate(income_statement):
-            if i == 0:
-                row_style = "background-color: #f5f5f5; font-weight: 500;"
-            else:
-                row_style = "background-color: #f9f9f9;"
-            cells = ""
-            for j, cell in enumerate(row):
-                if j == 0:
-                    cells += f"<td style='padding: 8px; border: 1px solid #ddd; text-align: left; font-size: 14px;'>{cell}</td>"
-                else:
-                    # 添加增长箭头（模拟数据）
-                    growth = None
-                    if i == 1 and j == 1: growth = "+3.05%"
-                    elif i == 1 and j == 2: growth = "-6.00%"
-                    elif i == 2 and j == 1: growth = "+3.05%"
-                    elif i == 2 and j == 2: growth = "-6.00%"
-                    elif i == 3 and j == 1: growth = "+3.05%"
-                    elif i == 3 and j == 2: growth = "-6.00%"
-                    elif i == 4 and j == 1: growth = "+29.17%"
-                    elif i == 4 and j == 2: growth = "+29.17%"
-                    elif i == 5 and j == 1: growth = "-13.05%"
-                    elif i == 5 and j == 2: growth = "+29.17%"
-
-                    if growth:
-                        arrow = "▲" if growth.startswith("+") else "▼"
-                        color = "green" if growth.startswith("+") else "red"
-                        cells += f"<td style='padding: 8px; border: 1px solid #ddd; text-align: right; font-size: 14px; position: relative;'><div>{cell}</div><div style='position: absolute; bottom: -5px; right: 5px; font-size: 10px; color: {color};'>{arrow}{growth}</div></td>"
-                    else:
-                        cells += f"<td style='padding: 8px; border: 1px solid #ddd; text-align: right; font-size: 14px;'>{cell}</td>"
-            table_rows += f"<tr style='{row_style}'>{cells}</tr>"
+    # def build_income_table(income_statement):
+    #     table_rows = ""
+    #     for i, row in enumerate(income_statement):
+    #         if i == 0:
+    #             row_style = "background-color: #f5f5f5; font-weight: 500;"
+    #         else:
+    #             row_style = "background-color: #f9f9f9;"
+    #         cells = ""
+    #         for j, cell in enumerate(row):
+    #             if j == 0:
+    #                 cells += f"<td style='padding: 8px; border: 1px solid #ddd; text-align: center; font-size: 14px;'>{cell}</td>"
+    #             else:
+    #                 # 添加增长箭头（模拟数据）
+    #                 growth = None
+    #                 if i == 1 and j == 1: growth = "+3.05%"
+    #                 elif i == 1 and j == 2: growth = "-6.00%"
+    #                 elif i == 2 and j == 1: growth = "+3.05%"
+    #                 elif i == 2 and j == 2: growth = "-6.00%"
+    #                 elif i == 3 and j == 1: growth = "+3.05%"
+    #                 elif i == 3 and j == 2: growth = "-6.00%"
+    #                 elif i == 4 and j == 1: growth = "+29.17%"
+    #                 elif i == 4 and j == 2: growth = "+29.17%"
+    #                 elif i == 5 and j == 1: growth = "-13.05%"
+    #                 elif i == 5 and j == 2: growth = "+29.17%"
+
+    #                 if growth:
+    #                     arrow = "▲" if growth.startswith("+") else "▼"
+    #                     color = "green" if growth.startswith("+") else "red"
+    #                     cells += f"""<td style='padding: 8px; border: 1px solid #ddd; text-align: center; font-size: 14px; position: relative;'>
+    #                     <div>{cell}</div>
+    #                     <div style='position: absolute; bottom: -5px; right: 5px; font-size: 10px; color: {color};'>{arrow}{growth}</div>
+    #                     </td>"""
+    #                 else:
+    #                     cells += f"<td style='padding: 8px; border: 1px solid #ddd; text-align: center; font-size: 14px;'>{cell}</td>"
+    #         table_rows += f"<tr style='{row_style}'>{cells}</tr>"
         
-        html = f"""
-        <div style="width: 600px;height: 300px !important;border: 1px solid #e0e0e0; border-radius: 8px; padding: 16px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); font-family: 'Segoe UI', sans-serif;">
-            <div style="display: flex; align-items: center; gap: 8px; margin-bottom: 16px;">
-                <svg width="16" height="16" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
-                    <path d="M12 2L15.09 8.26L19 9.07L16 14L16 19L12 19L8 14L8 9.07L4.91 8.26L8 2L12 2Z" fill="#0066cc"/>
-                </svg>
-                <div style="font-size: 18px; font-weight: 600;">Income Statement and Cash Flow</div>
-            </div>
-            <table style="width: 100%; border-collapse: collapse; font-size: 14px;">
-                {table_rows}
-            </table>
-        </div>
-        """
-        return html
+    #     html = f"""
+    #     <div style="width: 600px;height: 300px !important;border: 1px solid #e0e0e0; border-radius: 8px; padding: 16px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); font-family: 'Segoe UI', sans-serif;">
+    #         <div style="display: flex; align-items: center; gap: 8px; margin-bottom: 16px;">
+    #             <svg width="16" height="16" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+    #                 <path d="M12 2L15.09 8.26L19 9.07L16 14L16 19L12 19L8 14L8 9.07L4.91 8.26L8 2L12 2Z" fill="#0066cc"/>
+    #             </svg>
+    #             <div style="font-size: 18px; font-weight: 600;">Income Statement and Cash Flow</div>
+    #         </div>
+    #         <table style="width: 100%; border-collapse: collapse; font-size: 14px;">
+    #             {table_rows}
+    #         </table>
+    #     </div>
+    #     """
+    #     return html
 
     # 返回三个HTML组件的内容
-    return build_stock_card(), build_financial_metrics(), build_income_table()
+    return build_stock_card(company_info), build_financial_metrics(yearly_data), build_income_table(table_data)
             # gr.Column(scale=1, min_width=250)
             # gr.HTML(f'''
             #     <div class="metric-card-item" style="{card_custom_style}width:300px;">
@@ -1521,16 +1842,16 @@ def main():
             def update_metrics_dashboard_wrapper(company_name):
                 if company_name:
                     # 显示loading状态
-                    loading_html = '''
+                    loading_html = f'''
                     <div style="display: flex; justify-content: center; align-items: center; height: 300px;">
                         <div style="text-align: center;">
                             <div class="loading-spinner" style="width: 40px; height: 40px; border: 4px solid #f3f3f3; border-top: 4px solid #3498db; border-radius: 50%; animation: spin 1s linear infinite; margin: 0 auto;"></div>
                             <p style="margin-top: 20px; color: #666;">Loading financial data for {company_name}...</p>
                             <style>
-                                @keyframes spin {
-                                    0% { transform: rotate(0deg); }
-                                    100% { transform: rotate(360deg); }
-                                }
+                                @keyframes spin {{
+                                    0% {{ transform: rotate(0deg); }}
+                                    100% {{ transform: rotate(360deg); }}
+                                }}
                             </style>
                         </div>
                     </div>

requirements.txt

@@ -2,5 +2,12 @@ gradio>=5.49.1
 huggingface_hub>=0.20.0
 python-dotenv>=1.0.0
 SQLAlchemy>=2.0.44
+plotly>=5.24.1
 pandas>=2.2.2
-numpy>=1.26.4
+numpy>=1.26.4
+aiohttp>=3.8.1
+pdfplumber>=0.7.0
+beautifulsoup4>=4.11.0
+requests>=2.32.0
+urllib3>=2.5.0
+httpx>=0.23.0

service/chat_service.py

@@ -126,6 +126,18 @@ def search_news(user_input: str):
         return choices
     except json.JSONDecodeError:
         return []
+
+def get_stock_price_from_bailian(user_input: str):
+    if not user_input.strip():
+        return {}
+    # 获取非流式响应
+    response = chat_with_bailian_non_streaming(user_input, "d2b919f9b32e4fa28a75234cbb78a787")
+    print(f"查询结果：{response}{user_input}")
+    try:
+        parsed_data = json.loads(response)
+        return parsed_data if isinstance(parsed_data, dict) else {}
+    except json.JSONDecodeError:
+        return {}
 # def search_company(user_input: str):
 #     if not user_input.strip():
 #         return []  # 返回空列表，而不是 yield

service/report_mcp.py

@@ -0,0 +1,475 @@
+import gradio as gr
+import requests
+import json
+import os
+
+MCP_SPACE = "JC321/EasyReportsMCPServer"
+MCP_URL = "https://jc321-easyreportsmcpserver.hf.space"
+
+# 设置请求头
+HEADERS = {
+    "Content-Type": "application/json",
+    "User-Agent": "SEC-Query-Assistant/1.0 ([email protected])"
+}
+
+# 格式化数值显示
+def format_value(value, value_type="money"):
+    """
+    格式化数值：0显示为N/A，其他显示为带单位的格式
+    value_type: "money" (金额), "eps" (每股收益), "number" (普通数字)
+    """
+    if value is None or value == 0:
+        return "N/A"
+    
+    if value_type == "money":
+        return f"${value:.2f}B"
+    elif value_type == "eps":
+        return f"${value:.2f}"
+    else:  # number
+        return f"{value:.2f}"
+
+def normalize_cik(cik):
+    """
+    格式化 CIK 为标准的 10 位格式
+    """
+    if not cik:
+        return None
+    # 转换为字符串并移除非数字字符
+    cik_str = str(cik).replace('-', '').replace(' ', '')
+    # 仅保留数字
+    cik_str = ''.join(c for c in cik_str if c.isdigit())
+    # 填充前导 0 至 10 位
+    return cik_str.zfill(10) if cik_str else None
+
+def parse_mcp_response(response_data):
+    """
+    解析 MCP 协议响应数据
+    支持格式:
+    1. {"result": {"content": [{"type": "text", "text": "{...}"}]}}
+    2. {"content": [{"type": "text", "text": "{...}"}]}
+    3. 直接的 JSON 数据
+    """
+    if not isinstance(response_data, dict):
+        return response_data
+    
+    # 格式 1: {"result": {"content": [...]}}
+    if "result" in response_data and "content" in response_data["result"]:
+        content = response_data["result"]["content"]
+        if content and len(content) > 0:
+            text_content = content[0].get("text", "{}")
+            # 直接解析 JSON（MCP Server 已移除 emoji 前缀）
+            try:
+                return json.loads(text_content)
+            except json.JSONDecodeError:
+                return text_content
+        return {}
+    
+    # 格式 2: {"content": [...]}
+    elif "content" in response_data:
+        content = response_data.get("content", [])
+        if content and len(content) > 0:
+            text_content = content[0].get("text", "{}")
+            # 直接解析 JSON
+            try:
+                return json.loads(text_content)
+            except json.JSONDecodeError:
+                return text_content
+        return {}
+    
+    # 格式 3: 直接返回
+    return response_data
+
+# MCP 工具定义
+def create_mcp_tools():
+    """创建 MCP 工具列表"""
+    return [
+        {
+            "name": "query_financial_data",
+            "description": "Query SEC financial data for US listed companies",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "company_name": {
+                        "type": "string",
+                        "description": "Company name or stock symbol (e.g., Apple, NVIDIA, AAPL)"
+                    },
+                    "query_type": {
+                        "type": "string",
+                        "enum": ["Latest Financial Data", "3-Year Trends", "5-Year Trends"],
+                        "description": "Type of financial query"
+                    }
+                },
+                "required": ["company_name", "query_type"]
+            }
+        }
+    ]
+
+# 工具执行函数
+def execute_tool(tool_name, **kwargs):
+    """执行 MCP 工具"""
+    if tool_name == "query_financial_data":
+        return query_financial_data(kwargs.get("company_name"), kwargs.get("query_type"))
+    return f"Unknown tool: {tool_name}"
+# 创建超链接
+def create_source_link(source_form, source_url=None):
+    """为Source Form创建超链接，使用MCP后端返回的URL"""
+    if not source_form or source_form == 'N/A':
+        return source_form
+    
+    # 如果后端提供了URL，使用后端的URL
+    if source_url and source_url != 'N/A':
+        return f"[{source_form}]({source_url})"
+    
+    # 如果没有URL，只显示文本
+    return source_form
+
+def query_financial_data(company_name, query_type):
+    """查询财务数据的主函数"""
+    
+    if not company_name:
+        return "Please enter a company name or stock symbol"
+    
+    # 翻译英文查询类型为中文（用于后端处理）
+    query_type_mapping = {
+        "Latest": "最新财务数据",
+        "3-Year": "3年趋势",
+        "5-Year": "5年趋势",
+        "Filings": "公司报表列表"
+    }
+    internal_query_type = query_type_mapping.get(query_type, query_type)
+    
+    try:
+        # 使用 MCP 协议调用工具
+        # 先搜索公司（使用 advanced_search_company）
+        search_resp = requests.post(
+            f"{MCP_URL}/message",
+            json={
+                "method": "tools/call",
+                "params": {
+                    "name": "advanced_search_company",
+                    "arguments": {"company_input": company_name}
+                }
+            },
+            headers=HEADERS,
+            timeout=30
+        )
+
+        print(f"搜索公司：{company_name}，search_resp.status_code: {search_resp.status_code}\nSearch Response: {search_resp.text}")
+        
+        if search_resp.status_code != 200:
+            print(f"❌ Server Error: HTTP {search_resp.status_code}\n\nResponse: {search_resp.text[:500]}")
+            return []
+        
+        try:
+            result = search_resp.json()
+            # 使用统一的 MCP 响应解析函数
+            company = parse_mcp_response(result)
+        except (ValueError, KeyError, json.JSONDecodeError) as e:
+            return f"❌ JSON Parse Error: {str(e)}\n\nResponse: {search_resp.text[:500]}"
+        
+        if isinstance(company, dict) and company.get("error"):
+            return f"❌ Error: {company['error']}"
+        
+        # advanced_search 返回的字段: cik, name, ticker
+        # 注意: 不是 tickers 和 sic_description
+        company_name = company.get('name', 'Unknown')
+        ticker = company.get('ticker', 'N/A')
+        
+        result = f"# {company_name}\n\n"
+        result += f"**Stock Symbol**: {ticker}\n"
+        # sic_description 需要后续通过 get_company_info 获取，这里暂时不显示
+        result += "\n---\n\n"
+        
+        # 获取并格式化 CIK 为 10 位标准格式
+        cik = normalize_cik(company.get('cik'))
+        if not cik:
+            return result + f"❌ Error: Invalid CIK from company search\n\nDebug: company data = {json.dumps(company, indent=2)}"
+        
+        # 根据查询类型获取数据
+        if internal_query_type == "最新财务数据":
+            data_resp = requests.post(
+                f"{MCP_URL}/message",
+                json={
+                    "method": "tools/call",
+                    "params": {
+                        "name": "get_latest_financial_data",
+                        "arguments": {"cik": cik}
+                    }
+                },
+                headers=HEADERS,
+                timeout=30
+            )
+            
+            if data_resp.status_code != 200:
+                return result + f"❌ Server Error: HTTP {data_resp.status_code}\n\n{data_resp.text[:500]}"
+            
+            try:
+                data_result = data_resp.json()
+                # 使用统一的 MCP 响应解析函数
+                data = parse_mcp_response(data_result)
+            except (ValueError, KeyError, json.JSONDecodeError) as e:
+                return result + f"❌ JSON Parse Error: {str(e)}\n\n{data_resp.text[:500]}"
+            
+            if isinstance(data, dict) and data.get("error"):
+                return result + f"❌ {data['error']}"
+            
+            cik = data.get('cik')
+            result += f"## Fiscal Year {data.get('period', 'N/A')}\n\n"
+            
+            total_revenue = data.get('total_revenue', 0) / 1e9 if data.get('total_revenue') else 0
+            net_income = data.get('net_income', 0) / 1e9 if data.get('net_income') else 0
+            eps = data.get('earnings_per_share', 0) if data.get('earnings_per_share') else 0
+            opex = data.get('operating_expenses', 0) / 1e9 if data.get('operating_expenses') else 0
+            ocf = data.get('operating_cash_flow', 0) / 1e9 if data.get('operating_cash_flow') else 0
+            
+            result += f"- **Total Revenue**: {format_value(total_revenue)}\n"
+            result += f"- **Net Income**: {format_value(net_income)}\n"
+            result += f"- **Earnings Per Share**: {format_value(eps, 'eps')}\n"
+            result += f"- **Operating Expenses**: {format_value(opex)}\n"
+            result += f"- **Operating Cash Flow**: {format_value(ocf)}\n"
+            # 使用后端返回的 source_url
+            source_form = data.get('source_form', 'N/A')
+            source_url = data.get('source_url', None)  # 从后端获取URL
+            result += f"- **Source Form**: {create_source_link(source_form, source_url)}\n"
+        
+        elif internal_query_type == "3年趋势":
+            metrics_resp = requests.post(
+                f"{MCP_URL}/message",
+                json={
+                    "method": "tools/call",
+                    "params": {
+                        "name": "extract_financial_metrics",
+                        "arguments": {"cik": cik, "years": 3}
+                    }
+                },
+                headers=HEADERS,
+                timeout=60
+            )
+            
+            # 调试：显示 HTTP 响应状态
+            result += f"\n**Debug Info (3-Year)**:\n- HTTP Status: {metrics_resp.status_code}\n"
+            
+            if metrics_resp.status_code != 200:
+                return result + f"❌ Server Error: HTTP {metrics_resp.status_code}\n\n{metrics_resp.text[:500]}"
+            
+            try:
+                metrics_result = metrics_resp.json()
+                # 调试：显示原始 JSON 响应
+                result += f"- Raw Response Length: {len(metrics_resp.text)} chars\n"
+                result += f"- Response Preview: {metrics_resp.text[:200]}...\n\n"
+                
+                # 使用统一的 MCP 响应解析函数
+                metrics = parse_mcp_response(metrics_result)
+                
+                # 调试：显示解析后的数据类型和内容
+                result += f"- Parsed Type: {type(metrics).__name__}\n"
+                if isinstance(metrics, dict):
+                    result += f"- Parsed Keys: {list(metrics.keys())}\n"
+                    result += f"- Periods: {metrics.get('periods', 'N/A')}\n"
+                    result += f"- Data Length: {len(metrics.get('data', []))}\n\n"
+            except (ValueError, KeyError, json.JSONDecodeError) as e:
+                return result + f"❌ JSON Parse Error: {str(e)}\n\nResponse: {metrics_resp.text[:500]}"
+            
+            if isinstance(metrics, dict) and metrics.get("error"):
+                return result + f"❌ {metrics['error']}"
+            
+            # 调试：显示原始响应
+            if not isinstance(metrics, dict):
+                return result + f"❌ Invalid response format\n\nDebug: {str(metrics)[:500]}"
+            
+            result += f"## 3-Year Financial Trends ({metrics.get('periods', 0)} periods)\n\n"
+            
+            # 显示所有数据（包括年度和季度）
+            all_data = metrics.get('data', [])  # MCP Server 返回的字段是 'data'
+            
+            # 调试：检查是否有数据
+            if not all_data:
+                return result + f"❌ No data returned from MCP Server\n\nDebug: metrics keys = {list(metrics.keys())}\n\nFull response: {json.dumps(metrics, indent=2, ensure_ascii=False)[:1000]}"
+            
+            # 去重：根据period和source_form去重
+            seen = set()
+            unique_data = []
+            for m in all_data:
+                key = (m.get('period', 'N/A'), m.get('source_form', 'N/A'))
+                if key not in seen:
+                    seen.add(key)
+                    unique_data.append(m)
+            
+            # 按期间降序排序，确保显示最近的3年数据
+            # 使用更智能的排序：先按年份，再按是否是季度
+            # 正确顺序：FY2024 → 2024Q3 → 2024Q2 → 2024Q1 → FY2023
+            def sort_key(x):
+                period = x.get('period', '0000')
+                # 提取年份（前4位）
+                year = period[:4] if len(period) >= 4 else '0000'
+                # 如果有Q，提取季度号
+                if 'Q' in period:
+                    quarter = period[period.index('Q')+1] if period.index('Q')+1 < len(period) else '0'
+                    return (year, 1, 4 - int(quarter))  # Q在FY后面：Q3, Q2, Q1 (4-3=1, 4-2=2, 4-1=3)
+                else:
+                    return (year, 0, 0)  # FY 排在同年的所有Q之前
+            
+            unique_data = sorted(unique_data, key=sort_key, reverse=True)
+            
+            result += "| Period | Revenue (B) | Net Income (B) | EPS | Operating Expenses (B) | Operating Cash Flow (B) | Source Form |\n"
+            result += "|--------|-------------|----------------|-----|------------------------|-------------------------|-------------|\n"
+            
+            for m in unique_data:
+                period = m.get('period', 'N/A')
+                rev = (m.get('total_revenue') or 0) / 1e9
+                inc = (m.get('net_income') or 0) / 1e9
+                eps_val = m.get('earnings_per_share') or 0
+                opex = (m.get('operating_expenses') or 0) / 1e9
+                ocf = (m.get('operating_cash_flow') or 0) / 1e9
+                source_form = m.get('source_form', 'N/A')
+                source_url = m.get('source_url', None)  # 从后端获取URL
+                
+                # 区分年度和季度，修复双重FY前缀问题
+                if 'Q' in period:
+                    # 季度数据，不添加前缀
+                    display_period = period
+                else:
+                    # 年度数据，只在没有FY的情况下添加
+                    display_period = period if period.startswith('FY') else f"FY{period}"
+                
+                source_link = create_source_link(source_form, source_url)
+                
+                result += f"| {display_period} | {format_value(rev)} | {format_value(inc)} | {format_value(eps_val, 'eps')} | {format_value(opex)} | {format_value(ocf)} | {source_link} |\n"
+        
+        elif internal_query_type == "5年趋势":
+            metrics_resp = requests.post(
+                f"{MCP_URL}/message",
+                json={
+                    "method": "tools/call",
+                    "params": {
+                        "name": "extract_financial_metrics",
+                        "arguments": {"cik": cik, "years": 5}
+                    }
+                },
+                headers=HEADERS,
+                timeout=60
+            )
+            
+            # 调试：显示 HTTP 响应状态
+            result += f"\n**Debug Info (5-Year)**:\n- HTTP Status: {metrics_resp.status_code}\n"
+            
+            if metrics_resp.status_code != 200:
+                return result + f"❌ Server Error: HTTP {metrics_resp.status_code}\n\n{metrics_resp.text[:500]}"
+            
+            try:
+                metrics_result = metrics_resp.json()
+                # 调试：显示原始 JSON 响应
+                result += f"- Raw Response Length: {len(metrics_resp.text)} chars\n"
+                result += f"- Response Preview: {metrics_resp.text[:200]}...\n\n"
+                
+                # 使用统一的 MCP 响应解析函数
+                metrics = parse_mcp_response(metrics_result)
+                
+                # 调试：显示解析后的数据类型和内容
+                result += f"- Parsed Type: {type(metrics).__name__}\n"
+                if isinstance(metrics, dict):
+                    result += f"- Parsed Keys: {list(metrics.keys())}\n"
+                    result += f"- Periods: {metrics.get('periods', 'N/A')}\n"
+                    result += f"- Data Length: {len(metrics.get('data', []))}\n\n"
+            except (ValueError, KeyError, json.JSONDecodeError) as e:
+                return result + f"❌ JSON Parse Error: {str(e)}\n\nResponse: {metrics_resp.text[:500]}"
+            
+            if isinstance(metrics, dict) and metrics.get("error"):
+                return result + f"❌ {metrics['error']}"
+            
+            # 调试：显示原始响应
+            if not isinstance(metrics, dict):
+                return result + f"❌ Invalid response format\n\nDebug: {str(metrics)[:500]}"
+            
+            # 显示所有数据（包括年度和季度）
+            all_data = metrics.get('data', [])  # MCP Server 返回的字段是 'data'
+            
+            # 调试：检查是否有数据
+            if not all_data:
+                return result + f"❌ No data returned from MCP Server\n\nDebug: metrics keys = {list(metrics.keys())}\n\nFull response: {json.dumps(metrics, indent=2, ensure_ascii=False)[:1000]}"
+            
+            # 去重：根据period和source_form去重
+            seen = set()
+            unique_data = []
+            for m in all_data:
+                key = (m.get('period', 'N/A'), m.get('source_form', 'N/A'))
+                if key not in seen:
+                    seen.add(key)
+                    unique_data.append(m)
+            
+            # 按期间降序排序，确保显示最近的5年数据
+            # 使用更智能的排序：先按年份，再按是否是季度
+            # 正确顺序：FY2024 → 2024Q3 → 2024Q2 → 2024Q1 → FY2023
+            def sort_key(x):
+                period = x.get('period', '0000')
+                # 提取年份（前4位）
+                year = period[:4] if len(period) >= 4 else '0000'
+                # 如果有Q，提取季度号
+                if 'Q' in period:
+                    quarter = period[period.index('Q')+1] if period.index('Q')+1 < len(period) else '0'
+                    return (year, 1, 4 - int(quarter))  # Q在FY后面：Q3, Q2, Q1 (4-3=1, 4-2=2, 4-1=3)
+                else:
+                    return (year, 0, 0)  # FY 排在同年的所有Q之前
+            
+            unique_data = sorted(unique_data, key=sort_key, reverse=True)
+            print(f'5年数据：:{unique_data}')
+            result = unique_data    
+        
+        elif internal_query_type == "公司报表列表":
+            # 查询公司所有报表
+            filings_resp = requests.post(
+                f"{MCP_URL}/message",
+                json={
+                    "method": "tools/call",
+                    "params": {
+                        "name": "get_company_filings",
+                        "arguments": {"cik": cik, "limit": 50}
+                    }
+                },
+                headers=HEADERS,
+                timeout=60
+            )
+            
+            if filings_resp.status_code != 200:
+                return result + f"❌ Server Error: HTTP {filings_resp.status_code}\n\n{filings_resp.text[:500]}"
+            
+            try:
+                filings_result = filings_resp.json()
+                # 使用统一的 MCP 响应解析函数
+                filings_data = parse_mcp_response(filings_result)
+            except (ValueError, KeyError, json.JSONDecodeError) as e:
+                return result + f"❌ JSON Parse Error: {str(e)}\n\n{filings_resp.text[:500]}"
+            
+            if isinstance(filings_data, dict) and filings_data.get("error"):
+                return result + f"❌ {filings_data['error']}"
+            
+            filings = filings_data.get('filings', []) if isinstance(filings_data, dict) else filings_data
+            
+            result += f"## Company Filings ({len(filings)} records)\n\n"
+            result += "| Form Type | Filing Date | Accession Number | Primary Document |\n"
+            result += "|-----------|-------------|------------------|------------------|\n"
+            
+            for filing in filings:
+                form_type = filing.get('form_type', 'N/A')
+                filing_date = filing.get('filing_date', 'N/A')
+                accession_num = filing.get('accession_number', 'N/A')
+                primary_doc = filing.get('primary_document', 'N/A')
+                filing_url = filing.get('filing_url', None)  # 从后端获取URL
+                
+                # 使用后端返回的URL创建链接
+                if filing_url and filing_url != 'N/A':
+                    form_link = f"[{form_type}]({filing_url})"
+                    primary_doc_link = f"[{primary_doc}]({filing_url})"
+                else:
+                    form_link = form_type
+                    primary_doc_link = primary_doc
+                
+                result += f"| {form_link} | {filing_date} | {accession_num} | {primary_doc_link} |\n"
+        
+        return result
+        
+    except requests.exceptions.RequestException as e:
+        return f"❌ Network Error: {str(e)}\n\nMCP Server: {MCP_URL}"
+    except Exception as e:
+        import traceback
+        return f"❌ Unexpected Error: {str(e)}\n\nTraceback:\n{traceback.format_exc()}"

service/report_tools.py

@@ -0,0 +1,674 @@
+import json
+
+def extract_last_three_with_fallback(data_list):
+    # 定义年份范围（当前最新是 FY2025，所以前三年是 2025, 2024, 2023）
+    years = [2025, 2024, 2023]
+    
+    # 构建 period 映射：按优先级
+    priority_levels = [
+        ("FY", [f"FY{y}" for y in years]),
+        ("Q4", [f"{y}Q4" for y in years]),
+        ("Q3", [f"{y}Q3" for y in years]),
+        ("Q2", [f"{y}Q2" for y in years]),
+        ("Q1", [f"{y}Q1" for y in years]),
+    ]
+    
+    # 转为字典便于查找
+    data_map = {item["period"]: item for item in data_list if "period" in item}
+    
+    # 按优先级尝试
+    for level_name, periods in priority_levels:
+        records = []
+        valid = True
+        
+        for period in periods:
+            item = data_map.get(period)
+            if item is None or item.get("total_revenue") is None:
+                valid = False
+                break
+            # 提取关键字段
+            clean_item = {
+                "period": period,
+                "fiscal_year": int(period[:4]) if level_name != "FY" else int(period[2:]),
+                "level": level_name,
+                "total_revenue": item["total_revenue"],
+                "net_income": item.get("net_income"),
+                "earnings_per_share": item.get("earnings_per_share"),
+                "operating_expenses": item.get("operating_expenses"),
+                "operating_cash_flow": item.get("operating_cash_flow"),
+                "source_url": item.get("source_url")
+            }
+            records.append(clean_item)
+        
+        if valid:
+            # 找到完整三年数据，返回
+            return records
+    
+    # 如果所有层级都不完整，可选择返回最高优先级中有效的部分（或抛异常）
+    # 这里我们返回最高优先级中非空的记录（保守策略）
+    for level_name, periods in priority_levels:
+        records = []
+        for period in periods:
+            item = data_map.get(period)
+            if item and item.get("total_revenue") is not None:
+                clean_item = {
+                    "period": period,
+                    "fiscal_year": int(period[:4]) if level_name != "FY" else int(period[2:]),
+                    "level": level_name,
+                    "total_revenue": item["total_revenue"],
+                    "net_income": item.get("net_income"),
+                    "earnings_per_share": item.get("earnings_per_share"),
+                    "operating_expenses": item.get("operating_expenses"),
+                    "operating_cash_flow": item.get("operating_cash_flow"),
+                    "source_url": item.get("source_url")
+                }
+                records.append(clean_item)
+        if records:
+            return records  # 返回第一个有数据的层级（即使不全）
+    
+    return []  # 完全无数据
+
+
+def format_number(value):
+    """将大数字格式化为 $XM 或 $XB"""
+    if value >= 1_000_000_000:
+        return f"${value / 1_000_000_000:.2f}B".replace(".00B", "B").replace(".0B", "B")
+    elif value >= 1_000_000:
+        return f"${value / 1_000_000:.1f}M".replace(".0M", "M")
+    else:
+        return f"${value:,.0f}"
+
+def format_eps(value):
+    """EPS 保留两位小数"""
+    return f"${value:.2f}"
+
+def safe_int(val):
+    """安全转换为 int，支持字符串或 None"""
+    if val is None:
+        return 0
+    try:
+        return int(float(val))  # 兼容字符串或 float
+    except (ValueError, TypeError):
+        return 0
+
+def calculate_change(current, previous):
+    """计算同比变化百分比，返回如 '+12.4%' 或 '-3.2%'"""
+    if previous == 0:
+        return "+0.0%" if current >= 0 else "-0.0%"
+    change = (current - previous) / abs(previous) * 100
+    sign = "+" if change >= 0 else "-"
+    return f"{sign}{abs(change):.1f}%"
+
+def build_financial_metrics_three_year_data(three_year_data):
+    # 确保按 fiscal_year 降序排列（最新在前）
+    sorted_data = sorted(three_year_data, key=lambda x: x["fiscal_year"], reverse=True)
+    if len(sorted_data) < 2:
+        raise ValueError("至少需要两年数据来计算同比变化")
+
+    latest = sorted_data[0]
+    previous = sorted_data[1]
+
+    # 提取并转为 int
+    rev_curr = safe_int(latest.get("total_revenue"))
+    rev_prev = safe_int(previous.get("total_revenue"))
+
+    net_curr = safe_int(latest.get("net_income"))
+    net_prev = safe_int(previous.get("net_income"))
+
+    eps_curr = float(latest.get("earnings_per_share", 0) or 0)
+    eps_prev = float(previous.get("earnings_per_share", 0) or 0)
+
+    opex_curr = safe_int(latest.get("operating_expenses"))
+    opex_prev = safe_int(previous.get("operating_expenses"))
+
+    cash_curr = safe_int(latest.get("operating_cash_flow"))
+    cash_prev = safe_int(previous.get("operating_cash_flow"))
+
+    metrics = [
+        {
+            "label": "Total Revenue",
+            "value": format_number(rev_curr),
+            "change": calculate_change(rev_curr, rev_prev),
+            "color": "green" if rev_curr >= rev_prev else "red"
+        },
+        {
+            "label": "Net Income",
+            "value": format_number(net_curr),
+            "change": calculate_change(net_curr, net_prev),
+            "color": "green" if net_curr >= net_prev else "red"
+        },
+        {
+            "label": "Earnings Per Share",
+            "value": format_eps(eps_curr),
+            "change": calculate_change(eps_curr, eps_prev),
+            "color": "green" if eps_curr >= eps_prev else "red"
+        },
+        {
+            "label": "Operating Expenses",
+            "value": format_number(opex_curr),
+            "change": calculate_change(opex_curr, opex_prev),
+            "color": "green" if opex_curr >= opex_prev else "red"
+        },
+        {
+            "label": "Cash Flow",
+            "value": format_number(cash_curr),
+            "change": calculate_change(cash_curr, cash_prev),
+            "color": "green" if cash_curr >= cash_prev else "red"
+        }
+    ]
+
+    return metrics
+# 假设你的原始数据变量名为 raw_data（即你提供的大列表）
+# raw_data = [ {...}, ... ]
+
+# 执行
+# result = extract_last_three_with_fallback(raw_data)
+
+# # 输出 JSON
+# json_output = json.dumps(result, indent=2)
+# print(json_output)
+
+# ==========
+
+from collections import defaultdict
+import re
+
+def parse_period(period):
+    """解析 period 字符串，返回 (year, type, quarter)"""
+    if period.startswith('FY'):
+        year = int(period[2:])
+        return year, 'FY', None
+    elif re.match(r'Q[1-4]-\d{4}', period):
+        q, year = period.split('-')
+        return int(year), 'Q', int(q[1])
+    else:
+        raise ValueError(f"Unknown period format: {period}")
+
+def get_best_value_for_year(year_data, key):
+    """
+    year_data: dict like {'FY': value, 'Q1': val, 'Q2': val, ...}
+    返回该财年该指标的最佳可用值（优先 FY，其次 Q4->Q3->Q2->Q1）
+    """
+    if year_data.get('FY') is not None:
+        return year_data['FY']
+    # 否则从 Q4 到 Q1 找第一个非 None
+    for q in ['Q4', 'Q3', 'Q2', 'Q1']:
+        if year_data.get(q) is not None:
+            return year_data[q]
+    return None
+# def get_yearly_data(data_json):
+#     metrics_list = data_json['metrics']
+    
+#     # 按年份组织数据：year -> { 'FY': {...}, 'Q1': {...}, ... }
+#     yearly_data = "N/A"
+    
+#     for metric in metrics_list:
+#         period = metric['period']
+#         year, ptype, quarter = parse_period(period)
+#         if ptype == 'FY':
+#             yearly_data = f"{year} {ptype}"
+#         else:
+#             yearly_data = f"{year} {ptype} Q{quarter}"
+#     return yearly_data
+import re
+
+def parse_period_year_data(period):
+    """
+    支持以下格式：
+    - FY2025
+    - Q1-2025
+    - 2025Q1  （新增支持）
+    """
+    if not isinstance(period, str):
+        return None, None, None
+
+    # 格式 1: FY2025
+    if period.startswith('FY'):
+        try:
+            year = int(period[2:])
+            return year, 'FY', None
+        except ValueError:
+            pass
+
+    # 格式 2: Q1-2025
+    match = re.match(r'Q([1-4])-(\d{4})', period)
+    if match:
+        quarter = int(match.group(1))
+        year = int(match.group(2))
+        return year, 'Q', quarter
+
+    # 格式 3: 2025Q1 （新增）
+    match = re.match(r'(\d{4})Q([1-4])', period)
+    if match:
+        year = int(match.group(1))
+        quarter = int(match.group(2))
+        return year, 'Q', quarter
+
+    # 无法解析
+    return None, None, None
+def get_yearly_data(data_json):
+    metrics_list = data_json.get('metrics', [])
+    latest_desc = "N/A"
+    
+    for metric in metrics_list:
+        period = metric.get('period')
+        if not period:
+            continue
+        year, ptype, quarter = parse_period_year_data(period)
+        if year is None:
+            continue  # 跳过无法解析的
+        
+        if ptype == 'FY':
+            desc = f"{year} FY"
+        else:
+            desc = f"{year} Q{quarter}"
+        
+        # 简单认为列表顺序是时间顺序，最后一条最新
+        latest_desc = desc
+    
+    return latest_desc
+def parse_period_yoy(period):
+    """解析 period 为 (year, type, quarter)"""
+    if period.startswith('FY'):
+        year = int(period[2:])
+        return year, 'FY', None
+    elif re.match(r'Q[1-4]-\d{4}', period):
+        q_part, year_str = period.split('-')
+        return int(year_str), 'Q', int(q_part[1])
+    else:
+        # 忽略无法解析的 period
+        return None, None, None
+
+def get_best_value_for_year_yoy(values_dict, key):
+    """
+    从年度数据中获取指定指标的最佳值（优先 FY，其次 Q4 → Q1）
+    values_dict: {'FY': {...}, 'Q1': {...}, ...}
+    """
+    order = ['FY', 'Q4', 'Q3', 'Q2', 'Q1']
+    for q in order:
+        metric = values_dict.get(q)
+        if metric is not None and isinstance(metric, dict):
+            val = metric.get(key)
+            if val is not None:
+                return val
+    return None
+import json
+def calculate_yoy_comparison(data_json):
+    metrics_list = data_json.get('metrics', [])
+    if not metrics_list:
+        return []
+    if not isinstance(metrics_list, list):
+        return []
+    if not isinstance(metrics_list[0], dict):
+        return []
+    # 安全处理：确保每个 metric 是字典（防止双重 JSON 编码）
+    cleaned_metrics = []
+    for i, metric in enumerate(metrics_list):
+        if isinstance(metric, str):
+            try:
+                metric = json.loads(metric)
+                # metric = metric
+            except Exception as e:
+                raise ValueError(f"Failed to parse metrics[{i}] as JSON string: {metric}") from e
+        if not isinstance(metric, dict):
+            raise TypeError(f"metrics[{i}] is not a dictionary or valid JSON string. Type: {type(metric)}")
+        cleaned_metrics.append(metric)
+
+    # 按年份组织数据：year -> { 'FY': {...}, 'Q1': {...}, ... }
+    yearly_data = defaultdict(lambda: defaultdict(dict))
+    
+    for metric in cleaned_metrics:
+        period = metric.get('period')
+        if not period:
+            continue  # 跳过没有 period 的条目
+        
+        year, ptype, quarter = parse_period_yoy(period)
+        if year is None:
+            continue  # 跳过无法解析的 period
+        
+        if ptype == 'FY':
+            yearly_data[year]['FY'] = metric
+        elif ptype == 'Q':
+            yearly_data[year][f'Q{quarter}'] = metric
+        # 否则忽略
+
+    # 获取所有年份并排序（最新在前）
+    years = sorted(yearly_data.keys(), reverse=True)
+    if len(years) < 2:
+        raise ValueError("至少需要两个财年的数据")
+
+    latest_year = years[0]
+    prev_year = years[1]
+
+    result = []
+    indicators = [
+        ("Total Revenue", "total_revenue"),
+        ("Net Income", "net_income"),
+        ("Earnings Per Share", "earnings_per_share"),
+        ("Operating Expenses", "operating_expenses"),
+        ("Cash Flow", "operating_cash_flow")
+    ]
+
+    def format_value(val):
+        if val is None:
+            return "N/A"
+        try:
+            val = float(val)
+        except (TypeError, ValueError):
+            return "N/A"
+        abs_val = abs(val)
+        if abs_val >= 1e9:
+            return f"${val / 1e9:.2f}B"
+        elif abs_val >= 1e6:
+            return f"${val / 1e6:.1f}M"
+        elif abs_val >= 1e3:
+            return f"${val / 1e3:.1f}K"
+        else:
+            return f"${val:.2f}"
+
+    for label, key in indicators:
+        # 获取本财年最佳值
+        current_val = get_best_value_for_year_yoy(yearly_data[latest_year], key)
+        # 获取去年财年最佳值
+        prev_val = get_best_value_for_year_yoy(yearly_data[prev_year], key)
+
+        if current_val is None or prev_val is None or prev_val == 0:
+            change_str = "N/A"
+            color = "N/A"
+        else:
+            try:
+                current_val = float(current_val)
+                prev_val = float(prev_val)
+            except (TypeError, ValueError):
+                change_str = "N/A"
+                color = "N/A"
+            else:
+                change_pct = (current_val - prev_val) / abs(prev_val) * 100
+                if change_pct > 0:
+                    change_str = f"+{change_pct:.1f}%"
+                    color = "green"
+                elif change_pct < 0:
+                    change_str = f"{change_pct:.1f}%"
+                    color = "red"
+                else:
+                    change_str = "0.0%"
+                    color = "N/A"
+
+        formatted_value = format_value(current_val)
+
+        result.append({
+            "label": label,
+            "value": formatted_value,
+            "change": change_str,
+            "color": color
+        })
+
+    return result
+# def parse_period_yoy(period):
+#     """解析 period 为 (year, type, quarter)"""
+#     if period.startswith('FY'):
+#         year = int(period[2:])
+#         return year, 'FY', None
+#     elif re.match(r'Q[1-4]-\d{4}', period):
+#         q_part, year_str = period.split('-')
+#         return int(year_str), 'Q', int(q_part[1])
+#     else:
+#         # 忽略无法解析的 period
+#         return None, None, None
+# def calculate_yoy_comparison(data_json):
+#     metrics_list = data_json['metrics']
+    
+#     # 按年份组织数据：year -> { 'FY': {...}, 'Q1': {...}, ... }
+#     yearly_data = defaultdict(lambda: defaultdict(dict))
+    
+#     for metric in metrics_list:
+#         period = metric['period']
+#         year, ptype, quarter = parse_period_yoy(period)
+#         if ptype == 'FY':
+#             yearly_data[year]['FY'] = metric
+#         else:
+#             yearly_data[year][f'Q{quarter}'] = metric
+
+#     # 获取所有年份并排序（最新在前）
+#     years = sorted(yearly_data.keys(), reverse=True)
+#     if len(years) < 2:
+#         raise ValueError("至少需要两个财年的数据")
+
+#     latest_year = years[0]
+#     prev_year = years[1]
+
+#     result = []
+#     indicators = [
+#         ("Total Revenue", "total_revenue"),
+#         ("Net Income", "net_income"),
+#         ("Earnings Per Share", "earnings_per_share"),
+#         ("Operating Expenses", "operating_expenses"),
+#         ("Cash Flow", "operating_cash_flow")
+#     ]
+
+#     def format_value(val):
+#         if val is None:
+#             return "N/A"
+#         abs_val = abs(val)
+#         if abs_val >= 1e9:
+#             return f"${val / 1e9:.2f}B"
+#         elif abs_val >= 1e6:
+#             return f"${val / 1e6:.1f}M"
+#         elif abs_val >= 1e3:
+#             return f"${val / 1e3:.1f}K"
+#         else:
+#             return f"${val:.2f}"
+
+#     for label, key in indicators:
+#         # 获取本财年最佳值
+#         current_val = get_best_value_for_year(
+#             {k: v.get(key) for k, v in yearly_data[latest_year].items()},
+#             key
+#         )
+#         # 获取去年财年最佳值
+#         prev_val = get_best_value_for_year(
+#             {k: v.get(key) for k, v in yearly_data[prev_year].items()},
+#             key
+#         )
+
+#         if current_val is None or prev_val is None or prev_val == 0:
+#             change_str = "N/A"
+#             color = "N/A"
+#         else:
+#             change_pct = (current_val - prev_val) / abs(prev_val) * 100
+#             if change_pct > 0:
+#                 change_str = f"+{change_pct:.1f}%"
+#                 color = "green"
+#             elif change_pct < 0:
+#                 change_str = f"{change_pct:.1f}%"
+#                 color = "red"
+#             else:
+#                 change_str = "0.0%"
+#                 color = "N/A"
+
+#         formatted_value = format_value(current_val)
+
+#         result.append({
+#             "label": label,
+#             "value": formatted_value,
+#             "change": change_str,
+#             "color": color
+#         })
+
+#     return result
+
+
+
+
+import re
+import json
+from collections import defaultdict
+
+def parse_period_three_year(period):
+    """解析 period 为 (year, type, quarter)"""
+    if period.startswith('FY'):
+        year = int(period[2:])
+        return year, 'FY', None
+    elif re.match(r'Q[1-4]-\d{4}', period):
+        q_part, year_str = period.split('-')
+        return int(year_str), 'Q', int(q_part[1])
+    else:
+        # 忽略无法解析的 period
+        return None, None, None
+
+def extract_financial_table(data_json):
+    metrics_list = data_json.get('metrics', [])
+    if not metrics_list:
+        return []
+    if not isinstance(metrics_list, list):
+        return []
+    if not isinstance(metrics_list[0], dict):
+        return []
+    # === 安全清洗：确保每个 metric 是字典 ===
+    cleaned_metrics = []
+    for i, metric in enumerate(metrics_list):
+        if isinstance(metric, str):
+            try:
+                metric = json.loads(metric)
+            except Exception as e:
+                raise ValueError(f"Failed to parse metrics[{i}] as JSON string: {metric}") from e
+        if not isinstance(metric, dict):
+            raise TypeError(f"metrics[{i}] is not a dictionary or valid JSON string. Type: {type(metric)}")
+        cleaned_metrics.append(metric)
+
+    # 按年份组织所有报告：year -> { 'FY': metric_dict, 'Q1': ..., 'Q2': ... }
+    yearly_reports = defaultdict(dict)
+    all_years = set()
+
+    for metric in cleaned_metrics:
+        period = metric.get('period')
+        if not period:
+            continue  # 跳过无 period 的条目
+        
+        year, ptype, quarter = parse_period_three_year(period)
+        if year is None:
+            continue
+        all_years.add(year)
+        if ptype == 'FY':
+            yearly_reports[year]['FY'] = metric
+        elif ptype == 'Q':
+            yearly_reports[year][f'Q{quarter}'] = metric
+
+    if not all_years:
+        raise ValueError("未找到任何有效报告期")
+
+    # 取最近三个财年（倒序）
+    sorted_years = sorted(all_years, reverse=True)[:3]
+    # 补齐到3年（如果不足）
+    while len(sorted_years) < 3:
+        sorted_years.append(None)
+
+    # 为每个年份获取最佳值（优先 FY，其次 Q4→Q1）
+    def get_best_value(year, key):
+        if year is None:
+            return None
+        reports = yearly_reports.get(year, {})
+        # 确保 reports[q] 是 dict
+        fy_report = reports.get('FY')
+        if fy_report and isinstance(fy_report, dict):
+            fy_val = fy_report.get(key)
+            if fy_val is not None:
+                return fy_val
+        # 否则 Q4 → Q1
+        for q in ['Q4', 'Q3', 'Q2', 'Q1']:
+            q_report = reports.get(q)
+            if q_report and isinstance(q_report, dict):
+                q_val = q_report.get(key)
+                if q_val is not None:
+                    return q_val
+        return None
+
+    # 指标定义
+    indicators = [
+        ("Total", "total_revenue"),
+        ("Net Income", "net_income"),
+        ("Earnings Per Share", "earnings_per_share"),
+        ("Operating Expenses", "operating_expenses"),
+        ("Cash Flow", "operating_cash_flow")
+    ]
+
+    # 格式化函数
+    def format_to_m(value):
+        if value is None:
+            return "N/A"
+        try:
+            val = float(value)
+        except (TypeError, ValueError):
+            return "N/A"
+        val_in_m = val / 1e6
+        if abs(val_in_m - round(val_in_m)) < 1e-6:
+            return f"{int(round(val_in_m))}M"
+        else:
+            return f"{val_in_m:.1f}M"
+
+    def format_eps(value):
+        if value is None:
+            return "N/A"
+        try:
+            val = float(value)
+        except (TypeError, ValueError):
+            return "N/A"
+        return f"{val:.2f}"
+
+    # 构建 list_data
+    header = ["Category"] + [f"{year}/FY" for year in sorted_years if year is not None]
+    list_data = [header]
+
+    for label, key in indicators:
+        row = [label]
+        for year in sorted_years:
+            if year is None:
+                row.append("N/A")
+            else:
+                val = get_best_value(year, key)
+                if label == "Earnings Per Share":
+                    row.append(format_eps(val))
+                else:
+                    row.append(format_to_m(val))
+        list_data.append(row)
+
+    # 构建 yoy_rates
+    valid_years = [y for y in sorted_years if y is not None]
+    yoy_header = ["Category"]
+    yoy_pairs = []
+
+    if len(valid_years) >= 2:
+        yoy_header.append(f"{valid_years[0]}/FY")
+        yoy_pairs.append((valid_years[0], valid_years[1]))
+    if len(valid_years) >= 3:
+        yoy_header.append(f"{valid_years[1]}/FY")
+        yoy_pairs.append((valid_years[1], valid_years[2]))
+
+    yoy_rates = [yoy_header]
+
+    for label, key in indicators:
+        row = [label]
+        for curr_y, prev_y in yoy_pairs:
+            curr_val = get_best_value(curr_y, key)
+            prev_val = get_best_value(prev_y, key)
+
+            if curr_val is None or prev_val is None or prev_val == 0:
+                row.append("N/A")
+            else:
+                try:
+                    curr_val = float(curr_val)
+                    prev_val = float(prev_val)
+                except (TypeError, ValueError):
+                    row.append("N/A")
+                else:
+                    pct = (curr_val - prev_val) / abs(prev_val) * 100
+                    if pct >= 0:
+                        row.append(f"+{pct:.2f}%")
+                    else:
+                        row.append(f"{pct:.2f}%")
+        yoy_rates.append(row)
+
+    return {
+        "list_data": list_data,
+        "yoy_rates": yoy_rates
+    }

service/three_year_table_tool.py

@@ -0,0 +1,106 @@
+def format_number_for_table(value):
+    """用于表格的数字格式化：自动 B/M，保留适当小数"""
+    if value is None or value == 0:
+        return "0"
+    if abs(value) >= 1_000_000_000:
+        num = value / 1_000_000_000
+        if num.is_integer():
+            return f"{int(num)}B"
+        else:
+            return f"{num:.2f}B".rstrip('0').rstrip('.')
+    elif abs(value) >= 1_000_000:
+        num = value / 1_000_000
+        if num.is_integer():
+            return f"{int(num)}M"
+        else:
+            return f"{num:.1f}M".rstrip('0').rstrip('.')
+    else:
+        return f"{value:,.1f}".rstrip('0').rstrip('.')
+
+def safe_float_or_zero(val):
+    if val is None:
+        return 0.0
+    try:
+        return float(val)
+    except (ValueError, TypeError):
+        return 0.0
+
+def calculate_yoy_rate(current, previous):
+    if previous == 0:
+        return "+0.0%" if current >= 0 else "-0.0%"
+    rate = (current - previous) / abs(previous) * 100
+    sign = "+" if rate >= 0 else "-"
+    return f"{sign}{abs(rate):.1f}%"
+
+def build_table_format(three_year_data):
+    # 按 fiscal_year 降序排列（最新在前）
+    sorted_data = sorted(three_year_data, key=lambda x: x["fiscal_year"], reverse=True)
+    
+    # 生成年份标签：如 "2025 FY" 或 "2025 Q3"
+    year_labels = [f"{item['fiscal_year']} {item['level']}" for item in sorted_data]
+    
+    # 提取数值（确保至少三年，不足用 0 补齐）
+    while len(sorted_data) < 3:
+        sorted_data.append({
+            "fiscal_year": 0,
+            "level": "N/A",
+            "total_revenue": 0,
+            "net_income": 0,
+            "earnings_per_share": 0.0,
+            "operating_expenses": 0,
+            "operating_cash_flow": 0
+        })
+        year_labels.append("N/A")
+
+    # 取前三
+    y0, y1, y2 = sorted_data[0], sorted_data[1], sorted_data[2]
+
+    # 构建 list_data
+    list_data = [
+        ["Category"] + year_labels[:3],
+        ["Total Revenue",
+         format_number_for_table(safe_float_or_zero(y0["total_revenue"])),
+         format_number_for_table(safe_float_or_zero(y1["total_revenue"])),
+         format_number_for_table(safe_float_or_zero(y2["total_revenue"]))],
+        ["Net Income",
+         format_number_for_table(safe_float_or_zero(y0["net_income"])),
+         format_number_for_table(safe_float_or_zero(y1["net_income"])),
+         format_number_for_table(safe_float_or_zero(y2["net_income"]))],
+        ["Earnings Per Share",
+         f"{safe_float_or_zero(y0['earnings_per_share']):.2f}",
+         f"{safe_float_or_zero(y1['earnings_per_share']):.2f}",
+         f"{safe_float_or_zero(y2['earnings_per_share']):.2f}"],
+        ["Operating Expenses",
+         format_number_for_table(safe_float_or_zero(y0["operating_expenses"])),
+         format_number_for_table(safe_float_or_zero(y1["operating_expenses"])),
+         format_number_for_table(safe_float_or_zero(y2["operating_expenses"]))],
+        ["Operating Cash Flow",
+         format_number_for_table(safe_float_or_zero(y0["operating_cash_flow"])),
+         format_number_for_table(safe_float_or_zero(y1["operating_cash_flow"])),
+         format_number_for_table(safe_float_or_zero(y2["operating_cash_flow"]))]
+    ]
+
+    # 构建 yoy_rates（只有两列：2025 vs 2024，2024 vs 2023）
+    yoy_rates = [
+        ["Category"] + year_labels[:2],  # 只取前两年标签
+        ["Total Revenue",
+         calculate_yoy_rate(safe_float_or_zero(y0["total_revenue"]), safe_float_or_zero(y1["total_revenue"])),
+         calculate_yoy_rate(safe_float_or_zero(y1["total_revenue"]), safe_float_or_zero(y2["total_revenue"]))],
+        ["Net Income",
+         calculate_yoy_rate(safe_float_or_zero(y0["net_income"]), safe_float_or_zero(y1["net_income"])),
+         calculate_yoy_rate(safe_float_or_zero(y1["net_income"]), safe_float_or_zero(y2["net_income"]))],
+        ["Earnings Per Share",
+         calculate_yoy_rate(safe_float_or_zero(y0["earnings_per_share"]), safe_float_or_zero(y1["earnings_per_share"])),
+         calculate_yoy_rate(safe_float_or_zero(y1["earnings_per_share"]), safe_float_or_zero(y2["earnings_per_share"]))],
+        ["Operating Expenses",
+         calculate_yoy_rate(safe_float_or_zero(y0["operating_expenses"]), safe_float_or_zero(y1["operating_expenses"])),
+         calculate_yoy_rate(safe_float_or_zero(y1["operating_expenses"]), safe_float_or_zero(y2["operating_expenses"]))],
+        ["Operating Cash Flow",
+         calculate_yoy_rate(safe_float_or_zero(y0["operating_cash_flow"]), safe_float_or_zero(y1["operating_cash_flow"])),
+         calculate_yoy_rate(safe_float_or_zero(y1["operating_cash_flow"]), safe_float_or_zero(y2["operating_cash_flow"]))]
+    ]
+
+    return {
+        "list_data": list_data,
+        "yoy_rates": yoy_rates
+    }

service/three_year_tool.py

@@ -0,0 +1,249 @@
+import json
+
+# -----------------------------
+# 辅助函数：安全转换数值
+# -----------------------------
+def safe_int(val, default=0):
+    if val is None:
+        return default
+    try:
+        return int(float(val))
+    except (ValueError, TypeError):
+        return default
+
+def safe_float(val, default=0.0):
+    if val is None:
+        return default
+    try:
+        return float(val)
+    except (ValueError, TypeError):
+        return default
+
+# -----------------------------
+# 步骤1：提取带 fallback 的三年数据
+# -----------------------------
+def extract_last_three_with_fallback(data_list):
+    years = [2025, 2024, 2023]
+    priority_levels = [
+        ("FY", [f"FY{y}" for y in years]),
+        ("Q4", [f"{y}Q4" for y in years]),
+        ("Q3", [f"{y}Q3" for y in years]),
+        ("Q2", [f"{y}Q2" for y in years]),
+        ("Q1", [f"{y}Q1" for y in years]),
+    ]
+    
+    data_map = {item["period"]: item for item in data_list if isinstance(item, dict) and "period" in item}
+    
+    for level_name, periods in priority_levels:
+        records = []
+        valid = True
+        for period in periods:
+            item = data_map.get(period)
+            if not isinstance(item, dict) or item.get("total_revenue") is None:
+                valid = False
+                break
+            fiscal_year = int(period[:4]) if level_name != "FY" else int(period[2:])
+            records.append({
+                "period": period,
+                "fiscal_year": fiscal_year,
+                "level": level_name,
+                "total_revenue": item.get("total_revenue"),
+                "net_income": item.get("net_income"),
+                "earnings_per_share": item.get("earnings_per_share"),
+                "operating_expenses": item.get("operating_expenses"),
+                "operating_cash_flow": item.get("operating_cash_flow")
+            })
+        if valid:
+            return records
+    
+    # Fallback: 返回第一个有数据的层级（即使不全）
+    for level_name, periods in priority_levels:
+        records = []
+        for period in periods:
+            item = data_map.get(period)
+            if isinstance(item, dict) and item.get("total_revenue") is not None:
+                fiscal_year = int(period[:4]) if level_name != "FY" else int(period[2:])
+                records.append({
+                    "period": period,
+                    "fiscal_year": fiscal_year,
+                    "level": level_name,
+                    "total_revenue": item.get("total_revenue"),
+                    "net_income": item.get("net_income"),
+                    "earnings_per_share": item.get("earnings_per_share"),
+                    "operating_expenses": item.get("operating_expenses"),
+                    "operating_cash_flow": item.get("operating_cash_flow")
+                })
+        if records:
+            return records
+    return []
+
+# -----------------------------
+# 步骤2：格式化数字（B / M）
+# -----------------------------
+def format_number(value):
+    if value >= 1_000_000_000:
+        num = value / 1_000_000_000
+        if num == int(num):
+            return f"${int(num)}B"
+        else:
+            return f"${num:.2f}B".rstrip('0').rstrip('.')
+    elif value >= 1_000_000:
+        num = value / 1_000_000
+        if num == int(num):
+            return f"${int(num)}M"
+        else:
+            return f"${num:.1f}M".rstrip('0').rstrip('.')
+    elif value >= 1_000:
+        return f"${value:,.0f}"
+    else:
+        return f"${value}"
+
+def format_eps(value):
+    return f"${value:.2f}"
+
+def calculate_change(current, previous):
+    if previous == 0:
+        return "+0.0%" if current >= 0 else "-0.0%"
+    change = (current - previous) / abs(previous) * 100
+    sign = "+" if change >= 0 else "-"
+    return f"{sign}{abs(change):.1f}%"
+
+# -----------------------------
+# 步骤3：构建最终 metrics
+# -----------------------------
+def build_financial_metrics(three_year_data):
+    if len(three_year_data) < 2:
+        raise ValueError("至少需要两年数据来计算同比变化")
+
+    sorted_data = sorted(three_year_data, key=lambda x: x["fiscal_year"], reverse=True)
+    latest = sorted_data[0]
+    previous = sorted_data[1]
+
+    rev_curr = safe_int(latest["total_revenue"])
+    rev_prev = safe_int(previous["total_revenue"])
+
+    net_curr = safe_int(latest["net_income"])
+    net_prev = safe_int(previous["net_income"])
+
+    eps_curr = safe_float(latest["earnings_per_share"])
+    eps_prev = safe_float(previous["earnings_per_share"])
+
+    opex_curr = safe_int(latest["operating_expenses"])
+    opex_prev = safe_int(previous["operating_expenses"])
+
+    cash_curr = safe_int(latest["operating_cash_flow"])
+    cash_prev = safe_int(previous["operating_cash_flow"])
+
+    return [
+        {
+            "label": "Total Revenue",
+            "value": format_number(rev_curr),
+            "change": calculate_change(rev_curr, rev_prev),
+            "color": "green" if rev_curr >= rev_prev else "red"
+        },
+        {
+            "label": "Net Income",
+            "value": format_number(net_curr),
+            "change": calculate_change(net_curr, net_prev),
+            "color": "green" if net_curr >= net_prev else "red"
+        },
+        {
+            "label": "Earnings Per Share",
+            "value": format_eps(eps_curr),
+            "change": calculate_change(eps_curr, eps_prev),
+            "color": "green" if eps_curr >= eps_prev else "red"
+        },
+        {
+            "label": "Operating Expenses",
+            "value": format_number(opex_curr),
+            "change": calculate_change(opex_curr, opex_prev),
+            "color": "green" if opex_curr >= opex_prev else "red"
+        },
+        {
+            "label": "Operating Cash Flow",
+            "value": format_number(cash_curr),
+            "change": calculate_change(cash_curr, cash_prev),
+            "color": "green" if cash_curr >= cash_prev else "red"
+        }
+    ]
+
+# -----------------------------
+# 主流程：输入 raw_data，输出 financial_metrics
+# -----------------------------
+# def process_financial_data(raw_data):
+#     # 如果是字符串，先解析 JSON
+#     if isinstance(raw_data, str):
+#         raw_data = json.loads(raw_data)
+    
+#     # 确保是列表
+#     if not isinstance(raw_data, list):
+#         raise TypeError("raw_data 必须是列表或 JSON 字符串表示的列表")
+    
+#     # 提取三年数据
+#     three_years = extract_last_three_with_fallback(raw_data)
+    
+#     if not three_years:
+#         raise ValueError("无法提取有效的三年财务数据")
+    
+#     # 构建指标
+#     return build_financial_metrics(three_years)
+
+def process_financial_data_with_metadata(raw_data):
+    """
+    返回包含 financial_metrics + year_data + three_year_data 的完整结果
+    """
+    return_value = {"financial_metrics": [], "year_data": "N/A", "three_year_data": []}
+    if not raw_data:
+        return return_value
+    if not isinstance(raw_data, list):
+        return return_value
+    if not isinstance(raw_data[0], dict):
+        return {"financial_metrics": [], "year_data": "N/A", "three_year_data": []}
+    # 1. 解析输入
+    if isinstance(raw_data, str):
+        raw_data = json.loads(raw_data)
+    if not isinstance(raw_data, list):
+        raise TypeError("raw_data 必须是列表或 JSON 字符串")
+
+    # 2. 提取三年数据（带 fallback）
+    three_years = extract_last_three_with_fallback(raw_data)
+    if not three_years:
+        print("无法提取有效的三年财务数据")
+        # raise ValueError("无法提取有效的三年财务数据")
+        return return_value
+
+    # 3. 获取最新年份和报告类型（用于 year_data）
+    latest = max(three_years, key=lambda x: x["fiscal_year"])
+    year = latest["fiscal_year"]
+    level = latest["level"]
+    
+    if level == "FY":
+        year_data = f"{year} FY"
+    else:  # Q1, Q2, Q3, Q4
+        year_data = f"{year} {level}"
+
+    # 4. 构建指标
+    financial_metrics = build_financial_metrics(three_years)
+
+    # 5. 返回完整结构
+    return {
+        "financial_metrics": financial_metrics,
+        "year_data": year_data,
+        "three_year_data": three_years  # 已包含 period, fiscal_year, level, 各项财务字段
+    }
+
+# -----------------------------
+# 示例使用（替换为你的真实数据）
+# -----------------------------
+# if __name__ == "__main__":
+#     # 👇 在这里粘贴你的原始数据（可以是字符串或变量）
+#     # 示例：从文件读取或直接赋值
+#     with open("financial_data.json", "r", encoding="utf-8") as f:
+#         raw_input = f.read()  # 或者直接赋值为你的数据变量
+
+#     # 处理
+#     try:
+#         financial_metrics = process_financial_data(raw_input)
+#         print(json.dumps(financial_metrics, indent=2))
+#     except Exception as e:
+#         print("处理失败:", e)