TCPDashboard/docs/modules/data_collectors.md

Enhanced Data Collector System

This documentation describes the enhanced data collector system, featuring a modular architecture, centralized management, and robust health monitoring.

Table of Contents

• Overview
• System Architecture
• Core Components
• Exchange Factory
• Health Monitoring
• API Reference
• Troubleshooting

Overview

Key Features

• Modular Exchange Integration: Easily add new exchanges without impacting core logic
• Centralized Management: CollectorManager for system-wide control
• Robust Health Monitoring: Automatic restarts and failure detection
• Factory Pattern: Standardized creation of collector instances
• Asynchronous Operations: High-performance data collection
• Comprehensive Logging: Detailed component-level logging

Supported Exchanges

• OKX: Full implementation with WebSocket support
• Binance (Future): Planned support
• Coinbase (Future): Planned support

For exchange-specific documentation, see Exchange Implementations (./exchanges/).

System Architecture

┌─────────────────────────────────────────────────────────────┐
│                   TCP Dashboard Platform                    │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐    │
│  │              CollectorManager                       │    │
│  │  • Centralized start/stop/status control           │    │
│  │                                                     │    │
│  │  ┌─────────────────────────────────────────────────┐│    │
│  │  │            Global Health Monitor               ││    │
│  │  │  • System-wide health checks                   ││    │
│  │  │  • Auto-restart coordination                   ││    │
│  │  │  • Performance analytics                       ││    │
│  │  └─────────────────────────────────────────────────┘│    │
│  │                         │                           │    │
│  │ ┌─────────────┐ ┌─────────────┐ ┌────────────────┐  │    │
│  │ │OKX Collector│ │Binance Coll.│ │Custom Collector│  │    │
│  │ │• Health Mon │ │• Health Mon │ │• Health Monitor│  │    │
│  │ │• Auto-restart│ │• Auto-restart│ │• Auto-restart │  │    │
│  │ │• Data Valid │ │• Data Valid │ │• Data Validate │  │    │
│  │ └─────────────┘ └─────────────┘ └────────────────┘  │    │
│  └─────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────┘

Core Components

1. BaseDataCollector

An abstract base class that defines the common interface for all exchange collectors. It now orchestrates specialized components for connection management, state and telemetry, and callback dispatching.

Key Responsibilities:

• Standardized start, stop, restart methods.
• Orchestrates connection handling via ConnectionManager.
• Delegates state, health, and statistics management to CollectorStateAndTelemetry.
• Utilizes CallbackDispatcher for managing and notifying data subscribers.
• Defines abstract methods for exchange-specific implementations (e.g., _actual_connect, _actual_disconnect, _subscribe_channels, _process_message).

2. CollectorStateAndTelemetry

Manages the operational state, health, and performance statistics of a data collector.

Key Responsibilities:

• Tracks CollectorStatus (e.g., RUNNING, STOPPED, ERROR).
• Monitors health metrics like heartbeat and data silence.
• Collects and provides operational statistics (e.g., messages processed, errors).
• Provides centralized logging functionality for the collector.

3. ConnectionManager

Handles the WebSocket connection lifecycle and resilience for a data collector.

Key Responsibilities:

• Establishes and terminates WebSocket connections.
• Manages automatic reconnection attempts with exponential backoff.
• Handles connection-related errors and ensures robust connectivity.
• Tracks WebSocket connection state and statistics.

4. CallbackDispatcher

Manages and dispatches real-time data to registered callbacks.

Key Responsibilities:

• Registers and unregisters data callbacks for different DataTypes.
• Notifies all subscribed listeners when new data points are received.
• Ensures efficient and reliable distribution of processed market data.

5. CollectorLifecycleManager

Manages the lifecycle of individual data collectors, including adding, removing, starting, stopping, and restarting them.

Key Responsibilities:

• Handles add_collector, remove_collector, enable_collector, disable_collector.
• Manages _start_collector, restart_collector, restart_all_collectors operations.

6. ManagerHealthMonitor

Encapsulates the logic for global system health monitoring.

Key Responsibilities:

• Implements the _global_health_monitor logic.
• Provides system-wide health checks and auto-restart coordination.

7. ManagerStatsTracker

Manages the collection and retrieval of performance statistics for the CollectorManager.

Key Responsibilities:

• Updates and provides statistics via get_status.
• Utilizes CachedStatsManager for optimized, periodic updates of statistics.

8. ManagerLogger

Centralizes all logging operations for the CollectorManager.

Key Responsibilities:

• Provides wrapper methods for logging at different levels (_log_debug, _log_info, _log_warning, _log_error, _log_critical).
• Ensures consistent log formatting and includes exc_info=True for error logs.

9. CollectorManager

A singleton class that orchestrates all active data collectors and their associated components. It now delegates specific responsibilities to dedicated component classes.

Key Responsibilities:

• Centralized control and coordination of CollectorLifecycleManager.
• Aggregation of system-wide status from ManagerHealthMonitor and ManagerStatsTracker.
• Unified logging through ManagerLogger.
• Overall system-wide status aggregation.

10. Exchange-Specific Collectors

Concrete implementations of BaseDataCollector for each exchange (e.g., OKXCollector).

Key Responsibilities:

• Handle exchange-specific WebSocket protocols
• Parse and standardize incoming data
• Implement exchange-specific authentication
• Define subscription messages for different data types

For more details, see OKX Collector Documentation (./exchanges/okx.md).

11. ServiceConfig

Handles the loading, creation, and validation of service configurations.

Key Responsibilities:

• Manages _load_config and _create_default_config logic.
• Implements schema validation for configuration files and file permission validation.

12. CollectorFactory

Encapsulates the logic for creating individual data collector instances.

Key Responsibilities:

• Manages the _create_collector logic, decoupling collector creation from the DataCollectionService.

13. AsyncTaskManager

Provides a comprehensive utility for managing and tracking asynchronous tasks.

Key Responsibilities:

• Manages asyncio.Task instances, preventing potential memory leaks and ensuring proper task lifecycle.
• Used by CollectorManager and DataCollectionService for robust asynchronous operations.

Exchange Factory

The ExchangeFactory provides a standardized way to create data collectors, decoupling the client code from specific implementations.

Features

• Simplified Creation: Single function to create any supported collector
• Configuration Driven: Uses ExchangeCollectorConfig for flexible setup
• Validation: Validates configuration before creating a collector
• Extensible: Easily register new exchange collectors

Usage

from data.exchanges import ExchangeFactory, ExchangeCollectorConfig
from data.common import DataType

# Create config for OKX collector
config = ExchangeCollectorConfig(
    exchange="okx",
    symbol="BTC-USDT",
    data_types=[DataType.TRADE, DataType.ORDERBOOK],
    auto_restart=True
)

# Create collector using the factory
try:
    collector = ExchangeFactory.create_collector(config)
    # Use the collector
    await collector.start()
except ValueError as e:
    print(f"Error creating collector: {e}")

# Create multiple collectors
configs = [...]
collectors = ExchangeFactory.create_multiple_collectors(configs)

Health Monitoring

The system includes a robust, two-level health monitoring system, now enhanced with cached statistics management.

1. Collector-Level Monitoring

Each BaseDataCollector instance has its own health monitoring.

Key Metrics:

• Heartbeat: Regular internal signal to confirm the collector is responsive
• Data Silence: Tracks time since last message to detect frozen connections
• Restart Count: Number of automatic restarts
• Connection Status: Tracks WebSocket connection state

2. Manager-Level Monitoring

The CollectorManager provides a global view of system health, leveraging ManagerHealthMonitor and ManagerStatsTracker.

Key Metrics:

• Aggregate Status: Overview of all collectors (running, stopped, failed)
• System Uptime: Total uptime for the collector system
• Failed Collectors: List of collectors that failed to restart
• Resource Usage: System-level CPU and memory monitoring

Health Status API

# Get status of a single collector
status = collector.get_status()
health = collector.get_health_status()

# Get status of the entire system
system_status = manager.get_status()

For detailed status schemas, refer to the Reference Documentation (../../reference/README.md).

API Reference

BaseDataCollector

• async start()
• async stop()
• async restart()
• get_status() -> dict
• get_health_status() -> dict

CollectorManager

• add_collector(collector)
• remove_collector(collector_id)
• enable_collector(collector_id)
• disable_collector(collector_id)
• restart_collector(collector_id)
• async start_all()
• async stop_all()
• get_status() -> dict
• list_collectors() -> list

DataCollectionService

• async run()
• async stop()

ExchangeFactory

• create_collector(config) -> BaseDataCollector
• create_multiple_collectors(configs) -> list
• get_supported_exchanges() -> list

CollectorLifecycleManager

• add_collector(collector)
• remove_collector(collector_id)
• enable_collector(collector_id)
• disable_collector(collector_id)
• _start_collector(collector)
• restart_collector(collector_id)
• restart_all_collectors()

ManagerHealthMonitor

• _global_health_monitor()

ManagerStatsTracker

• get_status() -> dict
• _update_stats()

ManagerLogger

• _log_debug(message, exc_info=False)
• _log_info(message, exc_info=False)
• _log_warning(message, exc_info=False)
• _log_error(message, exc_info=True)
• _log_critical(message, exc_info=True)

ServiceConfig

• _load_config(config_path)
• _create_default_config()
• validate_permissions(file_path)

CollectorFactory

• _create_collector(config)

AsyncTaskManager

• add_task(task)
• remove_task(task_id)
• cancel_all_tasks()
• wait_for_all_tasks()

Troubleshooting

Common Issues

1. Collector fails to start

   • Cause: Invalid symbol, incorrect API keys, or network issues.
   • Solution: Check logs for error messages. Verify configuration and network connectivity.

2. Collector stops receiving data

   • Cause: WebSocket connection dropped, exchange issues.
   • Solution: Health monitor should automatically restart. If not, check logs for reconnect errors.

3. "Exchange not supported" error

   • Cause: Trying to create a collector for an exchange not registered in the factory.
   • Solution: Implement the collector and register it in data/exchanges/__init__.py.

4. Error information leakage in logs/responses

   • Cause: Raw exception details being exposed.
   • Solution: Ensure error messages are sanitized using _sanitize_error before logging or returning to external calls.

5. Configuration file permission issues

   • Cause: Improper file permissions preventing the service from reading configuration.
   • Solution: Verify file permissions for configuration files. The ServiceConfig now includes validation for this.

Best Practices

• Use the CollectorManager for lifecycle management, delegating to its components.
• Always validate configurations before creating collectors, leveraging ServiceConfig.
• Monitor system status regularly using manager.get_status().
• Refer to logs for detailed error analysis, paying attention to exc_info=True for critical errors.
• Ensure AsyncTaskManager is used for all long-running asynchronous operations to prevent resource leaks.

---

Back to [Modules Documentation (../README.md)]