"""Cryptocurrency mention detection and Kraken price lookup."""
from __future__ import annotations
import asyncio
import logging
import re
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import aiohttp
from .fetch_common import _HEADERS, _TIMEOUT
logger = logging.getLogger(__name__)
CRYPTO_SYMBOLS: dict[str, tuple[str, str]] = {
"btc": ("XXBTZUSD", "Bitcoin"), "bitcoin": ("XXBTZUSD", "Bitcoin"),
"sats": ("XXBTZUSD", "Bitcoin"), "satoshi": ("XXBTZUSD", "Bitcoin"),
"satoshis": ("XXBTZUSD", "Bitcoin"), "xbt": ("XXBTZUSD", "Bitcoin"),
"eth": ("XETHZUSD", "Ethereum"), "ethereum": ("XETHZUSD", "Ethereum"),
"ether": ("XETHZUSD", "Ethereum"), "gwei": ("XETHZUSD", "Ethereum"),
"xmr": ("XXMRZUSD", "Monero"), "monero": ("XXMRZUSD", "Monero"),
"sol": ("SOLUSD", "Solana"), "solana": ("SOLUSD", "Solana"),
"doge": ("XDGUSD", "Dogecoin"), "dogecoin": ("XDGUSD", "Dogecoin"),
"shibe": ("XDGUSD", "Dogecoin"),
"ltc": ("XLTCZUSD", "Litecoin"), "litecoin": ("XLTCZUSD", "Litecoin"),
"ada": ("ADAUSD", "Cardano"), "cardano": ("ADAUSD", "Cardano"),
"xrp": ("XXRPZUSD", "XRP"), "ripple": ("XXRPZUSD", "XRP"),
"polkadot": ("DOTUSD", "Polkadot"),
"chainlink": ("LINKUSD", "Chainlink"),
"avax": ("AVAXUSD", "Avalanche"), "avalanche": ("AVAXUSD", "Avalanche"),
"matic": ("MATICUSD", "Polygon"), "polygon": ("MATICUSD", "Polygon"),
"atom": ("ATOMUSD", "Cosmos"), "cosmos": ("ATOMUSD", "Cosmos"),
"uni": ("UNIUSD", "Uniswap"), "uniswap": ("UNIUSD", "Uniswap"),
}
CASE_SENSITIVE_CRYPTO: dict[str, tuple[str, str]] = {
"LINK": ("LINKUSD", "Chainlink"),
"DOT": ("DOTUSD", "Polkadot"),
}
[docs]
def detect_crypto_mentions(text: str) -> List[tuple]:
if not text:
return []
found: dict[str, str] = {}
for w in re.findall(r"\b[a-zA-Z]+\b", text.lower()):
if w in CRYPTO_SYMBOLS:
pair, name = CRYPTO_SYMBOLS[w]
found.setdefault(pair, name)
for w in re.findall(r"\b[a-zA-Z]+\b", text):
if w in CASE_SENSITIVE_CRYPTO:
pair, name = CASE_SENSITIVE_CRYPTO[w]
found.setdefault(pair, name)
return list(found.items())
[docs]
async def get_crypto_prices(pairs: List[tuple]) -> Optional[Dict[str, Any]]:
if not pairs:
return None
try:
pair_names = [p[0] for p in pairs]
pair_str = ",".join(pair_names)
api = f"https://api.kraken.com/0/public/Ticker?pair={pair_str}"
async with aiohttp.ClientSession() as s:
async with s.get(api, timeout=_TIMEOUT, headers=_HEADERS) as r:
if r.status != 200:
return None
d = await r.json()
if d.get("error"):
return None
rd = d.get("result", {})
prices = []
sym_map = {
"XBT": "BTC", "XDG": "DOGE", "XLM": "XLM",
"ETH": "ETH", "XMR": "XMR", "LTC": "LTC",
"XRP": "XRP", "SOL": "SOL", "ADA": "ADA",
"DOT": "DOT", "LINK": "LINK", "AVAX": "AVAX",
"MATIC": "MATIC", "ATOM": "ATOM", "UNI": "UNI",
}
for kp, dn in pairs:
td = rd.get(kp)
if not td:
for k in rd:
if kp in k or k in kp:
td = rd[k]
break
if td:
sym = kp.replace("ZUSD", "").replace("USD", "")
if sym.startswith("X"):
sym = sym[1:]
sym = sym_map.get(sym, sym)
cur = float(td["c"][0])
opn = float(td["o"])
chg = ((cur - opn) / opn * 100) if opn else 0.0
prices.append({
"name": dn, "symbol": sym, "price": cur,
"change_24h": round(chg, 2),
"high_24h": float(td["h"][1]),
"low_24h": float(td["l"][1]),
"volume_24h": float(td["v"][1]),
})
if not prices:
return None
ts = datetime.now(timezone.utc).isoformat()
return {"prices": prices, "timestamp": ts}
except asyncio.TimeoutError:
logger.error("Timeout fetching crypto prices")
except Exception:
logger.exception("Error fetching crypto prices")
return None