"""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