diff --git a/backend/alembic/versions/20261013_bf0ffbfcaf9a_default_um_symbols.py b/backend/alembic/versions/20261013_bf0ffbfcaf9a_default_um_symbols.py new file mode 100644 index 0000000..3332dce --- /dev/null +++ b/backend/alembic/versions/20261013_bf0ffbfcaf9a_default_um_symbols.py @@ -0,0 +1,35 @@ +"""default um symbols + +Revision ID: bf0ffbfcaf9a +Revises: c7e1beb72c75 +Create Date: 2026-10-13 06:01:27.204671 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'bf0ffbfcaf9a' +down_revision: Union[str, Sequence[str], None] = 'c7e1beb72c75' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # U 本位永续的默认交易对;精度取自 2026-10-13 币安 fapi exchangeInfo(tickSize 0.10/0.01/0.010,stepSize 0.001/0.001/0.01) + op.execute(sa.text(""" + INSERT INTO market_symbols (market, symbol, base, quote, price_precision, volume_precision, status, is_default) + VALUES ('um', 'BTCUSDT', 'BTC', 'USDT', 1, 3, 'pending', true), + ('um', 'ETHUSDT', 'ETH', 'USDT', 2, 3, 'pending', true), + ('um', 'BNBUSDT', 'BNB', 'USDT', 2, 2, 'pending', true) + ON CONFLICT (market, symbol) DO UPDATE SET is_default = true + """)) + + +def downgrade() -> None: + """Downgrade schema.""" + op.execute(sa.text("UPDATE market_symbols SET is_default = false WHERE market = 'um' AND symbol IN ('BTCUSDT', 'ETHUSDT', 'BNBUSDT')")) diff --git a/backend/src/alphapulse/market/binance.py b/backend/src/alphapulse/market/binance.py index f4751f5..a4f4419 100644 --- a/backend/src/alphapulse/market/binance.py +++ b/backend/src/alphapulse/market/binance.py @@ -15,7 +15,8 @@ REST = { "um": {"base": "https://fapi.binance.com", "klines": "/fapi/v1/klines", "exchange_info": "/fapi/v1/exchangeInfo", "ticker": "/fapi/v1/ticker/24hr", "weight_limit": 2400, "kline_limit": 1500}, } -WS = {"spot": "wss://stream.binance.com:9443/stream", "um": "wss://fstream.binance.com/stream"} +# 合约旧路由 fstream.binance.com/stream 能连上但不推送数据,K 线/行情流在 /market 路由下 +WS = {"spot": "wss://stream.binance.com:9443/stream", "um": "wss://fstream.binance.com/market/stream"} # 权重用量超过上限的这个比例就等到下一分钟 WEIGHT_RATIO = 0.5 diff --git a/backend/src/alphapulse/market/worker.py b/backend/src/alphapulse/market/worker.py index 9f447d4..61f88d5 100644 --- a/backend/src/alphapulse/market/worker.py +++ b/backend/src/alphapulse/market/worker.py @@ -26,6 +26,8 @@ logger = logging.getLogger("alphapulse.market.worker") PAGES_PER_TURN = 20 CATCHUP_SECONDS = 3600 SYMBOL_CHECK_SECONDS = 10 +# miniTicker 约每秒推送,这么久没消息视为连接卡死 +STALL_SECONDS = 60 ACTIVE_STATUSES = ("pending", "backfilling", "ready") client = BinanceClient() @@ -209,8 +211,19 @@ async def stream_loop(market: str) -> None: logger.info("%s WS 已连接:%d 个交易对,%d 个流", market, len(symbols), len(streams)) catchup_needed.set() next_check = time.monotonic() + SYMBOL_CHECK_SECONDS - async for raw in ws: - await handle_message(market, raw) + last_message = time.monotonic() + while True: + # 带超时接收:没有消息时也要按时检查交易对变化和连接是否卡死 + try: + raw = await asyncio.wait_for(ws.recv(), timeout=SYMBOL_CHECK_SECONDS) + except TimeoutError: + pass + else: + last_message = time.monotonic() + await handle_message(market, raw) + if time.monotonic() - last_message >= STALL_SECONDS: + logger.warning("%s WS %ds 未收到消息,重连", market, STALL_SECONDS) + break if time.monotonic() >= next_check: next_check = time.monotonic() + SYMBOL_CHECK_SECONDS if await active_symbols(market) != symbols: