fix(backfill): 增加重试和自动重连,解决长连接超时问题 (#91)

- 单只股票查询失败自动重试 3 次(指数退避 2s/4s/8s)
- 每 200 只股票主动重连 baostock,防止长连接被服务端断开
- backfill 写入也加 chunksize=500 防止 SQLite 变量超限
- 中断后重跑可自动续传(已有数据的股票自动 skip)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
YuDong Yang
2026-05-09 15:18:30 +08:00
co-authored by Claude Opus 4.6
parent 418d1bdf49
commit 444c0db69f
+82 -23
View File
@@ -156,20 +156,36 @@ class DataEngine:
return count
def backfill(self, symbols: list[str]) -> None:
"""通过 baostock 批量回填历史日 K 线数据(后复权)。已入库的自动 skip。"""
import baostock as bs
"""通过 baostock 批量回填历史日 K 线数据(后复权)。
容错机制:
- 单只股票失败自动重试 3 次,间隔递增(2s/4s/8s)
- 每 200 只股票自动重连 baostock(防止长连接超时)
- 已入库的自动 skip,中断后可重跑续传
"""
import time
from datetime import date, timedelta
today_str = date.today().strftime("%Y-%m-%d")
import baostock as bs
lg = bs.login()
if lg.error_code != "0":
logger.error(f"baostock 登录失败: {lg.error_msg}")
today_str = date.today().strftime("%Y-%m-%d")
max_retries = 3
reconnect_interval = 200 # 每处理 N 只股票重连一次
def _login():
lg = bs.login()
if lg.error_code != "0":
logger.error(f"baostock 登录失败: {lg.error_msg}")
return False
return True
if not _login():
return
success = 0
skipped = 0
failed = 0
since_reconnect = 0
try:
for i, symbol in enumerate(symbols):
@@ -177,32 +193,69 @@ class DataEngine:
if last_date and last_date >= today_str:
skipped += 1
if (i + 1) % 500 == 0:
logger.info(f"已处理 {i + 1}/{len(symbols)},成功 {success} 跳过 {skipped} 失败 {failed}")
logger.info(
f"已处理 {i + 1}/{len(symbols)},"
f"成功 {success} 跳过 {skipped} 失败 {failed}"
)
continue
# 定期重连,防止长连接超时
since_reconnect += 1
if since_reconnect >= reconnect_interval:
bs.logout()
time.sleep(1)
if not _login():
logger.error("重连失败,终止回填")
return
since_reconnect = 0
start = last_date or self.start_date
if last_date:
start = (date.fromisoformat(last_date) + timedelta(days=1)).strftime("%Y-%m-%d")
bs_code = self._to_baostock_code(symbol)
rs = bs.query_history_k_data_plus(
bs_code,
"date,open,high,low,close,volume,amount",
start_date=start,
end_date=today_str,
frequency="d",
adjustflag="1", # 后复权
)
if rs.error_code != "0":
logger.warning(f"[{symbol}] baostock 查询失败: {rs.error_msg}")
# 带重试的查询
rows = []
query_ok = False
for attempt in range(max_retries):
try:
rs = bs.query_history_k_data_plus(
bs_code,
"date,open,high,low,close,volume,amount",
start_date=start,
end_date=today_str,
frequency="d",
adjustflag="1", # 后复权
)
if rs.error_code != "0":
raise RuntimeError(rs.error_msg)
rows = []
while rs.next():
rows.append(rs.get_row_data())
query_ok = True
break
except Exception as exc:
if attempt < max_retries - 1:
wait = 2 ** (attempt + 1)
logger.warning(
f"[{symbol}] 第{attempt + 1}次失败: {exc},{wait}s 后重试"
)
time.sleep(wait)
# 重连 baostock
bs.logout()
time.sleep(1)
_login()
else:
logger.warning(f"[{symbol}] {max_retries}次重试均失败,跳过")
if not query_ok:
failed += 1
continue
rows = []
while rs.next():
rows.append(rs.get_row_data())
if not rows:
skipped += 1
continue
@@ -223,14 +276,20 @@ class DataEngine:
try:
with sqlite3.connect(self.db_path) as conn:
df.to_sql("stock_daily", conn, if_exists="append", index=False, method="multi")
df.to_sql(
"stock_daily", conn, if_exists="append",
index=False, method="multi", chunksize=500,
)
except sqlite3.IntegrityError:
pass
success += 1
if (i + 1) % 500 == 0:
logger.info(f"已处理 {i + 1}/{len(symbols)},成功 {success} 跳过 {skipped} 失败 {failed}")
logger.info(
f"已处理 {i + 1}/{len(symbols)},"
f"成功 {success} 跳过 {skipped} 失败 {failed}"
)
finally:
bs.logout()