From c3560fdd637c4f034cac4f7371aeed65d018bc91 Mon Sep 17 00:00:00 2001 From: TheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com> Date: Wed, 1 Apr 2026 16:12:40 +0900 Subject: feat(services): integrate structlog, healthcheck, metrics, and Telegram Replace logging.basicConfig/getLogger with structlog setup_logging in all four service entry points. Add HealthCheckServer, ServiceMetrics, and TelegramNotifier initialization to each service. Update OrderExecutor to accept a notifier parameter and send order notifications. Add RedisBroker.ping() for health checks. Update executor tests with notifier=AsyncMock(). --- .../order-executor/src/order_executor/executor.py | 18 ++++++++---- services/order-executor/src/order_executor/main.py | 33 ++++++++++++++++------ 2 files changed, 37 insertions(+), 14 deletions(-) (limited to 'services/order-executor/src/order_executor') diff --git a/services/order-executor/src/order_executor/executor.py b/services/order-executor/src/order_executor/executor.py index 16ae52c..099520d 100644 --- a/services/order-executor/src/order_executor/executor.py +++ b/services/order-executor/src/order_executor/executor.py @@ -1,5 +1,5 @@ """Order execution logic.""" -import logging +import structlog from datetime import datetime, timezone from decimal import Decimal from typing import Any, Optional @@ -8,10 +8,11 @@ from shared.broker import RedisBroker from shared.db import Database from shared.events import OrderEvent from shared.models import Order, OrderSide, OrderStatus, OrderType, Signal +from shared.notifier import TelegramNotifier from order_executor.risk_manager import RiskManager -logger = logging.getLogger(__name__) +logger = structlog.get_logger() class OrderExecutor: @@ -23,12 +24,14 @@ class OrderExecutor: risk_manager: RiskManager, broker: RedisBroker, db: Database, + notifier: TelegramNotifier, dry_run: bool = True, ) -> None: self.exchange = exchange self.risk_manager = risk_manager self.broker = broker self.db = db + self.notifier = notifier self.dry_run = dry_run async def execute(self, signal: Signal) -> Optional[Order]: @@ -56,7 +59,7 @@ class OrderExecutor: if not result.allowed: logger.warning( - "Risk check rejected signal %s: %s", signal.id, result.reason + "risk_check_rejected", signal_id=str(signal.id), reason=result.reason ) return None @@ -74,7 +77,7 @@ class OrderExecutor: if self.dry_run: order.status = OrderStatus.FILLED order.filled_at = datetime.now(timezone.utc) - logger.info("[DRY RUN] Order filled: %s %s %s", order.side, order.quantity, order.symbol) + logger.info("order_filled_dry_run", side=str(order.side), quantity=str(order.quantity), symbol=order.symbol) else: try: await self.exchange.create_order( @@ -85,10 +88,10 @@ class OrderExecutor: ) order.status = OrderStatus.FILLED order.filled_at = datetime.now(timezone.utc) - logger.info("Order filled: %s %s %s", order.side, order.quantity, order.symbol) + logger.info("order_filled", side=str(order.side), quantity=str(order.quantity), symbol=order.symbol) except Exception as exc: order.status = OrderStatus.FAILED - logger.error("Order failed for signal %s: %s", signal.id, exc) + logger.error("order_failed", signal_id=str(signal.id), error=str(exc)) # Persist to DB await self.db.insert_order(order) @@ -97,4 +100,7 @@ class OrderExecutor: event = OrderEvent(data=order) await self.broker.publish("orders", event.to_dict()) + # Notify via Telegram + await self.notifier.send_order(order) + return order diff --git a/services/order-executor/src/order_executor/main.py b/services/order-executor/src/order_executor/main.py index b57c513..7f0578d 100644 --- a/services/order-executor/src/order_executor/main.py +++ b/services/order-executor/src/order_executor/main.py @@ -1,6 +1,5 @@ """Order Executor Service entry point.""" import asyncio -import logging from decimal import Decimal import ccxt.async_support as ccxt @@ -8,18 +7,21 @@ import ccxt.async_support as ccxt from shared.broker import RedisBroker from shared.db import Database from shared.events import Event, EventType +from shared.healthcheck import HealthCheckServer +from shared.logging import setup_logging +from shared.metrics import ServiceMetrics +from shared.notifier import TelegramNotifier from order_executor.config import ExecutorConfig from order_executor.executor import OrderExecutor from order_executor.risk_manager import RiskManager -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - async def run() -> None: config = ExecutorConfig() - logging.getLogger().setLevel(config.log_level) + log = setup_logging("order-executor", config.log_level, config.log_format) + metrics = ServiceMetrics("order_executor") + notifier = TelegramNotifier(bot_token=config.telegram_bot_token, chat_id=config.telegram_chat_id) db = Database(config.database_url) await db.connect() @@ -45,12 +47,19 @@ async def run() -> None: risk_manager=risk_manager, broker=broker, db=db, + notifier=notifier, dry_run=config.dry_run, ) last_id = "$" stream = "signals" - logger.info("Order executor started, listening on stream=%s dry_run=%s", stream, config.dry_run) + + health = HealthCheckServer("order-executor", port=config.health_port + 2) + health.register_check("redis", broker.ping) + await health.start() + metrics.service_up.labels(service="order-executor").set(1) + + log.info("service_started", stream=stream, dry_run=config.dry_run) try: while True: @@ -60,16 +69,24 @@ async def run() -> None: event = Event.from_dict(msg) if event.type == EventType.SIGNAL: signal = event.data - logger.info("Processing signal %s for %s", signal.id, signal.symbol) + log.info("processing_signal", signal_id=str(signal.id), symbol=signal.symbol) await executor.execute(signal) + metrics.events_processed.labels(service="order-executor", event_type="signal").inc() except Exception as exc: - logger.error("Failed to process message: %s", exc) + log.error("message_processing_failed", error=str(exc)) + metrics.errors_total.labels(service="order-executor", error_type="processing").inc() if messages: # Advance last_id to avoid re-reading — broker.read returns decoded dicts, # so we track progress by re-reading with "0" for replaying or "$" for new only. # Since we block on "$" we get only new messages each iteration. pass + except Exception as exc: + log.error("fatal_error", error=str(exc)) + await notifier.send_error(str(exc), "order-executor") + raise finally: + metrics.service_up.labels(service="order-executor").set(0) + await notifier.close() await broker.close() await db.close() await exchange.close() -- cgit v1.2.3