Harden background loops and API edge cases from code review

- crawler manager: retry transient errors inside each loop, never let
  log writes break supervision, run tasks in a TaskGroup so any escaped
  error restarts the container instead of leaving a half-dead process
- market worker: tolerate symbols deleted mid-backfill, retry task
  selection and symbol queries, run loops in a TaskGroup
- market WS: validate subscription messages and close the connection
  when either the receive or forward side ends
- news pipeline: make first insert concurrency-safe with ON CONFLICT
- market API: only proxy ticker/1s klines for added symbols, map Binance
  rate limits to 503 and upstream errors to 502 without blocking
- news API: escape LIKE wildcards in search, return 422 for bad cursors
- market page: cancel pending reconnect timer on unmount

backend/src/alphapulse/api/market.py

@@ -1,11 +1,15 @@
 import asyncio
 import json
+import logging
 import time
+from collections.abc import Awaitable
 from datetime import datetime
 from typing import Annotated, Literal
 
+import httpx
 from fastapi import APIRouter, Depends, HTTPException, Query, WebSocket, WebSocketDisconnect, status
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, ValidationError, field_validator
+from starlette.websockets import WebSocketState
 from sqlalchemy import delete, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -13,7 +17,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from alphapulse.core.auth import require_admin, websocket_admin
 from alphapulse.core.db import get_session
 from alphapulse.core.redis import KEY_PREFIX, redis
-from alphapulse.market.binance import BinanceClient
+from alphapulse.market.binance import BinanceClient, RateLimited
 from alphapulse.market.channels import kline_channel, ticker_channel
 from alphapulse.models import Kline, KlineSyncState, MarketSymbol
 from alphapulse.models.market import INTERVALS
@@ -26,7 +30,9 @@ Market = Literal["spot", "um"]
 SYMBOLS_CACHE_SECONDS = 3600
 MAX_KLINES = 1500
 
-client = BinanceClient()
+logger = logging.getLogger("alphapulse.api.market")
+# API 请求不等待限流，直接报 503，避免 HTTP 请求被挂起
+client = BinanceClient(wait=False)
 
 
 class IntervalProgress(BaseModel):
@@ -62,11 +68,43 @@ class TradableSymbol(BaseModel):
     volume_precision: int
 
 
+class Subscription(BaseModel):
+    market: Market
+    symbol: str = Field(pattern=r"^[A-Z0-9]{2,30}$")
+    interval: str
+
+    @field_validator("interval")
+    @classmethod
+    def check_interval(cls, value: str) -> str:
+        if value not in (*INTERVALS, "1s"):
+            raise ValueError(f"不支持的周期：{value}")
+        return value
+
+
+async def call_binance[T](request: Awaitable[T]) -> T:
+    """把币安调用的失败转换成明确的 HTTP 错误，而不是 500"""
+    try:
+        return await request
+    except RateLimited as e:
+        raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, str(e)) from None
+    except httpx.HTTPStatusError as e:
+        raise HTTPException(status.HTTP_502_BAD_GATEWAY, f"币安返回 {e.response.status_code}：{e.response.text[:200]}") from None
+    except httpx.TransportError as e:
+        raise HTTPException(status.HTTP_502_BAD_GATEWAY, f"无法连接币安：{e!r}") from None
+
+
+async def ensure_symbol(session: AsyncSession, market: str, symbol: str) -> None:
+    """只允许查询已添加的交易对，不把任意 symbol 转发给币安"""
+    exists = await session.scalar(select(MarketSymbol.id).where(MarketSymbol.market == market, MarketSymbol.symbol == symbol))
+    if exists is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, f"未添加交易对 {market} {symbol}")
+
+
 async def tradable_symbols(market: str) -> list[dict]:
     key = f"{KEY_PREFIX}market:symbols:{market}"
     if cached := await redis.get(key):
         return json.loads(cached)
-    symbols = await client.tradable_symbols(market)
+    symbols = await call_binance(client.tradable_symbols(market))
     await redis.set(key, json.dumps(symbols), ex=SYMBOLS_CACHE_SECONDS)
     return symbols
 
@@ -136,8 +174,9 @@ async def klines(
         # 1s 不落库，直接取币安最近数据
         if market != "spot":
             raise HTTPException(status.HTTP_400_BAD_REQUEST, "合约没有 1s 周期")
+        await ensure_symbol(session, market, symbol)
         params = {"symbol": symbol, "interval": "1s", "limit": limit, **({"endTime": end - 1} if end else {})}
-        rows = await client.get("spot", "klines", params)
+        rows = await call_binance(client.get("spot", "klines", params))
         return [[r[0], float(r[1]), float(r[2]), float(r[3]), float(r[4]), float(r[5]), float(r[7])] for r in rows]
     if interval not in INTERVALS:
         raise HTTPException(status.HTTP_400_BAD_REQUEST, f"不支持的周期：{interval}")
@@ -154,8 +193,9 @@ async def klines(
 
 
 @router.get("/ticker")
-async def ticker(market: Market, symbol: str) -> dict:
-    return await client.ticker(market, symbol)
+async def ticker(session: Annotated[AsyncSession, Depends(get_session)], market: Market, symbol: str) -> dict:
+    await ensure_symbol(session, market, symbol)
+    return await call_binance(client.ticker(market, symbol))
 
 
 @ws_router.websocket("/ws")
@@ -168,6 +208,19 @@ async def market_ws(websocket: WebSocket) -> None:
     pubsub = redis.pubsub()
     subscribed = asyncio.Event()
 
+    async def receive() -> None:
+        while True:
+            raw = await websocket.receive_text()
+            try:
+                sub = Subscription.model_validate_json(raw)
+            except ValidationError as e:
+                await websocket.send_text(json.dumps({"type": "error", "message": f"订阅参数无效：{e.errors()[0]['msg']}"}, ensure_ascii=False))
+                continue
+            if subscribed.is_set():
+                await pubsub.unsubscribe()
+            await pubsub.subscribe(kline_channel(sub.market, sub.symbol, sub.interval), ticker_channel(sub.market, sub.symbol))
+            subscribed.set()
+
     async def forward() -> None:
         await subscribed.wait()
         while True:
@@ -184,19 +237,18 @@ async def market_ws(websocket: WebSocket) -> None:
                 "data": json.loads(message["data"]),
             }))
 
-    task = asyncio.create_task(forward())
+    # 收、发任一方向结束（客户端断开或出错）就整体收尾；非正常断开时主动关闭，让前端重连，
+    # 避免转发任务已死而连接看起来仍正常
+    tasks = [asyncio.create_task(receive()), asyncio.create_task(forward())]
     try:
-        while True:
-            sub = await websocket.receive_json()
-            if subscribed.is_set():
-                await pubsub.unsubscribe()
-            await pubsub.subscribe(
-                kline_channel(sub["market"], sub["symbol"], sub["interval"]),
-                ticker_channel(sub["market"], sub["symbol"]),
-            )
-            subscribed.set()
-    except WebSocketDisconnect:
-        pass
+        done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+        for task in done:
+            error = task.exception()
+            if error is not None and not isinstance(error, WebSocketDisconnect):
+                logger.warning("行情 WS 连接异常结束：%r", error)
     finally:
-        task.cancel()
+        for task in tasks:
+            task.cancel()
         await pubsub.aclose()
+        if websocket.application_state == WebSocketState.CONNECTED and websocket.client_state == WebSocketState.CONNECTED:
+            await websocket.close(code=1011)

backend/src/alphapulse/api/news.py

@@ -69,15 +69,23 @@ async def list_news(
     if source:
         stmt = stmt.where(NewsItem.source_code.in_(source))
     if q:
-        pattern = f"%{q}%"
-        stmt = stmt.where(or_(NewsItem.title.ilike(pattern), NewsItem.content_text.ilike(pattern)))
+        # 转义 LIKE 通配符，按字面搜索
+        escaped = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+        pattern = f"%{escaped}%"
+        stmt = stmt.where(or_(NewsItem.title.ilike(pattern, escape="\\"), NewsItem.content_text.ilike(pattern, escape="\\")))
     if start:
         stmt = stmt.where(NewsItem.published_at >= start)
     if end:
         stmt = stmt.where(NewsItem.published_at < end)
     if cursor:
-        published_at, news_id = cursor.rsplit("|", 1)
-        stmt = stmt.where(tuple_(NewsItem.published_at, NewsItem.id) < (datetime.fromisoformat(published_at), int(news_id)))
+        try:
+            published_at, news_id = cursor.rsplit("|", 1)
+            position = (datetime.fromisoformat(published_at), int(news_id))
+            if position[0].tzinfo is None:
+                raise ValueError("cursor 时间缺少时区")
+        except ValueError:
+            raise HTTPException(status.HTTP_422_UNPROCESSABLE_CONTENT, "cursor 格式无效") from None
+        stmt = stmt.where(tuple_(NewsItem.published_at, NewsItem.id) < position)
     rows = (await session.execute(stmt.order_by(NewsItem.published_at.desc(), NewsItem.id.desc()).limit(limit))).all()
 
     items = [NewsBrief.model_validate({**news.__dict__, "excerpt": excerpt}) for news, excerpt in rows]

backend/src/alphapulse/crawler/manager.py

@@ -24,6 +24,7 @@ RESTART_DELAY_MAX = 300
 # 子进程稳定运行超过该时长后，异常退出的重启延迟重新从最小值开始
 STABLE_SECONDS = 60
 STOP_TIMEOUT = 15
+CONTROL_RECONNECT_SECONDS = 5
 
 
 @dataclass
@@ -40,8 +41,12 @@ class Worker:
 
 
 async def log(code: str, level: str, message: str) -> None:
+    """写 stdout 和 crawl_logs；写库失败只记 stdout，不能让守护任务因为日志而中断"""
     logger.log(logging.getLevelName(level), "[%s] %s", code, message)
-    await write_logs([(code, None, level, message)])
+    try:
+        await write_logs([(code, None, level, message)])
+    except Exception:
+        logger.exception("[%s] 日志落库失败", code)
 
 
 class Manager:
@@ -58,26 +63,22 @@ class Manager:
         for sig in (signal.SIGTERM, signal.SIGINT):
             loop.add_signal_handler(sig, self.stopping.set)
 
-        tasks = [asyncio.create_task(self.supervise(w)) for w in self.workers.values()]
-        tasks += [asyncio.create_task(c) for c in (self.schedule_loop(), self.control_loop(), self.status_loop())]
-        await self.stopping.wait()
-        logger.info("收到退出信号，停止全部子进程")
-        await asyncio.gather(*(self.stop_process(w) for w in self.workers.values()))
-        for task in tasks:
-            task.cancel()
+        # 各循环自行重试可恢复的错误；仍有异常逃出时 TaskGroup 让整个进程退出，由 docker 重启，
+        # 避免出现"进程活着但调度或守护任务已经死掉"的状态
+        async with asyncio.TaskGroup() as tg:
+            for w in self.workers.values():
+                tg.create_task(self.supervise(w))
+            loops = [tg.create_task(c) for c in (self.schedule_loop(), self.control_loop(), self.status_loop())]
+            await self.stopping.wait()
+            logger.info("收到退出信号，停止全部子进程")
+            await asyncio.gather(*(self.stop_process(w) for w in self.workers.values()))
+            for task in loops:
+                task.cancel()
         await redis.delete(STATUS_KEY)
 
     async def supervise(self, w: Worker) -> None:
         while not self.stopping.is_set():
-            # 上次进程遗留的请求和未收尾的轮次都已失效，清理后从干净状态开始
-            await redis.delete(f"{KEY_PREFIX}{w.code}:requests", start_urls_key(w.code))
-            async with SessionLocal() as session:
-                await session.execute(
-                    update(CrawlRun)
-                    .where(CrawlRun.source_code == w.code, CrawlRun.status == "running")
-                    .values(status="interrupted", finished_at=datetime.now(UTC), error="子进程退出时本轮未完成")
-                )
-                await session.commit()
+            await self.cleanup_before_spawn(w)
             w.proc = await asyncio.create_subprocess_exec("scrapy", "crawl", w.code)
             w.started_at = datetime.now(UTC)
             # 新进程的第一轮立即执行，并做全量回溯扫描
@@ -104,6 +105,20 @@ class Manager:
                 pass
             w.restart_delay = min(w.restart_delay * 2, RESTART_DELAY_MAX)
 
+    async def cleanup_before_spawn(self, w: Worker) -> None:
+        """上次进程遗留的请求和未收尾的轮次都已失效；清理失败不影响拉起子进程"""
+        try:
+            await redis.delete(f"{KEY_PREFIX}{w.code}:requests", start_urls_key(w.code))
+            async with SessionLocal() as session:
+                await session.execute(
+                    update(CrawlRun)
+                    .where(CrawlRun.source_code == w.code, CrawlRun.status == "running")
+                    .values(status="interrupted", finished_at=datetime.now(UTC), error="子进程退出时本轮未完成")
+                )
+                await session.commit()
+        except Exception:
+            logger.exception("[%s] 拉起前清理失败，继续启动子进程", w.code)
+
     async def stop_process(self, w: Worker) -> None:
         if w.proc is None:
             return
@@ -115,38 +130,66 @@ class Manager:
             w.proc.kill()
 
     async def schedule_loop(self) -> None:
+        unknown: set[str] = set()
         while True:
-            async with SessionLocal() as session:
-                sources = (await session.scalars(select(Source))).all()
-            now = datetime.now(UTC)
-            for s in sources:
-                w = self.workers[s.code]
-                if not s.enabled or w.proc is None or w.restarting:
-                    continue
-                if s.paused_until and s.paused_until > now:
-                    w.next_run_at = max(w.next_run_at, s.paused_until)
-                    continue
-                if now < w.next_run_at:
-                    continue
-                # 子进程还没取走上一条触发时不再堆积
-                if await redis.llen(start_urls_key(s.code)) == 0:
-                    await redis.lpush(start_urls_key(s.code), json.dumps({"trigger": w.trigger, "backfill_hours": s.backfill_hours}))
-                    w.trigger = "schedule"
-                w.next_run_at = now + timedelta(seconds=s.interval_seconds)
+            try:
+                await self.schedule_once(unknown)
+            except Exception:
+                logger.exception("调度出错，1s 后重试")
             await asyncio.sleep(1)
 
+    async def schedule_once(self, unknown: set[str]) -> None:
+        async with SessionLocal() as session:
+            sources = (await session.scalars(select(Source))).all()
+        now = datetime.now(UTC)
+        for s in sources:
+            w = self.workers.get(s.code)
+            if w is None:
+                # 启动后才新增的来源，需要重启管理进程才会接管
+                if s.code not in unknown:
+                    unknown.add(s.code)
+                    logger.warning("来源 %s 不在本进程的守护列表中，重启管理进程后生效", s.code)
+                continue
+            if not s.enabled or w.proc is None or w.restarting:
+                continue
+            if s.paused_until and s.paused_until > now:
+                w.next_run_at = max(w.next_run_at, s.paused_until)
+                continue
+            if now < w.next_run_at:
+                continue
+            # 子进程还没取走上一条触发时不再堆积
+            if await redis.llen(start_urls_key(s.code)) == 0:
+                await redis.lpush(start_urls_key(s.code), json.dumps({"trigger": w.trigger, "backfill_hours": s.backfill_hours}))
+                w.trigger = "schedule"
+            w.next_run_at = now + timedelta(seconds=s.interval_seconds)
+
     async def control_loop(self) -> None:
-        async with redis.pubsub() as pubsub:
-            await pubsub.subscribe(CONTROL_CHANNEL)
-            async for message in pubsub.listen():
-                if message["type"] != "message":
-                    continue
-                command = json.loads(message["data"])
-                if command["action"] != "restart":
-                    logger.warning("未知指令：%s", command)
-                    continue
-                codes = list(self.workers) if command["source"] == "all" else [command["source"]]
-                await asyncio.gather(*(self.restart(self.workers[code]) for code in codes))
+        while True:
+            try:
+                async with redis.pubsub() as pubsub:
+                    await pubsub.subscribe(CONTROL_CHANNEL)
+                    async for message in pubsub.listen():
+                        if message["type"] == "message":
+                            await self.handle_command(message["data"])
+            except Exception:
+                logger.exception("控制通道断开，%ds 后重新订阅", CONTROL_RECONNECT_SECONDS)
+                await asyncio.sleep(CONTROL_RECONNECT_SECONDS)
+
+    async def handle_command(self, raw: str) -> None:
+        """单条指令出错只记日志，不影响后续指令"""
+        try:
+            command = json.loads(raw)
+            if command.get("action") != "restart":
+                logger.warning("未知指令：%s", raw)
+                return
+            source = command.get("source")
+            codes = list(self.workers) if source == "all" else [source]
+            if any(code not in self.workers for code in codes):
+                logger.warning("重启指令中的来源不存在：%s", raw)
+                return
+            await asyncio.gather(*(self.restart(self.workers[code]) for code in codes))
+        except Exception:
+            logger.exception("处理指令失败：%s", raw)
 
     async def restart(self, w: Worker) -> None:
         await log(w.code, "INFO", "收到重启指令")
@@ -166,20 +209,26 @@ class Manager:
 
     async def status_loop(self) -> None:
         while True:
-            mapping = {
-                w.code: json.dumps({
-                    "state": "restarting" if w.restarting else ("running" if w.proc else "stopped"),
-                    "pid": w.proc.pid if w.proc else None,
-                    "started_at": w.started_at.isoformat() if w.started_at else None,
-                    "next_run_at": w.next_run_at.isoformat(),
-                })
-                for w in self.workers.values()
-            }
-            async with redis.pipeline() as pipe:
-                # 管理进程挂掉后状态自动过期，前端据此显示"管理进程未运行"
-                await pipe.hset(STATUS_KEY, mapping=mapping).expire(STATUS_KEY, 10).execute()
+            try:
+                await self.publish_status()
+            except Exception:
+                logger.exception("发布状态失败，2s 后重试")
             await asyncio.sleep(2)
 
+    async def publish_status(self) -> None:
+        mapping = {
+            w.code: json.dumps({
+                "state": "restarting" if w.restarting else ("running" if w.proc else "stopped"),
+                "pid": w.proc.pid if w.proc else None,
+                "started_at": w.started_at.isoformat() if w.started_at else None,
+                "next_run_at": w.next_run_at.isoformat(),
+            })
+            for w in self.workers.values()
+        }
+        async with redis.pipeline() as pipe:
+            # 管理进程挂掉后状态自动过期，前端据此显示"管理进程未运行"
+            await pipe.hset(STATUS_KEY, mapping=mapping).expire(STATUS_KEY, 10).execute()
+
 
 if __name__ == "__main__":
     logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")

backend/src/alphapulse/crawler/pipelines.py

@@ -4,6 +4,7 @@ from typing import Any
 
 from scrapy.crawler import Crawler
 from sqlalchemy import func, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from alphapulse.core.db import SessionLocal
 from alphapulse.core.redis import redis
@@ -32,17 +33,26 @@ class NewsPipeline:
         spider = self.crawler.spider
         digest = content_hash(item)
         raw_response_id = item.pop("raw_response_id")
+        locked = (
+            select(NewsItem)
+            .where(NewsItem.source_code == item["source_code"], NewsItem.external_id == item["external_id"])
+            .with_for_update()
+        )
         async with SessionLocal() as session:
-            news = await session.scalar(
-                select(NewsItem)
-                .where(NewsItem.source_code == item["source_code"], NewsItem.external_id == item["external_id"])
-                .with_for_update()
-            )
+            news = await session.scalar(locked)
+            inserted_id = None
             if news is None:
+                # 行不存在时 FOR UPDATE 锁不住任何东西；用 ON CONFLICT 处理同一条目被并发首次写入
+                inserted_id = await session.scalar(
+                    pg_insert(NewsItem)
+                    .values(**item, content_hash=digest, revision_no=1)
+                    .on_conflict_do_nothing(index_elements=[NewsItem.source_code, NewsItem.external_id])
+                    .returning(NewsItem.id)
+                )
+                # 插入成功取新行；冲突说明另一处刚写入，按已存在条目走比较/修订流程
+                news = await session.scalar(locked)
+            if inserted_id is not None:
                 kind = "new"
-                news = NewsItem(**item, content_hash=digest, revision_no=1)
-                session.add(news)
-                await session.flush()
             elif news.content_hash != digest:
                 kind = "revised"
                 for field in UPDATABLE_FIELDS:

backend/src/alphapulse/market/binance.py

@@ -27,8 +27,14 @@ def decimals(step: str) -> int:
     return len(step.rstrip("0").split(".")[1]) if "." in step.rstrip("0") else 0
 
 
+class RateLimited(Exception):
+    """非等待模式下被币安限流"""
+
+
 class BinanceClient:
-    def __init__(self) -> None:
+    def __init__(self, wait: bool = True) -> None:
+        # wait=False 用于 API 请求：被限流时立即抛 RateLimited，不让 HTTP 请求挂起等待
+        self.wait = wait
         self.http = httpx.AsyncClient(timeout=20)
 
     async def get(self, market: str, path_key: str, params: dict[str, Any] | None = None) -> Any:
@@ -45,6 +51,8 @@ class BinanceClient:
             if resp.status_code in (418, 429):
                 # 418 表示 IP 已被临时封禁，必须按 Retry-After 等待
                 wait = int(resp.headers.get("Retry-After", "60"))
+                if not self.wait:
+                    raise RateLimited(f"币安限流 {resp.status_code}，需等待 {wait}s")
                 logger.warning("币安限流 %d，暂停 %ds", resp.status_code, wait)
                 await asyncio.sleep(wait)
                 continue
@@ -53,7 +61,7 @@ class BinanceClient:
                 continue
             resp.raise_for_status()
             used = int(resp.headers.get("X-MBX-USED-WEIGHT-1M", "0"))
-            if used > conf["weight_limit"] * WEIGHT_RATIO:
+            if self.wait and used > conf["weight_limit"] * WEIGHT_RATIO:
                 wait = 61 - time.time() % 60
                 logger.info("%s 权重已用 %d，等待 %.0fs 到下一分钟", market, used, wait)
                 await asyncio.sleep(wait)

backend/src/alphapulse/market/worker.py

@@ -100,6 +100,9 @@ async def next_backfill_task(market: str) -> tuple[MarketSymbol, str] | None:
 async def set_symbol_status(symbol_id: int, status: str, error: str | None = None) -> None:
     async with SessionLocal() as session:
         s = await session.get(MarketSymbol, symbol_id)
+        if s is None:
+            # 补齐过程中交易对被用户删除
+            return
         s.status = status
         s.last_error = error
         await session.commit()
@@ -107,15 +110,20 @@ async def set_symbol_status(symbol_id: int, status: str, error: str | None = Non
 
 async def backfill_loop(market: str) -> None:
     while True:
-        task = await next_backfill_task(market)
+        try:
+            task = await next_backfill_task(market)
+        except Exception:
+            logger.exception("%s 选取补齐任务出错，稍后重试", market)
+            await asyncio.sleep(10)
+            continue
         if task is None:
             await asyncio.sleep(5)
             continue
         s, interval = task
-        if s.status == "pending":
-            await set_symbol_status(s.id, "backfilling")
-            logger.info("%s %s 开始补齐", market, s.symbol)
         try:
+            if s.status == "pending":
+                await set_symbol_status(s.id, "backfilling")
+                logger.info("%s %s 开始补齐", market, s.symbol)
             for _ in range(PAGES_PER_TURN):
                 if await sync_page(market, s.symbol, interval):
                     logger.info("%s %s %s 补齐完成", market, s.symbol, interval)
@@ -141,12 +149,16 @@ async def catchup_loop() -> None:
         except TimeoutError:
             pass
         catchup_needed.clear()
-        async with SessionLocal() as session:
-            states = (await session.execute(
-                select(KlineSyncState.market, KlineSyncState.symbol, KlineSyncState.interval)
-                .join(MarketSymbol, (MarketSymbol.market == KlineSyncState.market) & (MarketSymbol.symbol == KlineSyncState.symbol))
-                .where(KlineSyncState.backfill_done, MarketSymbol.status.in_(ACTIVE_STATUSES))
-            )).all()
+        try:
+            async with SessionLocal() as session:
+                states = (await session.execute(
+                    select(KlineSyncState.market, KlineSyncState.symbol, KlineSyncState.interval)
+                    .join(MarketSymbol, (MarketSymbol.market == KlineSyncState.market) & (MarketSymbol.symbol == KlineSyncState.symbol))
+                    .where(KlineSyncState.backfill_done, MarketSymbol.status.in_(ACTIVE_STATUSES))
+                )).all()
+        except Exception:
+            logger.exception("读取追平列表失败，下次再试")
+            continue
         started = time.monotonic()
         for market, symbol, interval in states:
             try:
@@ -186,13 +198,13 @@ async def handle_message(market: str, raw: str | bytes) -> None:
 
 async def stream_loop(market: str) -> None:
     while True:
-        symbols = await active_symbols(market)
-        if not symbols:
-            await asyncio.sleep(SYMBOL_CHECK_SECONDS)
-            continue
-        intervals = INTERVALS + (["1s"] if market == "spot" else [])
-        streams = [f"{s.lower()}@kline_{i}" for s in symbols for i in intervals] + [f"{s.lower()}@miniTicker" for s in symbols]
         try:
+            symbols = await active_symbols(market)
+            if not symbols:
+                await asyncio.sleep(SYMBOL_CHECK_SECONDS)
+                continue
+            intervals = INTERVALS + (["1s"] if market == "spot" else [])
+            streams = [f"{s.lower()}@kline_{i}" for s in symbols for i in intervals] + [f"{s.lower()}@miniTicker" for s in symbols]
             async with websockets.connect(f"{WS[market]}?streams={'/'.join(streams)}", max_size=2**22) as ws:
                 logger.info("%s WS 已连接：%d 个交易对，%d 个流", market, len(symbols), len(streams))
                 catchup_needed.set()
@@ -205,16 +217,17 @@ async def stream_loop(market: str) -> None:
                             logger.info("%s 交易对列表变化，重新订阅", market)
                             break
         except Exception:
-            logger.exception("%s WS 断开，5s 后重连", market)
+            logger.exception("%s WS 循环出错，5s 后重连", market)
             await asyncio.sleep(5)
 
 
 async def main() -> None:
-    await asyncio.gather(
-        *(backfill_loop(m) for m in MARKETS),
-        *(stream_loop(m) for m in MARKETS),
-        catchup_loop(),
-    )
+    # 各循环自行重试可恢复的错误；仍有异常逃出时整个进程退出，由 docker 重启
+    async with asyncio.TaskGroup() as tg:
+        for m in MARKETS:
+            tg.create_task(backfill_loop(m))
+            tg.create_task(stream_loop(m))
+        tg.create_task(catchup_loop())
 
 
 if __name__ == "__main__":