Logging System

Overview

The Quantstrip logging system provides centralized, thread-safe logging for all application components. It automatically handles log rotation, console output, and separate error tracking for the web interface.

Key Features

• Automatic daily log rotation with 30-day retention
• Thread-aware logging (tracks which thread generated each log)
• Separate error logs for web interface issues
• Console output for real-time monitoring
• Simulation console buffer for web IDE
• Singleton pattern ensures consistent logging across application

---

Using the Logger

Basic Setup

The logger is automatically initialized when you import it. Simply get a logger instance for your module:

import logging

logger = logging.getLogger(__name__)

# Now you can log messages
logger.info("Strategy initialized")
logger.warning("Low liquidity detected")
logger.error("Order failed")

---

Log Levels

The logger supports standard Python logging levels:

Level	When to Use	Example
DEBUG	Detailed diagnostic information	Variable values, state transitions
INFO	General informational messages	Trade executions, strategy signals
WARNING	Warning messages for unexpected situations	Missing data, API rate limits
ERROR	Error messages for failures	Order rejections, connection failures
CRITICAL	Critical failures requiring immediate attention	Database corruption, system crashes

Example:

import logging

logger = logging.getLogger(__name__)

# Different log levels
logger.debug(f"Calculating signal with price={price}, volume={volume}")
logger.info(f"Generated BUY signal for {symbol}")
logger.warning(f"Spread wider than expected: {spread:.4f}")
logger.error(f"Failed to place order: {error_msg}")
logger.critical(f"Database connection lost!")

---

Logging in Trading Clients

Here's how to use logging in your custom trading clients:

import logging
from base_client import BaseClient

logger = logging.getLogger(__name__)

class Client(BaseClient):
    def __init__(self):
        super().__init__()
        logger.info("Client initialized")

    def on_bar(self, symbol, bar):
        logger.debug(f"Processing bar for {symbol}: close={bar.close}")

        # Your strategy logic
        if self.should_buy(bar):
            logger.info(f"BUY signal generated for {symbol} at {bar.close}")
            self.place_order(symbol, 100, "BUY")

    def on_order_filled(self, order):
        logger.info(f"Order filled: {order.symbol} {order.side} "
                   f"{order.quantity}@{order.fill_price}")

    def on_error(self, error):
        logger.error(f"Error occurred: {error}")

---

Structured Logging

Use f-strings to create clear, structured log messages:

# Good - Clear and structured
logger.info(f"Position opened: symbol={symbol}, qty={qty}, price={price:.2f}")

# Good - Include context
logger.warning(f"Order retry {attempt}/{max_attempts} for {symbol}")

# Bad - Vague message
logger.info("Something happened")

# Bad - Missing context
logger.error("Failed")

---

Logging Exceptions

Always log the full exception traceback for debugging:

import logging
import traceback

logger = logging.getLogger(__name__)

try:
    result = risky_operation()
except Exception as e:
    # Log with traceback
    logger.error(f"Operation failed: {e}")
    logger.error(traceback.format_exc())

    # Or use exception() method for automatic traceback
    logger.exception(f"Operation failed for {symbol}")

---

Logger Output Format

Logs are automatically formatted with: - Timestamp (YYYY-MM-DD HH:MM:SS) - Log level (INFO, WARNING, ERROR, etc.) - Module name - Thread name - Message

Example output:

2025-01-17 14:30:45 - INFO - my_strategy - MainThread - Strategy initialized
2025-01-17 14:30:46 - INFO - my_strategy - MainThread - BUY signal for AAPL at 150.25
2025-01-17 14:30:47 - WARNING - my_strategy - MainThread - API rate limit approaching
2025-01-17 14:30:48 - ERROR - my_strategy - MainThread - Order rejected: Insufficient funds

---

Log Files

Service Log

Main application log with all system messages.

• Location: Data/logs/service.log
• Rotation: Daily at midnight
• Retention: 30 days
• Contents: All application logs (INFO and above)

---

Dash Error Log

Separate log for web interface errors.

• Location: Data/logs/dash_errors.log
• Rotation: Daily at midnight
• Retention: 7 days
• Contents: Dash/Flask/Werkzeug errors only

Tip

If your web interface isn't working, check dash_errors.log first.

---

Common Patterns

Strategy Logging

import logging

logger = logging.getLogger(__name__)

class Client(BaseClient):
    def __init__(self):
        super().__init__()
        self.strategy_name = "Momentum v2"
        logger.info(f"Initialized {self.strategy_name}")

    def run_client(self):
        logger.info(f"Starting {self.strategy_name}")

        try:
            self.connect()
            self.subscribe_data()
            self.main_loop()
        except Exception as e:
            logger.error(f"{self.strategy_name} crashed: {e}")
            logger.exception("Full traceback:")
        finally:
            logger.info(f"Shutting down {self.strategy_name}")
            self.cleanup()

---

Trade Execution Logging

def place_order(self, symbol, quantity, side):
    logger.info(f"Placing {side} order: {quantity} {symbol}")

    try:
        order_id = self.broker.submit_order(symbol, quantity, side)
        logger.info(f"Order submitted: ID={order_id}")
        return order_id
    except Exception as e:
        logger.error(f"Order failed for {symbol}: {e}")
        logger.exception("Order failure details:")
        return None

def on_order_filled(self, execution):
    logger.info(
        f"FILL: {execution.side} {execution.shares} {execution.symbol} "
        f"@ ${execution.price:.2f} (Order ID: {execution.order_id})"
    )

---

Position Management Logging

def update_position(self, symbol, new_position):
    old_position = self.positions.get(symbol, 0)

    if new_position != old_position:
        change = new_position - old_position
        logger.info(
            f"Position change: {symbol} {old_position} -> {new_position} "
            f"({change:+d})"
        )

        self.positions[symbol] = new_position

        # Log position action
        if old_position == 0:
            logger.info(f"OPENED position in {symbol}")
        elif new_position == 0:
            logger.info(f"CLOSED position in {symbol}")
        elif old_position * new_position < 0:
            logger.info(f"REVERSED position in {symbol}")

---

Data Feed Logging

def on_bar(self, symbol, bar):
    # Debug-level for frequent events
    logger.debug(
        f"Bar: {symbol} O={bar.open:.2f} H={bar.high:.2f} "
        f"L={bar.low:.2f} C={bar.close:.2f} V={bar.volume}"
    )

def on_connection_lost(self):
    logger.warning("Market data connection lost, attempting reconnect...")

def on_connection_restored(self):
    logger.info("Market data connection restored")

---

Error Recovery Logging

def execute_with_retry(self, operation, max_attempts=3):
    for attempt in range(1, max_attempts + 1):
        try:
            logger.debug(f"Attempt {attempt}/{max_attempts}: {operation}")
            result = operation()
            logger.info(f"Operation succeeded on attempt {attempt}")
            return result
        except Exception as e:
            logger.warning(
                f"Attempt {attempt}/{max_attempts} failed: {e}"
            )

            if attempt == max_attempts:
                logger.error(f"All {max_attempts} attempts failed")
                logger.exception("Final failure details:")
                raise

            time.sleep(2 ** attempt)  # Exponential backoff

---

Performance Monitoring

import time
import logging

logger = logging.getLogger(__name__)

def monitor_performance(func):
    """Decorator to log function execution time"""
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        elapsed = time.time() - start

        if elapsed > 1.0:  # Warn if slow
            logger.warning(
                f"{func.__name__} took {elapsed:.2f}s (slow!)"
            )
        else:
            logger.debug(
                f"{func.__name__} took {elapsed:.3f}s"
            )

        return result
    return wrapper

@monitor_performance
def calculate_signals(self, data):
    # Your calculation logic
    pass

---

Best Practices

Do's

✅ Use appropriate log levels

logger.debug("Detailed variable state")  # For debugging only
logger.info("Normal operations")         # Standard events
logger.warning("Unexpected but handled") # Warnings
logger.error("Operation failed")         # Errors

✅ Include context in messages

logger.info(f"Order filled: {symbol} {qty}@{price} (ID: {order_id})")

✅ Log exceptions with traceback

logger.exception("Failed to process order")

✅ Use module-level loggers

logger = logging.getLogger(__name__)  # At module level

---

Don'ts

❌ Don't log sensitive information

# BAD - Exposes credentials
logger.info(f"Connecting with password: {password}")

# GOOD - Mask sensitive data
logger.info("Connecting to broker")

❌ Don't log excessively in tight loops

# BAD - Logs every iteration
for i in range(1000000):
    logger.info(f"Processing {i}")  # Creates huge logs!

# GOOD - Log summary
for i in range(1000000):
    process(i)
logger.info(f"Processed 1,000,000 items")

❌ Don't use print() instead of logging

# BAD - Won't be captured in log files
print("Order placed")

# GOOD - Use logger
logger.info("Order placed")

❌ Don't create logger in init

# BAD
class Client:
    def __init__(self):
        self.logger = logging.getLogger(__name__)  # Creates new instance

# GOOD
logger = logging.getLogger(__name__)  # Module-level

class Client:
    def __init__(self):
        logger.info("Initialized")

---

Thread-Safe Logging

The logging system is fully thread-safe. Each log message includes the thread name automatically:

import threading
import logging

logger = logging.getLogger(__name__)

def worker_task(task_id):
    logger.info(f"Task {task_id} started")
    # Do work...
    logger.info(f"Task {task_id} completed")

# Run multiple threads
threads = []
for i in range(5):
    t = threading.Thread(target=worker_task, args=(i,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()

Output shows thread names:

2025-01-17 14:30:45 - INFO - module - Thread-1 - Task 0 started
2025-01-17 14:30:45 - INFO - module - Thread-2 - Task 1 started
2025-01-17 14:30:46 - INFO - module - Thread-1 - Task 0 completed
2025-01-17 14:30:46 - INFO - module - Thread-2 - Task 1 completed

---

Simulation Console

When running strategies in the web IDE, logs are captured in real-time and displayed in the console.

The simulation console shows: - Logger output from simulation_thread - Print statements from your code - Errors and exceptions

Example client with console output:

import logging

logger = logging.getLogger(__name__)

class Client(BaseClient):
    def run_client(self):
        logger.info("Strategy starting...")  # Appears in console
        print("Loading data...")             # Also appears in console

        try:
            self.execute_strategy()
            logger.info("Strategy completed")
        except Exception as e:
            logger.error(f"Strategy failed: {e}")  # Red in console

---

Troubleshooting

Logs Not Appearing

Check log level:

import logging

# Ensure logger level is set correctly
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)  # Will show DEBUG and above

Verify logger name:

# Use __name__ for automatic module naming
logger = logging.getLogger(__name__)

# Or use explicit name
logger = logging.getLogger("my_strategy")

---

Too Many Logs

Reduce log level:

# Only show warnings and errors
logger.setLevel(logging.WARNING)

Add conditional logging:

# Only log every 100th iteration
if iteration % 100 == 0:
    logger.debug(f"Progress: {iteration} iterations")

---

Performance Impact

Logging has minimal performance impact, but be cautious in high-frequency scenarios:

# BAD - Logs every tick
def on_tick(self, tick):
    logger.info(f"Tick: {tick.price}")  # Too frequent!

# GOOD - Log aggregates
def on_tick(self, tick):
    self.tick_count += 1
    if self.tick_count % 1000 == 0:
        logger.info(f"Processed {self.tick_count} ticks")

---

Log File Locations

All log files are stored in the Data/logs/ directory:

Data/
└── logs/
    ├── service.log              # Current main log
    ├── service.log.2025-01-16   # Previous day's log
    ├── service.log.2025-01-15   # 2 days ago
    ├── ...
    ├── dash_errors.log          # Current web errors
    └── dash_errors.log.2025-01-16

Log Retention

• Main logs are kept for 30 days
• Dash error logs are kept for 7 days
• Logs rotate automatically at midnight