""" Overlay API endpoints - attention overlay scores for retail-investor interest signals. Route ordering is intentional: static paths (/bulk, /top-movers, /admin/*) must be registered BEFORE the parameterized /{symbol} routes to prevent FastAPI from treating those literal path segments as symbol values. """ import asyncio import logging from datetime import datetime, timedelta, timezone from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import Response from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, desc, and_, func from app.core.config import settings from app.core.database import get_db from app.schemas.overlay import ( OverlayScoreResponse, BulkOverlayResponse, TopMoversResponse, OverlayTopMover, HeadlinesResponse, HeadlineItem, YouTubeResponse, VideoItem, WikiResponse, WikiPageviewPoint, CrowdingResponse, TrendsResponse, TrendPoint, OverlayHistoryResponse, OverlayHistoryPoint, AdminHealthResponse, SourceHealthItem, TriggerPipelineResponse, JobLogResponse, JobLogEntry, OverlayFeatures, OverlaySourcePresence, OverlaySourceDetails, YahooSourceDetail, YouTubeSourceDetail, WikiSourceDetail, FinraSourceDetail, OverlayMetadata, ) from app.models.overlay_feature import OverlayFeatureRecord, OverlayJobLog from app.models.overlay_raw_event import ( OverlayHeadlineEvent, OverlayVideoEvent, OverlayWikiPageview, OverlayTrendObservation, ) from app.models.overlay_registry import ThemeTopicMap from app.services.overlay.overlay_pipeline import OverlayPipeline from app.utils.cache import with_cache logger = logging.getLogger(__name__) router = APIRouter() def _utc(dt) -> datetime: """Ensure datetime is UTC-aware (SQLite returns naive datetimes).""" if dt is None: return dt if isinstance(dt, datetime) and dt.tzinfo is None: return dt.replace(tzinfo=timezone.utc) return dt # Shared pipeline instance (stateless — safe to share) _pipeline = OverlayPipeline() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _overlay_enabled() -> bool: return getattr(settings, "OVERLAY_ENABLED", True) def _record_to_response(record: OverlayFeatureRecord) -> OverlayScoreResponse: """Convert an OverlayFeatureRecord ORM object to the API response schema.""" mask = record.source_presence_mask or {} source_presence = OverlaySourcePresence( yahoo=mask.get("yahoo", False), youtube=mask.get("youtube", False), wikimedia=mask.get("wikimedia", False), google_trends=mask.get("google_trends", False), finra=mask.get("finra", False), ) features = OverlayFeatures( headline_burst_z=record.headline_burst_z, youtube_influence_z=record.youtube_influence_z, wiki_attention_z=record.wiki_attention_z, theme_heat_z=record.theme_heat_z, crowding_stress_z=record.crowding_stress_z, ) yahoo_detail = ( YahooSourceDetail( headline_count_6h=record.headline_count_6h or 0, headline_count_24h=record.headline_count_24h or 0, publisher_breadth_24h=record.publisher_breadth_24h or 0, ) if source_presence.yahoo else None ) yt_detail = ( YouTubeSourceDetail( mentions_24h=record.youtube_mentions_24h or 0, weighted_views_24h=record.youtube_weighted_views_24h or 0.0, ) if source_presence.youtube else None ) wiki_detail = ( WikiSourceDetail( page_views_1d=record.wiki_views_1d, page_views_7d_avg=record.wiki_views_7d_avg, ) if source_presence.wikimedia else None ) finra_detail = ( FinraSourceDetail( short_volume_ratio=record.short_volume_ratio, short_volume_spike_zscore=record.short_volume_spike_zscore, ) if source_presence.finra else None ) source_details = OverlaySourceDetails( yahoo=yahoo_detail, youtube=yt_detail, wikimedia=wiki_detail, finra=finra_detail, ) next_update = ( record.as_of_ts + timedelta(hours=24) if record.as_of_ts else None ) return OverlayScoreResponse( symbol=record.symbol, as_of_ts=record.as_of_ts, overlay_score=record.overlay_score, overlay_confidence=record.overlay_confidence, overlay_band=record.overlay_band, hold_extension_hint=record.hold_extension_hint, add_on_eligibility=record.add_on_eligibility, features=features, source_presence=source_presence, source_details=source_details, metadata=OverlayMetadata( feature_version=record.feature_version or "v1", data_freshness=record.as_of_ts, next_update_expected=next_update, ), ) # =========================================================================== # Static routes (MUST come before /{symbol} to avoid path shadowing) # =========================================================================== @router.get( "/bulk", response_model=BulkOverlayResponse, summary="Bulk overlay scores", description="Comma-separated symbols (max 50). Returns overlay scores for each.", ) @with_cache(namespace="overlay:bulk", ttl=1800, key_params=["symbols"]) async def get_bulk_overlay( symbols: str, response: Response, force_refresh: bool = Query(False), db: AsyncSession = Depends(get_db), ): if not _overlay_enabled(): raise HTTPException(status_code=503, detail="Overlay feature is disabled") sym_list = [s.strip().upper() for s in symbols.split(",") if s.strip()] if len(sym_list) > 50: raise HTTPException(status_code=400, detail="Max 50 symbols per request") if not sym_list: raise HTTPException(status_code=400, detail="No valid symbols provided") results = [] for sym in sym_list: record = await _pipeline.get_or_build(db, sym) if record: results.append(_record_to_response(record)) return BulkOverlayResponse( results=results, total_count=len(results), metadata={"requested": len(sym_list), "returned": len(results)}, ) @router.get( "/top-movers", response_model=TopMoversResponse, summary="Top overlay movers", description="Symbols with highest overlay scores in the last 24 hours.", ) @with_cache(namespace="overlay:top-movers", ttl=900, key_params=["limit"]) async def get_top_movers( response: Response, limit: int = Query(20, ge=1, le=100), force_refresh: bool = Query(False), db: AsyncSession = Depends(get_db), ): if not _overlay_enabled(): raise HTTPException(status_code=503, detail="Overlay feature is disabled") cutoff = datetime.now(timezone.utc) - timedelta(hours=24) # Get the latest record per symbol, then rank by score latest_per_symbol = ( select( OverlayFeatureRecord.symbol, func.max(OverlayFeatureRecord.as_of_ts).label("max_ts"), ) .where(OverlayFeatureRecord.as_of_ts >= cutoff) .group_by(OverlayFeatureRecord.symbol) .subquery() ) result = await db.execute( select(OverlayFeatureRecord) .join( latest_per_symbol, and_( OverlayFeatureRecord.symbol == latest_per_symbol.c.symbol, OverlayFeatureRecord.as_of_ts == latest_per_symbol.c.max_ts, ), ) .order_by(desc(OverlayFeatureRecord.overlay_score)) .limit(limit) ) records = result.scalars().all() movers = [ OverlayTopMover( symbol=r.symbol, overlay_score=r.overlay_score, overlay_band=r.overlay_band, as_of_ts=r.as_of_ts, ) for r in records ] return TopMoversResponse( top_movers=movers, total_count=len(movers), metadata={"as_of": datetime.now(timezone.utc).isoformat()}, ) # --------------------------------------------------------------------------- # Admin routes (static, before /{symbol}) # --------------------------------------------------------------------------- @router.get( "/admin/health", response_model=AdminHealthResponse, summary="Overlay system health", tags=["overlay-admin"], ) async def admin_health(db: AsyncSession = Depends(get_db)): overlay_enabled = _overlay_enabled() # Last pipeline run result = await db.execute( select(OverlayJobLog).order_by(desc(OverlayJobLog.started_at)).limit(1) ) last_job = result.scalars().first() # Per-job-type status (collect_all runs all sources; feature_build computes scores) sources = [] for source_name in ["collect_all", "feature_build"]: result_s = await db.execute( select(OverlayJobLog) .where(OverlayJobLog.job_type == source_name) .order_by(desc(OverlayJobLog.started_at)) .limit(1) ) job = result_s.scalars().first() sources.append( SourceHealthItem( source=source_name, last_collected_at=job.completed_at if job else None, status=job.status if job else "never_run", records_24h=job.records_processed if job else 0, ) ) return AdminHealthResponse( overlay_enabled=overlay_enabled, sources=sources, last_pipeline_run=last_job.started_at if last_job else None, metadata={"as_of": datetime.now(timezone.utc).isoformat()}, ) @router.post( "/admin/trigger-pipeline", response_model=TriggerPipelineResponse, summary="Trigger overlay pipeline manually", tags=["overlay-admin"], ) async def trigger_pipeline(): from app.core.database import AsyncSessionLocal async def _run(): async with AsyncSessionLocal() as db: await _pipeline.run_full_pipeline(db) asyncio.create_task(_run()) return TriggerPipelineResponse( status="triggered", message="Overlay pipeline started in background (seeds topic maps, collects data, builds features)", job_ids=[], ) @router.post( "/admin/seed-topics", summary="Seed ThemeTopicMap with default topic mappings", tags=["overlay-admin"], ) async def seed_topics(db: AsyncSession = Depends(get_db)): """Create default ThemeTopicMap entries for all TOP_50_SYMBOLS (safe to re-run; skips existing).""" inserted = await _pipeline.seed_topic_maps(db) return {"status": "ok", "inserted": inserted, "message": f"Seeded {inserted} new topic mappings"} @router.get( "/admin/job-log", response_model=JobLogResponse, summary="Overlay job log", tags=["overlay-admin"], ) async def get_job_log( limit: int = Query(50, ge=1, le=500), db: AsyncSession = Depends(get_db), ): result = await db.execute( select(OverlayJobLog) .order_by(desc(OverlayJobLog.started_at)) .limit(limit) ) logs = result.scalars().all() entries = [ JobLogEntry( id=str(log.id), job_type=log.job_type, status=log.status, started_at=log.started_at, completed_at=log.completed_at, records_processed=log.records_processed or 0, error_message=log.error_message, ) for log in logs ] return JobLogResponse( logs=entries, total_count=len(entries), metadata={}, ) # =========================================================================== # Parameterized routes (/{symbol} and sub-paths) # =========================================================================== @router.get( "/{symbol}", response_model=OverlayScoreResponse, summary="Overlay score for a symbol", description="Returns attention overlay score, z-scored features, and source details.", ) @with_cache(namespace="overlay:score", ttl=1800, key_params=["symbol"]) async def get_overlay_score( symbol: str, response: Response, force_refresh: bool = Query(False, description="Bypass cache and trigger on-demand rebuild"), db: AsyncSession = Depends(get_db), ): if not _overlay_enabled(): raise HTTPException(status_code=503, detail="Overlay feature is disabled") symbol = symbol.upper() record = await _pipeline.get_or_build(db, symbol) if record is None: raise HTTPException( status_code=404, detail=f"No overlay data available for {symbol}. Data collection may not have run yet.", ) return _record_to_response(record) @router.get( "/{symbol}/headlines", response_model=HeadlinesResponse, summary="Recent headlines for a symbol", ) @with_cache(namespace="overlay:headlines", ttl=600, key_params=["symbol", "hours"]) async def get_headlines( symbol: str, response: Response, hours: int = Query(24, ge=1, le=168), force_refresh: bool = Query(False), db: AsyncSession = Depends(get_db), ): symbol = symbol.upper() cutoff = datetime.now(timezone.utc) - timedelta(hours=hours) cutoff_6h = datetime.now(timezone.utc) - timedelta(hours=6) result = await db.execute( select(OverlayHeadlineEvent) .where(OverlayHeadlineEvent.published_at >= cutoff) .order_by(desc(OverlayHeadlineEvent.published_at)) .limit(500) ) all_events = result.scalars().all() sym_events = [e for e in all_events if symbol in (e.matched_symbols or [])] headlines = [ HeadlineItem( title=e.title, publisher=e.publisher, published_at=e.published_at, article_guid=e.article_guid, ) for e in sym_events ] publishers = {e.publisher for e in sym_events if e.publisher} count_6h = sum(1 for e in sym_events if _utc(e.published_at) >= cutoff_6h) return HeadlinesResponse( symbol=symbol, headlines=headlines, headline_count_6h=count_6h, headline_count_24h=len(sym_events), publisher_breadth_24h=len(publishers), metadata={"hours_requested": hours}, ) @router.get( "/{symbol}/youtube", response_model=YouTubeResponse, summary="YouTube mentions for a symbol", ) @with_cache(namespace="overlay:youtube", ttl=1200, key_params=["symbol"]) async def get_youtube( symbol: str, response: Response, force_refresh: bool = Query(False), db: AsyncSession = Depends(get_db), ): symbol = symbol.upper() cutoff_48h = datetime.now(timezone.utc) - timedelta(hours=48) cutoff_24h = datetime.now(timezone.utc) - timedelta(hours=24) result = await db.execute( select(OverlayVideoEvent) .where(OverlayVideoEvent.published_at >= cutoff_48h) .order_by(desc(OverlayVideoEvent.published_at)) .limit(200) ) all_events = result.scalars().all() sym_events = [e for e in all_events if symbol in (e.matched_symbols or [])] events_24h = [e for e in sym_events if _utc(e.published_at) >= cutoff_24h] videos = [ VideoItem( video_id=e.video_id, channel_id=e.channel_id, title=e.title, view_count=e.view_count, comment_count=e.comment_count, published_at=e.published_at, channel_weight=e.channel_weight, ) for e in sym_events ] weighted_views = sum((e.view_count or 0) * (e.channel_weight or 0.5) for e in events_24h) return YouTubeResponse( symbol=symbol, videos=videos, mentions_24h=len(events_24h), weighted_views_24h=round(weighted_views, 2), metadata={}, ) @router.get( "/{symbol}/wiki", response_model=WikiResponse, summary="Wikipedia pageview time series for a symbol", ) @with_cache(namespace="overlay:wiki", ttl=3600, key_params=["symbol", "days"]) async def get_wiki( symbol: str, response: Response, days: int = Query(30, ge=1, le=90), force_refresh: bool = Query(False), db: AsyncSession = Depends(get_db), ): symbol = symbol.upper() cutoff = datetime.now(timezone.utc) - timedelta(days=days) result = await db.execute( select(OverlayWikiPageview) .where( and_( OverlayWikiPageview.mapped_symbol == symbol, OverlayWikiPageview.date >= cutoff, ) ) .order_by(OverlayWikiPageview.date) ) rows = result.scalars().all() pageviews = [ WikiPageviewPoint(date=r.date, views=r.views, page_title=r.page_title) for r in rows ] views_1d = rows[-1].views if rows else None recent_7 = rows[-7:] if len(rows) >= 7 else rows views_7d_avg = sum(r.views for r in recent_7) / len(recent_7) if recent_7 else None return WikiResponse( symbol=symbol, pageviews=pageviews, views_1d=views_1d, views_7d_avg=round(views_7d_avg, 2) if views_7d_avg else None, metadata={"days_requested": days}, ) @router.get( "/{symbol}/crowding", response_model=CrowdingResponse, summary="FINRA crowding metrics for a symbol", ) @with_cache(namespace="overlay:crowding", ttl=3600, key_params=["symbol"]) async def get_crowding( symbol: str, response: Response, force_refresh: bool = Query(False), db: AsyncSession = Depends(get_db), ): symbol = symbol.upper() from app.services.overlay.finra_overlay_loader import FinraOverlayLoader loader = FinraOverlayLoader() metrics = await loader.get_crowding_metrics(db, symbol) return CrowdingResponse( symbol=symbol, short_volume_ratio=metrics.get("short_volume_ratio"), short_volume_spike_zscore=metrics.get("short_volume_spike_zscore"), crowding_stress_z=metrics.get("crowding_stress_z"), metadata={}, ) @router.get( "/{symbol}/trends", response_model=TrendsResponse, summary="Google Trends data for a symbol", ) @with_cache(namespace="overlay:trends", ttl=7200, key_params=["symbol"]) async def get_trends( symbol: str, response: Response, force_refresh: bool = Query(False), db: AsyncSession = Depends(get_db), ): symbol = symbol.upper() cutoff = datetime.now(timezone.utc) - timedelta(days=30) # Resolve topic IDs for this symbol result_topics = await db.execute( select(ThemeTopicMap).where(ThemeTopicMap.active == True) ) all_topics = result_topics.scalars().all() relevant_topics = { t.topic_id: t.topic_label for t in all_topics if symbol in (t.mapped_symbols or []) } if not relevant_topics: return TrendsResponse( symbol=symbol, trends=[], theme_heat_z=None, metadata={"note": "No topic mappings found for this symbol"}, ) result = await db.execute( select(OverlayTrendObservation) .where( and_( OverlayTrendObservation.topic_id.in_(list(relevant_topics.keys())), OverlayTrendObservation.observed_at >= cutoff, ) ) .order_by(OverlayTrendObservation.observed_at) ) rows = result.scalars().all() trends = [ TrendPoint( observed_at=r.observed_at, interest_value=r.interest_value, topic_id=r.topic_id, topic_label=relevant_topics.get(r.topic_id), ) for r in rows ] # Pull theme_heat_z from the latest feature record latest = await _pipeline.get_latest_feature(db, symbol) theme_heat_z = latest.theme_heat_z if latest else None return TrendsResponse( symbol=symbol, trends=trends, theme_heat_z=theme_heat_z, metadata={"topics": list(relevant_topics.keys())}, ) @router.get( "/{symbol}/history", response_model=OverlayHistoryResponse, summary="Overlay score history for a symbol", description="Time-series of overlay scores (useful for backtesting).", ) async def get_history( symbol: str, days: int = Query(30, ge=1, le=365), db: AsyncSession = Depends(get_db), ): symbol = symbol.upper() cutoff = datetime.now(timezone.utc) - timedelta(days=days) result = await db.execute( select(OverlayFeatureRecord) .where( and_( OverlayFeatureRecord.symbol == symbol, OverlayFeatureRecord.as_of_ts >= cutoff, ) ) .order_by(OverlayFeatureRecord.as_of_ts) ) records = result.scalars().all() history = [ OverlayHistoryPoint( as_of_ts=r.as_of_ts, overlay_score=r.overlay_score, overlay_confidence=r.overlay_confidence, overlay_band=r.overlay_band, ) for r in records ] return OverlayHistoryResponse( symbol=symbol, history=history, metadata={"days_requested": days, "data_points": len(history)}, )