logging channels; training steps storage
This commit is contained in:
Dobromir Popov
73
utils/log_control.py
Normal file
73
utils/log_control.py
Normal file
@@ -0,0 +1,73 @@
|
||||
"""
|
||||
Runtime Logging Control Utility
|
||||
|
||||
Allows enabling/disabling logging channels at runtime without restarting the application.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add parent directory to path
|
||||
parent_dir = Path(__file__).parent.parent
|
||||
sys.path.insert(0, str(parent_dir))
|
||||
|
||||
from utils.logging_config import (
|
||||
enable_channel,
|
||||
disable_channel,
|
||||
get_enabled_channels,
|
||||
print_channel_status,
|
||||
LogChannel
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
"""Interactive logging control"""
|
||||
print("\n" + "="*50)
|
||||
print(" Logging Channel Control")
|
||||
print("="*50)
|
||||
|
||||
while True:
|
||||
print("\nCommands:")
|
||||
print(" status - Show current channel status")
|
||||
print(" enable - Enable a channel")
|
||||
print(" disable - Disable a channel")
|
||||
print(" list - List all available channels")
|
||||
print(" quit - Exit")
|
||||
|
||||
cmd = input("\n> ").strip().lower()
|
||||
|
||||
if cmd == 'quit' or cmd == 'exit' or cmd == 'q':
|
||||
break
|
||||
|
||||
elif cmd == 'status':
|
||||
print_channel_status()
|
||||
|
||||
elif cmd == 'list':
|
||||
print("\nAvailable Channels:")
|
||||
print(f" - {LogChannel.CORE} (Core system operations)")
|
||||
print(f" - {LogChannel.TRADING} (Trading operations)")
|
||||
print(f" - {LogChannel.TRAINING} (Model training)")
|
||||
print(f" - {LogChannel.INFERENCE} (Model inference)")
|
||||
print(f" - {LogChannel.PIVOTS} (Pivot calculations)")
|
||||
print(f" - {LogChannel.DATA} (Data fetching/caching)")
|
||||
print(f" - {LogChannel.WEBSOCKET} (WebSocket communications)")
|
||||
print(f" - {LogChannel.API} (API requests/responses)")
|
||||
print(f" - {LogChannel.WEBUI} (Web UI chart requests)")
|
||||
print(f" - {LogChannel.PERFORMANCE} (Performance metrics)")
|
||||
print(f" - {LogChannel.DEBUG} (Debug information)")
|
||||
|
||||
elif cmd == 'enable':
|
||||
channel = input("Channel name: ").strip()
|
||||
enable_channel(channel)
|
||||
|
||||
elif cmd == 'disable':
|
||||
channel = input("Channel name: ").strip()
|
||||
disable_channel(channel)
|
||||
|
||||
else:
|
||||
print("Unknown command. Type 'quit' to exit.")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
233
utils/logging_config.py
Normal file
233
utils/logging_config.py
Normal file
@@ -0,0 +1,233 @@
|
||||
"""
|
||||
Modular Logging Configuration System
|
||||
|
||||
Provides granular control over logging channels for different subsystems.
|
||||
Configure which channels to enable/disable at startup.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
from typing import Dict, Set
|
||||
from pathlib import Path
|
||||
|
||||
# Define logging channels
|
||||
class LogChannel:
|
||||
"""Available logging channels"""
|
||||
CORE = "core" # Core system operations
|
||||
TRADING = "trading" # Trading operations
|
||||
TRAINING = "training" # Model training
|
||||
INFERENCE = "inference" # Model inference
|
||||
PIVOTS = "pivots" # Pivot calculations (Williams structure)
|
||||
DATA = "data" # Data fetching/caching
|
||||
WEBSOCKET = "websocket" # WebSocket communications
|
||||
API = "api" # API requests/responses
|
||||
WEBUI = "webui" # Web UI requests/responses
|
||||
PERFORMANCE = "performance" # Performance metrics
|
||||
DEBUG = "debug" # Debug information
|
||||
|
||||
# Default channel configuration (which channels are enabled)
|
||||
DEFAULT_CHANNEL_CONFIG = {
|
||||
LogChannel.CORE: True,
|
||||
LogChannel.TRADING: True,
|
||||
LogChannel.TRAINING: True,
|
||||
LogChannel.INFERENCE: True,
|
||||
LogChannel.PIVOTS: False, # Disabled by default (too verbose)
|
||||
LogChannel.DATA: True,
|
||||
LogChannel.WEBSOCKET: False, # Disabled by default
|
||||
LogChannel.API: False, # Disabled by default
|
||||
LogChannel.WEBUI: False, # Disabled by default (too verbose)
|
||||
LogChannel.PERFORMANCE: True,
|
||||
LogChannel.DEBUG: False # Disabled by default
|
||||
}
|
||||
|
||||
class ChannelLogger:
|
||||
"""Logger with channel-based filtering"""
|
||||
|
||||
_instance = None
|
||||
_initialized = False
|
||||
|
||||
def __new__(cls):
|
||||
if cls._instance is None:
|
||||
cls._instance = super().__new__(cls)
|
||||
return cls._instance
|
||||
|
||||
def __init__(self):
|
||||
if not self._initialized:
|
||||
self.enabled_channels: Set[str] = set()
|
||||
self.loggers: Dict[str, logging.Logger] = {}
|
||||
self._load_config()
|
||||
ChannelLogger._initialized = True
|
||||
|
||||
def _load_config(self):
|
||||
"""Load channel configuration from environment or use defaults"""
|
||||
# Load from environment variables (e.g., LOG_CHANNELS="core,trading,inference")
|
||||
env_channels = os.getenv('LOG_CHANNELS', None)
|
||||
|
||||
if env_channels:
|
||||
# Use environment config
|
||||
self.enabled_channels = set(env_channels.split(','))
|
||||
print(f"Logging channels (from env): {', '.join(sorted(self.enabled_channels))}")
|
||||
else:
|
||||
# Use default config
|
||||
self.enabled_channels = {
|
||||
channel for channel, enabled in DEFAULT_CHANNEL_CONFIG.items()
|
||||
if enabled
|
||||
}
|
||||
print(f"Logging channels (default): {', '.join(sorted(self.enabled_channels))}")
|
||||
|
||||
def get_logger(self, name: str, channel: str = LogChannel.CORE) -> logging.Logger:
|
||||
"""
|
||||
Get a logger for a specific channel
|
||||
|
||||
Args:
|
||||
name: Logger name (usually __name__)
|
||||
channel: Logging channel (from LogChannel)
|
||||
|
||||
Returns:
|
||||
Logger instance with channel filtering
|
||||
"""
|
||||
logger_key = f"{name}:{channel}"
|
||||
|
||||
if logger_key not in self.loggers:
|
||||
logger = logging.getLogger(name)
|
||||
|
||||
# Wrap logger to check channel before logging
|
||||
wrapped_logger = ChannelFilteredLogger(logger, channel, self)
|
||||
self.loggers[logger_key] = wrapped_logger
|
||||
|
||||
return self.loggers[logger_key]
|
||||
|
||||
def is_channel_enabled(self, channel: str) -> bool:
|
||||
"""Check if a channel is enabled"""
|
||||
return channel in self.enabled_channels
|
||||
|
||||
def enable_channel(self, channel: str):
|
||||
"""Enable a logging channel at runtime"""
|
||||
self.enabled_channels.add(channel)
|
||||
print(f"Enabled logging channel: {channel}")
|
||||
|
||||
def disable_channel(self, channel: str):
|
||||
"""Disable a logging channel at runtime"""
|
||||
self.enabled_channels.discard(channel)
|
||||
print(f"Disabled logging channel: {channel}")
|
||||
|
||||
def set_channels(self, channels: Set[str]):
|
||||
"""Set enabled channels"""
|
||||
self.enabled_channels = channels
|
||||
print(f"Logging channels updated: {', '.join(sorted(channels))}")
|
||||
|
||||
def get_enabled_channels(self) -> Set[str]:
|
||||
"""Get currently enabled channels"""
|
||||
return self.enabled_channels.copy()
|
||||
|
||||
|
||||
class ChannelFilteredLogger:
|
||||
"""Wrapper around logging.Logger that filters by channel"""
|
||||
|
||||
def __init__(self, logger: logging.Logger, channel: str, channel_logger: ChannelLogger):
|
||||
self.logger = logger
|
||||
self.channel = channel
|
||||
self.channel_logger = channel_logger
|
||||
|
||||
def _should_log(self) -> bool:
|
||||
"""Check if this channel should log"""
|
||||
return self.channel_logger.is_channel_enabled(self.channel)
|
||||
|
||||
def debug(self, msg, *args, **kwargs):
|
||||
if self._should_log():
|
||||
self.logger.debug(f"[{self.channel}] {msg}", *args, **kwargs)
|
||||
|
||||
def info(self, msg, *args, **kwargs):
|
||||
if self._should_log():
|
||||
self.logger.info(f"[{self.channel}] {msg}", *args, **kwargs)
|
||||
|
||||
def warning(self, msg, *args, **kwargs):
|
||||
if self._should_log():
|
||||
self.logger.warning(f"[{self.channel}] {msg}", *args, **kwargs)
|
||||
|
||||
def error(self, msg, *args, **kwargs):
|
||||
# Errors always log regardless of channel
|
||||
self.logger.error(f"[{self.channel}] {msg}", *args, **kwargs)
|
||||
|
||||
def exception(self, msg, *args, **kwargs):
|
||||
# Exceptions always log regardless of channel
|
||||
self.logger.exception(f"[{self.channel}] {msg}", *args, **kwargs)
|
||||
|
||||
def critical(self, msg, *args, **kwargs):
|
||||
# Critical always logs regardless of channel
|
||||
self.logger.critical(f"[{self.channel}] {msg}", *args, **kwargs)
|
||||
|
||||
|
||||
# Global instance
|
||||
_channel_logger = ChannelLogger()
|
||||
|
||||
|
||||
def get_channel_logger(name: str, channel: str = LogChannel.CORE) -> ChannelFilteredLogger:
|
||||
"""
|
||||
Get a channel-filtered logger
|
||||
|
||||
Usage:
|
||||
from utils.logging_config import get_channel_logger, LogChannel
|
||||
|
||||
logger = get_channel_logger(__name__, LogChannel.PIVOTS)
|
||||
logger.info("Pivot calculated") # Only logs if PIVOTS channel is enabled
|
||||
|
||||
Args:
|
||||
name: Logger name (usually __name__)
|
||||
channel: Logging channel
|
||||
|
||||
Returns:
|
||||
Channel-filtered logger
|
||||
"""
|
||||
return _channel_logger.get_logger(name, channel)
|
||||
|
||||
|
||||
def configure_logging_channels(channels: Set[str]):
|
||||
"""
|
||||
Configure which logging channels are enabled
|
||||
|
||||
Args:
|
||||
channels: Set of channel names to enable
|
||||
"""
|
||||
_channel_logger.set_channels(channels)
|
||||
|
||||
|
||||
def enable_channel(channel: str):
|
||||
"""Enable a specific logging channel"""
|
||||
_channel_logger.enable_channel(channel)
|
||||
|
||||
|
||||
def disable_channel(channel: str):
|
||||
"""Disable a specific logging channel"""
|
||||
_channel_logger.disable_channel(channel)
|
||||
|
||||
|
||||
def get_enabled_channels() -> Set[str]:
|
||||
"""Get currently enabled channels"""
|
||||
return _channel_logger.get_enabled_channels()
|
||||
|
||||
|
||||
def print_channel_status():
|
||||
"""Print status of all logging channels"""
|
||||
print("\n=== Logging Channel Status ===")
|
||||
all_channels = [
|
||||
LogChannel.CORE,
|
||||
LogChannel.TRADING,
|
||||
LogChannel.TRAINING,
|
||||
LogChannel.INFERENCE,
|
||||
LogChannel.PIVOTS,
|
||||
LogChannel.DATA,
|
||||
LogChannel.WEBSOCKET,
|
||||
LogChannel.API,
|
||||
LogChannel.WEBUI,
|
||||
LogChannel.PERFORMANCE,
|
||||
LogChannel.DEBUG
|
||||
]
|
||||
|
||||
enabled = _channel_logger.get_enabled_channels()
|
||||
|
||||
for channel in all_channels:
|
||||
status = "ENABLED" if channel in enabled else "DISABLED"
|
||||
print(f" {channel:15s} : {status}")
|
||||
print("=" * 31 + "\n")
|
||||
|
||||
Reference in New Issue
Block a user