diff --git a/app.py b/app.py index 6b13db6..64be5ca 100644 --- a/app.py +++ b/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}") diff --git a/config.py.back b/config.py.back new file mode 100644 index 0000000..49cbfe5 --- /dev/null +++ b/config.py.back @@ -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 +} \ No newline at end of file diff --git a/handlers/crypto_handler.py b/handlers/crypto_handler.py index f2b3f0c..cef298b 100644 --- a/handlers/crypto_handler.py +++ b/handlers/crypto_handler.py @@ -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 [{ diff --git a/services/crypto_service.py b/services/crypto_service.py index e85cb06..82aa361 100644 --- a/services/crypto_service.py +++ b/services/crypto_service.py @@ -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数据 diff --git a/test.py b/test.py deleted file mode 100644 index 33661d0..0000000 --- a/test.py +++ /dev/null @@ -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() \ No newline at end of file diff --git a/utils/rss_generator.py b/utils/rss_generator.py index bb76c52..570f4af 100644 --- a/utils/rss_generator.py +++ b/utils/rss_generator.py @@ -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