File size: 2,322 Bytes
5cd2b89 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 |
import os
import logging
import ccxt
from logging.handlers import RotatingFileHandler
from datetime import datetime
import json
from typing import List, Dict, Any
def setup_logger():
"""Configure and set up the logger"""
# Create logs directory if it doesn't exist
os.makedirs("logs", exist_ok=True)
# Configure logger
logger = logging.getLogger("crypto_data_source")
logger.setLevel(logging.INFO)
# Create console handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# Create file handler
log_file = f"logs/crypto_data_source_{datetime.now().strftime('%Y%m%d')}.log"
file_handler = RotatingFileHandler(
log_file, maxBytes=10*1024*1024, backupCount=5
)
file_handler.setLevel(logging.INFO)
# Create formatter
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# Add formatter to handlers
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)
# Add handlers to logger
logger.addHandler(console_handler)
logger.addHandler(file_handler)
return logger
def get_available_exchanges() -> List[str]:
"""Get list of available exchanges with OHLCV support"""
exchanges = []
# Focus on major exchanges that are reliable and have good API support
preferred_exchanges = [
'binance', 'kucoin', 'coinbase', 'kraken', 'okx', 'bybit'
]
for exchange_id in preferred_exchanges:
try:
exchange_class = getattr(ccxt, exchange_id)
exchange = exchange_class()
# Only include exchanges that support OHLCV data
if exchange.has['fetchOHLCV']:
exchanges.append(exchange_id)
except:
# Skip exchanges that don't initialize properly
continue
return exchanges
def format_timestamp(timestamp_ms: int) -> str:
"""Format timestamp to ISO string"""
return datetime.fromtimestamp(timestamp_ms / 1000).isoformat()
def safe_json_serialize(obj):
"""Custom JSON serializer for handling non-serializable types"""
if isinstance(obj, (datetime)):
return obj.isoformat()
raise TypeError(f"Type {type(obj)} not serializable")
|