#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 股票数据API接口 提供便捷的股票数据获取接口，支持完整的降级机制 """ import sys import os from typing import Dict, List, Optional, Any from datetime import datetime, timedelta # 添加dataflows目录到路径 dataflows_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'dataflows') if dataflows_path not in sys.path: sys.path.append(dataflows_path) try: from stock_data_service import get_stock_data_service SERVICE_AVAILABLE = True except ImportError as e: print(f"⚠️ 股票数据服务不可用: {e}") SERVICE_AVAILABLE = False def get_stock_info(stock_code: str) -> Dict[str, Any]: """ 获取单个股票的基础信息 Args: stock_code: 股票代码（如 '000001'） Returns: Dict: 股票基础信息 Example: >>> info = get_stock_info('000001') >>> print(info['name']) # 平安银行 """ if not SERVICE_AVAILABLE: return { 'error': '股票数据服务不可用', 'code': stock_code, 'suggestion': '请检查服务配置' } service = get_stock_data_service() result = service.get_stock_basic_info(stock_code) if result is None: return { 'error': f'未找到股票{stock_code}的信息', 'code': stock_code, 'suggestion': '请检查股票代码是否正确' } return result def get_all_stocks() -> List[Dict[str, Any]]: """ 获取所有股票的基础信息 Returns: List[Dict]: 所有股票的基础信息列表 Example: >>> stocks = get_all_stocks() >>> print(f"共有{len(stocks)}只股票") """ if not SERVICE_AVAILABLE: return [{ 'error': '股票数据服务不可用', 'suggestion': '请检查服务配置' }] service = get_stock_data_service() result = service.get_stock_basic_info() if result is None or (isinstance(result, dict) and 'error' in result): return [{ 'error': '无法获取股票列表', 'suggestion': '请检查网络连接和数据库配置' }] return result if isinstance(result, list) else [result] def get_stock_data(stock_code: str, start_date: str = None, end_date: str = None) -> str: """ 获取股票历史数据（带降级机制） Args: stock_code: 股票代码 start_date: 开始日期（格式：YYYY-MM-DD），默认为30天前 end_date: 结束日期（格式：YYYY-MM-DD），默认为今天 Returns: str: 股票数据的字符串表示或错误信息 Example: >>> data = get_stock_data('000001', '2024-01-01', '2024-01-31') >>> print(data) """ if not SERVICE_AVAILABLE: return "❌ 股票数据服务不可用，请检查服务配置" # 设置默认日期 if end_date is None: end_date = datetime.now().strftime('%Y-%m-%d') if start_date is None: start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d') service = get_stock_data_service() return service.get_stock_data_with_fallback(stock_code, start_date, end_date) def search_stocks(keyword: str) -> List[Dict[str, Any]]: """ 根据关键词搜索股票 Args: keyword: 搜索关键词（股票代码或名称的一部分） Returns: List[Dict]: 匹配的股票信息列表 Example: >>> results = search_stocks('平安') >>> for stock in results: ... print(f"{stock['code']}: {stock['name']}") """ all_stocks = get_all_stocks() if not all_stocks or (len(all_stocks) == 1 and 'error' in all_stocks[0]): return all_stocks # 搜索匹配的股票 matches = [] keyword_lower = keyword.lower() for stock in all_stocks: if 'error' in stock: continue code = stock.get('code', '').lower() name = stock.get('name', '').lower() if keyword_lower in code or keyword_lower in name: matches.append(stock) return matches def get_market_summary() -> Dict[str, Any]: """ 获取市场概览信息 Returns: Dict: 市场统计信息 Example: >>> summary = get_market_summary() >>> print(f"沪市股票数量: {summary['shanghai_count']}") """ all_stocks = get_all_stocks() if not all_stocks or (len(all_stocks) == 1 and 'error' in all_stocks[0]): return { 'error': '无法获取市场数据', 'suggestion': '请检查网络连接和数据库配置' } # 统计市场信息 shanghai_count = 0 shenzhen_count = 0 category_stats = {} for stock in all_stocks: if 'error' in stock: continue market = stock.get('market', '') category = stock.get('category', '未知') if market == '上海': shanghai_count += 1 elif market == '深圳': shenzhen_count += 1 category_stats[category] = category_stats.get(category, 0) + 1 return { 'total_count': len([s for s in all_stocks if 'error' not in s]), 'shanghai_count': shanghai_count, 'shenzhen_count': shenzhen_count, 'category_stats': category_stats, 'data_source': all_stocks[0].get('source', 'unknown') if all_stocks else 'unknown', 'updated_at': datetime.now().isoformat() } def check_service_status() -> Dict[str, Any]: """ 检查服务状态 Returns: Dict: 服务状态信息 Example: >>> status = check_service_status() >>> print(f"MongoDB状态: {status['mongodb_status']}") """ if not SERVICE_AVAILABLE: return { 'service_available': False, 'error': '股票数据服务不可用', 'suggestion': '请检查服务配置和依赖' } service = get_stock_data_service() # 检查MongoDB状态 mongodb_status = 'disconnected' if service.db_manager and service.db_manager.mongodb_db: try: # 尝试执行一个简单的查询来测试连接 service.db_manager.mongodb_db.list_collection_names() mongodb_status = 'connected' except Exception: mongodb_status = 'error' # 检查通达信API状态 tdx_status = 'unavailable' if service.tdx_provider: try: # 尝试获取一个股票名称来测试API test_name = service.tdx_provider._get_stock_name('000001') if test_name and test_name != '000001': tdx_status = 'available' else: tdx_status = 'limited' except Exception: tdx_status = 'error' return { 'service_available': True, 'mongodb_status': mongodb_status, 'tdx_api_status': tdx_status, 'enhanced_fetcher_available': hasattr(service, '_get_from_tdx_api'), 'fallback_available': True, 'checked_at': datetime.now().isoformat() } # 便捷的别名函数 get_stock = get_stock_info # 别名 get_stocks = get_all_stocks # 别名 search = search_stocks # 别名 status = check_service_status # 别名 if __name__ == '__main__': # 简单的命令行测试 print("🔍 股票数据API测试") print("=" * 50) # 检查服务状态 print("\n📊 服务状态检查:") status_info = check_service_status() for key, value in status_info.items(): print(f" {key}: {value}") # 测试获取单个股票信息 print("\n🏢 获取平安银行信息:") stock_info = get_stock_info('000001') if 'error' not in stock_info: print(f" 代码: {stock_info.get('code')}") print(f" 名称: {stock_info.get('name')}") print(f" 市场: {stock_info.get('market')}") print(f" 类别: {stock_info.get('category')}") print(f" 数据源: {stock_info.get('source')}") else: print(f" 错误: {stock_info.get('error')}") # 测试搜索功能 print("\n🔍 搜索'平安'相关股票:") search_results = search_stocks('平安') for i, stock in enumerate(search_results[:3]): # 只显示前3个结果 if 'error' not in stock: print(f" {i+1}. {stock.get('code')}: {stock.get('name')}") # 测试市场概览 print("\n📈 市场概览:") summary = get_market_summary() if 'error' not in summary: print(f" 总股票数: {summary.get('total_count')}") print(f" 沪市股票: {summary.get('shanghai_count')}") print(f" 深市股票: {summary.get('shenzhen_count')}") print(f" 数据源: {summary.get('data_source')}") else: print(f" 错误: {summary.get('error')}")