from typing import List, Dict, Optional from datetime import datetime, timedelta import uuid from app.schemas.schemas import ( Alert, AlertType, AlertSeverity, AlertsResponse, PriceData, NewsPriceCorrelation, CorrelationAnalysisResponse, ) from app.config import settings class AlertService: def __init__(self): self.alerts: List[Alert] = [] self.price_history: List[PriceData] = [] self.last_price: Optional[float] = None self.support_levels: List[float] = [] self.resistance_levels: List[float] = [] def set_support_resistance(self, support: List[float], resistance: List[float]): """Set support and resistance levels for breach detection""" self.support_levels = support self.resistance_levels = resistance def add_price_data(self, price_data: PriceData): """Add new price data and check for alerts""" self.price_history.append(price_data) # Keep only last 1000 data points if len(self.price_history) > 1000: self.price_history = self.price_history[-1000:] current_price = price_data.close if self.last_price: self._check_price_alerts(current_price, self.last_price) self._check_volatility_alerts(price_data) self._check_support_resistance_breach(current_price) self.last_price = current_price def _check_price_alerts(self, current_price: float, last_price: float): """Check for significant price movements""" change_percent = ((current_price - last_price) / last_price) * 100 threshold = settings.PRICE_ALERT_THRESHOLD if abs(change_percent) >= threshold: if change_percent > 0: alert_type = AlertType.PRICE_SPIKE title = f"Gold Price Spike: +{change_percent:.2f}%" severity = AlertSeverity.HIGH if change_percent > 2.0 else AlertSeverity.MEDIUM else: alert_type = AlertType.PRICE_DROP title = f"Gold Price Drop: {change_percent:.2f}%" severity = AlertSeverity.HIGH if change_percent < -2.0 else AlertSeverity.MEDIUM alert = Alert( id=str(uuid.uuid4()), type=alert_type, severity=severity, title=title, message=f"Gold price moved from ${last_price:.2f} to ${current_price:.2f} ({change_percent:+.2f}%)", price=current_price, change_percent=change_percent, timestamp=datetime.now(), action_required=severity == AlertSeverity.HIGH, ) self.alerts.append(alert) def _check_volatility_alerts(self, price_data: PriceData): """Check for high volatility conditions""" if len(self.price_history) < 20: return # Calculate ATR-like volatility recent_data = self.price_history[-20:] ranges = [d.high - d.low for d in recent_data] avg_range = sum(ranges) / len(ranges) current_range = price_data.high - price_data.low # Alert if current range is 2x average if current_range > avg_range * 2: alert = Alert( id=str(uuid.uuid4()), type=AlertType.HIGH_VOLATILITY, severity=AlertSeverity.MEDIUM, title="High Volatility Detected", message=f"Current price range ${current_range:.2f} is significantly higher than average ${avg_range:.2f}", price=price_data.close, timestamp=datetime.now(), ) self.alerts.append(alert) def _check_support_resistance_breach(self, current_price: float): """Check if price breached support or resistance levels""" if not self.last_price: return # Check resistance breach (upward) for resistance in self.resistance_levels: if self.last_price < resistance <= current_price: alert = Alert( id=str(uuid.uuid4()), type=AlertType.RESISTANCE_BREACH, severity=AlertSeverity.HIGH, title=f"Resistance Breached: ${resistance:.2f}", message=f"Gold price broke above resistance level of ${resistance:.2f}", price=current_price, timestamp=datetime.now(), action_required=True, ) self.alerts.append(alert) # Check support breach (downward) for support in self.support_levels: if self.last_price > support >= current_price: alert = Alert( id=str(uuid.uuid4()), type=AlertType.SUPPORT_BREACH, severity=AlertSeverity.HIGH, title=f"Support Breached: ${support:.2f}", message=f"Gold price broke below support level of ${support:.2f}", price=current_price, timestamp=datetime.now(), action_required=True, ) self.alerts.append(alert) def add_news_alert(self, news_title: str, impact: str, sentiment: str): """Add alert for breaking news""" severity_map = { "HIGH": AlertSeverity.CRITICAL, "MEDIUM": AlertSeverity.HIGH, "LOW": AlertSeverity.MEDIUM, } alert = Alert( id=str(uuid.uuid4()), type=AlertType.NEWS_BREAKING, severity=severity_map.get(impact, AlertSeverity.MEDIUM), title=f"Breaking: {news_title[:50]}...", message=f"High-impact news detected: {news_title}", timestamp=datetime.now(), action_required=impact == "HIGH", ) self.alerts.append(alert) def add_economic_event_alert(self, event_title: str, importance: str): """Add alert for upcoming economic event""" severity_map = { "HIGH": AlertSeverity.HIGH, "MEDIUM": AlertSeverity.MEDIUM, "LOW": AlertSeverity.LOW, } alert = Alert( id=str(uuid.uuid4()), type=AlertType.ECONOMIC_EVENT, severity=severity_map.get(importance, AlertSeverity.MEDIUM), title=f"Upcoming: {event_title}", message=f"Important economic event scheduled: {event_title}", timestamp=datetime.now(), action_required=importance == "HIGH", ) self.alerts.append(alert) def get_alerts(self, limit: int = 50) -> AlertsResponse: """Get recent alerts""" # Sort by timestamp (newest first) sorted_alerts = sorted(self.alerts, key=lambda x: x.timestamp, reverse=True) # Limit results recent_alerts = sorted_alerts[:limit] # Count critical alerts critical_count = sum(1 for a in recent_alerts if a.severity == AlertSeverity.CRITICAL) # For MVP, all alerts are unread unread_count = len(recent_alerts) return AlertsResponse( alerts=recent_alerts, critical_count=critical_count, unread_count=unread_count, ) def clear_old_alerts(self, hours: int = 24): """Remove alerts older than specified hours""" cutoff = datetime.now() - timedelta(hours=hours) self.alerts = [a for a in self.alerts if a.timestamp > cutoff] def analyze_news_price_correlation( self, news_articles: List, price_data: List[PriceData], ) -> CorrelationAnalysisResponse: """Analyze correlation between news and price movements""" correlations = [] for article in news_articles: news_time = article.published_at # Find price before and after news price_before = None price_after = None for i, data in enumerate(price_data): data_time = datetime.fromtimestamp(data.time) # Price before news (within 1 hour before) if data_time < news_time and (news_time - data_time).total_seconds() < 3600: price_before = data.close # Price after news (within 1 hour after) if data_time > news_time and (data_time - news_time).total_seconds() < 3600: if not price_after: # Take first price after price_after = data.close if price_before and price_after: price_change = price_after - price_before price_change_percent = (price_change / price_before) * 100 time_delta = 60 # Approximate minutes # Determine correlation strength if abs(price_change_percent) > 1.0: strength = "STRONG" elif abs(price_change_percent) > 0.5: strength = "MODERATE" else: strength = "WEAK" correlation = NewsPriceCorrelation( news_id=article.id, news_title=article.title, news_time=news_time, price_before=price_before, price_after=price_after, price_change=price_change, price_change_percent=price_change_percent, time_delta_minutes=time_delta, correlation_strength=strength, ) correlations.append(correlation) # Calculate statistics significant_events = sum(1 for c in correlations if c.correlation_strength in ["STRONG", "MODERATE"]) avg_impact = ( sum(abs(c.price_change_percent) for c in correlations) / len(correlations) if correlations else 0.0 ) return CorrelationAnalysisResponse( correlations=correlations[:20], # Limit to 20 most recent significant_events=significant_events, avg_price_impact=avg_impact, ) # Global instance alert_service = AlertService()