"""Social Momentum Spike Detector — tracks social media growth, community sentiment, and developer activity to detect momentum spikes that often precede price moves. Analyses AssetProfile social metrics (twitter, reddit, telegram, github) and sentiment votes alongside news sentiment from AssetExtendedData to produce a per-asset momentum score (0-100). Data sources (all lazy-imported): - Asset: id, symbol, name, asset_type, is_active - AssetProfile: twitter_followers, reddit_subscribers, telegram_users, github_stars, sentiment_votes_up, profile_json (contains developer_score, community_score, liquidity_score, github_forks, github_commits_4w, sentiment_votes_down_percentage, coingecko_score), fetched_at - AssetExtendedData (data_type='news'): news sentiment data Scoring: social_dominance_score 0-100 (z-score rank of social metrics vs peers) developer_activity_score 0-100 (github stars + forks + dev score) sentiment_score 0-100 (vote ratio + news sentiment) community_score 0-100 (CoinGecko community score normalized) momentum_score 0-100 (weighted: social 35%, dev 25%, sentiment 25%, community 15%) Signals: HOT >= 80 WARMING >= 60 NEUTRAL >= 40 COOLING >= 20 COLD < 20 Alert types: VIRAL_GROWTH — social dominance >= 85 DEVELOPER_SURGE — developer activity >= 85 SENTIMENT_SHIFT — sentiment score >= 85 COMMUNITY_SPIKE — community score >= 85 NONE — no spike detected """ from __future__ import annotations import json import logging import math import statistics from datetime import datetime, timedelta from typing import Any, Optional logger = logging.getLogger(__name__) # ── Weight configuration ───────────────────────────────────────────── _W_SOCIAL = 0.35 _W_DEVELOPER = 0.25 _W_SENTIMENT = 0.25 _W_COMMUNITY = 0.15 # Signal thresholds _SIGNAL_HOT = 80 _SIGNAL_WARMING = 60 _SIGNAL_NEUTRAL = 40 _SIGNAL_COOLING = 20 # Alert thresholds _ALERT_THRESHOLD = 85 # Normalization caps for individual sub-scores _MAX_TWITTER = 500_000 # 500k followers -> 100 _MAX_REDDIT = 200_000 # 200k subscribers -> 100 _MAX_TELEGRAM = 100_000 # 100k users -> 100 _MAX_GITHUB_STARS = 10_000 # 10k stars -> 100 _MAX_GITHUB_FORKS = 5_000 # 5k forks -> 100 _MAX_DEV_SCORE = 100.0 # CoinGecko developer_score max _MAX_COMMUNITY_SCORE = 100.0 # CoinGecko community_score max # News sentiment keywords _POSITIVE_KEYWORDS = { 'bullish', 'surge', 'rally', 'soar', 'jump', 'gain', 'positive', 'upgrade', 'growth', 'outperform', 'partnership', 'adoption', 'launch', 'breakthrough', 'record', 'milestone', 'strong', 'profit', 'revenue', 'beat', 'exceed', 'innovation', 'approval', } _NEGATIVE_KEYWORDS = { 'bearish', 'crash', 'plunge', 'drop', 'decline', 'negative', 'downgrade', 'loss', 'underperform', 'hack', 'exploit', 'scam', 'fraud', 'lawsuit', 'regulation', 'ban', 'warning', 'weak', 'miss', 'delay', 'cancel', 'fail', 'risk', 'concern', 'sell-off', } def _safe_float(val, default: float = 0.0) -> float: """Safely convert a DB Numeric / Decimal / None to float.""" if val is None: return default try: f = float(val) return default if (math.isnan(f) or math.isinf(f)) else f except (ValueError, TypeError): return default def _safe_int(val, default: int = 0) -> int: """Safely convert to int.""" if val is None: return default try: return int(val) except (ValueError, TypeError): return default def _clamp(value: float, lo: float = 0.0, hi: float = 100.0) -> float: """Clamp a value to [lo, hi] range.""" return max(lo, min(hi, value)) def _normalize(value: float, maximum: float) -> float: """Normalize value to 0-100 scale given a maximum reference.""" if maximum <= 0: return 0.0 return _clamp((value / maximum) * 100.0) def _zscore(value: float, mean: float, std: float) -> float: """Compute z-score, returning 0.0 when std is zero.""" if std <= 0: return 0.0 return (value - mean) / std def _zscore_to_percentile(z: float) -> float: """Convert z-score to approximate 0-100 percentile score. Uses a simple sigmoid-like mapping: z=0 -> 50, z=2 -> ~88, z=-2 -> ~12. """ return _clamp(50.0 + z * 20.0) class SocialMomentumDetector: """Detects social momentum spikes across all tracked assets.""" # ────────────────────────────────────────────────────────────────── # 1. Main entry point # ────────────────────────────────────────────────────────────────── def get_social_momentum(self, limit: int = 50) -> dict: """Return social momentum analysis for active assets. Returns dict with keys: signals, summary, timestamp. """ try: from app.models.asset import Asset, AssetProfile from app.models.asset_extended_data import AssetExtendedData from app.helpers.asset_filter import get_asset_mode from app.extensions import db from sqlalchemy import func as sa_func asset_mode = get_asset_mode() # ── Load latest AssetProfile per asset ────────────────────── latest_sq = ( db.session.query( AssetProfile.asset_id, sa_func.max(AssetProfile.id).label('max_id'), ) .group_by(AssetProfile.asset_id) .subquery() ) profiles = ( AssetProfile.query .join(latest_sq, AssetProfile.id == latest_sq.c.max_id) .join(Asset, Asset.id == AssetProfile.asset_id) .filter( Asset.is_active.is_(True), Asset.asset_type == asset_mode, ) .all() ) if not profiles: return self._empty_result() # Build asset map asset_ids = [p.asset_id for p in profiles] coins_map = { c.id: c for c in Asset.query.filter(Asset.id.in_(asset_ids)).all() } # ── Load news data from AssetExtendedData ───────────────── news_map: dict[str, Any] = {} try: news_rows = ( AssetExtendedData.query .filter( AssetExtendedData.asset_id.in_(asset_ids), AssetExtendedData.data_type == 'news', ) .all() ) for nr in news_rows: if nr.data_json: news_map[nr.asset_id] = nr.data_json except Exception as e: logger.warning(f'Could not load news data: {e}') # ── Compute peer statistics for z-score ranking ─────────── peer_stats = self._compute_peer_stats(profiles) # ── Score each asset ────────────────────────────────────── signals = [] for profile in profiles: asset = coins_map.get(profile.asset_id) if not asset: continue pj = profile.profile_json or {} # Extract raw metrics twitter = _safe_int(profile.twitter_followers) reddit = _safe_int(profile.reddit_subscribers) telegram = _safe_int(profile.telegram_users) github_stars = _safe_int(profile.github_stars) github_forks = _safe_int(pj.get('github_forks', 0)) votes_up = _safe_float(profile.sentiment_votes_up) votes_down = _safe_float(pj.get('sentiment_votes_down_percentage', 0)) cg_community = _safe_float(pj.get('community_score', 0)) cg_developer = _safe_float(pj.get('developer_score', 0)) cg_coingecko = _safe_float(pj.get('coingecko_score', 0)) current_price = _safe_float( profile.current_price_idr if profile.current_price_idr else profile.current_price_usd ) # Compute sub-scores social_dom = self._compute_social_dominance( profile, peer_stats, ) dev_activity = self._compute_developer_activity( profile, pj, ) news_data = news_map.get(profile.asset_id) sentiment = self._compute_sentiment( profile, pj, news_data, ) community = _normalize(cg_community, _MAX_COMMUNITY_SCORE) # Combined momentum momentum = self._compute_momentum( social_dom, dev_activity, sentiment, community, ) # Determine signal and alert signal_label = self._get_signal(momentum) metrics = { 'social_dominance_score': social_dom, 'developer_activity_score': dev_activity, 'sentiment_score': sentiment, 'community_score_norm': community, 'momentum_score': momentum, 'twitter_followers': twitter, 'reddit_subscribers': reddit, 'telegram_users': telegram, 'github_stars': github_stars, 'github_forks': github_forks, 'cg_community': cg_community, 'cg_developer': cg_developer, } alert_type = self._detect_alert_type(metrics) reasons = self._build_reasons(metrics, signal_label, alert_type) # Sentiment ratio total_votes = votes_up + votes_down sentiment_ratio = round( (votes_up / total_votes * 100.0) if total_votes > 0 else 50.0, 1, ) # News sentiment label news_sentiment = self._parse_news_sentiment(news_data) signals.append({ 'symbol': asset.symbol, 'name': asset.name, 'current_price': current_price, 'asset_type': asset.asset_type, 'twitter_followers': twitter, 'reddit_subscribers': reddit, 'telegram_users': telegram, 'github_stars': github_stars, 'community_score': round(cg_community, 1), 'developer_score': round(cg_developer, 1), 'sentiment_ratio': sentiment_ratio, 'social_dominance_score': round(social_dom, 1), 'developer_activity_score': round(dev_activity, 1), 'sentiment_score': round(sentiment, 1), 'news_sentiment': news_sentiment, 'momentum_score': round(momentum, 1), 'signal': signal_label, 'alert_type': alert_type, 'reasons': reasons, }) # Sort by momentum_score descending signals.sort(key=lambda x: x['momentum_score'], reverse=True) # Apply limit signals = signals[:limit] # ── Summary ─────────────────────────────────────────────── hot_count = sum(1 for s in signals if s['signal'] == 'HOT') viral_alerts = sum( 1 for s in signals if s['alert_type'] != 'NONE' ) sentiment_scores = [ s['sentiment_score'] for s in signals if s['sentiment_score'] > 0 ] avg_sentiment = round( statistics.mean(sentiment_scores) if sentiment_scores else 0.0, 1, ) return { 'signals': signals, 'summary': { 'total_analyzed': len(signals), 'hot_count': hot_count, 'viral_alerts': viral_alerts, 'avg_sentiment': avg_sentiment, }, 'timestamp': datetime.utcnow().isoformat(), } except Exception as e: logger.error(f'Social momentum error: {e}', exc_info=True) return { 'signals': [], 'summary': { 'total_analyzed': 0, 'hot_count': 0, 'viral_alerts': 0, 'avg_sentiment': 0.0, }, 'error': str(e), 'timestamp': datetime.utcnow().isoformat(), } # ────────────────────────────────────────────────────────────────── # 2. Peer statistics (for z-score ranking) # ────────────────────────────────────────────────────────────────── def _compute_peer_stats(self, profiles) -> dict: """Compute mean and stdev for each social metric across all profiles. Returns dict with keys for each metric, each containing 'mean' and 'std' values used for z-score computation. """ twitter_vals = [] reddit_vals = [] telegram_vals = [] github_stars_vals = [] github_forks_vals = [] community_vals = [] developer_vals = [] for p in profiles: pj = p.profile_json or {} tw = _safe_int(p.twitter_followers) rd = _safe_int(p.reddit_subscribers) tg = _safe_int(p.telegram_users) gs = _safe_int(p.github_stars) gf = _safe_int(pj.get('github_forks', 0)) cs = _safe_float(pj.get('community_score', 0)) ds = _safe_float(pj.get('developer_score', 0)) if tw > 0: twitter_vals.append(float(tw)) if rd > 0: reddit_vals.append(float(rd)) if tg > 0: telegram_vals.append(float(tg)) if gs > 0: github_stars_vals.append(float(gs)) if gf > 0: github_forks_vals.append(float(gf)) if cs > 0: community_vals.append(cs) if ds > 0: developer_vals.append(ds) def _stats(vals): if len(vals) < 2: return {'mean': statistics.mean(vals) if vals else 0.0, 'std': 0.0} return { 'mean': statistics.mean(vals), 'std': statistics.stdev(vals), } return { 'twitter': _stats(twitter_vals), 'reddit': _stats(reddit_vals), 'telegram': _stats(telegram_vals), 'github_stars': _stats(github_stars_vals), 'github_forks': _stats(github_forks_vals), 'community': _stats(community_vals), 'developer': _stats(developer_vals), } # ────────────────────────────────────────────────────────────────── # 3. Social dominance (z-score relative to peers) # ────────────────────────────────────────────────────────────────── def _compute_social_dominance(self, profile, peer_stats: dict) -> float: """Rank social metrics vs peers using z-scores. Combines z-scores from twitter, reddit, telegram into a 0-100 score. Assets with more social presence relative to peers score higher. """ pj = profile.profile_json or {} twitter = _safe_float(profile.twitter_followers) reddit = _safe_float(profile.reddit_subscribers) telegram = _safe_float(profile.telegram_users) z_scores = [] weights = [] # Twitter z-score (weight 0.40) tw_stats = peer_stats.get('twitter', {}) if twitter > 0 and tw_stats.get('std', 0) > 0: z = _zscore(twitter, tw_stats['mean'], tw_stats['std']) z_scores.append(z) weights.append(0.40) elif twitter > 0: # No peer variance — use simple normalization z_scores.append(_normalize(twitter, _MAX_TWITTER) / 50.0 - 1.0) weights.append(0.40) # Reddit z-score (weight 0.35) rd_stats = peer_stats.get('reddit', {}) if reddit > 0 and rd_stats.get('std', 0) > 0: z = _zscore(reddit, rd_stats['mean'], rd_stats['std']) z_scores.append(z) weights.append(0.35) elif reddit > 0: z_scores.append(_normalize(reddit, _MAX_REDDIT) / 50.0 - 1.0) weights.append(0.35) # Telegram z-score (weight 0.25) tg_stats = peer_stats.get('telegram', {}) if telegram > 0 and tg_stats.get('std', 0) > 0: z = _zscore(telegram, tg_stats['mean'], tg_stats['std']) z_scores.append(z) weights.append(0.25) elif telegram > 0: z_scores.append(_normalize(telegram, _MAX_TELEGRAM) / 50.0 - 1.0) weights.append(0.25) if not z_scores: return 0.0 # Weighted average z-score total_weight = sum(weights) weighted_z = sum(z * w for z, w in zip(z_scores, weights)) / total_weight return round(_zscore_to_percentile(weighted_z), 1) # ────────────────────────────────────────────────────────────────── # 4. Developer activity score # ────────────────────────────────────────────────────────────────── def _compute_developer_activity(self, profile, pj: dict) -> float: """Score developer activity from github metrics and CoinGecko dev score. Components: - github_stars: normalized 0-100 (weight 0.30) - github_forks: normalized 0-100 (weight 0.20) - github_commits_4w: normalized 0-100 (weight 0.20) - developer_score (CoinGecko): normalized 0-100 (weight 0.30) """ github_stars = _safe_int(profile.github_stars) github_forks = _safe_int(pj.get('github_forks', 0)) github_commits_4w = _safe_int(pj.get('github_commits_4w', 0)) cg_dev_score = _safe_float(pj.get('developer_score', 0)) stars_norm = _normalize(float(github_stars), _MAX_GITHUB_STARS) forks_norm = _normalize(float(github_forks), _MAX_GITHUB_FORKS) # Commits: 100 commits in 4 weeks -> 100 commits_norm = _normalize(float(github_commits_4w), 100.0) # CoinGecko developer_score is typically 0-100 dev_score_norm = _normalize(cg_dev_score, _MAX_DEV_SCORE) # Check if we have any developer data at all has_data = ( github_stars > 0 or github_forks > 0 or github_commits_4w > 0 or cg_dev_score > 0 ) if not has_data: return 0.0 score = ( stars_norm * 0.30 + forks_norm * 0.20 + commits_norm * 0.20 + dev_score_norm * 0.30 ) return round(_clamp(score), 1) # ────────────────────────────────────────────────────────────────── # 5. Sentiment score (votes + news) # ────────────────────────────────────────────────────────────────── def _compute_sentiment( self, profile, pj: dict, news_data: Any, ) -> float: """Combine vote-based sentiment with news sentiment. Components: - vote_ratio: sentiment_votes_up / total * 100 -> normalized (weight 0.50) - news_sentiment: positive/negative keyword scoring (weight 0.30) - coingecko_score: overall CG score as quality proxy (weight 0.20) """ # ── Vote-based sentiment ────────────────────────────────────── votes_up = _safe_float(profile.sentiment_votes_up) votes_down = _safe_float(pj.get('sentiment_votes_down_percentage', 0)) total_votes = votes_up + votes_down if total_votes > 0: vote_ratio = (votes_up / total_votes) * 100.0 else: vote_ratio = 50.0 # neutral default # Map vote_ratio (0-100) to score # 50% -> 50 (neutral), 100% -> 100 (very positive), 0% -> 0 (very negative) vote_score = _clamp(vote_ratio) # ── News sentiment ──────────────────────────────────────────── news_score = self._score_news_sentiment(news_data) # ── CoinGecko overall score (quality proxy) ─────────────────── cg_score = _safe_float(pj.get('coingecko_score', 0)) cg_norm = _normalize(cg_score, 100.0) # ── Combine ────────────────────────────────────────────────── # If news data is available, give it weight; otherwise redistribute if news_data: score = ( vote_score * 0.50 + news_score * 0.30 + cg_norm * 0.20 ) else: # No news data: redistribute news weight to votes and CG score = ( vote_score * 0.65 + cg_norm * 0.35 ) return round(_clamp(score), 1) def _score_news_sentiment(self, news_data: Any) -> float: """Score news items for positive/negative sentiment. Returns 0-100 where 50 is neutral, >50 positive, <50 negative. """ if not news_data: return 50.0 articles = [] if isinstance(news_data, list): articles = news_data elif isinstance(news_data, dict): articles = news_data.get('articles', news_data.get('items', [])) if not articles and 'title' in news_data: articles = [news_data] if not articles: return 50.0 positive_count = 0 negative_count = 0 total_articles = 0 for article in articles: if not isinstance(article, dict): continue title = str(article.get('title', '')).lower() snippet = str(article.get('description', '')).lower() combined = f'{title} {snippet}' if not combined.strip(): continue total_articles += 1 pos_hits = sum(1 for kw in _POSITIVE_KEYWORDS if kw in combined) neg_hits = sum(1 for kw in _NEGATIVE_KEYWORDS if kw in combined) if pos_hits > neg_hits: positive_count += 1 elif neg_hits > pos_hits: negative_count += 1 if total_articles == 0: return 50.0 # Net sentiment: -1.0 (all negative) to +1.0 (all positive) net = (positive_count - negative_count) / total_articles # Map to 0-100 where 0.0 -> 50 return _clamp(50.0 + net * 50.0) # ────────────────────────────────────────────────────────────────── # 6. Combined momentum score # ────────────────────────────────────────────────────────────────── def _compute_momentum( self, social: float, developer: float, sentiment: float, community: float, ) -> float: """Weighted combination of sub-scores into final momentum score. Weights: social 35%, developer 25%, sentiment 25%, community 15%. If a sub-score is zero (missing data), redistribute its weight proportionally to the other components. """ components = [ (social, _W_SOCIAL), (developer, _W_DEVELOPER), (sentiment, _W_SENTIMENT), (community, _W_COMMUNITY), ] active_weight = 0.0 weighted_sum = 0.0 for score, weight in components: if score > 0: weighted_sum += score * weight active_weight += weight if active_weight <= 0: return 0.0 # Redistribute: scale up to account for missing components raw = weighted_sum / active_weight * (active_weight / 1.0) # But also penalize for missing data — fewer data sources = less # confidence, so scale by coverage ratio coverage = active_weight / 1.0 # 1.0 = sum of all weights penalty = 0.7 + 0.3 * coverage # 70% floor, 100% at full coverage return round(_clamp(raw * penalty), 1) # ────────────────────────────────────────────────────────────────── # 7. Alert type detection # ────────────────────────────────────────────────────────────────── def _detect_alert_type(self, metrics: dict) -> str: """Detect the most prominent spike type based on sub-score thresholds. Priority order: VIRAL_GROWTH > DEVELOPER_SURGE > SENTIMENT_SHIFT > COMMUNITY_SPIKE > NONE. """ social = metrics.get('social_dominance_score', 0) dev = metrics.get('developer_activity_score', 0) sentiment = metrics.get('sentiment_score', 0) community = metrics.get('community_score_norm', 0) if social >= _ALERT_THRESHOLD: return 'VIRAL_GROWTH' if dev >= _ALERT_THRESHOLD: return 'DEVELOPER_SURGE' if sentiment >= _ALERT_THRESHOLD: return 'SENTIMENT_SHIFT' if community >= _ALERT_THRESHOLD: return 'COMMUNITY_SPIKE' return 'NONE' # ────────────────────────────────────────────────────────────────── # 8. Signal label # ────────────────────────────────────────────────────────────────── def _get_signal(self, score: float) -> str: """Map momentum score to signal label.""" if score >= _SIGNAL_HOT: return 'HOT' if score >= _SIGNAL_WARMING: return 'WARMING' if score >= _SIGNAL_NEUTRAL: return 'NEUTRAL' if score >= _SIGNAL_COOLING: return 'COOLING' return 'COLD' # ────────────────────────────────────────────────────────────────── # 9. News sentiment label # ────────────────────────────────────────────────────────────────── def _parse_news_sentiment(self, news_data: Any) -> str: """Extract overall sentiment label from news data. Returns POSITIVE, NEGATIVE, or NEUTRAL. """ if not news_data: return 'NEUTRAL' score = self._score_news_sentiment(news_data) if score >= 60.0: return 'POSITIVE' if score <= 40.0: return 'NEGATIVE' return 'NEUTRAL' # ────────────────────────────────────────────────────────────────── # 10. Human-readable reasons # ────────────────────────────────────────────────────────────────── def _build_reasons( self, metrics: dict, signal: str, alert_type: str, ) -> list[str]: """Generate human-readable explanation strings for the signal.""" reasons: list[str] = [] social = metrics.get('social_dominance_score', 0) dev = metrics.get('developer_activity_score', 0) sentiment = metrics.get('sentiment_score', 0) community = metrics.get('community_score_norm', 0) momentum = metrics.get('momentum_score', 0) twitter = metrics.get('twitter_followers', 0) reddit = metrics.get('reddit_subscribers', 0) telegram = metrics.get('telegram_users', 0) github_stars = metrics.get('github_stars', 0) github_forks = metrics.get('github_forks', 0) # ── Social dominance ───────────────────────────────────────── if social >= 80: reasons.append( f'Very high social dominance ({social:.0f}/100) — ' f'strong community presence across platforms' ) elif social >= 60: reasons.append( f'Above-average social presence ({social:.0f}/100)' ) elif social > 0 and social < 20: reasons.append( f'Low social presence ({social:.0f}/100) — ' f'limited community visibility' ) # ── Social platform highlights ──────────────────────────────── if twitter >= 100_000: reasons.append( f'Large Twitter following ({twitter:,} followers)' ) if reddit >= 50_000: reasons.append( f'Active Reddit community ({reddit:,} subscribers)' ) if telegram >= 20_000: reasons.append( f'Strong Telegram presence ({telegram:,} users)' ) # ── Developer activity ──────────────────────────────────────── if dev >= 80: reasons.append( f'Very active development ({dev:.0f}/100) — ' f'{github_stars:,} stars, {github_forks:,} forks' ) elif dev >= 60: reasons.append( f'Active development ({dev:.0f}/100)' ) elif dev == 0: reasons.append('No developer activity data available') # ── Sentiment ───────────────────────────────────────────────── if sentiment >= 75: reasons.append( f'Very positive sentiment ({sentiment:.0f}/100)' ) elif sentiment >= 60: reasons.append( f'Positive sentiment ({sentiment:.0f}/100)' ) elif sentiment <= 25: reasons.append( f'Negative sentiment ({sentiment:.0f}/100) — proceed with caution' ) # ── Community score ─────────────────────────────────────────── cg_community = metrics.get('cg_community', 0) if cg_community >= 70: reasons.append( f'High CoinGecko community score ({cg_community:.0f})' ) # ── Alert-specific reasons ──────────────────────────────────── if alert_type == 'VIRAL_GROWTH': reasons.append( 'ALERT: Social metrics indicate potential viral growth phase' ) elif alert_type == 'DEVELOPER_SURGE': reasons.append( 'ALERT: Unusual spike in developer activity detected' ) elif alert_type == 'SENTIMENT_SHIFT': reasons.append( 'ALERT: Significant positive sentiment shift detected' ) elif alert_type == 'COMMUNITY_SPIKE': reasons.append( 'ALERT: Community engagement spike detected' ) # ── Overall signal summary ──────────────────────────────────── if signal == 'HOT': reasons.append( f'Overall momentum is HOT ({momentum:.0f}/100) — ' f'strong social activity across multiple metrics' ) elif signal == 'COLD': reasons.append( f'Overall momentum is COLD ({momentum:.0f}/100) — ' f'minimal social activity detected' ) # If no reasons were generated, add a generic one if not reasons: reasons.append( f'Momentum score: {momentum:.0f}/100 ({signal})' ) return reasons # ────────────────────────────────────────────────────────────────── # 11. Empty result helper # ────────────────────────────────────────────────────────────────── def _empty_result(self) -> dict: """Return empty result structure.""" return { 'signals': [], 'summary': { 'total_analyzed': 0, 'hot_count': 0, 'viral_alerts': 0, 'avg_sentiment': 0.0, }, 'timestamp': datetime.utcnow().isoformat(), }