John Patrick TeruelandCopilot
4cefdb9083
- 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]>
139 lines
5.0 KiB
Python
139 lines
5.0 KiB
Python
"""
|
|
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
|
|
import os
|
|
from pybreaker import CircuitBreaker
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class DecisionEngine:
|
|
"""Makes final approve/block decisions with resilience patterns"""
|
|
|
|
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,
|
|
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}")
|