Initial PayGuard fraud detection system implementation
- Complete dual-layer architecture for fraud detection - Phase 1: Docker Compose infrastructure (Kafka, Zookeeper, Redis) - Phase 2: Mock ML models (Layer 1 fast, Layer 2 deep) - Phase 3: FastAPI transaction ingestion with JWT auth - Phase 4: Kafka consumer pipeline with feature extraction - Phase 5: Test transaction producer with demo scenarios - Phase 6: Clean code structure and comprehensive documentation - Phase 7: End-to-end validation and demo readiness Demonstrates key system design patterns: - Microservices architecture with decoupled components - Asynchronous event processing via Kafka - Tiered decision-making (fast path vs deep analysis) - Resilience patterns (circuit breaker for graceful degradation) - Behavioral profiling with Redis caching Ready for academic demonstration with clear pipeline observability. Co-authored-by: Copilot <[email protected]>
This commit is contained in:
John Patrick TeruelandCopilot committed
commit
aa4dd497e2
24 files changed
+2343
No files matched your search
Whitespace-only changes.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,131 @@
|
||||
"""
|
||||
Decision Engine & Circuit Breaker
|
||||
|
||||
Responsibilities:
|
||||
- Combine model scores into final decision
|
||||
- Apply circuit breaker pattern for graceful degradation
|
||||
- Store decision results in Redis
|
||||
|
||||
Academic Point:
|
||||
The circuit breaker is a critical resilience pattern. If the deep analysis (Layer 2)
|
||||
becomes slow or unavailable, the system can gracefully degrade to using only Layer 1
|
||||
scores, rather than failing entirely. This ensures uptime even during partial outages.
|
||||
"""
|
||||
|
||||
import redis
|
||||
import logging
|
||||
from pybreaker import CircuitBreaker
|
||||
|
||||
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):
|
||||
self.redis_client = redis.Redis(
|
||||
host=redis_host,
|
||||
port=redis_port,
|
||||
decode_responses=True
|
||||
)
|
||||
|
||||
# Circuit breaker for Layer 2 calls
|
||||
# Fails after 3 consecutive errors, resets after 30 seconds
|
||||
self.layer2_breaker = CircuitBreaker(
|
||||
fail_max=3,
|
||||
reset_timeout=30,
|
||||
listeners=[] # Could add monitoring here
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"[DECISION ENGINE] Initialized with circuit breaker: "
|
||||
"fail_max=3, reset_timeout=30s"
|
||||
)
|
||||
|
||||
def make_decision(self,
|
||||
transaction_id: str,
|
||||
layer1_score: float,
|
||||
layer2_scorer=None,
|
||||
features=None) -> str:
|
||||
"""
|
||||
Make final decision: "approved" or "blocked"
|
||||
|
||||
Logic:
|
||||
1. If Layer 1 score < 0.4: Approve immediately (fast path)
|
||||
2. If Layer 1 score >= 0.4: Try Layer 2 with circuit breaker
|
||||
- If Layer 2 available: Use Layer 2 score (threshold: 0.7)
|
||||
- If Layer 2 fails: Fall back to conservative Layer 1 estimate
|
||||
"""
|
||||
|
||||
logger.info(f"\n[DECISION ENGINE] Processing transaction {transaction_id}")
|
||||
|
||||
# Fast path: low risk
|
||||
if layer1_score < 0.4:
|
||||
decision = "approved"
|
||||
logger.info(
|
||||
f"[DECISION] ✓ APPROVED (Layer 1 score {layer1_score:.4f} < 0.4 threshold)"
|
||||
)
|
||||
return decision
|
||||
|
||||
# Suspicious: try Layer 2
|
||||
logger.info(f"[DECISION] Suspicious score from Layer 1 ({layer1_score:.4f})")
|
||||
logger.info("[DECISION] Attempting Layer 2 deep analysis...")
|
||||
|
||||
if layer2_scorer is None or features is None:
|
||||
logger.warning("[DECISION] Layer 2 scorer not provided, using Layer 1 only")
|
||||
decision = "approved" if layer1_score < 0.5 else "blocked"
|
||||
return decision
|
||||
|
||||
# Try Layer 2 with circuit breaker
|
||||
try:
|
||||
@self.layer2_breaker
|
||||
def run_layer2():
|
||||
return layer2_scorer.score_layer2(features)
|
||||
|
||||
layer2_score = run_layer2()
|
||||
|
||||
# Layer 2 threshold
|
||||
if layer2_score > 0.7:
|
||||
decision = "blocked"
|
||||
logger.info(
|
||||
f"[DECISION] ✗ BLOCKED (Layer 2 score {layer2_score:.4f} > 0.7 threshold)"
|
||||
)
|
||||
else:
|
||||
decision = "approved"
|
||||
logger.info(
|
||||
f"[DECISION] ✓ APPROVED (Layer 2 score {layer2_score:.4f} <= 0.7 threshold)"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
# Circuit breaker is open or Layer 2 failed
|
||||
logger.warning(f"[CIRCUIT BREAKER] ⚠ Layer 2 unavailable: {str(e)}")
|
||||
logger.warning("[CIRCUIT BREAKER] Falling back to conservative Layer 1 estimate")
|
||||
|
||||
# Conservative fallback: if Layer 1 is uncertain, block
|
||||
if layer1_score > 0.6:
|
||||
decision = "blocked"
|
||||
logger.info(
|
||||
f"[DECISION] ✗ BLOCKED (Layer 2 failed, Layer 1 score {layer1_score:.4f} too high)"
|
||||
)
|
||||
else:
|
||||
decision = "approved"
|
||||
logger.info(
|
||||
f"[DECISION] ✓ APPROVED (Layer 2 failed, Layer 1 score {layer1_score:.4f} acceptable)"
|
||||
)
|
||||
|
||||
return decision
|
||||
|
||||
def store_result(self, transaction_id: str, decision: str,
|
||||
layer1_score: float, layer2_score: float = None) -> None:
|
||||
"""Store decision result in Redis for retrieval"""
|
||||
key = f"result:{transaction_id}"
|
||||
value = {
|
||||
"decision": decision,
|
||||
"layer1_score": round(layer1_score, 4),
|
||||
}
|
||||
if layer2_score is not None:
|
||||
value["layer2_score"] = round(layer2_score, 4)
|
||||
|
||||
# Store with 1-hour TTL
|
||||
self.redis_client.setex(key, 3600, str(value))
|
||||
|
||||
logger.info(f"[RESULT STORAGE] Stored decision in Redis: {key} = {decision}")
|
||||
@@ -0,0 +1,133 @@
|
||||
"""
|
||||
Feature Extraction Module
|
||||
|
||||
Responsibilities:
|
||||
- Extract raw transaction data into ML-friendly features
|
||||
- Query Redis for user behavioral history
|
||||
- Construct feature vectors for ML models
|
||||
|
||||
Academic Point:
|
||||
Feature engineering is the bridge between raw business data and ML models.
|
||||
This module demonstrates how to enrich simple transaction data with historical
|
||||
context to improve model decision quality.
|
||||
"""
|
||||
|
||||
import redis
|
||||
import numpy as np
|
||||
import logging
|
||||
from typing import Dict, List
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class FeatureExtractor:
|
||||
"""Extracts and engineers features from transactions"""
|
||||
|
||||
def __init__(self, redis_host: str = "localhost", redis_port: int = 6379):
|
||||
self.redis_client = redis.Redis(
|
||||
host=redis_host,
|
||||
port=redis_port,
|
||||
decode_responses=True
|
||||
)
|
||||
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.
|
||||
|
||||
Returns:
|
||||
{
|
||||
'avg_amount': float,
|
||||
'tx_count': int,
|
||||
'last_timestamp': float,
|
||||
'days_since_last_tx': float
|
||||
}
|
||||
"""
|
||||
key = f"user:{user_id}:profile"
|
||||
profile = self.redis_client.hgetall(key)
|
||||
|
||||
if not profile:
|
||||
logger.debug(f"[REDIS LOOKUP] No profile for user={user_id}, using defaults")
|
||||
return {
|
||||
'avg_amount': 0.0,
|
||||
'tx_count': 0,
|
||||
'last_timestamp': 0.0,
|
||||
'days_since_last_tx': 999.0
|
||||
}
|
||||
|
||||
logger.debug(
|
||||
f"[REDIS LOOKUP] Found profile for user={user_id}: "
|
||||
f"avg=${profile.get('avg_amount', 0)}, count={profile.get('tx_count', 0)}"
|
||||
)
|
||||
|
||||
return {
|
||||
'avg_amount': float(profile.get('avg_amount', 0.0)),
|
||||
'tx_count': int(profile.get('tx_count', 0)),
|
||||
'last_timestamp': float(profile.get('last_timestamp', 0.0)),
|
||||
'days_since_last_tx': float(profile.get('days_since_last_tx', 999.0))
|
||||
}
|
||||
|
||||
def extract_features(self, transaction: Dict) -> np.ndarray:
|
||||
"""
|
||||
Extract feature vector from transaction + user profile.
|
||||
|
||||
Feature vector: [amount, avg_amount, tx_count, days_since_last_tx]
|
||||
"""
|
||||
user_id = transaction['user_id']
|
||||
amount = transaction['amount']
|
||||
|
||||
# Get user history
|
||||
profile = self.get_user_profile(user_id)
|
||||
|
||||
# Construct feature vector
|
||||
features = np.array([[
|
||||
amount,
|
||||
profile['avg_amount'],
|
||||
profile['tx_count'],
|
||||
profile['days_since_last_tx']
|
||||
]])
|
||||
|
||||
logger.info(
|
||||
f"[FEATURE EXTRACTION] tx_id={transaction['transaction_id'][:8]}... "
|
||||
f"user={user_id} features=[${amount:.0f}, "
|
||||
f"avg=${profile['avg_amount']:.0f}, "
|
||||
f"count={profile['tx_count']}, "
|
||||
f"days_since={profile['days_since_last_tx']:.1f}]"
|
||||
)
|
||||
|
||||
return features
|
||||
|
||||
def update_user_profile(self, transaction: Dict) -> None:
|
||||
"""Update user's behavioral profile with new transaction"""
|
||||
user_id = transaction['user_id']
|
||||
amount = transaction['amount']
|
||||
timestamp = transaction['timestamp']
|
||||
|
||||
key = f"user:{user_id}:profile"
|
||||
profile = self.redis_client.hgetall(key)
|
||||
|
||||
# Calculate new statistics
|
||||
old_count = int(profile.get('tx_count', 0))
|
||||
new_count = old_count + 1
|
||||
|
||||
old_avg = float(profile.get('avg_amount', 0.0))
|
||||
new_avg = (old_avg * old_count + amount) / new_count
|
||||
|
||||
old_timestamp = float(profile.get('last_timestamp', 0.0))
|
||||
days_since = 0.0 # Just made a transaction
|
||||
|
||||
# Update profile in Redis
|
||||
self.redis_client.hset(
|
||||
key,
|
||||
mapping={
|
||||
'tx_count': new_count,
|
||||
'avg_amount': round(new_avg, 2),
|
||||
'last_timestamp': timestamp,
|
||||
'days_since_last_tx': days_since
|
||||
}
|
||||
)
|
||||
|
||||
logger.info(
|
||||
f"[PROFILE UPDATE] user={user_id} "
|
||||
f"count={old_count}→{new_count}, "
|
||||
f"avg=${old_avg:.2f}→${new_avg:.2f}"
|
||||
)
|
||||
@@ -0,0 +1,72 @@
|
||||
"""
|
||||
Model Scoring Modules
|
||||
|
||||
Responsibilities:
|
||||
- Load pre-trained models
|
||||
- Run Layer 1 (fast) and Layer 2 (deep) scoring
|
||||
- Return fraud probabilities
|
||||
|
||||
Academic Point:
|
||||
Dual-layer architecture allows high-throughput systems to make most decisions
|
||||
(>95% of transactions) with a fast, lightweight model, while complex edge cases
|
||||
are delegated to a more expensive deep model. This optimizes latency vs. accuracy.
|
||||
"""
|
||||
|
||||
import pickle
|
||||
import logging
|
||||
from typing import Tuple
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class ModelScorer:
|
||||
"""Loads and runs fraud detection models"""
|
||||
|
||||
def __init__(self, layer1_path: str = "models/layer1.pkl",
|
||||
layer2_path: str = "models/layer2.pkl"):
|
||||
# Load Layer 1 model
|
||||
with open(layer1_path, 'rb') as f:
|
||||
self.layer1_model = pickle.load(f)
|
||||
logger.info(f"[MODEL LOADER] ✓ Layer 1 model loaded from {layer1_path}")
|
||||
|
||||
# Load Layer 2 model
|
||||
with open(layer2_path, 'rb') as f:
|
||||
self.layer2_model = pickle.load(f)
|
||||
logger.info(f"[MODEL LOADER] ✓ Layer 2 model loaded from {layer2_path}")
|
||||
|
||||
def score_layer1(self, features) -> float:
|
||||
"""
|
||||
Layer 1: Fast, shallow model for initial risk assessment.
|
||||
|
||||
Decision threshold: 0.4
|
||||
- score < 0.4 → Low risk (approve immediately)
|
||||
- score >= 0.4 → Suspicious (escalate to Layer 2)
|
||||
"""
|
||||
probs = self.layer1_model.predict_proba(features)
|
||||
fraud_probability = probs[0][1] # Index 1 is fraud class
|
||||
|
||||
logger.info(
|
||||
f"[LAYER 1 SCORING] Fraud probability: {fraud_probability:.4f} "
|
||||
f"{'→ LOW RISK, fast-track approval' if fraud_probability < 0.4 else '→ SUSPICIOUS, escalate to Layer 2'}"
|
||||
)
|
||||
|
||||
return fraud_probability
|
||||
|
||||
def score_layer2(self, features) -> float:
|
||||
"""
|
||||
Layer 2: Deep, complex model for suspicious transactions.
|
||||
|
||||
Decision threshold: 0.7
|
||||
- score <= 0.7 → Approve
|
||||
- score > 0.7 → Block
|
||||
|
||||
Only called if Layer 1 score >= 0.4 (suspicious).
|
||||
"""
|
||||
probs = self.layer2_model.predict_proba(features)
|
||||
fraud_probability = probs[0][1] # Index 1 is fraud class
|
||||
|
||||
logger.info(
|
||||
f"[LAYER 2 SCORING] Fraud probability: {fraud_probability:.4f} "
|
||||
f"{'→ BLOCK (high confidence in fraud)' if fraud_probability > 0.7 else '→ APPROVE (acceptable risk)'}"
|
||||
)
|
||||
|
||||
return fraud_probability
|
||||
@@ -0,0 +1,227 @@
|
||||
"""
|
||||
Kafka Consumer Pipeline Orchestrator
|
||||
|
||||
Responsibilities:
|
||||
- Connect to Kafka and consume transactions
|
||||
- Orchestrate the entire fraud detection pipeline
|
||||
- Coordinate all steps: feature extraction → scoring → decision → storage
|
||||
|
||||
Academic Point:
|
||||
This is where the architecture comes alive. By clearly separating concerns
|
||||
(features, models, decisions), the code becomes testable and maintainable.
|
||||
Each step is independently verifiable, making it suitable for academic demonstration.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from kafka import KafkaConsumer
|
||||
from kafka.errors import KafkaError
|
||||
|
||||
from features import FeatureExtractor
|
||||
from models import ModelScorer
|
||||
from decision import DecisionEngine
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
# CONFIGURATION
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
KAFKA_BROKER = "localhost:9092"
|
||||
KAFKA_TOPIC = "transactions"
|
||||
KAFKA_GROUP = "fraud-detection-consumer-group"
|
||||
|
||||
# Logging setup
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s [%(levelname)s] %(message)s'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
# PIPELINE INITIALIZATION
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
def initialize_pipeline():
|
||||
"""Initialize all components of the fraud detection pipeline"""
|
||||
|
||||
logger.info("="*80)
|
||||
logger.info("PayGuard Fraud Detection Consumer - INITIALIZATION")
|
||||
logger.info("="*80)
|
||||
|
||||
# Feature extraction
|
||||
feature_extractor = FeatureExtractor(
|
||||
redis_host="localhost",
|
||||
redis_port=6379
|
||||
)
|
||||
|
||||
# Model scoring
|
||||
model_scorer = ModelScorer(
|
||||
layer1_path="models/layer1.pkl",
|
||||
layer2_path="models/layer2.pkl"
|
||||
)
|
||||
|
||||
# Decision engine with circuit breaker
|
||||
decision_engine = DecisionEngine(
|
||||
redis_host="localhost",
|
||||
redis_port=6379
|
||||
)
|
||||
|
||||
# Kafka consumer
|
||||
consumer = KafkaConsumer(
|
||||
KAFKA_TOPIC,
|
||||
bootstrap_servers=KAFKA_BROKER,
|
||||
group_id=KAFKA_GROUP,
|
||||
value_deserializer=lambda m: json.loads(m.decode('utf-8')),
|
||||
auto_offset_reset='earliest',
|
||||
enable_auto_commit=True
|
||||
)
|
||||
|
||||
logger.info(f"[KAFKA CONSUMER] Connected to {KAFKA_BROKER}")
|
||||
logger.info(f"[KAFKA CONSUMER] Subscribed to topic '{KAFKA_TOPIC}'")
|
||||
logger.info(f"[KAFKA CONSUMER] Consumer group: {KAFKA_GROUP}")
|
||||
logger.info("="*80)
|
||||
logger.info("Waiting for transactions...\n")
|
||||
|
||||
return {
|
||||
'consumer': consumer,
|
||||
'feature_extractor': feature_extractor,
|
||||
'model_scorer': model_scorer,
|
||||
'decision_engine': decision_engine
|
||||
}
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
# PIPELINE EXECUTION
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
def process_transaction(transaction, components):
|
||||
"""
|
||||
Execute the complete fraud detection pipeline for one transaction.
|
||||
|
||||
Steps:
|
||||
1. Feature extraction (with Redis behavioral profiling)
|
||||
2. Layer 1 fast model scoring
|
||||
3. Behavioral anomaly detection
|
||||
4. Layer 2 deep model scoring (if suspicious)
|
||||
5. Final decision with circuit breaker
|
||||
6. Update user profile
|
||||
7. Store result
|
||||
"""
|
||||
|
||||
transaction_id = transaction['transaction_id']
|
||||
user_id = transaction['user_id']
|
||||
amount = transaction['amount']
|
||||
merchant = transaction['merchant']
|
||||
|
||||
logger.info(f"\n{'='*80}")
|
||||
logger.info(f"[PIPELINE START] Transaction: {transaction_id[:8]}...")
|
||||
logger.info(f"[TRANSACTION] User: {user_id} | Amount: ${amount:.2f} | Merchant: {merchant}")
|
||||
logger.info(f"{'='*80}\n")
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
# STEP 1: FEATURE EXTRACTION
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
logger.info("[STEP 1/6] Feature Extraction & Behavioral Profiling")
|
||||
logger.info("-" * 80)
|
||||
|
||||
features = components['feature_extractor'].extract_features(transaction)
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
# STEP 2: LAYER 1 FAST MODEL
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
logger.info("\n[STEP 2/6] Layer 1 - Fast Model Scoring")
|
||||
logger.info("-" * 80)
|
||||
|
||||
layer1_score = components['model_scorer'].score_layer1(features)
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
# STEP 3: BEHAVIORAL ANOMALY DETECTION
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
logger.info("\n[STEP 3/6] Behavioral Anomaly Detection")
|
||||
logger.info("-" * 80)
|
||||
|
||||
user_profile = components['feature_extractor'].get_user_profile(user_id)
|
||||
if user_profile['avg_amount'] > 0:
|
||||
ratio = amount / user_profile['avg_amount']
|
||||
logger.info(f"[ANOMALY] Amount/Avg Ratio: {ratio:.2f}x")
|
||||
if ratio > 3.0:
|
||||
logger.warning(f"[ANOMALY] ⚠ Significant deviation detected (>{3.0}x)")
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
# STEP 4: LAYER 2 DEEP MODEL (if needed)
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
layer2_score = None
|
||||
if layer1_score >= 0.4:
|
||||
logger.info("\n[STEP 4/6] Layer 2 - Deep Model Scoring")
|
||||
logger.info("-" * 80)
|
||||
|
||||
layer2_score = components['model_scorer'].score_layer2(features)
|
||||
else:
|
||||
logger.info("\n[STEP 4/6] Layer 2 - Deep Model Scoring")
|
||||
logger.info("-" * 80)
|
||||
logger.info("[LAYER 2 SCORING] Skipped (Layer 1 score indicates low risk)")
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
# STEP 5: FINAL DECISION
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
logger.info("\n[STEP 5/6] Final Decision Engine")
|
||||
logger.info("-" * 80)
|
||||
|
||||
decision = components['decision_engine'].make_decision(
|
||||
transaction_id=transaction_id,
|
||||
layer1_score=layer1_score,
|
||||
layer2_scorer=components['model_scorer'] if layer1_score >= 0.4 else None,
|
||||
features=features if layer1_score >= 0.4 else None
|
||||
)
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
# STEP 6: UPDATE PROFILE & STORE RESULT
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
logger.info("\n[STEP 6/6] Profile Update & Result Storage")
|
||||
logger.info("-" * 80)
|
||||
|
||||
components['feature_extractor'].update_user_profile(transaction)
|
||||
components['decision_engine'].store_result(
|
||||
transaction_id=transaction_id,
|
||||
decision=decision,
|
||||
layer1_score=layer1_score,
|
||||
layer2_score=layer2_score
|
||||
)
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
# PIPELINE COMPLETE
|
||||
# ─────────────────────────────────────────────────────────────────────────
|
||||
logger.info(f"\n{'='*80}")
|
||||
logger.info(f"[PIPELINE COMPLETE] Transaction {transaction_id[:8]}... → {decision.upper()}")
|
||||
logger.info(f"{'='*80}\n")
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
# MAIN LOOP
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
def run_consumer():
|
||||
"""Main loop: consume and process transactions from Kafka"""
|
||||
|
||||
components = initialize_pipeline()
|
||||
|
||||
try:
|
||||
logger.info("Consumer ready. Processing messages...\n")
|
||||
|
||||
for message in components['consumer']:
|
||||
if message.value is None:
|
||||
continue
|
||||
|
||||
transaction = message.value
|
||||
|
||||
try:
|
||||
process_transaction(transaction, components)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[ERROR] Failed to process transaction: {str(e)}", exc_info=True)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
logger.info("\nShutting down consumer...")
|
||||
finally:
|
||||
components['consumer'].close()
|
||||
logger.info("Consumer closed.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_consumer()
|
||||
Reference in new issue
Block a user