| """
|
| Utils package - Consolidated utility functions
|
| Provides logging setup and other utility functions for the application
|
| """
|
|
|
|
|
| try:
|
| from .logger import setup_logger
|
| except ImportError as e:
|
| print(f"ERROR: Failed to import setup_logger from .logger: {e}")
|
| import logging
|
| def setup_logger(name: str, level: str = "INFO") -> logging.Logger:
|
| """Fallback setup_logger if import fails"""
|
| logger = logging.getLogger(name)
|
| if not logger.handlers:
|
| handler = logging.StreamHandler()
|
| handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
|
| logger.addHandler(handler)
|
| logger.setLevel(getattr(logging, level.upper()))
|
| return logger
|
|
|
|
|
|
|
| def setup_logging():
|
| """
|
| Setup logging for the application
|
| This is a compatibility wrapper around setup_logger
|
|
|
| Returns:
|
| logging.Logger: Configured logger instance
|
| """
|
| return setup_logger("crypto_aggregator", level="INFO")
|
|
|
|
|
|
|
|
|
| import sys
|
| import os
|
|
|
|
|
| parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
| if parent_dir not in sys.path:
|
| sys.path.insert(0, parent_dir)
|
|
|
|
|
| try:
|
|
|
| import importlib.util
|
| utils_path = os.path.join(parent_dir, 'utils.py')
|
| spec = importlib.util.spec_from_file_location("utils_standalone", utils_path)
|
| if spec and spec.loader:
|
| utils_standalone = importlib.util.module_from_spec(spec)
|
| spec.loader.exec_module(utils_standalone)
|
|
|
|
|
| format_number = utils_standalone.format_number
|
| calculate_moving_average = utils_standalone.calculate_moving_average
|
| calculate_rsi = utils_standalone.calculate_rsi
|
| extract_coins_from_text = utils_standalone.extract_coins_from_text
|
| export_to_csv = utils_standalone.export_to_csv
|
| validate_price_data = utils_standalone.validate_price_data
|
| is_data_stale = utils_standalone.is_data_stale
|
| cache_with_ttl = utils_standalone.cache_with_ttl
|
| safe_float = utils_standalone.safe_float
|
| safe_int = utils_standalone.safe_int
|
| truncate_string = utils_standalone.truncate_string
|
| percentage_change = utils_standalone.percentage_change
|
| except Exception as e:
|
| print(f"Warning: Could not import from standalone utils.py: {e}")
|
|
|
| def format_number(num, decimals=2):
|
| return str(num)
|
| def calculate_moving_average(prices, period):
|
| return None
|
| def calculate_rsi(prices, period=14):
|
| return None
|
| def extract_coins_from_text(text):
|
| return []
|
| def export_to_csv(data, filename):
|
| return False
|
| def validate_price_data(price_data):
|
| return True
|
| def is_data_stale(timestamp_str, max_age_minutes=30):
|
| return False
|
| def cache_with_ttl(ttl_seconds=300):
|
| def decorator(func):
|
| return func
|
| return decorator
|
| def safe_float(value, default=0.0):
|
| return default
|
| def safe_int(value, default=0):
|
| return default
|
| def truncate_string(text, max_length=100, suffix="..."):
|
| return text
|
| def percentage_change(old_value, new_value):
|
| return None
|
|
|
|
|
| __all__ = [
|
| 'setup_logging',
|
| 'setup_logger',
|
| 'format_number',
|
| 'calculate_moving_average',
|
| 'calculate_rsi',
|
| 'extract_coins_from_text',
|
| 'export_to_csv',
|
| 'validate_price_data',
|
| 'is_data_stale',
|
| 'cache_with_ttl',
|
| 'safe_float',
|
| 'safe_int',
|
| 'truncate_string',
|
| 'percentage_change',
|
| ]
|
|
|