- Add base API client and cache manager classes - Implement FMP and YFinance specific clients and cache managers - Add API factory for managing multiple data providers - Add test suite for API configuration and caching - Add logging configuration for API operations
87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Cache System Test
|
|
|
|
Tests the caching behavior with consecutive API calls.
|
|
"""
|
|
|
|
import logging
|
|
import time
|
|
from api_client import APIClient
|
|
|
|
# Setup logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def run_test(api_key: str):
|
|
"""Run cache test with consecutive API calls."""
|
|
# Initialize API client
|
|
api_client = APIClient(api_key=api_key)
|
|
|
|
# Test data
|
|
test_tickers = ["SPY", "QQQ", "VTI"]
|
|
|
|
logger.info("\n=== First Run (Should make API calls) ===")
|
|
for ticker in test_tickers:
|
|
logger.info(f"\nTesting {ticker}:")
|
|
|
|
# Get profile
|
|
start_time = time.time()
|
|
profile = api_client.get_profile(ticker)
|
|
duration = time.time() - start_time
|
|
logger.info(f"Profile data received in {duration:.2f}s")
|
|
|
|
# Get historical data
|
|
start_time = time.time()
|
|
historical = api_client.get_historical_data(ticker, timeframe="1d")
|
|
duration = time.time() - start_time
|
|
logger.info(f"Historical data received in {duration:.2f}s")
|
|
|
|
# Get holdings
|
|
start_time = time.time()
|
|
holdings = api_client.get_holdings(ticker)
|
|
duration = time.time() - start_time
|
|
logger.info(f"Holdings data received in {duration:.2f}s")
|
|
|
|
# Wait a moment between runs
|
|
time.sleep(2)
|
|
|
|
logger.info("\n=== Second Run (Should use cache) ===")
|
|
for ticker in test_tickers:
|
|
logger.info(f"\nTesting {ticker}:")
|
|
|
|
# Get profile
|
|
start_time = time.time()
|
|
profile = api_client.get_profile(ticker)
|
|
duration = time.time() - start_time
|
|
logger.info(f"Profile data received in {duration:.2f}s")
|
|
|
|
# Get historical data
|
|
start_time = time.time()
|
|
historical = api_client.get_historical_data(ticker, timeframe="1d")
|
|
duration = time.time() - start_time
|
|
logger.info(f"Historical data received in {duration:.2f}s")
|
|
|
|
# Get holdings
|
|
start_time = time.time()
|
|
holdings = api_client.get_holdings(ticker)
|
|
duration = time.time() - start_time
|
|
logger.info(f"Holdings data received in {duration:.2f}s")
|
|
|
|
# Get cache statistics
|
|
from cache_manager import cache_manager
|
|
stats = cache_manager.get_stats()
|
|
logger.info("\n=== Cache Statistics ===")
|
|
logger.info(f"Cache hits: {stats['hits']}")
|
|
logger.info(f"Cache misses: {stats['misses']}")
|
|
logger.info(f"Hit rate: {stats['hit_rate']:.2%}")
|
|
logger.info(f"Total cache size: {stats['total_size']} bytes")
|
|
logger.info(f"Number of cache files: {stats['cache_files']}")
|
|
|
|
if __name__ == "__main__":
|
|
# Use your FMP API key
|
|
API_KEY = "fmp_live_8c8c8c8c8c8c8c8c8c8c8c8c8c8c8c8c" # Replace with your actual API key
|
|
run_test(API_KEY) |