etf/archive/tests/test_trading_calendar.py

#!/usr/bin/env python3
"""
测试交易日历 API
"""

import sys
from pathlib import Path
import requests

# Flask 服务地址
FLASK_API_URL = "http://localhost:80"

def test_calendar_api():
    """测试交易日历 API"""
    print("\n" + "="*80)
    print("📅 交易日历 API 测试")
    print("="*80)
    
    # 测试 1: A 股
    print("\n[1] 测试 A 股交易日历...")
    url = f"{FLASK_API_URL}/api/v1/trading-calendar"
    params = {"market": "A", "start": "2024-01-01", "end": "2024-01-31"}
    
    try:
        response = requests.get(url, params=params, timeout=10)
        if response.status_code == 200:
            data = response.json()
            print(f"   ✅ 成功: {data['count']} 个交易日")
            print(f"   市场: {data['market']}")
            print(f"   交易所: {data['exchange']}")
            print(f"   日期范围: {data['start']} ~ {data['end']}")
            print(f"   前5个交易日: {data['trading_dates'][:5]}")
        else:
            print(f"   ❌ 失败: {response.status_code}")
            print(f"   响应: {response.json()}")
    except Exception as e:
        print(f"   ❌ 异常: {e}")
    
    # 测试 2: 美股
    print("\n[2] 测试美股交易日历...")
    params = {"market": "US", "start": "2024-01-01", "end": "2024-01-31"}
    
    try:
        response = requests.get(url, params=params, timeout=10)
        if response.status_code == 200:
            data = response.json()
            print(f"   ✅ 成功: {data['count']} 个交易日")
            print(f"   市场: {data['market']}")
            print(f"   交易所: {data['exchange']}")
            print(f"   前5个交易日: {data['trading_dates'][:5]}")
        else:
            print(f"   ❌ 失败: {response.status_code}")
            print(f"   响应: {response.json()}")
    except Exception as e:
        print(f"   ❌ 异常: {e}")
    
    # 测试 3: 港股
    print("\n[3] 测试港股交易日历...")
    params = {"market": "HK", "start": "2024-01-01", "end": "2024-01-31"}
    
    try:
        response = requests.get(url, params=params, timeout=10)
        if response.status_code == 200:
            data = response.json()
            print(f"   ✅ 成功: {data['count']} 个交易日")
            print(f"   市场: {data['market']}")
            print(f"   交易所: {data['exchange']}")
            print(f"   前5个交易日: {data['trading_dates'][:5]}")
        else:
            print(f"   ❌ 失败: {response.status_code}")
            print(f"   响应: {response.json()}")
    except Exception as e:
        print(f"   ❌ 异常: {e}")
    
    # 测试 4: 日历信息
    print("\n[4] 测试日历信息...")
    url_info = f"{FLASK_API_URL}/api/v1/calendar/info"
    
    try:
        response = requests.get(url_info, timeout=10)
        if response.status_code == 200:
            data = response.json()
            print(f"   ✅ 成功")
            print(f"   支持的市场:")
            for market, info in data.get('supported_markets', {}).items():
                print(f"      {market}: {info['name']} ({info['method']})")
            print(f"   pandas_market_calendars: {'✅ 已安装' if data.get('pandas_market_calendars_installed') else '❌ 未安装'}")
        else:
            print(f"   ❌ 失败: {response.status_code}")
    except Exception as e:
        print(f"   ❌ 异常: {e}")

def test_local_fetcher():
    """测试本地 UniversalDataFetcher"""
    print("\n" + "="*80)
    print("🧪 本地 UniversalDataFetcher 测试")
    print("="*80)
    
    sys.path.insert(0, str(Path(__file__).parent.parent))
    
    try:
        from datasource.universal_fetcher import UniversalDataFetcher
        
        fetcher = UniversalDataFetcher()
        
        # 测试 A 股
        print("\n[1] A 股交易日历 (2024年)...")
        cal_a = fetcher.get_trading_calendar('A', '2024-01-01', '2024-12-31')
        print(f"   ✅ {len(cal_a)} 个交易日")
        print(f"   前5天: {list(cal_a[:5])}")
        
        # 测试美股
        print("\n[2] 美股交易日历 (2024年)...")
        cal_us = fetcher.get_trading_calendar('US', '2024-01-01', '2024-12-31')
        print(f"   ✅ {len(cal_us)} 个交易日")
        print(f"   前5天: {list(cal_us[:5])}")
        
        # 测试港股
        print("\n[3] 港股交易日历 (2024年)...")
        cal_hk = fetcher.get_trading_calendar('HK', '2024-01-01', '2024-12-31')
        print(f"   ✅ {len(cal_hk)} 个交易日")
        print(f"   前5天: {list(cal_hk[:5])}")
        
        # 日历信息
        print("\n[4] 日历支持信息...")
        info = fetcher.get_calendar_info()
        print(f"   ✅ 支持 {len(info['supported_markets'])} 个市场")
        
    except Exception as e:
        print(f"   ❌ 失败: {e}")
        import traceback
        traceback.print_exc()

def main():
    print("\n" + "="*80)
    print("📅 交易日历功能测试")
    print("="*80)
    
    # 测试 1: 本地 fetcher
    test_local_fetcher()
    
    # 测试 2: Flask API（如果服务在运行）
    print("\n" + "="*80)
    print("🌐 测试 Flask API 端点")
    print("="*80)
    print(f"\nAPI 地址: {FLASK_API_URL}")
    print("注意: 需要 Flask 服务正在运行")
    
    try:
        response = requests.get(f"{FLASK_API_URL}/health", timeout=3)
        if response.status_code == 200:
            print("✅ Flask 服务可访问")
            test_calendar_api()
        else:
            print(f"⚠️  Flask 服务返回 {response.status_code}，跳过 API 测试")
    except:
        print("⚠️  Flask 服务未运行，跳过 API 测试")
    
    print("\n" + "="*80)
    print("✅ 测试完成")
    print("="*80)

if __name__ == "__main__":
    main()