"fix:reduce-thread-count-and-add-exponential-backoff"

sync_data.py

@@ -49,14 +49,14 @@ def get_thread_config():
     import multiprocessing
     cpu_count = multiprocessing.cpu_count()
     
-    # 分层并发策略
+    # 分层并发策略（保守配置，避免 API 限流）
     config = {
-        'daily': _safe_int_env("MAX_WORKERS_DAILY", cpu_count * 4),
-        'fund': _safe_int_env("MAX_WORKERS_FUND", cpu_count * 3),
-        'valuation': _safe_int_env("MAX_WORKERS_VALUATION", cpu_count * 3),
-        'margin': _safe_int_env("MAX_WORKERS_MARGIN", cpu_count * 3),
-        'financial': _safe_int_env("MAX_WORKERS_FINANCIAL", cpu_count * 2),
-        'dividend': _safe_int_env("MAX_WORKERS_DIVIDEND", cpu_count * 2),
+        'daily': _safe_int_env("MAX_WORKERS_DAILY", max(2, cpu_count * 2)),
+        'fund': _safe_int_env("MAX_WORKERS_FUND", max(2, cpu_count)),
+        'valuation': _safe_int_env("MAX_WORKERS_VALUATION", max(2, cpu_count)),
+        'margin': _safe_int_env("MAX_WORKERS_MARGIN", max(2, cpu_count)),
+        'financial': _safe_int_env("MAX_WORKERS_FINANCIAL", max(1, cpu_count)),
+        'dividend': _safe_int_env("MAX_WORKERS_DIVIDEND", max(1, cpu_count)),
     }
     
     # 向后兼容
@@ -94,8 +94,9 @@ def get_stock_list() -> pd.DataFrame:
     logger.info("Fetching all-market target list...")
     all_lists = []
     
-    # A股列表获取（带重试）
-    max_retries = 3
+    # A股列表获取（带重试和指数退避）
+    max_retries = 5
+    base_delay = 2.0
     for attempt in range(max_retries):
         try:
             df_a = ak.stock_zh_a_spot_em()[['代码', '名称']]
@@ -108,8 +109,9 @@ def get_stock_list() -> pd.DataFrame:
             if attempt == max_retries - 1:
                 logger.error(f"Failed to fetch A-stock list after {max_retries} attempts: {e}")
             else:
-                logger.warning(f"Attempt {attempt + 1} failed, retrying... Error: {e}")
-                time.sleep(5)  # 重试前等待 5 秒
+                delay = base_delay * (2 ** attempt)
+                logger.warning(f"Attempt {attempt + 1} failed, retrying in {delay}s... Error: {e}")
+                time.sleep(delay)
 
     # ETF (增加容错)
     try:
@@ -164,9 +166,15 @@ def get_stock_list() -> pd.DataFrame:
 
 def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.DataFrame]:
     """抓取单只标的数据"""
-    max_retries = 3
+    max_retries = 5
+    base_delay = 1.0  # 寂础延迟秒
+    
     for attempt in range(max_retries):
         try:
+            # 指数退避：每次重试增加延迟
+            if attempt > 0:
+                delay = base_delay * (2 ** attempt)  # 1s, 2s, 4s, 8s
+                time.sleep(delay)
             end_date = get_beijing_time().strftime('%Y%m%d')
             fetch_start = start_date.replace('-', '')
             df = None
@@ -369,7 +377,8 @@ def sync_stock_daily(targets: List[Dict[str, str]], last_trade_day: str) -> int:
 
 def get_stock_fund_flow(code: str, market: str) -> Optional[pd.DataFrame]:
     """获取单只股票资金流向数据"""
-    max_retries = 3
+    max_retries = 5
+    base_delay = 1.0
     for attempt in range(max_retries):
         try:
             # 确定 market 参数
@@ -410,7 +419,10 @@ def get_stock_fund_flow(code: str, market: str) -> Optional[pd.DataFrame]:
         except Exception as e:
             if attempt == max_retries - 1:
                 pass  # 静默失败，很多股票可能没有资金流向数据
-            time.sleep(0.5)
+            # 指数退避：每次重试增加延迟
+            if attempt > 0:
+                delay = base_delay * (2 ** attempt)
+                time.sleep(delay)
     return None
 
 
@@ -499,7 +511,8 @@ def sync_fund_flow(targets: List[Dict[str, str]], last_trade_day: str) -> int:
 
 def get_stock_valuation(code: str) -> Optional[pd.DataFrame]:
     """获取单只股票估值指标数据"""
-    max_retries = 3
+    max_retries = 5
+    base_delay = 1.0
     for attempt in range(max_retries):
         try:
             df = ak.stock_a_lg_indicator(symbol=code)
@@ -525,7 +538,9 @@ def get_stock_valuation(code: str) -> Optional[pd.DataFrame]:
                 available_cols = [c for c in cols if c in df.columns]
                 return df[available_cols]
         except Exception:
-            time.sleep(0.5)
+            if attempt > 0:
+                delay = base_delay * (2 ** attempt)
+                time.sleep(delay)
     return None
 
 
@@ -612,7 +627,8 @@ def sync_valuation(targets: List[Dict[str, str]], last_trade_day: str) -> int:
 
 def get_stock_margin(code: str) -> Optional[pd.DataFrame]:
     """获取单只股票融资融券数据"""
-    max_retries = 3
+    max_retries = 5
+    base_delay = 1.0
     for attempt in range(max_retries):
         try:
             # 尝试上交所
@@ -643,7 +659,9 @@ def get_stock_margin(code: str) -> Optional[pd.DataFrame]:
         except Exception as e:
             if attempt == max_retries - 1:
                 pass
-            time.sleep(0.5)
+            if attempt > 0:
+                delay = base_delay * (2 ** attempt)
+                time.sleep(delay)
     return None
 
 
@@ -730,7 +748,8 @@ def sync_margin(targets: List[Dict[str, str]], last_trade_day: str) -> int:
 
 def get_stock_financial_indicator(code: str) -> Optional[pd.DataFrame]:
     """获取单只股票财务指标数据"""
-    max_retries = 3
+    max_retries = 5
+    base_delay = 1.0
     for attempt in range(max_retries):
         try:
             df = ak.stock_financial_analysis_indicator(symbol=code)
@@ -760,7 +779,9 @@ def get_stock_financial_indicator(code: str) -> Optional[pd.DataFrame]:
                 available_cols = [c for c in cols if c in df.columns]
                 return df[available_cols]
         except Exception:
-            time.sleep(0.5)
+            if attempt > 0:
+                delay = base_delay * (2 ** attempt)
+                time.sleep(delay)
     return None
 
 
@@ -890,7 +911,8 @@ def sync_holder_num() -> int:
 
 def get_stock_dividend(code: str) -> Optional[pd.DataFrame]:
     """获取单只股票分红数据"""
-    max_retries = 3
+    max_retries = 5
+    base_delay = 1.0
     for attempt in range(max_retries):
         try:
             df = ak.stock_history_dividend(symbol=code)
@@ -912,7 +934,9 @@ def get_stock_dividend(code: str) -> Optional[pd.DataFrame]:
                 available_cols = [c for c in cols if c in df.columns]
                 return df[available_cols]
         except Exception:
-            time.sleep(0.5)
+            if attempt > 0:
+                delay = base_delay * (2 ** attempt)
+                time.sleep(delay)
     return None