Binance WebSocket 多流订阅:Combined Stream 与 SUBSCRIBE 动态管理

Binance Spot WebSocket 的一个连接可以订阅多个代币的多个数据流,不需要为每个代币单独建连接。 方法一:Combined Stream URL 连接时直接在 URL 里拼接多个 stream,用 / 分隔: wss://stream.binance.com:9443/stream?streams=btcusdt@trade/ethusdt@tradewss://stream.binance.com:9443/stream?streams=btcusdt@bookTicker/ethusdt@bookTicker/solusdt@bookTicker收到的消息会包含 stream 字段标识来源: { "stream": "btcusdt@trade", "data": { ... } }方法二:SUBSCRIBE 消息(动态添加) 先连接基础 endpoint: wss://stream.binance.com:9443/ws连接建立后发送 JSON 消息订阅: { "method": "SUBSCRIBE", "params": [ "btcusdt@aggTrade", "ethusdt@aggTrade", "solusdt@aggTrade" ], "id": 1 }id 是请求 ID,服务端会返回对应的响应(result: null 表示成功): { "result": null, "id": 1 }取消订阅用 UNSUBSCRIBE,结构相同: { "method": "UNSUBSCRIBE", "params": ["btcusdt@aggTrade"], "id": 2 }常用 Stream 类型Stream 名称 说明<symbol>@trade 逐笔成交<symbol>@aggTrade 归集成交(合并同一时间同方向同价格的成交)<symbol>@bookTicker 最优挂单(bid/ask 实时更新)<symbol>@ticker 24 小时价格统计<symbol>@kline_1m K 线(1 分钟,其他周期类似)<symbol>@depth5 5 档深度快照<symbol>@depth@100ms 深度增量推送(100ms 间隔)<symbol> 全小写,如 btcusdt、ethusdt。 数量限制单个 WebSocket 连接最多订阅 1024 个 stream 超出上限需要新开连接 一般订阅几十到几百个代币的 bookTicker,一个连接完全够用选哪种方式场景 推荐启动时订阅固定代币列表 Combined Stream URL运行时动态增减代币 SUBSCRIBE / UNSUBSCRIBE需要区分消息来源 Combined Stream(有 stream 字段)两种方式可以混用:先用 Combined Stream URL 订阅一批,再用 SUBSCRIBE 动态添加新的代币。

Chrome/Edge 浏览器插件数据存放位置与读取方法

Chrome/Edge 插件安装目录 Windows Chrome: C:\Users\用户名\AppData\Local\Google\Chrome\User Data\Default\Extensions\Edge: C:\Users\用户名\AppData\Local\Microsoft\Edge\User Data\Default\Extensions\每个插件一个子目录,目录名是插件 ID(32 位小写字母字符串): Extensions\ ├── nkbihfbeogaeaoehlefnkodbefgpgknn\ ← MetaMask │ └── 12.3.0_0\ │ ├── manifest.json │ ├── background.js │ └── ... ├── cjpalhdlnbpafiamejdnhcphjbkeiagm\ ← uBlock Origin │ └── ...插件存储的数据位置 chrome.storage.local 和 chrome.storage.sync 数据以 LevelDB 格式存在: Default\Local Extension Settings\插件ID\Chrome 示例: C:\Users\用户名\AppData\Local\Google\Chrome\User Data\Default\ Local Extension Settings\ nkbihfbeogaeaoehlefnkodbefgpgknn\ ← MetaMask 存储 LOG CURRENT MANIFEST-000001 000003.ldb这是二进制 LevelDB 数据库,不能直接文本编辑器打开,需要 LevelDB 工具读取。 IndexedDB 存储 部分插件使用 IndexedDB: Default\IndexedDB\ chrome-extension_插件ID.indexeddb.leveldb\Session Storage / Cache Default\Session Storage\ Default\Cache\ Default\Code Cache\查找某个插件的 ID 在浏览器地址栏输入: chrome://extensions/开启开发者模式后,每个插件卡片下方显示 ID(32位字符串)。 在 DevTools 里查看插件存储内容打开 chrome://extensions/ 找到目标插件,点击「详情」 点击「检查视图」(Service Worker 或背景页) 在 Console 里执行:// 查看所有 chrome.storage.local 数据 chrome.storage.local.get(null, console.log);// 或用 async/await const data = await chrome.storage.local.get(); console.log(data);Manifest V3 的 Service Worker 和 V2 的 Background Page 都支持这个 API。 常见插件 ID(Chrome)插件 IDMetaMask nkbihfbeogaeaoehlefnkodbefgpgknnuBlock Origin cjpalhdlnbpafiamejdnhcphjbkeiagmTampermonkey dhdgffkkebhmkfjojejmpbldmpobfkfoSwitchyOmega padekgcemlokbadohgkifijomclgjgifEdge 的同款插件 ID 可能不同(Edge 有自己的扩展商店)。 macOS 位置 ~/Library/Application Support/Google/Chrome/Default/Extensions/ ~/Library/Application Support/Google/Chrome/Default/Local Extension Settings/

uvicorn reload 模式下找不到文件:用 __file__ 解决相对路径问题

uvicorn 加 --reload 参数时,每次代码变更都会重启子进程。子进程在 import 模块时如果代码顶层有相对路径操作,很容易报 FileNotFoundError。 问题复现 # main.py import cv2img = cv2.imread("small.png", 0) # 顶层直接读取uvicorn main:app --reload启动会报: FileNotFoundError: [Errno 2] No such file or directory: 'small.png'原因:uvicorn reload 子进程的 工作目录(cwd) 可能和你预期的不一致,相对路径 "small.png" 找的是 cwd 下的文件而不是脚本旁边的文件。 解决:用 __file__ 构造绝对路径 from pathlib import PathBASE_DIR = Path(__file__).parentimg = cv2.imread(str(BASE_DIR / "small.png"), 0)Path(__file__).parent 是脚本自身所在目录,不受 cwd 影响,任何启动方式都一致。 统一管理项目路径 from pathlib import PathBASE_DIR = Path(__file__).parent MODELS_DIR = BASE_DIR / "models" DATA_DIR = BASE_DIR / "data" CONFIG_FILE = BASE_DIR / "config.yaml"然后全局使用 BASE_DIR / "xxx" 而不是裸字符串路径。 uvicorn 启动方式 字符串形式(支持 reload): uvicorn.run( "main:app", host="0.0.0.0", port=8000, reload=True )对象形式(不支持 reload): uvicorn.run( app, # FastAPI 对象 host="0.0.0.0", port=8000 # reload=True 在这里无效,reload 必须用字符串形式 )reload 模式会 fork 子进程并 import 指定模块,所以 module 名必须是字符串。 GPU 模型服务的 worker 数 如果 FastAPI 服务里加载了 GPU 模型(SAM、YOLO、PyTorch 等),只开 1 个 worker: uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1多 worker 会导致每个进程加载一份模型,显存翻倍。CPU 服务才适合 workers = CPU核数 × 2 + 1。 Nginx upstream 与单进程 单进程 FastAPI 的 upstream 可以简化: upstream backend { server 127.0.0.1:18001; }location / { proxy_pass http://backend; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; }weight、least_conn 在只有一个节点时没有任何效果,max_fails + fail_timeout 仍然生效(健康检测)。

Python list 转 dict 的几种写法

Python 里"把 list 转成 map"通常指把列表转成字典,方便按 key 查询。几种常见场景: 1. 两个列表合并成 dict keys = ["a", "b", "c"] values = [1, 2, 3]d = dict(zip(keys, values)) print(d) # {'a': 1, 'b': 2, 'c': 3}2. 元组列表转 dict data = [("a", 1), ("b", 2), ("c", 3)] d = dict(data) print(d) # {'a': 1, 'b': 2, 'c': 3}3. 对象列表按字段索引(最常用) 把 list[dict] 转成按某个 key 索引的字典,避免每次用 next(x for x in lst if x["id"] == ...) 线性搜索: users = [ {"id": 1, "name": "Tom"}, {"id": 2, "name": "Jack"}, {"id": 3, "name": "Lucy"} ]user_map = {u["id"]: u for u in users}print(user_map[2]) # {'id': 2, 'name': 'Jack'}这是查询密集型代码里的常用优化:构建一次 O(n),之后每次查找 O(1)。 4. 统计频次(Counter) from collections import Counterlst = ["a", "b", "a", "c", "a"] freq = Counter(lst)print(freq) # Counter({'a': 3, 'b': 1, 'c': 1}) print(freq["a"]) # 3 print(freq.most_common(2)) # [('a', 3), ('b', 1)]Counter 继承自 dict,支持 +、- 合并,常用于词频统计。 5. map() 函数式处理 Python 的 map() 不是字典,而是将函数应用到每个元素: nums = [1, 2, 3, 4]# 不推荐(用 list comprehension 更清晰) result = list(map(lambda x: x * 2, nums))# 推荐 result = [x * 2 for x in nums] print(result) # [2, 4, 6, 8]map() 返回迭代器,需要 list() 转换。列表推导式可读性更好,PyCharm 等工具也不会提示替换。 6. 分组 dict 按某个字段分组: from collections import defaultdictrecords = [ {"type": "A", "val": 1}, {"type": "B", "val": 2}, {"type": "A", "val": 3}, ]grouped = defaultdict(list) for r in records: grouped[r["type"]].append(r)print(dict(grouped)) # {'A': [{'type': 'A', 'val': 1}, {'type': 'A', 'val': 3}], # 'B': [{'type': 'B', 'val': 2}]}快速对比场景 写法两个列表合并 dict(zip(keys, vals))元组列表 dict(data)按字段索引 {item["id"]: item for item in lst}统计频次 Counter(lst)函数处理 [f(x) for x in lst](推荐 comprehension)分组 defaultdict(list)

Python list 转 dict/map 的五种常见姿势

Python 里说"list 转 map"通常指的是把列表变成字典(dict)。到底怎么转,取决于原始 list 长什么样。 1. 两个平行 list 转 dict keys 和 values 是分开的两列: keys = ["a", "b", "c"] values = [1, 2, 3]m = dict(zip(keys, values)) # {'a': 1, 'b': 2, 'c': 3}2. 二元组 list 转 dict 已经是 (key, value) 对: data = [("a", 1), ("b", 2), ("c", 3)] m = dict(data) # {'a': 1, 'b': 2, 'c': 3}3. 字典 list 按字段索引(最常用) 后端返回一堆对象,前端要按 ID 查——最典型场景: users = [ {"id": 1, "name": "Tom"}, {"id": 2, "name": "Jack"}, {"id": 3, "name": "Lucy"}, ]m = {u["id"]: u for u in users}m[2] # {'id': 2, 'name': 'Jack'}如果 key 会重复,改成分组: from collections import defaultdictgroups = defaultdict(list) for u in users: groups[u["dept"]].append(u)4. Python 内建 map() 函数 如果说的是 map() 而不是"映射": nums = [1, 2, 3, 4] result = list(map(lambda x: x * 2, nums)) # [2, 4, 6, 8]更 Pythonic 的写法是列表推导: result = [x * 2 for x in nums]5. 统计次数 用 Counter: from collections import Counterlst = ["a", "b", "a", "c", "a"] Counter(lst) # Counter({'a': 3, 'b': 1, 'c': 1})记忆表输入形态 用法两列 keys + values dict(zip(keys, values))[(k, v), ...] dict(data)[{...}, ...] 按字段 {u[key]: u for u in list}对每个元素求值 [f(x) for x in list]统计次数 Counter(list)