1#!/usr/bin/env python
 2# -*- coding: utf-8 -*-
 3from loguru import logger
 4
 5from config import API, PROXY, cache
 6from networking import hx_req
 7from price.chart import generate_chart
 8from utils import number, ts_to_dt
 9
10
11@cache.memoize(ttl=3600)
12async def get_binance_symbols() -> dict[str, str]:
13    """Get all symbols from Binance."""
14    logger.info("Fetching Binance symbols...")
15    res = {}
16    # um
17    url = f"{API.BINANCE_UM}/fapi/v1/exchangeInfo"  # region restriction
18    response = await hx_req(url, proxy=PROXY.CRYPTO, check_keys=["symbols"], max_retry=0, timeout=5, silent=True)
19    if not response.get("hx_error"):
20        data = response["symbols"]
21        res = {coin["symbol"]: "UM" for coin in data if coin["status"] == "TRADING"}
22    # spot
23    url = f"{API.BINANCE_SPOT}/api/v3/exchangeInfo"
24    params = {"showPermissionSets": False, "symbolStatus": "TRADING"}
25    response = await hx_req(url, params=params, proxy=PROXY.CRYPTO, check_keys=["symbols"], silent=True)
26    if not response.get("hx_error"):
27        data = response["symbols"]
28        res |= {coin["symbol"]: "SPOT" for coin in data}
29    return res
30
31
32async def binance_supported(coin: str) -> tuple[str, str]:
33    """Check if the coin is supported by Binance.
34
35    If supported, return the supported symbol format and the market.
36
37    e.g. "BTC" -> ("BTCUSDT", "SPOT")
38    """
39    symbols = await get_binance_symbols()
40    symbol = coin.upper()
41    stablecoins = ["USDT", "USDC", "FDUSD", "TUSD"]
42    while symbol not in symbols and stablecoins:
43        symbol = f"{coin}{stablecoins.pop(0)}".upper()
44    return (symbol, symbols[symbol]) if symbol in symbols else ("", "")
45
46
47@cache.memoize(ttl=60)
48async def get_binance_price(coin: str, interval: str | None = None) -> dict:
49    """Get the price of a crypto asset from Binance."""
50    if interval is None:
51        interval = "30m"
52    # Binance interval unit: m, h, d, w, M
53    if interval.endswith(("H", "D", "W")):
54        interval = interval.lower()
55    if interval not in ["1m", "3m", "5m", "15m", "30m", "1h", "2h", "4h", "6h", "8h", "12h", "1d", "3d", "1w", "1M"]:
56        interval = "30m"
57    symbol, market = await binance_supported(coin)
58    if not symbol:
59        return {}
60
61    if market == "SPOT":
62        url = f"{API.BINANCE_SPOT}/api/v3/klines?symbol={symbol}&interval={interval}&limit=49"
63        klines = await hx_req(url, proxy=PROXY.CRYPTO, silent=True)
64    elif market == "UM":
65        url = f"{API.BINANCE_UM}/fapi/v1/klines?symbol={symbol}&interval={interval}&limit=49"
66        klines = await hx_req(url, proxy=PROXY.CRYPTO, silent=True)
67    else:
68        return {}
69    if isinstance(klines, dict) and klines.get("hx_error"):
70        return {"texts": f"Binance price failed: {coin}"}
71    klines = sorted(klines, key=lambda x: x[0])
72    high_price = max(float(number(x[2])) for x in klines)
73    low_price = min(float(number(x[3])) for x in klines)
74    open_price = float(number(klines[0][1]))
75    close_price = float(number(klines[-1][4]))
76    change_pct = (close_price - open_price) / open_price
77    amplitude = (high_price - low_price) / low_price
78    if close_price < open_price:
79        amplitude *= -1
80    title = f"{symbol}•{interval}•Binance"
81    subtitle = f"开: {open_price} 高: {high_price} 低: {low_price} 收: {close_price} 涨: {change_pct:+.2%} 振: {abs(amplitude):.2%}"
82    text = f"{title}\n"
83    text += f"时间段: {ts_to_dt(klines[0][0]):%m-%d %H:%M} → {ts_to_dt(klines[-1][0]):%m-%d %H:%M}\n"
84    text += f"开盘价: {open_price}\n"
85    text += f"最高价: {high_price}\n"
86    text += f"最低价: {low_price}\n"
87    text += f"收盘价: {close_price} \n"
88    text += f"涨跌幅: {change_pct:+.2%}\n"
89    text += f"振幅: {amplitude:.2%}"
90    chart = await generate_chart(klines, interval, title, subtitle)  # type:ignore
91    return {"texts": text, "media": [{"photo": chart}]}