""" Balance alert hook example. Monitors balance thresholds and sends alerts when they are crossed. POST hooks - run after transactions complete. """ import logging from decimal import Decimal from typing import Dict, Optional from wallet_utils.hooks import HookContext logger = logging.getLogger(__name__) class BalanceMonitor: """ Monitors wallet balances and triggers alerts on threshold crossings. Tracks balance states to avoid duplicate alerts. """ def __init__(self): # Track last alert state per user/point_type to avoid spam # Format: {(user_id, point_type): last_alert_level} self._alert_states: Dict[tuple, str] = {} # Define alert thresholds self.thresholds = { "critical": Decimal("10.00"), # Balance critically low "low": Decimal("100.00"), # Balance low "medium": Decimal("500.00"), # Balance getting low } self.logger = logging.getLogger("wallet.balance_monitor") def check_balance_level(self, balance: Decimal) -> Optional[str]: """ Determine alert level for given balance. Returns: 'critical', 'low', 'medium', or None if no alert needed """ if balance <= self.thresholds["critical"]: return "critical" elif balance <= self.thresholds["low"]: return "low" elif balance <= self.thresholds["medium"]: return "medium" return None def should_alert(self, user_id: int, point_type: str, current_level: Optional[str]) -> bool: """ Check if we should send alert based on state changes. Only alerts when: - First time crossing threshold - Moving to more critical level - Balance recovered and then dropped again """ key = (user_id, point_type) last_level = self._alert_states.get(key) # Update state if current_level: self._alert_states[key] = current_level else: # Balance recovered, clear state if key in self._alert_states: del self._alert_states[key] return False # Alert if: # 1. First time alerting (no previous state) # 2. Crossing to more critical level if last_level is None: return True level_severity = {"medium": 1, "low": 2, "critical": 3} if level_severity.get(current_level, 0) > level_severity.get(last_level, 0): return True return False def set_thresholds(self, critical: Decimal = None, low: Decimal = None, medium: Decimal = None): """Update threshold values.""" if critical is not None: self.thresholds["critical"] = critical if low is not None: self.thresholds["low"] = low if medium is not None: self.thresholds["medium"] = medium # Global monitor instance _monitor = BalanceMonitor() def check_low_balance(context: HookContext) -> None: """ POST hook to monitor balance and send alerts when thresholds are crossed. Monitors balance after transactions and sends graduated alerts: - Medium alert: Balance below 500 - Low alert: Balance below 100 - Critical alert: Balance below 10 Args: context: Hook context with transaction details Example: >>> from wallet_utils.hooks import register_hook, HookType >>> register_hook( ... name="balance_monitor", ... hook_type=HookType.POST, ... callback=check_low_balance, ... operation="*", # Monitor all operations ... priority=700, ... ) """ # Skip if no current balance info if context.current_balance is None: return user_id = context.user_id point_type = context.point_type balance = context.current_balance # Check alert level alert_level = _monitor.check_balance_level(balance) # Determine if we should alert if not _monitor.should_alert(user_id, point_type, alert_level): return # Build alert message based on level messages = { "critical": { "title": "⚠️ Critical Balance Alert", "message": ( f"Your {point_type} balance is critically low at {balance}. " f"Immediate action required to avoid transaction failures." ), "severity": "critical", }, "low": { "title": "⚠️ Low Balance Alert", "message": ( f"Your {point_type} balance is low at {balance}. " f"Please consider adding more points soon." ), "severity": "warning", }, "medium": { "title": "ℹ️ Balance Notice", "message": ( f"Your {point_type} balance is {balance}. " f"You may want to add more points soon." ), "severity": "info", }, } alert_info = messages.get(alert_level) if not alert_info: return # Log the alert _monitor.logger.warning( f"Balance alert [{alert_level.upper()}] for user {user_id}: " f"{point_type} = {balance}" ) # Store alert in metadata for potential notification hook context.metadata["balance_alert"] = { "level": alert_level, "threshold": float(_monitor.thresholds[alert_level]), "current_balance": float(balance), "title": alert_info["title"], "message": alert_info["message"], "severity": alert_info["severity"], } # In production, trigger actual notification # _send_notification(user_id, alert_info) def check_balance_milestone(context: HookContext) -> None: """ POST hook to celebrate positive balance milestones. Sends congratulatory messages when balance crosses certain positive thresholds. Args: context: Hook context with transaction details Example: >>> from wallet_utils.hooks import register_hook, HookType >>> register_hook( ... name="milestone_checker", ... hook_type=HookType.POST, ... callback=check_balance_milestone, ... operation="add", # Only for additions ... priority=700, ... ) """ # Only check on adds if context.operation != "add": return if context.current_balance is None: return milestones = [ Decimal("1000.00"), Decimal("5000.00"), Decimal("10000.00"), Decimal("50000.00"), Decimal("100000.00"), ] # Check if we just crossed a milestone balance_before = context.current_balance - context.amount for milestone in milestones: if balance_before < milestone <= context.current_balance: logger.info( f"🎉 User {context.user_id} reached {context.point_type} " f"milestone: {milestone}" ) context.metadata["milestone_reached"] = { "milestone": float(milestone), "point_type": context.point_type, "current_balance": float(context.current_balance), } # In production, send congratulatory notification # _send_milestone_notification(context.user_id, milestone, context.point_type) break def get_monitor() -> BalanceMonitor: """Get the global balance monitor instance.""" return _monitor def configure_balance_thresholds( critical: Decimal = None, low: Decimal = None, medium: Decimal = None, ) -> None: """ Configure balance alert thresholds. Args: critical: Critical balance threshold (default: 10.00) low: Low balance threshold (default: 100.00) medium: Medium balance threshold (default: 500.00) Example: >>> from decimal import Decimal >>> configure_balance_thresholds( ... critical=Decimal("5.00"), ... low=Decimal("50.00"), ... medium=Decimal("200.00"), ... ) """ _monitor.set_thresholds(critical, low, medium)