Upload three_year_tool.py

service/three_year_tool.py

@@ -20,10 +20,15 @@ def safe_float(val, default=0.0):
         return default
 
 # -----------------------------
-# 步骤1：提取带 fallback 的三年数据
+# 步骤1:提取带 fallback 的数据(支持最新年+历史3年)
 # -----------------------------
 def extract_last_three_with_fallback(data_list):
-    years = [2025, 2024, 2023]
+    """
+    从数据列表中提取最新的4年数据(最新1年 + 历史3年)
+    优先级: FY > Q4 > Q3 > Q2 > Q1
+    返回: 最新的4个period的数据
+    """
+    years = [2025, 2024, 2023, 2022]  # 扩展到4年
     priority_levels = [
         ("FY", [f"FY{y}" for y in years]),
         ("Q4", [f"{y}Q4" for y in years]),
@@ -34,6 +39,7 @@ def extract_last_three_with_fallback(data_list):
     
     data_map = {item["period"]: item for item in data_list if isinstance(item, dict) and "period" in item}
     
+    # 尝试找到完整的4年数据
     for level_name, periods in priority_levels:
         records = []
         valid = True
@@ -42,7 +48,12 @@ def extract_last_three_with_fallback(data_list):
             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:])
+            # 从period中提取fiscal_year
+            if level_name == "FY":
+                fiscal_year = int(period[2:])  # FY2025 -> 2025
+            else:
+                fiscal_year = int(period[:4])  # 2025Q3 -> 2025
+            
             records.append({
                 "period": period,
                 "fiscal_year": fiscal_year,
@@ -56,13 +67,17 @@ def extract_last_three_with_fallback(data_list):
         if valid:
             return records
     
-    # Fallback: 返回第一个有数据的层级（即使不全）
+    # 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:])
+                if level_name == "FY":
+                    fiscal_year = int(period[2:])
+                else:
+                    fiscal_year = int(period[:4])
+                
                 records.append({
                     "period": period,
                     "fiscal_year": fiscal_year,
@@ -191,6 +206,8 @@ def build_financial_metrics(three_year_data):
 def process_financial_data_with_metadata(raw_data):
     """
     返回包含 financial_metrics + year_data + three_year_data 的完整结果
+    financial_metrics: 最新1年的指标(与前一年对比)
+    three_year_data: 最新的前3年数据(排除最新年,用于Latest 3 Years表格)
     """
     return_value = {"financial_metrics": [], "year_data": "N/A", "three_year_data": []}
     if not raw_data:
@@ -199,37 +216,51 @@ def process_financial_data_with_metadata(raw_data):
         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("无法提取有效的三年财务数据")
+    # 2. 提取4年数据（最新1年 + 历史3年，带 fallback）
+    four_years = extract_last_three_with_fallback(raw_data)
+    if not four_years:
+        print("无法提取有效的财务数据")
         return return_value
-
-    # 3. 获取最新年份和报告类型（用于 year_data）
-    latest = max(three_years, key=lambda x: x["fiscal_year"])
-    year = latest["fiscal_year"]
-    level = latest["level"]
+    
+    # 按fiscal_year降序排序
+    four_years_sorted = sorted(four_years, key=lambda x: x["fiscal_year"], reverse=True)
+    
+    # 3. 分离最新1年和历史3年
+    if len(four_years_sorted) < 2:
+        print("数据不足,至少需要2年数据")
+        return return_value
+    
+    # 最新年用于financial_metrics
+    latest_year = four_years_sorted[0]
+    previous_year = four_years_sorted[1]
+    
+    # 历史3年用于three_year_data表格(排除最新年)
+    three_years_for_table = four_years_sorted[1:4] if len(four_years_sorted) >= 4 else four_years_sorted[1:]
+    
+    # 4. 获取最新年份和报告类型（用于 year_data）
+    year = latest_year["fiscal_year"]
+    level = latest_year["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. 构建最新年的financial_metrics(与前一年对比)
+    financial_metrics = build_financial_metrics([latest_year, previous_year])
 
-    # 5. 返回完整结构
+    # 6. 返回完整结构
     return {
-        "financial_metrics": financial_metrics,
-        "year_data": year_data,
-        "three_year_data": three_years  # 已包含 period, fiscal_year, level, 各项财务字段
+        "financial_metrics": financial_metrics,  # 最新1年的指标
+        "year_data": year_data,  # 最新年份标签
+        "three_year_data": three_years_for_table  # 历史3年数据(用于表格)
     }
 
 # -----------------------------