增加新渠道

README.md

@@ -105,7 +105,7 @@ python -m src.main --dividend --symbol 000001
 python -m src.main --intraday --start-date 20260101
 
 # 抓取全部频率分钟K线
-python -m src.main --intraday --freq all --start-date 20260508
+python -m src.main --intraday --freq all --start-date 20260101
 
 # 抓取行业+地域分类
 python -m src.main --sector

src/baostock_conn.py

@@ -4,6 +4,7 @@ BaoStock 的 query_xxx() 非线程安全，所有查询需通过同一把锁串
 """
 
 import threading
+import time
 from contextlib import contextmanager
 from datetime import datetime
 import baostock as bs
@@ -11,7 +12,8 @@ import baostock as bs
 _lock = threading.Lock()
 _logged_in = False
 
-QUERY_TIMEOUT = 60  # 单次查询超时秒数
+QUERY_TIMEOUT = 60
+MAX_RETRY = 3
 
 
 def bs_login():
@@ -33,25 +35,24 @@ def bs_logout():
 
 
 def _relogin():
-    """超时后重连"""
+    """断线重连（调用方需持有 _lock）"""
     global _logged_in
     try:
         bs.logout()
     except Exception:
         pass
     _logged_in = False
+    time.sleep(1)
     bs.login()
     _logged_in = True
+    print(f"    [{datetime.now().strftime('%H:%M:%S')}] [BS] 已重连", flush=True)
 
 
 @contextmanager
 def bs_query(query_fn, *args, **kwargs):
-    """加锁执行 BaoStock 查询，yield ResultData，超时自动重连
+    """加锁执行 BaoStock 查询，yield ResultData
 
-    用法:
-        with bs_query(bs.query_history_k_data_plus, code, fields, ...) as rs:
-            while rs.next():
-                row = rs.get_row_data()
+    连接断开或超时自动重连，最多重试 MAX_RETRY 次。
     """
     bs_login()
     short_name = query_fn.__name__.replace("query_", "")
@@ -70,22 +71,48 @@ def bs_query(query_fn, *args, **kwargs):
             exc_box[0] = e
 
     with _lock:
-        t = threading.Thread(target=_run, daemon=True)
-        t.start()
-        t.join(timeout=QUERY_TIMEOUT)
-        if t.is_alive():
-            print(f"    [{datetime.now().strftime('%H:%M:%S')}] [BS] 查询超时({QUERY_TIMEOUT}s)，重连...", flush=True)
-            _relogin()
-            raise TimeoutError(f"BaoStock query timeout: {short_name}({sig})")
-        if exc_box[0] is not None:
-            raise exc_box[0]
-        yield result_box[0]
+        last_err = None
+        for attempt in range(1, MAX_RETRY + 1):
+            result_box[0] = None
+            exc_box[0] = None
+
+            t = threading.Thread(target=_run, daemon=True)
+            t.start()
+            t.join(timeout=QUERY_TIMEOUT)
+
+            if t.is_alive():
+                print(f"    [{datetime.now().strftime('%H:%M:%S')}] [BS] 查询超时({QUERY_TIMEOUT}s)，重连(第{attempt}次)...", flush=True)
+                _relogin()
+                last_err = TimeoutError(f"BaoStock query timeout: {short_name}")
+                continue
+
+            if exc_box[0] is not None:
+                err_msg = str(exc_box[0])
+                if "10057" in err_msg or "连接" in err_msg or "socket" in err_msg.lower() or "接收数据" in err_msg:
+                    print(f"    [{datetime.now().strftime('%H:%M:%S')}] [BS] 连接断开，重连(第{attempt}次)...", flush=True)
+                    _relogin()
+                    last_err = exc_box[0]
+                    continue
+                raise exc_box[0]
+
+            # 检查返回结果是否有错误码
+            rs = result_box[0]
+            if hasattr(rs, 'error_code') and rs.error_code != "0" and "login" in rs.error_msg.lower():
+                print(f"    [{datetime.now().strftime('%H:%M:%S')}] [BS] 未登录，重连(第{attempt}次)...", flush=True)
+                _relogin()
+                last_err = RuntimeError(f"BaoStock not logged in: {rs.error_msg}")
+                continue
+
+            yield result_box[0]
+            return
+
+        raise last_err or RuntimeError("BaoStock query failed after retries")
 
 
 def code_to_bs(code: str) -> str:
     """纯数字代码转 BaoStock 格式: '600000' → 'sh.600000'"""
     if code.startswith("920"):
-        return None  # 北交所不支持
+        return None
     if code.startswith(("6", "9")):
         return f"sh.{code}"
     return f"sz.{code}"

src/fetchers/daily.py

@@ -1,10 +1,19 @@
-"""日线行情抓取模块 — 使用 BaoStock
+"""日线行情抓取模块 — 多数据源
+
+数据源优先级：
+  - baostock: BaoStock（默认，含换手率/振幅）
+  - sina:     新浪财经（通过 akshare）
+  - tencent:  腾讯财经
+  - eastmoney: 东方财富
 
 增量抓取：一次本地查询确定每只股票的缺口范围，只请求缺失日期段。
 """
 
 import time
 from datetime import datetime, timedelta
+from concurrent.futures import ThreadPoolExecutor, as_completed
+import threading
+import requests
 import baostock as bs
 from src.baostock_conn import bs_query, code_to_bs, bs_login
 from src.config import get_fetch_config
@@ -12,6 +21,13 @@ from src.db import StockInfo, StockDaily, batch_upsert, get_session, get_stock_c
 from src.fetchers.trading_day import get_trading_days
 from sqlalchemy import select, func, text
 
+VALID_SOURCES = ("baostock", "sina", "tencent", "eastmoney")
+
+_HEADERS = {
+    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+    "Referer": "https://quote.eastmoney.com/",
+}
+
 
 def _clean(val):
     if val is None:
@@ -21,105 +37,32 @@ def _clean(val):
     return val
 
 
-def _analyze_gaps(codes: list[str], start_date: str, end_date: str,
-                  trading_days: list[str]) -> dict[str, list[str]]:
-    """一条SQL统计每只股票每月行情数，与交易日对比找缺口。"""
-    if not trading_days:
-        return {}
+def _fill_derived_fields(rows: list[dict]) -> list[dict]:
+    """补算缺失的振幅、涨跌幅、涨跌额、换手率。
 
-    sd = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:8]}"
-    ed = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}"
+    rows 按日期升序，利用前一行的 close 作为 preclose。
+    """
+    if not rows:
+        return rows
+    for i, row in enumerate(rows):
+        close = row.get("close")
+        high = row.get("high")
+        low = row.get("low")
+        preclose = rows[i - 1]["close"] if i > 0 else None
 
-    t0 = time.time()
+        if close is not None and preclose is not None and float(preclose) != 0:
+            pc = float(preclose)
+            c = float(close)
+            if row.get("pct_change") is None:
+                row["pct_change"] = round((c - pc) / pc * 100, 2)
+            if row.get("change") is None:
+                row["change"] = round(c - pc, 3)
+            if row.get("amplitude") is None and high is not None and low is not None:
+                row["amplitude"] = round((float(high) - float(low)) / pc * 100, 2)
+    return rows
 
-    # 上市日期（进程内缓存，只查一次）
-    ipo_dates = get_ipo_dates()
-    print(f"  [1/3] 上市日期查询完成 {len(ipo_dates)} 只 {time.time()-t0:.1f}s", flush=True)
-
-    # 按月统计每只股票行情数（一条SQL）
-    t1 = time.time()
-    session = get_session()
-    try:
-        sql = text("""
-            SELECT code, DATE_FORMAT(date, '%Y-%m') AS month, COUNT(*) AS cnt
-            FROM stock_daily
-            WHERE date >= :sd AND date <= :ed
-            GROUP BY code, DATE_FORMAT(date, '%Y-%m')
-        """)
-        result = session.execute(sql, {"sd": sd, "ed": ed})
-        # code_month_cnt[code][month] = count
-        code_month_cnt: dict[str, dict[str, int]] = {}
-        for code, month, cnt in result:
-            code_month_cnt.setdefault(code, {})[month] = cnt
-    finally:
-        session.close()
-    print(f"  [2/3] 行情按月统计完成 {time.time()-t1:.1f}s", flush=True)
-
-    # 按月对比找缺口
-    t2 = time.time()
-    td_by_month: dict[str, list[str]] = {}
-    for d in trading_days:
-        td_by_month.setdefault(d[:7], []).append(d)
-
-    gap_codes: set[str] = set()
-    no_ipo_codes: set[str] = set()
-    for month_key, month_days in sorted(td_by_month.items()):
-        for code in codes:
-            if code in gap_codes:
-                continue
-            ipo = ipo_dates.get(code)
-            if not ipo:
-                # stock_info 里没有该股票，跳过避免误判全量缺口
-                no_ipo_codes.add(code)
-                continue
-            if ipo > month_days[-1]:
-                continue
-            expected = [d for d in month_days if d >= ipo]
-            if not expected:
-                continue
-            cnt = code_month_cnt.get(code, {}).get(month_key, 0)
-            if cnt < len(expected):
-                gap_codes.add(code)
-                # 调试：打印触发缺口的月份和计数
-                print(f"    [gap] {code} {month_key} 本地:{cnt} 期望:{len(expected)}", flush=True)
-
-    print(f"  [3/3] 缺口对比完成 缺口股票:{len(gap_codes)} 只 跳过(无上市日期):{len(no_ipo_codes)} 只 {time.time()-t2:.1f}s", flush=True)
-
-    if not gap_codes:
-        return {}
-
-    # 确定缺口范围
-    gaps: dict[str, list[str]] = {}
-    for code in gap_codes:
-        ipo = ipo_dates.get(code)
-        expected = [d for d in trading_days if not ipo or d >= ipo]
-        session = get_session()
-        try:
-            minmax = session.execute(
-                select(func.min(StockDaily.date), func.max(StockDaily.date))
-                .where(StockDaily.code == code)
-                .where(StockDaily.date >= sd)
-                .where(StockDaily.date <= ed)
-            ).fetchone()
-        finally:
-            session.close()
-
-        if minmax and minmax[0]:
-            min_d, max_d = str(minmax[0]), str(minmax[1])
-            front = [d for d in expected if d < min_d]
-            back = [d for d in expected if d > max_d]
-            if front and back:
-                gaps[code] = [expected[0], expected[-1]]
-            elif front:
-                gaps[code] = [front[0], front[-1]]
-            elif back:
-                gaps[code] = [back[0], back[-1]]
-            # else: 数据两端已覆盖，内部缺口属停牌，不重抓
-        else:
-            gaps[code] = [expected[0], expected[-1]]
-
-    return gaps
 
+# ==================== 数据源: BaoStock ====================
 
 def _fetch_baostock(code: str, start_date: str, end_date: str) -> list[dict] | None:
     bs_code = code_to_bs(code)
@@ -164,9 +107,297 @@ def _fetch_baostock(code: str, start_date: str, end_date: str) -> list[dict] | N
         return None
 
 
-def fetch_daily(start_date: str | None = None, end_date: str | None = None):
+# ==================== 数据源: 新浪 (akshare) ====================
+
+def _code_to_sina(code: str) -> str | None:
+    if code.startswith("920"):
+        return None
+    if code.startswith(("6", "9")):
+        return f"sh{code}"
+    return f"sz{code}"
+
+
+def _fetch_sina(code: str, start_date: str, end_date: str) -> list[dict] | None:
+    sina_code = _code_to_sina(code)
+    if not sina_code:
+        return None
+    sd = f"{start_date[:4]}{start_date[4:6]}{start_date[6:8]}"
+    ed = f"{end_date[:4]}{end_date[4:6]}{end_date[6:8]}"
+    try:
+        import akshare as ak
+        df = ak.stock_zh_a_daily(symbol=sina_code, start_date=sd, end_date=ed, adjust="qfq")
+        if df is None or df.empty:
+            return None
+        rows = []
+        for _, r in df.iterrows():
+            date_str = str(r["date"])[:10]
+            close = float(r["close"])
+            open_ = float(r["open"])
+            high = float(r["high"])
+            low = float(r["low"])
+            volume = float(r["volume"]) if "volume" in r else None
+            amount = float(r["amount"]) if "amount" in r else None
+            turnover_rate = float(r["turnover"]) if "turnover" in r else None
+            rows.append({
+                "code": code, "date": date_str,
+                "open": open_, "high": high, "low": low, "close": close,
+                "volume": volume, "turnover": amount,
+                "amplitude": None, "pct_change": None, "change": None,
+                "turnover_rate": turnover_rate,
+            })
+        return rows if rows else None
+    except Exception:
+        return None
+
+
+# ==================== 数据源: 腾讯 ====================
+
+def _code_to_tencent(code: str) -> str | None:
+    if code.startswith("920"):
+        return None
+    if code.startswith(("6", "9")):
+        return f"sh{code}"
+    return f"sz{code}"
+
+
+def _fetch_tencent(code: str, start_date: str, end_date: str) -> list[dict] | None:
+    tc_code = _code_to_tencent(code)
+    if not tc_code:
+        return None
+    sd = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:8]}"
+    ed = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}"
+    try:
+        url = "https://web.ifzq.gtimg.cn/appstock/app/fqkline/get"
+        params = {"param": f"{tc_code},day,{sd},{ed},640,qfq"}
+        r = requests.get(url, params=params, timeout=15)
+        data = r.json().get("data", {})
+        stock_data = data.get(tc_code, {})
+        klines = stock_data.get("qfqday") or stock_data.get("day")
+        if not klines:
+            return None
+        rows = []
+        for k in klines:
+            # [date, open, close, high, low, volume]
+            date_str = k[0]
+            open_ = float(k[1])
+            close = float(k[2])
+            high = float(k[3])
+            low = float(k[4])
+            volume = float(k[5]) if len(k) > 5 else None
+            rows.append({
+                "code": code, "date": date_str,
+                "open": open_, "high": high, "low": low, "close": close,
+                "volume": volume, "turnover": None,
+                "amplitude": None, "pct_change": None, "change": None,
+                "turnover_rate": None,
+            })
+        return rows if rows else None
+    except Exception:
+        return None
+
+
+# ==================== 数据源: 东方财富 ====================
+
+def _code_to_eastmoney(code: str) -> str | None:
+    if code.startswith("920"):
+        return None
+    if code.startswith(("6", "9")):
+        return f"1.{code}"
+    return f"0.{code}"
+
+
+def _fetch_eastmoney(code: str, start_date: str, end_date: str) -> list[dict] | None:
+    em_code = _code_to_eastmoney(code)
+    if not em_code:
+        return None
+    sd = f"{start_date[:4]}{start_date[4:6]}{start_date[6:8]}"
+    ed = f"{end_date[:4]}{end_date[4:6]}{end_date[6:8]}"
+    try:
+        url = "https://push2his.eastmoney.com/api/qt/stock/kline/get"
+        params = {
+            "secid": em_code,
+            "fields1": "f1,f2,f3,f4,f5,f6",
+            "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61",
+            "klt": 101, "fqt": 1, "beg": sd, "end": ed,
+            "ut": "fa5fd1943c7b386f172d6893dbfba10b",
+        }
+        r = requests.get(url, params=params, headers=_HEADERS, timeout=15)
+        data = r.json().get("data") or {}
+        klines = data.get("klines") or []
+        if not klines:
+            return None
+        rows = []
+        for line in klines:
+            # date,open,close,high,low,volume,amount,amplitude,pct_change,change,turnover_rate
+            parts = line.split(",")
+            rows.append({
+                "code": code, "date": parts[0],
+                "open": float(parts[1]), "high": float(parts[3]),
+                "low": float(parts[4]), "close": float(parts[2]),
+                "volume": float(parts[5]), "turnover": float(parts[6]),
+                "amplitude": float(parts[7]) if parts[7] != "" else None,
+                "pct_change": float(parts[8]) if parts[8] != "" else None,
+                "change": float(parts[9]) if parts[9] != "" else None,
+                "turnover_rate": float(parts[10]) if parts[10] != "" else None,
+            })
+        return rows if rows else None
+    except Exception:
+        return None
+
+
+# ==================== 数据源分发 ====================
+
+_SOURCE_FN = {
+    "baostock": _fetch_baostock,
+    "sina": _fetch_sina,
+    "tencent": _fetch_tencent,
+    "eastmoney": _fetch_eastmoney,
+}
+
+_SOURCE_LABEL = {
+    "baostock": "BaoStock",
+    "sina": "新浪",
+    "tencent": "腾讯",
+    "eastmoney": "东方财富",
+}
+
+
+# ==================== 缺口分析 ====================
+
+def _analyze_gaps(codes: list[str], start_date: str, end_date: str,
+                  trading_days: list[str]) -> dict[str, list[str]]:
+    """一条SQL统计每只股票每月行情数，与交易日对比找缺口。"""
+    if not trading_days:
+        return {}
+
+    sd = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:8]}"
+    ed = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}"
+
+    t0 = time.time()
+
+    # 上市日期（进程内缓存，只查一次）
+    ipo_dates = get_ipo_dates()
+    print(f"  [1/3] 上市日期查询完成 {len(ipo_dates)} 只 {time.time()-t0:.1f}s", flush=True)
+
+    # 按月统计每只股票行情数（一条SQL）
+    t1 = time.time()
+    session = get_session()
+    try:
+        sql = text("""
+            SELECT code, DATE_FORMAT(date, '%Y-%m') AS month, COUNT(*) AS cnt
+            FROM stock_daily
+            WHERE date >= :sd AND date <= :ed
+            GROUP BY code, DATE_FORMAT(date, '%Y-%m')
+        """)
+        result = session.execute(sql, {"sd": sd, "ed": ed})
+        code_month_cnt: dict[str, dict[str, int]] = {}
+        for code, month, cnt in result:
+            code_month_cnt.setdefault(code, {})[month] = cnt
+    finally:
+        session.close()
+    print(f"  [2/3] 行情按月统计完成 {time.time()-t1:.1f}s", flush=True)
+
+    # 按月对比找缺口
+    t2 = time.time()
+    td_by_month: dict[str, list[str]] = {}
+    for d in trading_days:
+        td_by_month.setdefault(d[:7], []).append(d)
+
+    gap_codes: set[str] = set()
+    no_ipo_codes: set[str] = set()
+    for month_key, month_days in sorted(td_by_month.items()):
+        for code in codes:
+            if code in gap_codes:
+                continue
+            ipo = ipo_dates.get(code)
+            if not ipo:
+                no_ipo_codes.add(code)
+                continue
+            if ipo > month_days[-1]:
+                continue
+            expected = [d for d in month_days if d >= ipo]
+            if not expected:
+                continue
+            cnt = code_month_cnt.get(code, {}).get(month_key, 0)
+            if cnt < len(expected):
+                gap_codes.add(code)
+
+    print(f"  [3/3] 缺口对比完成 缺口股票:{len(gap_codes)} 只 跳过(无上市日期):{len(no_ipo_codes)} 只 {time.time()-t2:.1f}s", flush=True)
+
+    if not gap_codes:
+        return {}
+
+    # 确定缺口范围
+    gaps: dict[str, list[str]] = {}
+    for code in gap_codes:
+        ipo = ipo_dates.get(code)
+        expected = [d for d in trading_days if not ipo or d >= ipo]
+        session = get_session()
+        try:
+            minmax = session.execute(
+                select(func.min(StockDaily.date), func.max(StockDaily.date))
+                .where(StockDaily.code == code)
+                .where(StockDaily.date >= sd)
+                .where(StockDaily.date <= ed)
+            ).fetchone()
+        finally:
+            session.close()
+
+        if minmax and minmax[0]:
+            min_d, max_d = str(minmax[0]), str(minmax[1])
+            front = [d for d in expected if d < min_d]
+            back = [d for d in expected if d > max_d]
+            if front and back:
+                gaps[code] = [expected[0], expected[-1]]
+            elif front:
+                gaps[code] = [front[0], front[-1]]
+            elif back:
+                gaps[code] = [back[0], back[-1]]
+            # else: 数据两端已覆盖，内部缺口属停牌，不重抓
+        else:
+            gaps[code] = [expected[0], expected[-1]]
+
+    return gaps
+
+
+# ==================== 单股票抓取+写入 ====================
+
+_print_lock = threading.Lock()
+
+
+def _fetch_and_save(code: str, gap_start: str, gap_end: str,
+                    source_key: str) -> tuple[str, str, int, bool]:
+    """抓取单只股票并写入，返回 (code, source_label, row_count, success)"""
+    fetch_fn = _SOURCE_FN[source_key]
+    label = _SOURCE_LABEL[source_key]
+    try:
+        rows = fetch_fn(code, gap_start, gap_end)
+        if rows is not None:
+            rows = _fill_derived_fields(rows)
+            batch_upsert(StockDaily, rows, ["code", "date"])
+            return code, label, len(rows), True
+        return code, label, 0, False
+    except Exception:
+        return code, label, 0, False
+
+
+# ==================== 主函数 ====================
+
+def fetch_daily(start_date: str | None = None, end_date: str | None = None,
+                source: str = "baostock"):
+    # 确定使用的数据源列表
+    if source == "all":
+        sources = list(VALID_SOURCES)
+    elif source in VALID_SOURCES:
+        sources = [source]
+    else:
+        print(f"  不支持的数据源 {source}，可选: {', '.join(VALID_SOURCES)}, all", flush=True)
+        return
+
+    source_names = ", ".join(_SOURCE_LABEL[s] for s in sources)
+    workers = len(sources)
+
     cfg = get_fetch_config()
-    delay = cfg.get("delay", 0.1)
 
     codes = get_stock_codes()
     if not codes:
@@ -178,12 +409,13 @@ def fetch_daily(start_date: str | None = None, end_date: str | None = None):
     if start_date is None:
         start_date = (datetime.now() - timedelta(days=30)).strftime("%Y%m%d")
 
-    bs_login()
+    if "baostock" in sources:
+        bs_login()
 
     # 优先使用本地交易日历；若覆盖不完整则自动补齐
     trading_days = get_trading_days(start_date, end_date)
     td_count = len(trading_days)
-    print(f"  交易日历: {start_date} ~ {end_date} 共 {td_count} 个交易日", flush=True)
+    print(f"  [数据源:{source_names}] [并发:{workers}] 交易日历: {start_date} ~ {end_date} 共 {td_count} 个交易日", flush=True)
 
     # 排除未上市股票
     ed_fmt = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}"
@@ -212,41 +444,67 @@ def fetch_daily(start_date: str | None = None, end_date: str | None = None):
         print("  所有股票数据已完整，无需抓取", flush=True)
         return
 
-    print(f"正在抓取日线行情 {start_date} ~ {end_date}，{total} 只需更新...", flush=True)
+    print(f"正在抓取日线行情({source_names}) {start_date} ~ {end_date}，{total} 只需更新，并发:{workers}...", flush=True)
+
+    # 为每只股票分配数据源（轮询）
+    gap_items = list(gaps.items())
+    tasks = []
+    for i, (code, g) in enumerate(gap_items):
+        src_key = sources[i % len(sources)]
+        gap_start = g[0].replace("-", "")
+        gap_end = g[-1].replace("-", "")
+        tasks.append((code, gap_start, gap_end, g, src_key))
 
     success = 0
     fail = 0
     nodata_count = 0
+    done = 0
     t_start = time.time()
 
-    for i, code in enumerate(gaps):
-        g = gaps[code]
-        gap_start = g[0].replace("-", "")
-        gap_end = g[-1].replace("-", "")
-
-        t0 = time.time()
-        rows = _fetch_baostock(code, gap_start, gap_end)
-        t_fetch = time.time() - t0
-        t1 = time.time()
-        if rows is not None:
-            try:
-                batch_upsert(StockDaily, rows, ["code", "date"])
+    if workers == 1:
+        # 单数据源串行
+        for code, gap_start, gap_end, g, src_key in tasks:
+            t0 = time.time()
+            code, label, row_count, ok = _fetch_and_save(code, gap_start, gap_end, src_key)
+            t_fetch = time.time() - t0
+            done += 1
+            if ok:
                 success += 1
-            except Exception as e:
-                print(f"  {code} 写入失败: {e}", flush=True)
+            elif row_count == 0:
+                nodata_count += 1
+            else:
                 fail += 1
-                continue
-        else:
-            nodata_count += 1
-        t_write = time.time() - t1
+            elapsed = time.time() - t_start
+            avg = elapsed / done
+            eta = avg * (total - done)
+            print(f"  [{done}/{total}] {code} [{label}] 缺口:{g[0]}~{g[-1]} "
+                  f"耗时:{t_fetch:.1f}s 行数:{row_count} "
+                  f"成功:{success} 剩余:{eta:.0f}s", flush=True)
+    else:
+        # 多数据源并发
+        with ThreadPoolExecutor(max_workers=workers) as pool:
+            future_map = {}
+            for code, gap_start, gap_end, g, src_key in tasks:
+                f = pool.submit(_fetch_and_save, code, gap_start, gap_end, src_key)
+                future_map[f] = (code, g, src_key)
 
-        elapsed = time.time() - t_start
-        avg = elapsed / (i + 1)
-        eta = avg * (total - i - 1)
-        print(f"  [{i+1}/{total}] {code} 缺口:{g[0]}~{g[-1]} "
-              f"网络:{t_fetch:.1f}s 写入:{t_write:.1f}s 行数:{len(rows) if rows else 0} "
-              f"成功:{success} 剩余:{eta:.0f}s", flush=True)
-        time.sleep(delay)
+            for future in as_completed(future_map):
+                code, g, src_key = future_map[future]
+                code_r, label, row_count, ok = future.result()
+                done += 1
+                if ok:
+                    success += 1
+                elif row_count == 0:
+                    nodata_count += 1
+                else:
+                    fail += 1
+                elapsed = time.time() - t_start
+                avg = elapsed / done
+                eta = avg * (total - done)
+                with _print_lock:
+                    print(f"  [{done}/{total}] {code_r} [{label}] 缺口:{g[0]}~{g[-1]} "
+                          f"行数:{row_count} 成功:{success} 剩余:{eta:.0f}s", flush=True)
 
     total_time = time.time() - t_start
-    print(f"  日线行情抓取完成，成功:{success} 失败:{fail} 停牌:{nodata_count} 总耗时:{total_time:.1f}s", flush=True)
+    print(f"  日线行情抓取完成 数据源:{source_names} 并发:{workers} "
+          f"成功:{success} 失败:{fail} 无数据:{nodata_count} 总耗时:{total_time:.1f}s", flush=True)

src/main.py

@@ -28,6 +28,9 @@ def main():
     parser.add_argument("--stock-info", action="store_true", help="抓取股票列表")
     parser.add_argument("--trading-day", action="store_true", help="抓取交易日历")
     parser.add_argument("--daily", action="store_true", help="抓取日线行情")
+    parser.add_argument("--source", type=str, default="baostock",
+                        choices=["baostock", "sina", "tencent", "eastmoney", "all"],
+                        help="日线数据源（默认baostock，all=全部并发）")
     parser.add_argument("--financial", action="store_true", help="抓取季频财务指标")
     parser.add_argument("--dividend", action="store_true", help="抓取分红送转")
     parser.add_argument("--intraday", action="store_true", help="抓取分钟K线行情")
@@ -67,7 +70,8 @@ def main():
 
         if args.daily:
             from src.fetchers.daily import fetch_daily
-            fetch_daily(start_date=args.start_date, end_date=args.end_date)
+            fetch_daily(start_date=args.start_date, end_date=args.end_date,
+                        source=args.source)
 
         if args.financial:
             from src.fetchers.financial import fetch_financial