Add default USD-M perpetual BTC/ETH/BNB and fix futures WS stream

Seed um BTCUSDT/ETHUSDT/BNBUSDT as default symbols. The legacy futures
route fstream.binance.com/stream accepts connections but no longer pushes
data, so switch to /market/stream. Receive with a timeout so symbol
changes are picked up and silent connections are reconnected.
This commit is contained in:
eezd committed 2026-10-13 14:10:18 +08:00
1 parent 554d224be2
commit 0831971e83
3 files changed
+51 -2

No files matched your search

@@ -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')"))
+2 -1
View File
@@ -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
+14 -1
View File
@@ -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:
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: