feat(buy_record): 多商家并发采集购买记录与全局请求限速优化

- 扩展监控商家列表由单商家 MERCHANT_ID 到多商家 MERCHANT_IDS 支持
- 引入线程池并发调度，提升购买记录采集吞吐率至全局限速近5次/秒
- 用全局令牌桶实现多线程安全的请求间隔限速，避免突刺防封
- 监控队列刷新支持多商家在售商品合并去重及动态增删商品
- 订单判重及采集状态相关内存结构加锁保证多线程环境数据安全
- 采集主流程改为主线程派发任务，worker 并发执行采集并更新状态
- 对订单滑动窗口对齐判重加锁，避免多线程竞争导致数据不一致
- 新增在途采集集合，确保同一商品同一时刻仅有一个worker采集
- 调整监控商品信息记录，增加商家名和监控商品数日志输出
- 失败采集任务自动延迟重试避免阻塞调度
- 常驻任务实现更短调度节奏，及时响应快车时段采集需求

deca_spider/buy_record_analysis/buy_record_spider.py

@@ -7,7 +7,7 @@
 功能：
   一个进程干两件事（方案A，2026/08/08 合并）：
   (1) 每分钟复用 daily 的免 token home/search 拉全站在售，落 deca_onsale_* 三张表（供 deca_on_sale_report 查库出报告）；
-  (2) 从库里取本商家(MERCHANT_ID)在售 code，采其购买记录写 deca_buy_record。
+  (2) 从库里取各监控商家(MERCHANT_IDS)在售 code，采其购买记录写 deca_buy_record。
   购买记录按「近 10 条购买的时间跨度」动态调节采集频率——
   卖得快（跨度短）密采、卖得慢（跨度长）稀采；商品下线、售卖结束或售罄自动停采（含白名单模式）。
   数据写 deca_buy_record；去重用「倒序滑动窗口序列对齐」：purchaseRecords 是最近 10 条按时间倒序、
@@ -49,6 +49,8 @@ import os
 import re
 import sys
 import time
+import threading
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, time as dtime
 
 # 挂靠项目根：复用核心签名/token/请求/代理层，让 application.yml、token.json 生效
@@ -67,8 +69,15 @@ from on_sale.deca_on_sale_daily_spider import get_shop_list, get_onsale_products
 # ==================== 配置 ====================
 core.USE_PROXY = True               # 详情接口(groupbuy/detail)走快代理隧道；全链路免 token
 
-MERCHANT_ID = "881226408"           # 监控商家：采其在售商品的购买记录（WATCH_CODES 为空时生效）
-WATCH_CODES = []                    # 指定商品白名单：非空则只盯这些 code、跳过在售发现；空则走 MERCHANT_ID 全在售（默认）
+# 监控商家列表：采这些商家在售商品的购买记录（WATCH_CODES 为空时生效）。
+# 2026/08/24 由单商家扩为多商家并行监控：三家同走「查库拿在售 code → 自适应频率采购买记录」同一套逻辑，
+# 各商品按自身售卖节奏独立调频，互不影响；快车时段/快车上限等参数对所有商家通用（按份数与滑窗跨度自适应）。
+MERCHANT_IDS = [
+    "881226408",   # 魔都兄弟球星卡
+    "274584650",   # 卡皇拆卡（2026/08/24 新增）
+    "538252487",   # 尼卡拆卡（2026/08/24 新增）
+]
+WATCH_CODES = []                    # 指定商品白名单：非空则只盯这些 code、跳过在售发现；空则走 MERCHANT_IDS 全在售（默认）
 ONSALE_INGEST_SEC = 60             # 全站在售落库 + 刷新监控队列间隔（常规时段 1 分钟，复用 daily 免 token home/search）
 # 快车高发时段（直播拆卡，通常 20:00~次日 06:00）专用：这类小批量拼团（本商家 31 份）上架后 2~3 分钟即售罄，
 # 寿命只比 60s 发现节奏大一个数量级，偶尔整场卡在两次全站扫描的缝隙里被漏掉（实测 2026/08/22 01:45 一辆 31 份车漏采，
@@ -85,8 +94,16 @@ MAX_INTERVAL_SEC = 600             # 单商品购买记录最大采集间隔（1
 # 否则「刚上架 span=0 → 排 600s」会让 3~8 分钟售罄的拼团在两次采集间隙整场漏采(2026/08/10 修复)
 FAST_LANE_MAX_COUNT = 100          # 份数<=此值视为「快车」（本商家快车=30份、大车=313+），涵盖未来 20/40/50 份的小批量快车
 FAST_LANE_MAX_INTERVAL_SEC = 20    # 快车最大采集间隔（实测最快 10 份滑窗~64s，20s 留约 3x 余量）
-GLOBAL_MIN_GAP_SEC = 0.2           # 全局相邻两次详情请求最小间隔（≈每秒 5 次上限）
+GLOBAL_MIN_GAP_SEC = 0.2           # 全局相邻两次详情请求最小间隔（≈每秒 5 次上限，防封的硬速率天花板）
+# 并发采集（2026/08/24）：实测单条详情请求 ~1~2.7s（含长尾），串行吞吐仅 0.4~0.9 次/秒，
+# 远低于 GLOBAL_MIN_GAP_SEC 允许的 5 次/秒——瓶颈是请求延迟(latency-bound)而非节流。多商家(21+ 在售)后，
+# 单线程一轮轮完 N 个商品要 N×延迟，追不上 31 份快车(最快 45~65s 售罄、翻窗 3 次)，实测魔都近 7 天捕获率
+# 均值仅 0.83、快车低至 0.35。故改线程池并发：把利用率从 ~0.5 次/秒填到接近 5 次/秒(约 10x)，
+# **但不抬高 5 次/秒的天花板**——峰值压力上限不变、防封风险几乎不增。worker 数需略大于「节流上限×平均延迟」
+# 才能填满时隙(5×1.5≈8)，长尾偶发占用也不至饿死；真正的速率上限由令牌桶(_slot_lock)守住，worker 多只是填满、不会超速。
+MAX_WORKERS = 8                    # 采集线程池大小（并发发详情请求，实际速率仍受 GLOBAL_MIN_GAP_SEC 令牌桶封顶）
 IDLE_SLEEP_SEC = 1                 # 主循环空转 sleep（没到点时）
+DISPATCH_SLEEP_SEC = 0.2           # 主循环派发轮询间隔（多线程下调度快节奏，及时把到点商品派给空闲 worker）
 DEDUP_TOLERANCE_SEC = 70           # 秒级基准判重容差：同(商品,user,份数)下反推时间戳相差<=此值视为同一笔；小时/天级按 _TOL_BY_UNIT 放大
 
 TABLE = "deca_buy_record"
@@ -114,8 +131,18 @@ _REL_RE = re.compile(r"^(\d+)\s*(秒|分钟|小时|天)前$")
 # 秒/分钟保持小容差(<同用户多笔的最小间隔~2min)以区分真实多笔；小时/天放大到略大于桶宽(3600/86400)。
 _TOL_BY_UNIT = {"秒": DEDUP_TOLERANCE_SEC, "分钟": 90, "小时": 3660, "天": 90000}
 
-# 全局请求节流：记录上次请求时刻，本轮请求前 sleep 到最小间隔
-_last_req_ts = 0.0
+# 全局请求节流（令牌桶/时隙预约）：多线程并发下用锁只保护「领下一个可发时隙」这步(μs 级)，sleep 在锁外，
+# 从而 N 个 worker 各领一个错开 GLOBAL_MIN_GAP_SEC 的时隙、稳定逼近 5 次/秒并发而不突刺。_next_slot=下一个可发时刻。
+_slot_lock = threading.Lock()
+_next_slot = 0.0
+
+# 调度状态锁：保护 monitored（监控队列）与 _inflight（在途 code 集合）的读改。这两个结构主线程(派发/刷新)与
+# worker(回写 next_run_ts/移出) 都会碰，须串行化；锁内只做内存操作(快)，绝不含网络/DB/长 sleep，故无死锁风险。
+_state_lock = threading.Lock()
+
+# 在途集合：正被 worker 采集的 code。派发前检查，保证「同一商品同一时刻只有一个 worker」——
+# 从而 _order_idx / _last_window 这类按 code 分片的结构对同一 code 无并发，不同 code 是不同 dict key(CPython 下并发安全)，无需额外加锁。
+_inflight: set = set()
 
 # 订单去重内存索引：{(product_code, user_id, card_count): [已入库订单的反推时间戳...]}
 # 判重靠「反推时间戳 ± 动态容差(_TOL_BY_UNIT)」——同一笔订单变老后的漂移能吸附回锚点，不同笔可区分
@@ -209,10 +236,12 @@ def is_sale_ended(data: dict, now_ts: int) -> tuple[bool, str]:
 
 
 def throttled_do_request(log, path: str, body: dict, need_auth: bool = False) -> dict | None:
-    """核心签名 POST 请求 + 全局最小间隔节流。
+    """核心签名 POST 请求 + 全局令牌桶限速（多线程安全）。
 
-    做请求前 sleep 到 `_last_req_ts + GLOBAL_MIN_GAP_SEC`，防止多商品并发瞬时打满。
-    详情/在售列表都走这里，共用同一根节流线。
+    用「时隙预约」实现全局限速：锁内只算出本次请求应发的时刻(领一个较上次错开 GLOBAL_MIN_GAP_SEC 的
+    时隙、并推进 _next_slot)，随后在**锁外** sleep 到该时刻再发请求。故锁只保护 μs 级的时隙计算、
+    不串行化整个请求，N 个 worker 能各领错峰时隙、稳定逼近 5 次/秒并发而不突刺——峰值速率天花板不变。
+    详情/在售列表都走这里，共用同一根令牌桶。
 
     Args:
         log: 日志对象。
@@ -224,14 +253,15 @@ def throttled_do_request(log, path: str, body: dict, need_auth: bool = False) ->
     Returns:
         dict | None: 响应 JSON。
     """
-    global _last_req_ts
-    now = time.time()
-    wait = _last_req_ts + GLOBAL_MIN_GAP_SEC - now
+    global _next_slot
+    with _slot_lock:
+        now = time.time()
+        start = now if now >= _next_slot else _next_slot   # 领本次时隙：不早于「现在」也不早于「上次时隙+间隔」
+        _next_slot = start + GLOBAL_MIN_GAP_SEC             # 推进下一个可发时隙，后续请求依次错峰
+    wait = start - time.time()
     if wait > 0:
-        time.sleep(wait)
-    resp = core.do_request(log, path, body, need_auth=need_auth)
-    _last_req_ts = time.time()
-    return resp
+        time.sleep(wait)                                    # 锁外 sleep，不阻塞其他 worker 领时隙
+    return core.do_request(log, path, body, need_auth=need_auth)
 
 
 def compute_next_interval(span_sec: int, card_count: int | None = None) -> int:
@@ -270,8 +300,10 @@ def load_order_index(pool, code: str):
     """
     rows = pool.select_all(
         f"SELECT user_id, card_count, purchased_at_ts FROM {TABLE} WHERE product_code=%s", (code,)) or []
-    for uid, cc, ts in rows:
-        _order_idx.setdefault((code, str(uid), cc), []).append(int(ts))
+    # DB 查询在锁外；仅把「写全局 _order_idx」这步收进 _state_lock，与 worker(其他 code)的并发追加互斥
+    with _state_lock:
+        for uid, cc, ts in rows:
+            _order_idx.setdefault((code, str(uid), cc), []).append(int(ts))
 
 
 def is_duplicate_order(code: str, uid: str, cc: int, ts: int, tol: int) -> bool:
@@ -518,58 +550,67 @@ def poll_product(log, pool, code: str, meta: dict) -> tuple:
     ts_old = rows[-1]["purchased_at_ts"]
     span_sec = max(0, ts_new - ts_old)
 
-    # 主判重：倒序滑动窗口序列对齐——只有从顶部新增的 k 条才是本轮新订单
+    # 主判重：倒序滑动窗口序列对齐——只有从顶部新增的 k 条才是本轮新订单。
+    # 判重段落收进 _state_lock（纯内存、μs 级）：与主线程 refresh 对同一 code 的 _last_window/_order_idx
+    # 增删互斥，避免多线程下同键并发读改。上方网络请求、下方 DB 写均在锁外，不被串行化、不拖慢吞吐。
     curr_keys = [(r["user_id"], r["card_count"]) for r in rows]
-    prev_keys = _last_window.get(code)
-    k, reliable = align_new_orders(prev_keys, curr_keys)
-
-    if reliable:
-        # 对齐可靠：顶部 k 条直接入库，不再走时间容差（否则会误杀同用户短间隔的多笔真实订单）
-        new_rows = rows[:k]
-        for row in new_rows:
-            _order_idx.setdefault((code, row["user_id"], row["card_count"]), []).append(row["purchased_at_ts"])
-    else:
-        # 无法对齐（进程重启首轮 / 窗口整体换新）：整窗退回动态容差 + DB 唯一键兜底
-        if prev_keys:
-            log.warning(f"[{code}] 窗口无重叠，疑似漏采（采集间隔内卖出≥{len(rows)}笔），建议提高采集频率")
-        new_rows = []
-        for row in rows:
-            tol = dedup_tolerance(row["purchased_at_text"])
-            if is_duplicate_order(code, row["user_id"], row["card_count"], row["purchased_at_ts"], tol):
-                continue
-            new_rows.append(row)
-            _order_idx.setdefault((code, row["user_id"], row["card_count"]), []).append(row["purchased_at_ts"])
+    with _state_lock:
+        prev_keys = _last_window.get(code)
+        k, reliable = align_new_orders(prev_keys, curr_keys)
+        if reliable:
+            # 对齐可靠：顶部 k 条直接入库，不再走时间容差（否则会误杀同用户短间隔的多笔真实订单）
+            new_rows = rows[:k]
+            for row in new_rows:
+                _order_idx.setdefault((code, row["user_id"], row["card_count"]), []).append(row["purchased_at_ts"])
+        else:
+            # 无法对齐（进程重启首轮 / 窗口整体换新）：整窗退回动态容差 + DB 唯一键兜底
+            if prev_keys:
+                log.warning(f"[{code}] 窗口无重叠，疑似漏采（采集间隔内卖出≥{len(rows)}笔），建议提高采集频率")
+            new_rows = []
+            for row in rows:
+                tol = dedup_tolerance(row["purchased_at_text"])
+                if is_duplicate_order(code, row["user_id"], row["card_count"], row["purchased_at_ts"], tol):
+                    continue
+                new_rows.append(row)
+                _order_idx.setdefault((code, row["user_id"], row["card_count"]), []).append(row["purchased_at_ts"])
+        _last_window[code] = curr_keys          # 更新窗口快照，供下轮对齐（锁内，与 refresh 的 pop 互斥）
 
     if new_rows:
-        # DB 唯一键(含 purchased_at_ts)兜底防并发/异常重复；主判重靠窗口对齐已完成
+        # DB 唯一键(含 purchased_at_ts)兜底防并发/异常重复；主判重靠窗口对齐已完成。DB 写在锁外，各线程借还连接池连接并发写。
         pool.insert_many(table=TABLE, data_list=new_rows, ignore=True)
 
-    _last_window[code] = curr_keys              # 更新窗口快照，供下轮对齐
     return span_sec, len(new_rows), ended, end_reason
 
 
 # ==================== 主循环 ====================
-def _refresh_monitored(log, pool, monitored: dict, merchant_id: str, now: float):
+def _refresh_monitored(log, pool, monitored: dict, merchant_ids: list, now: float):
     """刷新监控队列：新商品立即入队，下线商品移出队列。
 
     WATCH_CODES 非空 → 只监控白名单里的 code（跳过在售发现，无下线逻辑）。
-    WATCH_CODES 为空 → 查库拿 merchant_id 的在售商品（由 ingest_onsale 落库），动态增删。
+    WATCH_CODES 为空 → 查库拿 merchant_ids 各商家的在售商品（由 ingest_onsale 落库），合并去重后动态增删。
     每个商品首次纳入监控时，从库里恢复其订单去重索引（避免重启后重复入库）。
 
+    多商家（2026/08/24）：live_codes 取所有监控商家在售 code 的并集；因监控队列里的 code 只会从这些
+    商家的在售列表入队，故「不在并集内」即代表下线，移出逻辑对多商家天然正确、无需按商家分别对账。
+
     Args:
         log: 日志对象。
         pool: MySQL 连接池（用于恢复订单去重索引）。
         monitored (dict): 监控状态字典 {code: {next_run_ts, interval, title, ...}}，就地更新。
-        merchant_id (str): 商家 ID。
+        merchant_ids (list[str]): 商家 ID 列表。
         now (float): 当前时间戳。
     """
     if WATCH_CODES:
-        # 单/多商品白名单模式：只挂想盯的 code，不动其他
-        for code in WATCH_CODES:
-            if code in _ended_codes:
-                continue                      # 售卖已结束，不再重新纳入监控
-            if code not in monitored:
-                load_order_index(pool, code)  # 从库恢复订单去重索引
+        # 单/多商品白名单模式：只挂想盯的 code，不动其他。
+        # 先锁内挑出待新增的 code（读 monitored 需与 worker 互斥），锁外做 DB 恢复索引，再锁内加入监控。
+        with _state_lock:
+            to_add = [c for c in WATCH_CODES if c not in _ended_codes and c not in monitored]
+        for code in to_add:
+            load_order_index(pool, code)      # 从库恢复订单去重索引（DB 在锁外，内部自带锁写 _order_idx）
+        with _state_lock:
+            for code in to_add:
+                if code in monitored:         # 双检：避免与并发刷新重复加入
+                    continue
                 monitored[code] = {
                     "next_run_ts": now,
                     "interval": MIN_INTERVAL_SEC,
@@ -580,17 +621,31 @@ def _refresh_monitored(log, pool, monitored: dict, merchant_id: str, now: float)
                     "merchant_name": None,
                 }
                 log.info(f"[+] 纳入监控 {code} | 白名单模式 | 已恢复历史订单索引")
-        log.info(f"当前监控商品数：{len(monitored)}（白名单模式，共 {len(WATCH_CODES)} 个 code）")
+            cnt = len(monitored)
+        log.info(f"当前监控商品数：{cnt}（白名单模式，共 {len(WATCH_CODES)} 个 code）")
         return
 
-    products = fetch_on_sale_products(log, merchant_id, pool)
+    # 合并各监控商家的在售商品（DB 读，锁外；按 code 去重，正常不同商家不会共享 code）
+    products, seen = [], set()
+    for mid in merchant_ids:
+        for p in fetch_on_sale_products(log, mid, pool):
+            if p["code"] in seen:
+                continue
+            seen.add(p["code"])
+            products.append(p)
     live_codes = {p["code"] for p in products}
-    # 新增：立即到点采
-    for p in products:
-        if p["code"] in _ended_codes:
-            continue                          # 售卖已结束，跳过（正常也会自然从在售列表消失）
-        if p["code"] not in monitored:
-            load_order_index(pool, p["code"])  # 从库恢复订单去重索引
+
+    # 锁内挑出待新增商品（未结束、未在监控）；锁外逐个从库恢复订单索引；再锁内应用增删。
+    with _state_lock:
+        new_products = [p for p in products
+                        if p["code"] not in _ended_codes and p["code"] not in monitored]
+    for p in new_products:
+        load_order_index(pool, p["code"])     # 从库恢复订单去重索引（DB 在锁外）
+    with _state_lock:
+        # 新增：立即到点采
+        for p in new_products:
+            if p["code"] in monitored:        # 双检：避免与并发刷新重复加入
+                continue
             monitored[p["code"]] = {
                 "next_run_ts": now,
                 "interval": MIN_INTERVAL_SEC,
@@ -600,14 +655,15 @@ def _refresh_monitored(log, pool, monitored: dict, merchant_id: str, now: float)
                 "merchant_user_id": p["merchant_user_id"],
                 "merchant_name": p["merchant_name"],
             }
-            log.info(f"[+] 纳入监控 {p['code']} | {p['title']} | 售 {p['sold_count']}/{p['card_count']}")
-    # 下线：移出
-    for code in list(monitored.keys()):
-        if code not in live_codes:
-            log.info(f"[-] 移出监控 {code} | {monitored[code].get('title')}")
-            del monitored[code]
-            _last_window.pop(code, None)         # 下线同步清窗口快照，重新上架时按重启首轮兜底
-    log.info(f"当前监控商品数：{len(monitored)}")
+            log.info(f"[+] 纳入监控 {p['code']} | {p['merchant_name']} | {p['title']} | 售 {p['sold_count']}/{p['card_count']}")
+        # 下线：移出（不在任一监控商家在售并集内即下线）。in-flight 的 worker 完成时会双检 code 是否仍在监控，故安全。
+        for code in list(monitored.keys()):
+            if code not in live_codes:
+                log.info(f"[-] 移出监控 {code} | {monitored[code].get('title')}")
+                del monitored[code]
+                _last_window.pop(code, None)     # 下线同步清窗口快照，重新上架时按重启首轮兜底
+        cnt = len(monitored)
+    log.info(f"当前监控商品数：{cnt}（商家 {len(merchant_ids)} 家）")
 
 
 def _current_ingest_interval(now: datetime) -> int:
@@ -627,9 +683,51 @@ def _current_ingest_interval(now: datetime) -> int:
     return FAST_WINDOW_INGEST_SEC if in_fast_window else ONSALE_INGEST_SEC
 
 
+def _poll_worker(log, pool, monitored: dict, code: str):
+    """线程池 worker：并发采集单个商品一次并回写调度状态（多线程安全）。
+
+    先在 _state_lock 内取该 code 的快照 meta（若已被 refresh 下线则跳过），随后**锁外**执行 poll_product
+    （含网络请求与 DB 写，是耗时大头，故不占锁）；完成后按结果在锁内更新 next_run_ts，或（售罄/到期）移出监控
+    并拉黑。无论成败，finally 都从 _inflight 移除该 code，保证同一商品下一轮可再次被派发。
+
+    Args:
+        log: 日志对象。
+        pool: MySQL 连接池（线程安全，各 worker 借还各自连接）。
+        monitored (dict): 监控状态字典（共享，读改均在 _state_lock 内）。
+        code (str): 本次采集的商品编码。
+    """
+    try:
+        with _state_lock:
+            meta = monitored.get(code)
+            if meta is None:
+                return                       # 已被 refresh 移出（下线），本次跳过
+            meta = dict(meta)                # 快照，避免采集期间被并发改动
+        span_sec, n_rows, ended, end_reason = poll_product(log, pool, code, meta)
+        with _state_lock:
+            if ended:
+                # 售卖结束：本轮收尾记录已在 poll_product 入库，这里移出监控 + 拉黑，避免死循环重采固定的最后 10 条
+                _ended_codes.add(code)
+                _last_window.pop(code, None)     # 清窗口快照，避免复活时误对齐
+                monitored.pop(code, None)
+                log.info(f"[{code}] 售卖已结束（{end_reason}），停止采集并移出监控 | {meta.get('title')}")
+            elif code in monitored:          # 仍在监控（未被 refresh 下线）才回写下次采集时刻
+                interval = compute_next_interval(span_sec, monitored[code].get("card_count"))
+                monitored[code]["next_run_ts"] = time.time() + interval
+                monitored[code]["interval"] = interval
+                log.info(f"[{code}] 响应 {n_rows} 条 | 10条跨度 {span_sec}s | 下次 {interval}s 后 | {monitored[code]['title']}")
+    except Exception as e:
+        log.error(f"[{code}] 采集失败: {e}")
+        with _state_lock:
+            if code in monitored:
+                monitored[code]["next_run_ts"] = time.time() + 60   # 失败退避 60s，避免异常商品高频重试阻塞
+    finally:
+        with _state_lock:
+            _inflight.discard(code)          # 无论成败都释放在途标记，允许下轮重新派发
+
+
 @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
 def main_task(log):
-    """常驻主流程：每分钟落全站在售(免 token) + 采本商家在售商品的购买记录。
+    """常驻主流程：每分钟落全站在售(免 token) + 线程池并发采各监控商家(MERCHANT_IDS)在售商品的购买记录。
 
     Args:
         log: 日志对象。
@@ -637,8 +735,9 @@ def main_task(log):
     Raises:
         RuntimeError: 数据库连接池异常，触发外层每小时重试。
     """
-    mode = f"白名单模式 codes={WATCH_CODES}" if WATCH_CODES else f"商家模式 merchant={MERCHANT_ID}"
-    log.info(f"购买记录+在售常驻采集启动 | {mode} | 购买记录频率 [{MIN_INTERVAL_SEC}s, {MAX_INTERVAL_SEC}s] "
+    mode = f"白名单模式 codes={WATCH_CODES}" if WATCH_CODES else f"商家模式 merchants={MERCHANT_IDS}"
+    log.info(f"购买记录+在售常驻采集启动 | {mode} | 并发 {MAX_WORKERS} 线程(全局限速≈{1/GLOBAL_MIN_GAP_SEC:.0f}次/秒) "
+             f"| 购买记录频率 [{MIN_INTERVAL_SEC}s, {MAX_INTERVAL_SEC}s] "
              f"| 在售发现常规每 {ONSALE_INGEST_SEC}s、快车时段({FAST_WINDOW_START:%H:%M}~次日{FAST_WINDOW_END:%H:%M})每 {FAST_WINDOW_INGEST_SEC}s "
              f"| 全链路免 token | 详情代理={core.USE_PROXY}")
     pool = MySQLConnectionPool(log=log)
@@ -650,48 +749,40 @@ def main_task(log):
     last_ingest = 0.0                           # 上次「在售落库 + 刷新监控」时刻
     last_discover = 0.0                         # 上次「商家发现 + 补详情」时刻（按 SHOP_DISCOVER_SEC 降频）
 
-    while True:
-        now = time.time()
-
-        # 1) 每(动态)间隔：商家模式先落全站在售(免token)再查库刷新监控；白名单模式只按固定 code 刷新。
-        #    快车高发时段(20:00~次日06:00)把发现节奏加密到 FAST_WINDOW_INGEST_SEC，更快把 2~3 分钟即售罄的快车纳入监控。
-        if now - last_ingest >= _current_ingest_interval(datetime.now()):
-            try:
-                if not WATCH_CODES:
-                    discover = (now - last_discover >= SHOP_DISCOVER_SEC)
-                    ingest_onsale(log, pool, discover)
-                    if discover:
-                        last_discover = now
-                _refresh_monitored(log, pool, monitored, MERCHANT_ID, now)
-            except Exception as e:
-                log.error(f"在售落库/刷新监控异常: {e}")
-            last_ingest = now
-
-        # 2) 找到点的商品，取 next_run_ts 最早的一个跑
-        due_codes = [c for c, s in monitored.items() if s["next_run_ts"] <= now]
-        if not due_codes:
-            time.sleep(IDLE_SLEEP_SEC)
-            continue
-        code = min(due_codes, key=lambda c: monitored[c]["next_run_ts"])
-
-        try:
-            span_sec, n_rows, ended, end_reason = poll_product(log, pool, code, monitored[code])
-            if ended:
-                # 售卖结束：本轮收尾记录已在 poll_product 入库，这里移出监控 + 拉黑，避免死循环重采固定的最后 10 条
-                _ended_codes.add(code)
-                _last_window.pop(code, None)     # 清窗口快照，避免复活时误对齐
-                title = monitored[code]["title"]
-                monitored.pop(code, None)
-                log.info(f"[{code}] 售卖已结束（{end_reason}），停止采集并移出监控 | {title}")
-                continue
-            interval = compute_next_interval(span_sec, monitored[code].get("card_count"))
-            monitored[code]["next_run_ts"] = time.time() + interval
-            monitored[code]["interval"] = interval
-            log.info(f"[{code}] 响应 {n_rows} 条 | 10条跨度 {span_sec}s | 下次 {interval}s 后 | {monitored[code]['title']}")
-        except Exception as e:
-            log.error(f"[{code}] 采集失败: {e}")
-            # 失败退避：60s 后重试；避免异常商品阻塞全局
-            monitored[code]["next_run_ts"] = time.time() + 60
+    # 线程池常驻：主线程只负责「落在售/刷新监控 + 派发到点商品」，实际详情采集交由 worker 并发跑。
+    # 并发把串行时被请求延迟浪费的吞吐(仅 ~0.5 次/秒)填到接近全局限速上限(≈5 次/秒)，追上快车翻窗节奏、减少漏采。
+    with ThreadPoolExecutor(max_workers=MAX_WORKERS, thread_name_prefix="poll") as executor:
+        while True:
+            now = time.time()
+
+            # 1) 每(动态)间隔：商家模式先落全站在售(免token)再查库刷新监控；白名单模式只按固定 code 刷新。
+            #    快车高发时段(20:00~次日06:00)把发现节奏加密到 FAST_WINDOW_INGEST_SEC，更快把 2~3 分钟即售罄的快车纳入监控。
+            #    本步在主线程串行做（期间 worker 仍在并发采集），频率低(20~60s 一次)，不拖累采集吞吐。
+            if now - last_ingest >= _current_ingest_interval(datetime.now()):
+                try:
+                    if not WATCH_CODES:
+                        discover = (now - last_discover >= SHOP_DISCOVER_SEC)
+                        ingest_onsale(log, pool, discover)
+                        if discover:
+                            last_discover = now
+                    _refresh_monitored(log, pool, monitored, MERCHANT_IDS, now)
+                except Exception as e:
+                    log.error(f"在售落库/刷新监控异常: {e}")
+                last_ingest = now
+
+            # 2) 派发：把所有「到点且不在途」的商品按 next_run_ts 升序提交给线程池并发采集。
+            #    先在锁内挑出并标记在途（避免下一轮重复派发同一 code），再锁外 submit（submit 本身线程安全）。
+            with _state_lock:
+                due = sorted(
+                    (c for c, s in monitored.items() if s["next_run_ts"] <= now and c not in _inflight),
+                    key=lambda c: monitored[c]["next_run_ts"])
+                for code in due:
+                    _inflight.add(code)
+            for code in due:
+                executor.submit(_poll_worker, log, pool, monitored, code)
+
+            # 3) 有到点任务用快节奏回来继续喂满 worker；无则短歇省 CPU。真正的请求速率由令牌桶封顶，此处只是调度节拍。
+            time.sleep(DISPATCH_SLEEP_SEC if due else IDLE_SLEEP_SEC)
 
 
 def schedule_task():