52 lines
No EOL
2 KiB
Python
52 lines
No EOL
2 KiB
Python
from typing import Dict, Any, List
|
|
from datetime import datetime
|
|
from .base_handler import BaseHandler
|
|
from services.crypto_service import CryptoService
|
|
from config import Config # 导入配置
|
|
|
|
class CryptoHandler(BaseHandler):
|
|
"""加密货币价格处理器"""
|
|
|
|
def __init__(self, crypto_service: CryptoService, config=None):
|
|
self.crypto_service = crypto_service
|
|
self.config = config or Config # 使用传入的配置或默认配置
|
|
|
|
def handle(self, pair: str = None, **kwargs) -> List[Dict[str, Any]]:
|
|
"""处理加密货币价格请求"""
|
|
# 如果未指定交易对，则使用配置中的默认值
|
|
if pair is None:
|
|
pair = self.config.DEFAULT_CRYPTO_PAIR
|
|
|
|
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"${price:,.2f} {change_symbol} {abs(change_24h):.2f}%"
|
|
|
|
description = f"""
|
|
24小时涨跌: {change_symbol} {change_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" |