优化相关设备的格式显示
This commit is contained in:
Coldin04
parent
b18805aa26
commit
4a3eaf87e8
6 changed files with 67 additions and 142 deletions
12
app.py
12
app.py
|
|
@ -62,10 +62,18 @@ def create_app(config_name='default'):
|
|||
# 让处理器处理请求
|
||||
items = handler.handle(**request_params)
|
||||
|
||||
# 生成RSS XML
|
||||
rss_xml = rss_generator.generate_rss(items)
|
||||
if not items:
|
||||
return Response("No data available", status=204)
|
||||
|
||||
if handler_name == 'crypto':
|
||||
motdTitle = request_params.get('pair', app.config['DEFAULT_CRYPTO_PAIR'])
|
||||
motdTitle = motdTitle.replace('-', '/').upper()
|
||||
rss_xml = rss_generator.generate_rss(items, motdTitle=motdTitle)
|
||||
else:
|
||||
rss_xml = rss_generator.generate_rss(items)
|
||||
|
||||
return Response(rss_xml, mimetype='application/rss+xml')
|
||||
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"生成RSS时出错: {e}")
|
||||
|
|
|
|||
41
config.py.back
Normal file
41
config.py.back
Normal file
|
|
@ -0,0 +1,41 @@
|
|||
import os
|
||||
from datetime import timedelta
|
||||
|
||||
class Config:
|
||||
"""应用配置类"""
|
||||
|
||||
# Flask配置
|
||||
SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-secret-key-change-in-production')
|
||||
|
||||
# OKX配置（公共接口不需要API密钥）
|
||||
OKX_API_KEY = os.environ.get('OKX_API_KEY', '')
|
||||
OKX_SECRET_KEY = os.environ.get('OKX_SECRET_KEY', '')
|
||||
OKX_PASSPHRASE = os.environ.get('OKX_PASSPHRASE', '')
|
||||
|
||||
# RSS配置
|
||||
RSS_TITLE = "Crypto RSS Service"
|
||||
RSS_DESCRIPTION = "Real-time cryptocurrency prices and more"
|
||||
RSS_LANGUAGE = "zh-cn"
|
||||
RSS_LINK = "http://localhost:5000"
|
||||
|
||||
# 缓存配置
|
||||
CACHE_TIMEOUT = timedelta(seconds=30) # 30秒缓存，适合低频率刷新
|
||||
|
||||
# 服务配置
|
||||
DEFAULT_CRYPTO_PAIR = "ETH-USDT"
|
||||
|
||||
@staticmethod
|
||||
def init_app(app):
|
||||
pass
|
||||
|
||||
class DevelopmentConfig(Config):
|
||||
DEBUG = True
|
||||
|
||||
class ProductionConfig(Config):
|
||||
DEBUG = False
|
||||
|
||||
config = {
|
||||
'development': DevelopmentConfig,
|
||||
'production': ProductionConfig,
|
||||
'default': DevelopmentConfig
|
||||
}
|
||||
|
|
@ -2,15 +2,21 @@ 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):
|
||||
def __init__(self, crypto_service: CryptoService, config=None):
|
||||
self.crypto_service = crypto_service
|
||||
self.config = config or Config # 使用传入的配置或默认配置
|
||||
|
||||
def handle(self, pair: str = "ETH-USDT", **kwargs) -> List[Dict[str, Any]]:
|
||||
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:
|
||||
|
|
@ -25,17 +31,13 @@ class CryptoHandler(BaseHandler):
|
|||
# 格式化价格信息
|
||||
price = data['price']
|
||||
change_24h = data['change_24h']
|
||||
change_symbol = "📈" if change_24h >= 0 else "📉"
|
||||
change_symbol = "↑" if change_24h >= 0 else "↓"
|
||||
|
||||
title = f"{pair.replace('-', '/')} 当前价格: ${price:,.4f}"
|
||||
title = f"${price:,.2f} {change_symbol} {abs(change_24h):.2f}%"
|
||||
|
||||
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')}
|
||||
24小时涨跌: {change_symbol} {change_24h:+.2f}%
|
||||
更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
|
||||
""".strip()
|
||||
|
||||
return [{
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ from typing import Dict, Any, Optional
|
|||
import logging
|
||||
from okx import OkxRestClient
|
||||
from .base_service import BaseService
|
||||
from config import Config # 导入配置
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -13,7 +14,7 @@ class CryptoService(BaseService):
|
|||
# 对于公共数据，不需要API密钥
|
||||
self.client = OkxRestClient(api_key, secret_key, passphrase) if api_key else OkxRestClient()
|
||||
|
||||
def get_data(self, pair: str = "ETH-USDT") -> Optional[Dict[str, Any]]:
|
||||
def get_data(self, pair: str = Config.DEFAULT_CRYPTO_PAIR) -> Optional[Dict[str, Any]]:
|
||||
"""获取指定交易对的价格数据"""
|
||||
try:
|
||||
# 使用正确的marketdata属性获取ticker数据
|
||||
|
|
|
|||
127
test.py
127
test.py
|
|
@ -1,127 +0,0 @@
|
|||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
OKX marketdata属性测试脚本
|
||||
用于验证通过marketdata属性获取价格数据
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pprint import pprint
|
||||
|
||||
# 导入必要的库
|
||||
try:
|
||||
from okx import OkxRestClient
|
||||
print("成功导入 okx.OkxRestClient")
|
||||
except ImportError:
|
||||
print("错误: 无法导入OKX SDK, 请安装: pip install okx-sdk")
|
||||
sys.exit(1)
|
||||
|
||||
def test_marketdata():
|
||||
"""测试通过marketdata属性获取价格"""
|
||||
print("\n=== 测试 marketdata.get_ticker 方法 ===")
|
||||
|
||||
# 初始化客户端 (不使用API密钥)
|
||||
client = OkxRestClient()
|
||||
|
||||
# 检查是否有marketdata属性
|
||||
if not hasattr(client, 'marketdata'):
|
||||
print("❌ 错误: client对象没有marketdata属性")
|
||||
print(f"可用属性: {', '.join([a for a in dir(client) if not a.startswith('_')])}")
|
||||
return
|
||||
|
||||
print("✅ 找到marketdata属性")
|
||||
|
||||
# 检查marketdata对象是否有get_ticker方法
|
||||
marketdata = client.marketdata
|
||||
if not hasattr(marketdata, 'get_ticker'):
|
||||
print("❌ 错误: marketdata对象没有get_ticker方法")
|
||||
print(f"可用方法: {', '.join([a for a in dir(marketdata) if not a.startswith('_')])}")
|
||||
return
|
||||
|
||||
print("✅ 找到get_ticker方法")
|
||||
|
||||
# 测试获取ETH-USDT价格
|
||||
try:
|
||||
pair = "ETH-USDT"
|
||||
print(f"\n获取 {pair} 价格...")
|
||||
response = client.marketdata.get_ticker(instId=pair)
|
||||
|
||||
print(f"响应状态码: {response.get('code')}")
|
||||
if response and response.get('code') == '0' and response.get('data'):
|
||||
print("✅ 成功获取价格数据!")
|
||||
|
||||
ticker_data = response['data'][0]
|
||||
price = float(ticker_data['last'])
|
||||
|
||||
print(f"\n{pair} 当前价格: {price} USDT")
|
||||
print(f"24小时最高价: {ticker_data.get('high24h', 'N/A')}")
|
||||
print(f"24小时最低价: {ticker_data.get('low24h', 'N/A')}")
|
||||
print(f"24小时涨跌幅: {ticker_data.get('pctChange', 'N/A')}%") # 使用get方法避免KeyError
|
||||
|
||||
print("\n完整响应数据:")
|
||||
pprint(response)
|
||||
else:
|
||||
print("❌ 请求成功但返回错误:")
|
||||
pprint(response)
|
||||
except Exception as e:
|
||||
print(f"❌ 请求过程中出错: {e}")
|
||||
|
||||
def test_with_api_keys():
|
||||
"""测试使用API密钥的情况"""
|
||||
print("\n=== 测试使用API密钥 ===")
|
||||
|
||||
# 填写您的API密钥信息
|
||||
api_key = input("请输入API密钥 (或直接回车跳过): ").strip()
|
||||
if not api_key:
|
||||
print("跳过API密钥测试")
|
||||
return
|
||||
|
||||
secret_key = input("请输入Secret Key: ").strip()
|
||||
passphrase = input("请输入Passphrase: ").strip()
|
||||
|
||||
# 初始化带API密钥的客户端
|
||||
client = OkxRestClient(api_key, secret_key, passphrase)
|
||||
|
||||
# 测试获取ETH-USDT价格
|
||||
try:
|
||||
pair = "ETH-USDT"
|
||||
print(f"\n使用API密钥获取 {pair} 价格...")
|
||||
response = client.marketdata.get_ticker(instId=pair)
|
||||
|
||||
if response and response.get('code') == '0' and response.get('data'):
|
||||
print("✅ 使用API密钥成功获取价格!")
|
||||
price = float(response['data'][0]['last'])
|
||||
print(f"{pair} 当前价格: {price} USDT")
|
||||
else:
|
||||
print("❌ 使用API密钥请求失败:")
|
||||
pprint(response)
|
||||
except Exception as e:
|
||||
print(f"❌ 使用API密钥请求过程中出错: {e}")
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
print("OKX Marketdata属性测试")
|
||||
print("-" * 50)
|
||||
|
||||
# 检查SDK版本
|
||||
try:
|
||||
import pkg_resources
|
||||
print(f"SDK版本: {pkg_resources.get_distribution('okx-sdk').version}")
|
||||
except Exception as e:
|
||||
print(f"无法获取SDK版本: {e}")
|
||||
|
||||
# 测试不带API密钥
|
||||
test_marketdata()
|
||||
|
||||
# 询问是否测试API密钥
|
||||
print("\n要测试使用API密钥吗?")
|
||||
choice = input("输入 'y' 进行测试, 或任意键跳过: ")
|
||||
|
||||
if choice.lower() == 'y':
|
||||
test_with_api_keys()
|
||||
|
||||
print("\n测试完成!")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -11,14 +11,14 @@ class RSSGenerator:
|
|||
self.link = link
|
||||
self.language = language
|
||||
|
||||
def generate_rss(self, items: List[Dict[str, Any]]) -> str:
|
||||
def generate_rss(self, items: List[Dict[str, Any]], motdTitle: str = None) -> str:
|
||||
"""生成RSS XML"""
|
||||
# 创建根元素
|
||||
rss = ET.Element("rss", version="2.0")
|
||||
channel = ET.SubElement(rss, "channel")
|
||||
|
||||
# 频道信息
|
||||
ET.SubElement(channel, "title").text = self.title
|
||||
ET.SubElement(channel, "title").text = motdTitle or self.title
|
||||
ET.SubElement(channel, "description").text = self.description
|
||||
ET.SubElement(channel, "link").text = self.link
|
||||
ET.SubElement(channel, "language").text = self.language
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue