diff --git a/consumer/Dockerfile b/consumer/Dockerfile new file mode 100644 index 0000000..1e32b6d --- /dev/null +++ b/consumer/Dockerfile @@ -0,0 +1,29 @@ +FROM python:3.11-slim + +WORKDIR /app + +# Install system dependencies +RUN apt-get update && apt-get install -y --no-install-recommends \ + gcc \ + && rm -rf /var/lib/apt/lists/* + +# Copy requirements from root +COPY requirements.txt /tmp/ + +# Copy models module (includes __init__.py with class definitions and .pkl files) +COPY models /app/ml_models + +# Copy consumer code +COPY consumer/ /app/ + +# Install Python dependencies +RUN pip install --no-cache-dir -r /tmp/requirements.txt + +# Set environment for Docker network and models location +ENV KAFKA_BROKER=kafka:29092 +ENV REDIS_URL=redis://redis:6379 +ENV ML_MODELS_DIR=/app/ml_models +ENV PYTHONPATH=/app:/app/ml_models + +# Run the consumer pipeline +CMD ["python3", "pipeline.py"] diff --git a/consumer/decision.py b/consumer/decision.py index ec9003a..437d246 100644 --- a/consumer/decision.py +++ b/consumer/decision.py @@ -14,6 +14,7 @@ scores, rather than failing entirely. This ensures uptime even during partial ou import redis import logging +import os from pybreaker import CircuitBreaker logger = logging.getLogger(__name__) @@ -21,7 +22,13 @@ logger = logging.getLogger(__name__) class DecisionEngine: """Makes final approve/block decisions with resilience patterns""" - def __init__(self, redis_host: str = "localhost", redis_port: int = 6379): + def __init__(self, redis_host: str = None, redis_port: int = None): + # Support environment variables for Docker deployments + if redis_host is None: + redis_host = os.getenv("REDIS_HOST", "localhost") + if redis_port is None: + redis_port = int(os.getenv("REDIS_PORT", "6379")) + self.redis_client = redis.Redis( host=redis_host, port=redis_port, diff --git a/consumer/features.py b/consumer/features.py index a5067ee..9f5182f 100644 --- a/consumer/features.py +++ b/consumer/features.py @@ -15,6 +15,7 @@ context to improve model decision quality. import redis import numpy as np import logging +import os from typing import Dict, List logger = logging.getLogger(__name__) @@ -22,7 +23,13 @@ logger = logging.getLogger(__name__) class FeatureExtractor: """Extracts and engineers features from transactions""" - def __init__(self, redis_host: str = "localhost", redis_port: int = 6379): + def __init__(self, redis_host: str = None, redis_port: int = None): + # Support environment variables for Docker deployments + if redis_host is None: + redis_host = os.getenv("REDIS_HOST", "localhost") + if redis_port is None: + redis_port = int(os.getenv("REDIS_PORT", "6379")) + self.redis_client = redis.Redis( host=redis_host, port=redis_port, @@ -30,6 +37,7 @@ class FeatureExtractor: ) logger.info(f"[FEATURE EXTRACTOR] Initialized, redis={redis_host}:{redis_port}") + def get_user_profile(self, user_id: str) -> Dict[str, float]: """ Retrieve user's behavioral profile from Redis. diff --git a/consumer/models.py b/consumer/models.py index 8e2f8da..abf4ca7 100644 --- a/consumer/models.py +++ b/consumer/models.py @@ -14,15 +14,35 @@ are delegated to a more expensive deep model. This optimizes latency vs. accurac import pickle import logging +import os from typing import Tuple logger = logging.getLogger(__name__) +# Import model classes - they may be at /app/ml_models or /models in different contexts +try: + # Docker context + from ml_models import Layer1MockModel, Layer2MockModel +except ImportError: + try: + # Local context (when running from host) + from models import Layer1MockModel, Layer2MockModel + except ImportError: + pass + class ModelScorer: """Loads and runs fraud detection models""" - def __init__(self, layer1_path: str = "models/layer1.pkl", - layer2_path: str = "models/layer2.pkl"): + def __init__(self, layer1_path: str = None, layer2_path: str = None): + # Auto-detect paths based on context + if layer1_path is None: + models_dir = os.getenv("ML_MODELS_DIR", "models") + layer1_path = os.path.join(models_dir, "layer1.pkl") + + if layer2_path is None: + models_dir = os.getenv("ML_MODELS_DIR", "models") + layer2_path = os.path.join(models_dir, "layer2.pkl") + # Load Layer 1 model with open(layer1_path, 'rb') as f: self.layer1_model = pickle.load(f) @@ -70,3 +90,4 @@ class ModelScorer: ) return fraud_probability + diff --git a/consumer/pipeline.py b/consumer/pipeline.py index e606df1..f95cdd6 100644 --- a/consumer/pipeline.py +++ b/consumer/pipeline.py @@ -14,6 +14,7 @@ Each step is independently verifiable, making it suitable for academic demonstra import json import logging +import os from kafka import KafkaConsumer from kafka.errors import KafkaError @@ -25,10 +26,19 @@ from decision import DecisionEngine # CONFIGURATION # ══════════════════════════════════════════════════════════════════════════════ -KAFKA_BROKER = "localhost:9092" +KAFKA_BROKER = os.getenv("KAFKA_BROKER", "localhost:9092") KAFKA_TOPIC = "transactions" KAFKA_GROUP = "fraud-detection-consumer-group" +# Redis configuration +REDIS_HOST = os.getenv("REDIS_HOST", "localhost") +REDIS_PORT = int(os.getenv("REDIS_PORT", "6379")) + +# ML Models configuration +ML_MODELS_DIR = os.getenv("ML_MODELS_DIR", "models") +LAYER1_PATH = os.path.join(ML_MODELS_DIR, "layer1.pkl") +LAYER2_PATH = os.path.join(ML_MODELS_DIR, "layer2.pkl") + # Logging setup logging.basicConfig( level=logging.INFO, @@ -49,20 +59,20 @@ def initialize_pipeline(): # Feature extraction feature_extractor = FeatureExtractor( - redis_host="localhost", - redis_port=6379 + redis_host=REDIS_HOST, + redis_port=REDIS_PORT ) # Model scoring model_scorer = ModelScorer( - layer1_path="models/layer1.pkl", - layer2_path="models/layer2.pkl" + layer1_path=LAYER1_PATH, + layer2_path=LAYER2_PATH ) # Decision engine with circuit breaker decision_engine = DecisionEngine( - redis_host="localhost", - redis_port=6379 + redis_host=REDIS_HOST, + redis_port=REDIS_PORT ) # Kafka consumer diff --git a/docker-compose.yml b/docker-compose.yml index aaa5520..7d249ba 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -90,9 +90,26 @@ services: - NODE_ENV=production - VITE_API_URL=http://localhost:8000 restart: unless-stopped - # NOTE: All services now run in Docker on the same network - # Frontend: http://localhost:8000 (Docker service name resolution) - # Consumer: Still runs on host, connects to kafka:29092 + + consumer: + build: + context: . + dockerfile: consumer/Dockerfile + networks: + - payguard-network + depends_on: + kafka: + condition: service_healthy + redis: + condition: service_healthy + environment: + - KAFKA_BROKER=kafka:29092 + - REDIS_HOST=redis + - REDIS_PORT=6379 + - ML_MODELS_DIR=/app/ml_models + restart: unless-stopped + # Consumer runs indefinitely, consuming messages from Kafka and storing results in Redis + # Logs visible via: docker compose logs -f consumer volumes: redis_data: