""" Notification Service Handles notification creation, delivery, and management """ from datetime import datetime from typing import Optional, List, Dict from sqlalchemy.orm import Session from app.models.models import Notification import asyncio import logging logger = logging.getLogger(__name__) class NotificationService: """Service for managing notifications""" # Notification priority levels PRIORITY_LEVELS = { "critical": 1, "high": 2, "normal": 3, "low": 4, } # Delivery methods DELIVERY_METHODS = ["push", "email", "sms"] @staticmethod async def create_notification( db: Session, notification_type: str, title: str, message: str, priority: str = "normal", delivery_method: str = "push", data: Optional[Dict] = None, user_id: Optional[str] = None, ) -> Notification: """Create a new notification""" if priority not in NotificationService.PRIORITY_LEVELS: priority = "normal" if delivery_method not in NotificationService.DELIVERY_METHODS: delivery_method = "push" notification = Notification( user_id=user_id, notification_type=notification_type, title=title, message=message, priority=priority, delivery_method=delivery_method, data=data or {}, read=False, ) db.add(notification) db.commit() db.refresh(notification) # Send notification based on delivery method await NotificationService.send_notification(notification) return notification @staticmethod async def send_notification(notification: Notification): """Send notification via appropriate delivery method""" try: if notification.delivery_method == "push": await NotificationService.send_push(notification) elif notification.delivery_method == "email": await NotificationService.send_email(notification) elif notification.delivery_method == "sms": await NotificationService.send_sms(notification) except Exception as e: logger.error(f"Failed to send notification {notification.id}: {str(e)}") @staticmethod async def send_push(notification: Notification): """Send push notification (browser/WebSocket)""" # In production, this would integrate with push notification service # For now, log it logger.info( f"Push notification: {notification.title} - {notification.message}" ) @staticmethod async def send_email(notification: Notification): """Send email notification""" # In production, integrate with email service (SendGrid, Mailgun, etc.) logger.info( f"Email notification: {notification.title} - {notification.message}" ) @staticmethod async def send_sms(notification: Notification): """Send SMS notification""" # In production, integrate with SMS service (Twilio, etc.) logger.info( f"SMS notification: {notification.title} - {notification.message}" ) @staticmethod async def mark_as_read( db: Session, notification_id: int ) -> Notification: """Mark notification as read""" notification = db.query(Notification).filter( Notification.id == notification_id ).first() if notification: notification.read = True notification.read_at = datetime.utcnow() db.commit() db.refresh(notification) return notification @staticmethod async def mark_all_as_read(db: Session, user_id: Optional[str] = None): """Mark all notifications as read""" query = db.query(Notification).filter(Notification.read == False) if user_id: query = query.filter(Notification.user_id == user_id) query.update({ Notification.read: True, Notification.read_at: datetime.utcnow() }) db.commit() @staticmethod def get_unread_count(db: Session, user_id: Optional[str] = None) -> int: """Get count of unread notifications""" query = db.query(Notification).filter(Notification.read == False) if user_id: query = query.filter(Notification.user_id == user_id) return query.count() @staticmethod async def create_price_alert_notification( db: Session, price: float, threshold: float, direction: str, # "above" or "below" ): """Create notification for price alert""" title = f"Price Alert: Gold {direction} ${threshold:.2f}" message = f"Gold price has moved {direction} your alert level. Current: ${price:.2f}" await NotificationService.create_notification( db, notification_type="price_alert", title=title, message=message, priority="high", delivery_method="push", data={ "price": price, "threshold": threshold, "direction": direction } ) @staticmethod async def create_routine_notification( db: Session, routine_type: str, routine_name: str, ): """Create notification for routine execution""" messages = { "morning": f"Good morning! Time for your {routine_name} routine.", "active_trading": f"Active trading time! {routine_name} routine.", "evening": f"End of trading day. Time to review with {routine_name} routine.", } title = f"Daily Routine: {routine_name}" message = messages.get(routine_type, f"Time for {routine_name} routine") await NotificationService.create_notification( db, notification_type="routine", title=title, message=message, priority="normal", delivery_method="push", data={ "routine_type": routine_type, "routine_name": routine_name } ) @staticmethod async def create_news_notification( db: Session, news_title: str, summary: str, impact: str = "medium", # low, medium, high ): """Create notification for important news""" priority_map = { "low": "low", "medium": "normal", "high": "high", } await NotificationService.create_notification( db, notification_type="news", title=f"Market News: {news_title}", message=summary[:200], # Limit to 200 chars priority=priority_map.get(impact, "normal"), delivery_method="push", data={ "news_title": news_title, "impact": impact } ) @staticmethod async def create_reminder_notification( db: Session, reminder_type: str, title: str, message: str, ): """Create reminder notification""" await NotificationService.create_notification( db, notification_type="reminder", title=title, message=message, priority="normal", delivery_method="push", data={ "reminder_type": reminder_type } ) @staticmethod async def create_performance_notification( db: Session, metric: str, value: float, target: Optional[float] = None, ): """Create notification for performance metrics""" if target: message = f"Your {metric} is {value:.2f} (target: {target:.2f})" else: message = f"Your {metric} is {value:.2f}" await NotificationService.create_notification( db, notification_type="report", title=f"Performance: {metric}", message=message, priority="normal", delivery_method="push", data={ "metric": metric, "value": value, "target": target } ) class NotificationScheduler: """Schedule notification delivery and cleanup""" @staticmethod async def cleanup_old_notifications(db: Session, days: int = 30): """Remove notifications older than specified days""" from datetime import timedelta cutoff_date = datetime.utcnow() - timedelta(days=days) db.query(Notification).filter( Notification.created_at < cutoff_date ).delete() db.commit() @staticmethod async def batch_notifications( db: Session, user_id: Optional[str] = None, ) -> List[Notification]: """Get all unread notifications and batch them for delivery""" query = db.query(Notification).filter( Notification.read == False ).order_by(Notification.priority, Notification.created_at.desc()) if user_id: query = query.filter(Notification.user_id == user_id) return query.all()