|
|
""" |
|
|
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', |
|
|
] |
|
|
|