"perf:memory-optimization-with-explicit-gc"

sync_data.py

@@ -8,6 +8,7 @@ import sys
 import logging
 import time
 import threading
+import gc
 from datetime import datetime, timedelta
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import List, Optional, Dict, Any
@@ -456,6 +457,8 @@ def sync_stock_daily(targets: List[Dict[str, str]], last_trade_day: str) -> Dict
     
     if all_new_data:
         inc_df = pd.concat(all_new_data, ignore_index=True)
+        # 方案3：及时释放内存
+        del all_new_data
         total_records = len(inc_df)
         # 识别变动月份
         changed = inc_df.assign(yr=inc_df['trade_date'].dt.year, mo=inc_df['trade_date'].dt.month)[['yr', 'mo']].drop_duplicates().values
@@ -488,12 +491,20 @@ def sync_stock_daily(targets: List[Dict[str, str]], last_trade_day: str) -> Dict
             month_inc = inc_df[(inc_df['trade_date'].dt.year == yr) & (inc_df['trade_date'].dt.month == mo)]
             if old_df is not None:
                 final_month_df = pd.concat([old_df, month_inc]).drop_duplicates(subset=['code', 'trade_date'])
+                # 方案3：释放旧数据内存
+                del old_df, month_inc
             else:
                 final_month_df = month_inc
                 
             final_month_df.to_parquet(local_path)
             changed_files.append(filename)  # 记录变更的文件
             logger.info(f"Saved updated data for {filename}")
+            # 方案3：释放最终数据内存
+            del final_month_df
+        
+        # 方案3：循环结束后释放inc_df并触发GC
+        del inc_df
+        gc.collect()
     else:
         total_records = 0
             
@@ -654,6 +665,8 @@ def sync_fund_flow(targets: List[Dict[str, str]], last_trade_day: str) -> Dict[s
     # 5. 按月分表保存
     if all_data:
         new_df = pd.concat(all_data, ignore_index=True)
+        # 方案3：及时释放内存
+        del all_data
         total_records = len(new_df)
         
         # 确定需要更新的月份
@@ -682,16 +695,23 @@ def sync_fund_flow(targets: List[Dict[str, str]], last_trade_day: str) -> Dict[s
                     
                     if old_month_df is not None:
                         final_month_df = pd.concat([old_month_df, month_data]).drop_duplicates(subset=['code', 'trade_date'])
+                        # 方案3：释放旧数据内存
+                        del old_month_df, month_data
                     else:
                         final_month_df = month_data
                     
                     final_month_df.to_parquet(local_path)
                     changed_files.append(filename)  # 记录变更的文件
                     logger.info(f"Saved fund flow data for {filename}")
+                    # 方案3：释放最终数据内存
+                    del final_month_df
                 
                 current += 1
         
         logger.info(f"Fund flow updated: {len(new_df)} new records")
+        # 方案3：释放new_df并触发GC
+        del new_df
+        gc.collect()
     
     return {
         'count': len(success_codes),
@@ -829,6 +849,8 @@ def sync_valuation(targets: List[Dict[str, str]], last_trade_day: str) -> Dict[s
     # 5. 按月分表保存
     if all_data:
         new_df = pd.concat(all_data, ignore_index=True)
+        # 方案3：及时释放内存
+        del all_data
         total_records = len(new_df)
         
         if not new_df.empty:
@@ -855,16 +877,23 @@ def sync_valuation(targets: List[Dict[str, str]], last_trade_day: str) -> Dict[s
                     
                     if old_month_df is not None:
                         final_month_df = pd.concat([old_month_df, month_data]).drop_duplicates(subset=['code', 'trade_date'])
+                        # 方案3：释放旧数据内存
+                        del old_month_df, month_data
                     else:
                         final_month_df = month_data
                     
                     final_month_df.to_parquet(local_path)
                     changed_files.append(filename)  # 记录变更的文件
                     logger.info(f"Saved valuation data for {filename}")
+                    # 方案3：释放最终数据内存
+                    del final_month_df
                 
                 current += 1
         
         logger.info(f"Valuation updated: {len(new_df)} new records")
+        # 方案3：释放new_df并触发GC
+        del new_df
+        gc.collect()
     
     return {
         'count': success_count,