From 917d5bca2a8f461f2a616a7b232ae5648d90beba Mon Sep 17 00:00:00 2001
From: claude_dev <claude-dev@sanguo.local>
Date: Fri, 10 Jul 2026 07:10:36 +0800
Subject: [PATCH] =?UTF-8?q?fix(data):=20baostock=5Fdownload=E6=96=AD?=
 =?UTF-8?q?=E8=BF=9E=E8=87=AA=E5=8A=A8re-login=20retry(=E9=99=90=E6=B5=81?=
 =?UTF-8?q?=E9=B2=81=E6=A3=92)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

本机下131只后baostock broken pipe卡死(脚本无retry). 加reconnect()+download_one
max_retries=3(空结果/broken pipe→bs.logout+login重试). 下次跑断点续传skip 131.
---
 scripts/data_platform/baostock_download.py | 68 +++++++++++++++-------
 1 file changed, 47 insertions(+), 21 deletions(-)

diff --git a/scripts/data_platform/baostock_download.py b/scripts/data_platform/baostock_download.py
index 370b18d..e705556 100644
--- a/scripts/data_platform/baostock_download.py
+++ b/scripts/data_platform/baostock_download.py
@@ -64,27 +64,53 @@ def exists(code: str, adjust: str, start_year: int) -> bool:
         return False
 
 
-def download_one(code: str, start: str, end: str, adjustflag: str):
-    """baostock 15min，返回 (df, None) 或 (None, err)。"""
-    rs = bs.query_history_k_data_plus(
-        bs_symbol(code),
-        "date,time,open,high,low,close,volume,amount",
-        start_date=start, end_date=end, frequency="15", adjustflag=adjustflag,
-    )
-    rows, fields = [], rs.fields
-    while (rs.error_code == "0") & rs.next():
-        rows.append(rs.get_row_data())
-    if not rows:
-        return None, rs.error_msg
-    df = pd.DataFrame(rows, columns=fields)
-    for c in ("open", "high", "low", "close", "volume", "amount"):
-        df[c] = pd.to_numeric(df[c], errors="coerce")
-    t = df["time"].astype(str)
-    df["datetime"] = pd.to_datetime(
-        df["date"] + " " + t.str[8:10] + ":" + t.str[10:12], errors="coerce"
-    )
-    df = df.dropna(subset=["close", "datetime"]).sort_values("datetime")
-    return df, None
+def reconnect() -> bool:
+    """baostock 断连后重连（限流/broken pipe 恢复）。"""
+    try:
+        bs.logout()
+    except Exception:  # noqa: BLE001
+        pass
+    return bs.login().error_code == "0"
+
+
+def download_one(code: str, start: str, end: str, adjustflag: str, max_retries: int = 3):
+    """baostock 15min，返回 (df, None) 或 (None, err)。断连/限流自动 re-login retry。"""
+    last_err = ""
+    for attempt in range(max_retries):
+        try:
+            rs = bs.query_history_k_data_plus(
+                bs_symbol(code),
+                "date,time,open,high,low,close,volume,amount",
+                start_date=start, end_date=end, frequency="15", adjustflag=adjustflag,
+            )
+            rows, fields = [], rs.fields
+            while (rs.error_code == "0") & rs.next():
+                rows.append(rs.get_row_data())
+            if not rows:
+                last_err = rs.error_msg or "empty"
+                # 空可能是限流（"接收数据异常"）→ re-login 重试
+                if attempt < max_retries - 1:
+                    reconnect()
+                    time.sleep(2)
+                    continue
+                return None, last_err
+            df = pd.DataFrame(rows, columns=fields)
+            for c in ("open", "high", "low", "close", "volume", "amount"):
+                df[c] = pd.to_numeric(df[c], errors="coerce")
+            t = df["time"].astype(str)
+            df["datetime"] = pd.to_datetime(
+                df["date"] + " " + t.str[8:10] + ":" + t.str[10:12], errors="coerce"
+            )
+            df = df.dropna(subset=["close", "datetime"]).sort_values("datetime")
+            return df, None
+        except Exception as e:  # noqa: BLE001  broken pipe 等
+            last_err = f"{type(e).__name__}: {str(e)[:80]}"
+            if attempt < max_retries - 1:
+                reconnect()
+                time.sleep(2)
+                continue
+            return None, last_err
+    return None, last_err
 
 
 def save_one(code, df, adjust):