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_okx_symbols() -> dict[str, str]:
13    """Get all symbols from OKX."""
14    logger.info("Fetching OKX symbols...")
15    res = {}
16    url = f"{API.OKX}/api/v5/public/instruments"
17    for instType in ["SWAP", "SPOT"]:
18        params = {"instType": instType}
19        response = await hx_req(url, params=params, proxy=PROXY.CRYPTO, check_keys=["data"], check_kv={"code": "0"}, silent=True)
20        if response.get("hx_error"):
21            return {}
22        data = response["data"]
23        res |= {coin["instId"].replace("-", ""): coin["instId"] for coin in data if coin["state"] == "live"}
24    return res
25
26
27async def okx_supported(coin: str) -> tuple[str, str]:
28    """Check if the coin is supported by OKX.
29
30    If supported, return the supported symbol format and the instId.
31
32    e.g. "BTC" -> ("BTCUSDT", "BTC-USDT")
33    """
34    symbols = await get_okx_symbols()
35    coin = coin.upper().replace("-", "")
36    symbol = coin
37    suffixes = ["USDT", "USDC", "USDTSWAP", "USDCSWAP", "USDSWAP"]
38    while symbol not in symbols and suffixes:
39        symbol = f"{coin}{suffixes.pop(0)}".upper()
40    return (symbol, symbols[symbol]) if symbol in symbols else ("", "")
41
42
43@cache.memoize(ttl=60)
44async def get_okx_price(coin: str, interval: str | None = None) -> dict:
45    """Get the price of a crypto asset from OKX."""
46    if interval is None:
47        interval = "30m"
48    # OKX interval unit: m, H, D, W, M
49    if interval.endswith(("h", "d", "w")):
50        interval = interval.upper()
51    if interval == "8H":  # OKX does not support 8H interval
52        interval = "6H"
53    if interval not in ["1m", "3m", "5m", "15m", "30m", "1H", "2H", "4H", "6H", "12H", "1D", "2D", "3D", "1W", "1M", "3M"]:
54        interval = "30m"
55
56    symbol, inst_id = await okx_supported(coin)
57    if not symbol:
58        return {}
59    url = f"{API.OKX}/api/v5/market/candles?instId={inst_id}&bar={interval}&limit=49"
60    response = await hx_req(url, proxy=PROXY.CRYPTO, check_kv={"code": "0"}, silent=True)
61    if response.get("hx_error"):
62        return {"texts": f"OKX price for {coin} failed: {response['hx_error']}"}
63    klines = response["data"]  # ts, o, h, l, c
64    if not klines:
65        return {"texts": f"OKX price for {coin} failed: {response['hx_error']}"}
66    klines = sorted(klines, key=lambda x: x[0])
67    high_price = max(float(number(x[2])) for x in klines)
68    low_price = min(float(number(x[3])) for x in klines)
69    open_price = float(number(klines[0][1]))
70    close_price = float(number(klines[-1][4]))
71    change_pct = (close_price - open_price) / open_price
72    amplitude = (high_price - low_price) / low_price
73    if close_price < open_price:
74        amplitude *= -1
75    title = f"{symbol}•{interval}•OKX"
76    subtitle = f"开: {open_price} 高: {high_price} 低: {low_price} 收: {close_price} 涨: {change_pct:+.2%} 振: {abs(amplitude):.2%}"
77    text = f"{title}\n"
78    text += f"时间段: {ts_to_dt(klines[0][0]):%m-%d %H:%M} → {ts_to_dt(klines[-1][0]):%m-%d %H:%M}\n"
79    text += f"开盘价: {open_price}\n"
80    text += f"最高价: {high_price}\n"
81    text += f"最低价: {low_price}\n"
82    text += f"收盘价: {close_price} \n"
83    text += f"涨跌幅: {change_pct:+.2%}\n"
84    text += f"振幅: {amplitude:.2%}"
85    chart = await generate_chart(klines, interval, title, subtitle)
86    return {"texts": text, "media": [{"photo": chart}]}