from typing import Dict, Any, List from datetime import datetime from .base_handler import BaseHandler from services.crypto_service import CryptoService class CryptoHandler(BaseHandler): """加密货币价格处理器""" def __init__(self, crypto_service: CryptoService): self.crypto_service = crypto_service def handle(self, pair: str = "ETH-USDT", **kwargs) -> List[Dict[str, Any]]: """处理加密货币价格请求""" data = self.crypto_service.get_cached_data(f"{pair.lower()}_price", pair=pair) if not data: return [{ "title": f"{pair} 价格获取失败", "description": "无法获取当前价格,请稍后重试", "link": "https://www.okx.com", "pub_date": datetime.now().strftime("%a, %d %b %Y %H:%M:%S +0800"), "guid": f"{pair}-error-{int(datetime.now().timestamp())}" }] # 格式化价格信息 price = data['price'] change_24h = data['change_24h'] change_symbol = "📈" if change_24h >= 0 else "📉" title = f"{pair.replace('-', '/')} 当前价格: ${price:,.4f}" description = f""" 🔸 当前价格: ${price:,.4f} 🔸 24小时涨跌: {change_symbol} {change_24h:+.2f}% 🔸 24小时最高: ${data['high_24h']:,.4f} 🔸 24小时最低: ${data['low_24h']:,.4f} 🔸 24小时成交量: {data['volume_24h']:,.2f} 🔸 更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} """.strip() return [{ "title": title, "description": description, "link": f"https://www.okx.com/trade-spot/{pair.lower()}", "pub_date": datetime.now().strftime("%a, %d %b %Y %H:%M:%S +0800"), "guid": f"{pair}-{int(datetime.now().timestamp())}" }] def get_handler_name(self) -> str: return "crypto"