""" Smart Notification Scheduler for Phase 2 Intelligent scheduling to avoid notification fatigue """ from datetime import datetime, time, timedelta from typing import List, Optional from sqlalchemy.orm import Session from app.models.models import Notification, UserProfile from app.services.notification_service import NotificationService from apscheduler.schedulers.asyncio import AsyncIOScheduler import logging logger = logging.getLogger(__name__) class SmartNotificationScheduler: """Scheduler for intelligent notification delivery""" def __init__(self): self.scheduler: Optional[AsyncIOScheduler] = None async def initialize(self): """Initialize the scheduler""" self.scheduler = AsyncIOScheduler() self.scheduler.start() # Daily report at 5 PM self.scheduler.add_job( self.send_daily_reports, 'cron', hour=17, minute=0, id='daily_reports' ) # Weekly report every Friday at 6 PM self.scheduler.add_job( self.send_weekly_reports, 'cron', day_of_week=4, hour=18, minute=0, id='weekly_reports' ) # Check and batch notifications every hour self.scheduler.add_job( self.batch_and_send_notifications, 'interval', hours=1, id='batch_notifications' ) # Cleanup old notifications daily at 2 AM self.scheduler.add_job( self.cleanup_old_notifications, 'cron', hour=2, minute=0, id='cleanup_notifications' ) logger.info("Smart notification scheduler initialized") async def send_daily_reports(self, db: Session = None): """Send daily reports to users""" if not db: from app.db.database import SessionLocal db = SessionLocal() try: profiles = db.query(UserProfile).filter( UserProfile.email_reports == True, UserProfile.email != None ).all() for profile in profiles: from app.services.email_service import EmailService await EmailService.send_daily_report(db, profile.email) logger.info(f"Daily reports sent to {len(profiles)} users") except Exception as e: logger.error(f"Error sending daily reports: {str(e)}") finally: db.close() async def send_weekly_reports(self, db: Session = None): """Send weekly reports to users""" if not db: from app.db.database import SessionLocal db = SessionLocal() try: profiles = db.query(UserProfile).filter( UserProfile.email_reports == True, UserProfile.email != None ).all() for profile in profiles: from app.services.email_service import EmailService await EmailService.send_weekly_report(db, profile.email) logger.info(f"Weekly reports sent to {len(profiles)} users") except Exception as e: logger.error(f"Error sending weekly reports: {str(e)}") finally: db.close() async def batch_and_send_notifications(self, db: Session = None): """Batch notifications to avoid overwhelming users""" if not db: from app.db.database import SessionLocal db = SessionLocal() try: # Get all unread notifications grouped by priority from sqlalchemy import func # Count unread by priority unread_stats = db.query( Notification.priority, func.count(Notification.id) ).filter( Notification.read == False ).group_by( Notification.priority ).all() # Log batch statistics for priority, count in unread_stats: logger.info(f"Unread notifications - {priority}: {count}") # In production, implement batching logic: # - Group low-priority notifications # - Send digest emails instead of individual notifications # - Respect user's quiet hours # - Limit notification frequency except Exception as e: logger.error(f"Error batching notifications: {str(e)}") finally: db.close() async def cleanup_old_notifications(self, db: Session = None): """Clean up old notifications""" if not db: from app.db.database import SessionLocal db = SessionLocal() try: cutoff_date = datetime.utcnow() - timedelta(days=30) deleted = db.query(Notification).filter( Notification.created_at < cutoff_date ).delete() db.commit() logger.info(f"Cleaned up {deleted} old notifications") except Exception as e: logger.error(f"Error cleaning up notifications: {str(e)}") finally: db.close() async def shutdown(self): """Shutdown the scheduler""" if self.scheduler: self.scheduler.shutdown() logger.info("Notification scheduler shut down") class NotificationOptimizer: """Optimizes notification delivery timing and frequency""" @staticmethod def get_optimal_delivery_time( profile: UserProfile, notification_type: str, ) -> datetime: """ Calculate optimal delivery time for a notification Considers: - User's trading hours - Notification type priority - User's timezone - Quiet hours """ from pytz import timezone as tz_lib try: # Parse user's timezone user_tz = tz_lib(profile.timezone) now = datetime.now(user_tz) # Parse trading hours trading_start = datetime.strptime( profile.preferred_trading_start, "%H:%M" ).time() trading_end = datetime.strptime( profile.preferred_trading_end, "%H:%M" ).time() # Determine delivery time based on notification type if notification_type == "critical": # Critical: Send immediately return now elif notification_type == "price_alert": # Price alerts: During trading hours if trading_start <= now.time() <= trading_end: return now else: # Queue for next trading start next_start = now.replace( hour=trading_start.hour, minute=trading_start.minute, second=0 ) if next_start <= now: next_start += timedelta(days=1) return next_start elif notification_type == "routine": # Routines: At scheduled time return now elif notification_type == "report": # Reports: End of trading day return now.replace( hour=trading_end.hour, minute=trading_end.minute, second=0 ) else: # Default: Send immediately return now except Exception as e: logger.error(f"Error calculating optimal delivery time: {str(e)}") return datetime.now() @staticmethod def should_suppress_notification( notification_type: str, recent_notifications: List[Notification], minutes_back: int = 60, ) -> bool: """ Determine if notification should be suppressed Prevents notification fatigue by checking: - Recent notifications of same type - Notification frequency - User preferences """ cutoff_time = datetime.utcnow() - timedelta(minutes=minutes_back) similar_recent = [ n for n in recent_notifications if (n.notification_type == notification_type and n.created_at > cutoff_time) ] # Suppress if more than 5 similar notifications in last hour if len(similar_recent) > 5: logger.warning( f"Suppressing {notification_type} notification - " f"{len(similar_recent)} recent notifications" ) return True return False @staticmethod async def optimize_notification_chain( db: Session, notifications: List[dict], ) -> List[dict]: """ Optimize a batch of pending notifications Combines similar notifications and removes duplicates """ optimized = [] seen_types = set() for notif in notifications: notif_type = notif.get('notification_type') # Check if we've already added this type if notif_type in seen_types: continue optimized.append(notif) seen_types.add(notif_type) logger.info( f"Optimized {len(notifications)} notifications " f"to {len(optimized)} after deduplication" ) return optimized # Global scheduler instance notification_scheduler = SmartNotificationScheduler()