Containerize fraud detection consumer - full Docker integration

- Added consumer/Dockerfile with all dependencies
- Updated consumer pipeline.py to use environment variables
- Updated features.py and decision.py to support env var configuration
- Updated docker-compose.yml to include consumer service
- Consumer now runs automatically with 'docker compose up'
- Proper Kafka and Redis networking for Docker
- Environment variables for ML_MODELS_DIR, KAFKA_BROKER, REDIS_HOST/PORT

Now entire PayGuard system runs with: docker compose up -d

Co-authored-by: Copilot <[email protected]>

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"]

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,

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.

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", 
+    def __init__(self, layer1_path: str = None, layer2_path: str = None):
-                 layer2_path: str = "models/layer2.pkl"):
+        # 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
+

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_host=REDIS_HOST,
-        redis_port=6379
+        redis_port=REDIS_PORT
     )
     
     # Model scoring
     model_scorer = ModelScorer(
-        layer1_path="models/layer1.pkl",
+        layer1_path=LAYER1_PATH,
-        layer2_path="models/layer2.pkl"
+        layer2_path=LAYER2_PATH
     )
     
     # Decision engine with circuit breaker
     decision_engine = DecisionEngine(
-        redis_host="localhost",
+        redis_host=REDIS_HOST,
-        redis_port=6379
+        redis_port=REDIS_PORT
     )
     
     # Kafka consumer

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:
-    # Consumer: Still runs on host, connects to kafka:29092
+    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: