chore: sync shared modules and metrics across all repositories

- Updated Dockerfile for Prometheus integration
- Finalized metrics_server module and Prometheus counters/histograms
- Synchronized shared files: queue_handler.py, queue_sender.py, metrics.py
- Improved logging, retry logic, and structured metrics tracking
- Ensured consistency across repositories to prevent configuration drift

Author: mobious999

Dockerfile

@@ -37,6 +37,8 @@ ENV PYTHONPATH="/app"
 RUN useradd -m appuser && chown -R appuser /app
 USER appuser
 
+EXPOSE 8000
+
 HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
     CMD pgrep -f "app.main" > /dev/null || exit 1
 

src/app/output_handler.py

@@ -15,18 +15,9 @@
 from app import config_shared
 from app.queue_sender import publish_to_queue
 from app.utils.metrics import (
-    db_dispatch_counter,
-    db_dispatch_duration,
-    db_dispatch_failures,
-    output_counter,
-    paper_trade_counter,
-    paper_trade_failures,
-    rest_dispatch_counter,
-    rest_dispatch_duration,
-    rest_dispatch_failures,
-    s3_dispatch_counter,
-    s3_dispatch_duration,
-    s3_dispatch_failures,
+    record_output_metrics,
+    record_paper_trade_metrics,
+    record_sink_metrics,
 )
 from app.utils.setup_logger import setup_logger
 from app.utils.types import OutputMode, validate_list_of_dicts
@@ -42,11 +33,11 @@ def __init__(self) -> None:
         self.output_modes = config_shared.get_output_modes()
 
     def send(self, data: list[dict[str, Any]]) -> None:
-        """
-        Dispatch processed analysis output to one or more configured destinations.
+        """Dispatch processed analysis output to one or more configured destinations.
 
         Args:
             data (list[dict[str, Any]]): List of data payloads to send.
+
         """
         try:
             validate_list_of_dicts(data, required_keys=["text"])
@@ -72,11 +63,11 @@ def send(self, data: list[dict[str, Any]]) -> None:
             logger.error("❌ Failed to send output: %s", e)
 
     def send_trade_simulation(self, data: dict[str, Any]) -> None:
-        """
-        Send simulated trade data to the appropriate paper trade destination.
+        """Send simulated trade data to the appropriate paper trade destination.
 
         Args:
             data (dict[str, Any]): Simulated trade payload.
+
         """
         try:
             if config_shared.get_paper_trading_database_enabled():
@@ -85,19 +76,19 @@ def send_trade_simulation(self, data: dict[str, Any]) -> None:
                 self._output_paper_trade_to_queue(data)
         except Exception as e:
             logger.error("❌ Failed to send paper trade: %s", e)
-            self._record_metric("paper_trade_failure", 1)
+            record_paper_trade_metrics("queue", success=False, duration_sec=0)
 
     def _get_dispatch_method(
         self, mode: OutputMode
     ) -> Callable[[list[dict[str, Any]]], None] | None:
-        """
-        Resolve the output dispatch method based on the mode.
+        """Resolve the output dispatch method based on the mode.
 
         Args:
             mode (OutputMode): Output mode enum value.
 
         Returns:
             Callable or None: Method to handle the output.
+
         """
         return {
             OutputMode.LOG: self._output_to_log,
@@ -109,24 +100,46 @@ def _get_dispatch_method(
         }.get(mode)
 
     def _output_to_log(self, data: list[dict[str, Any]]) -> None:
-        """Log each item in the data list."""
+        """Log each item in the data list.
+
+        Args:
+            data (list[dict[str, Any]]): Data to log.
+
+        """
         for item in data:
             logger.info("📝 Processed message:\n%s", json.dumps(item, indent=4))
 
     def _output_to_stdout(self, data: list[dict[str, Any]]) -> None:
-        """Print each item in the data list to standard output."""
+        """Print each item in the data list to standard output.
+
+        Args:
+            data (list[dict[str, Any]]): Data to print.
+
+        """
         for item in data:
             print(json.dumps(item, indent=4))
 
     @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
     def _output_to_queue(self, data: list[dict[str, Any]]) -> None:
-        """Publish the data to the configured queue."""
+        """Publish the data to the configured queue.
+
+        Retries on failure using exponential backoff.
+
+        Args:
+            data (list[dict[str, Any]]): Data to publish.
+
+        """
         publish_to_queue(data)
         logger.info("✅ Output published to queue: %d message(s)", len(data))
-        self._record_metric("output_queue_success", len(data))
+        record_output_metrics("queue", success=True, duration_sec=0)
 
     def _output_to_rest(self, data: list[dict[str, Any]]) -> None:
-        """Send the data to the configured REST endpoint."""
+        """Send the data to the configured REST endpoint.
+
+        Args:
+            data (list[dict[str, Any]]): Data to post to REST API.
+
+        """
         import requests
 
         url = config_shared.get_rest_output_url()
@@ -135,20 +148,23 @@ def _output_to_rest(self, data: list[dict[str, Any]]) -> None:
         try:
             response = requests.post(url, json=data, headers=headers, timeout=10)
             duration = time.perf_counter() - start
-            rest_dispatch_duration.labels(status=str(response.status_code)).observe(duration)
+            record_sink_metrics("rest", str(response.status_code), duration, failed=not response.ok)
 
             if response.ok:
                 logger.info("🚀 Sent data to REST: HTTP %d", response.status_code)
-                rest_dispatch_counter.labels(status=str(response.status_code)).inc()
             else:
                 logger.error("❌ REST output failed: HTTP %d", response.status_code)
-                rest_dispatch_failures.labels(status=str(response.status_code)).inc()
         except Exception as e:
             logger.error("❌ REST output error: %s", e)
-            rest_dispatch_failures.labels(status="exception").inc()
+            record_sink_metrics("rest", "exception", 0, failed=True)
 
     def _output_to_s3(self, data: list[dict[str, Any]]) -> None:
-        """Upload the data as a JSON file to an S3 bucket."""
+        """Upload the data as a JSON file to an S3 bucket.
+
+        Args:
+            data (list[dict[str, Any]]): Data to upload.
+
+        """
         import boto3
 
         s3 = boto3.client("s3")
@@ -158,15 +174,19 @@ def _output_to_s3(self, data: list[dict[str, Any]]) -> None:
         try:
             s3.put_object(Bucket=bucket, Key=key, Body=json.dumps(data).encode("utf-8"))
             duration = time.perf_counter() - start
-            s3_dispatch_duration.labels(status="200").observe(duration)
-            s3_dispatch_counter.labels(status="200").inc()
+            record_sink_metrics("s3", "200", duration, failed=False)
             logger.info("🚚 Uploaded output to S3: %s/%s", bucket, key)
         except Exception as e:
             logger.error("❌ S3 upload failed: %s", e)
-            s3_dispatch_failures.labels(status="exception").inc()
+            record_sink_metrics("s3", "exception", 0, failed=True)
 
     def _output_to_database(self, data: list[dict[str, Any]]) -> None:
-        """Write the data to the configured database using raw SQL inserts."""
+        """Write the data to the configured database using raw SQL inserts.
+
+        Args:
+            data (list[dict[str, Any]]): Data records to insert.
+
+        """
         import sqlalchemy
 
         engine = sqlalchemy.create_engine(config_shared.get_database_output_url())
@@ -179,48 +199,46 @@ def _output_to_database(self, data: list[dict[str, Any]]) -> None:
                         continue
                     conn.execute(sqlalchemy.text(config_shared.get_database_insert_sql()), **item)
             duration = time.perf_counter() - start
-            db_dispatch_duration.labels(status="success").observe(duration)
-            db_dispatch_counter.labels(status="success").inc()
+            record_sink_metrics("db", "success", duration, failed=False)
             logger.info("📊 Wrote %d records to database", len(data))
         except Exception as e:
             logger.error("❌ Database output failed: %s", e)
-            db_dispatch_failures.labels(status="exception").inc()
+            record_sink_metrics("db", "exception", 0, failed=True)
 
     @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
     def _output_paper_trade_to_queue(self, data: dict[str, Any]) -> None:
-        """Send paper trade data to a paper trading queue."""
+        """Send paper trade data to a paper trading queue.
+
+        Args:
+            data (dict[str, Any]): Simulated trade to queue.
+
+        """
         queue_name = config_shared.get_paper_trading_queue_name()
         exchange = config_shared.get_paper_trading_exchange()
         publish_to_queue([data], queue=queue_name, exchange=exchange)
         logger.info("🪙 Paper trade sent to queue:\n%s", json.dumps(data, indent=4))
-        self._record_metric("paper_trade_sent", 1)
+        record_paper_trade_metrics("queue", success=True, duration_sec=0)
 
     def _output_paper_trade_to_database(self, data: dict[str, Any]) -> None:
-        """Placeholder for paper trade DB support (not implemented)."""
-        logger.warning("⚠️ Paper trading database integration not implemented.")
-        self._record_metric("paper_trade_skipped", 1)
+        """Placeholder for future paper trade DB integration.
+
+        Args:
+            data (dict[str, Any]): Simulated trade record.
 
-    def _record_metric(self, name: str, value: int) -> None:
-        """Record a metric for dispatch monitoring."""
-        if name == "output_queue_success":
-            output_counter.labels(mode="queue").inc(value)
-        elif name == "paper_trade_sent":
-            paper_trade_counter.labels(destination="queue").inc(value)
-        elif name == "paper_trade_failure":
-            paper_trade_failures.labels(destination="queue").inc(value)
-        else:
-            logger.debug("📊 Metric: %s = %d", name, value)
+        """
+        logger.warning("⚠️ Paper trading database integration not implemented.")
+        logger.info("📊 Skipped paper trade (DB output not implemented).")
 
 
 output_handler = OutputDispatcher()
 
 
 def send_to_output(data: list[dict[str, Any]]) -> None:
-    """
-    Send data using the default output handler instance.
+    """Send data using the default output handler instance.
 
     Args:
         data (list[dict[str, Any]]): List of messages to dispatch.
+
     """
     output_handler.send(data)
 

src/app/queue_handler.py

@@ -1,4 +1,9 @@
-"""Generic queue handler for RabbitMQ or SQS with batching and retries."""
+"""Generic queue handler for RabbitMQ or SQS with batching and retries.
+
+This module supports consuming messages from either RabbitMQ or Amazon SQS.
+It provides batching, retry logic, graceful shutdown handling, and clean logging
+with optional redaction of sensitive values.
+"""
 
 import json
 import signal
@@ -22,15 +27,29 @@
 
 
 def safe_log(msg: str) -> str:
-    """Standardized redacted log message."""
+    """Standardized redacted log message helper.
+
+    Args:
+        msg (str): Message to redact if sensitive logging is enabled.
+
+    Returns:
+        str: Either the original message or a redacted form.
+
+    """
     return f"{msg}: [REDACTED]" if REDACT_SENSITIVE_LOGS else msg
 
 
 def consume_messages(callback: Callable[[list[dict]], None]) -> None:
-    """Start the queue listener for the configured queue type.
+    """Start the message consumer using the configured QUEUE_TYPE.
+
+    This method determines whether to use RabbitMQ or SQS and invokes the
+    appropriate listener. It also registers signal handlers for graceful shutdown.
 
     Args:
-        callback: A function that takes a list of messages and processes them.
+        callback (Callable[[list[dict]], None]): Processing function for a batch of messages.
+
+    Raises:
+        ValueError: If QUEUE_TYPE is not supported.
 
     """
     signal.signal(signal.SIGINT, _graceful_shutdown)
@@ -46,17 +65,23 @@ def consume_messages(callback: Callable[[list[dict]], None]) -> None:
 
 
 def _graceful_shutdown(signum, frame) -> None:
-    """Handle shutdown signals to terminate listeners cleanly."""
+    """Gracefully signal shutdown of the consumer loop.
+
+    Args:
+        signum: Signal number.
+        frame: Current stack frame.
+
+    """
     logger.info("🛑 Shutdown signal received, stopping listener...")
     shutdown_event.set()
 
 
 @retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=10))
 def _start_rabbitmq_listener(callback: Callable[[list[dict]], None]) -> None:
-    """Connect to RabbitMQ and start consuming messages.
+    """Connect to RabbitMQ and start consuming messages from the configured queue.
 
     Args:
-        callback: Function to process received messages.
+        callback (Callable[[list[dict]], None]): Handler function for batches of messages.
 
     """
     connection = pika.BlockingConnection(
@@ -74,6 +99,15 @@ def _start_rabbitmq_listener(callback: Callable[[list[dict]], None]) -> None:
     channel.queue_declare(queue=queue_name, durable=True)
 
     def on_message(ch: BlockingChannel, method, properties, body: bytes) -> None:
+        """Callback invoked for each incoming RabbitMQ message.
+
+        Args:
+            ch (BlockingChannel): The channel object.
+            method: Delivery method.
+            properties: Message properties.
+            body (bytes): Raw message body.
+
+        """
         if shutdown_event.is_set():
             ch.stop_consuming()
             return
@@ -105,7 +139,7 @@ def _start_sqs_listener(callback: Callable[[list[dict]], None]) -> None:
     """Connect to AWS SQS and start polling messages.
 
     Args:
-        callback: Function to process a batch of messages.
+        callback (Callable[[list[dict]], None]): Handler function for a batch of messages.
 
     """
     sqs = boto3.client("sqs", region_name=config.get_sqs_region())