main
1#!/usr/bin/env python
2# -*- coding: utf-8 -*-
3import asyncio
4from collections import defaultdict
5
6from loguru import logger
7
8from config import PROXY, TOKEN, TZ, cache
9from messages.progress import modify_progress
10from networking import hx_req
11
12
13@cache.memoize(ttl=43200) # 12 hours
14async def get_tradingview_symbols() -> dict[str, list[tuple]]:
15 """Get all symbols from TradingView.
16
17 Returns: {
18 "AAPL": [("NASDAQ:AAPL", "america")] # (simple symbol)
19 "NASDAQ:AAPL": [("NASDAQ:AAPL", "america")], # (full symbol)
20 "000001": [("SSE:000001", "china"), ("SZSE:000001", "china")] # (multile tickers with same symbol)
21 }
22 """
23 logger.info("Fetching TradingView symbols...")
24 full = {}
25 for region in ["cfd", "america", "china", "hongkong"]: # priority: cfd > america > china > hongkong
26 url = f"https://scanner.tradingview.com/{region}/scan"
27 response = await hx_req(url, proxy=PROXY.CRYPTO, check_keys=["data"], silent=True)
28 if response.get("hx_error"):
29 continue
30 response = response["data"]
31 full |= {coin["s"]: [(coin["s"], region)] for coin in response}
32 simple = defaultdict(list)
33 for k, v in full.items():
34 if k.startswith("CRYPTOCAP"):
35 continue
36 simple[k.split(":")[-1]].extend(v)
37 return simple | full
38
39
40async def tradingview_supported(symbol: str) -> list[tuple[str, str]]:
41 """Check if the coin is supported by TradingView.
42
43 If supported, return the list of full symbol format.
44 e.g. [("SSE:000001", "china"), ("SZSE:000001", "china")]
45 """
46 symbols = await get_tradingview_symbols()
47 return symbols.get(symbol.upper(), [])
48
49
50async def get_tradingview_price(symbol: str, interval: str | None = None, **kwargs) -> dict:
51 """Get the price of a crypto asset from TradingView.
52
53 Returns: {
54 "url": "remote url of the chart image",
55 "symbol": "NADAQ:AAPL",
56 "interval": "5m"
57 }
58 """
59 if interval is None:
60 interval = "5m"
61 if interval not in ["1m", "3m", "5m", "15m", "30m", "45m", "1h", "2h", "3h", "4h", "1D", "1W", "1M", "3M", "6M", "1Y"]:
62 interval = "5m"
63 # TradingView interval unit: m, h, D, W, M, Y
64 if interval.endswith("H"):
65 interval = interval.lower()
66 if interval.endswith(("d", "w", "y")):
67 interval = interval.upper()
68
69 symbols = await tradingview_supported(symbol) # list of supported full symbols
70 if not symbols:
71 return {}
72 if not TOKEN.CHART_IMG:
73 await modify_progress(text="❌CHART_IMG_API is not set. Get it from: https://chart-img.com", force_update=True, **kwargs)
74 await asyncio.sleep(5)
75 return {}
76 query_symbol, market = symbols[0] # the first supported symbol
77 logger.info(f"Fetching TradingView chart for {query_symbol} @{interval} in {market.capitalize()}...")
78
79 params = {
80 "theme": "dark",
81 "interval": interval,
82 "session": "extended",
83 "symbol": query_symbol,
84 "timezone": TZ,
85 "studies": [{"name": "Volume", "forceOverlay": True}, {"name": "MA Cross", "override": {"PlotCrosses.visible": False}}],
86 "override": {"showStudyLastValue": False},
87 }
88 logger.trace(params)
89 resp = await hx_req("https://api.chart-img.com/v2/tradingview/advanced-chart/storage", "POST", max_retry=0, headers={"x-api-key": TOKEN.CHART_IMG}, json_data=params, check_keys=["url"])
90 if error := resp.get("hx_error"):
91 await modify_progress(text=f"❌Failed to fetch TradingView chart for {query_symbol} @{interval} in {market.capitalize()}\n{error}", force_update=True, **kwargs)
92 return {}
93 return {"url": resp["url"], "symbol": query_symbol, "interval": interval} if resp["url"] else {}