From 29fc870b379253ee035335deb8a14c52bd5b537d Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Fri, 13 Mar 2026 21:03:06 -0700 Subject: [PATCH] Add Phase 5 Attention Overlay API and fix Redis port mismatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 12 REST endpoints: overlay score, bulk, top-movers, headlines, youtube, wiki, crowding, trends, history, admin (health/trigger/job-log) - 10 services: entity_resolver, yahoo_rss_adapter, wikimedia_adapter, youtube_adapter, google_trends_adapter, finra_overlay_loader, feature_builder, overlay_scorer, overlay_pipeline, scheduler - 10 DB tables across overlay_registry, overlay_raw_event, overlay_feature models - APScheduler: collect @ 23:30 UTC + feature build @ 01:30 UTC weekdays - Fix Redis port mismatch: config default 16379 → 16380 to match docker-compose external port - 64 overlay tests covering cache utils, Redis config, all 12 endpoints, route ordering Co-Authored-By: Claude Sonnet 4.6 --- API_DOCUMENTATION.md | 144 ++++ CHANGELOG.md | 36 + README.md | 135 +++- alembic/env.py | 1 + app/api/v1/api.py | 5 +- app/api/v1/endpoints/overlay.py | 665 ++++++++++++++++++ app/core/config.py | 10 +- app/core/overlay_config.py | 89 +++ app/main.py | 94 ++- app/models/__init__.py | 14 + app/models/overlay_feature.py | 84 +++ app/models/overlay_raw_event.py | 89 +++ app/models/overlay_registry.py | 76 ++ app/schemas/overlay.py | 210 ++++++ app/services/overlay/__init__.py | 3 + app/services/overlay/entity_resolver.py | 87 +++ app/services/overlay/feature_builder.py | 244 +++++++ app/services/overlay/finra_overlay_loader.py | 85 +++ app/services/overlay/google_trends_adapter.py | 96 +++ app/services/overlay/overlay_pipeline.py | 230 ++++++ app/services/overlay/overlay_scorer.py | 97 +++ app/services/overlay/scheduler.py | 100 +++ app/services/overlay/wikimedia_adapter.py | 107 +++ app/services/overlay/yahoo_rss_adapter.py | 147 ++++ app/services/overlay/youtube_adapter.py | 168 +++++ docs/PYTHON_CLIENT.md | 64 ++ requirements-api.txt | 5 + tests/test_overlay.py | 396 +++++++++++ 28 files changed, 3473 insertions(+), 8 deletions(-) create mode 100644 app/api/v1/endpoints/overlay.py create mode 100644 app/core/overlay_config.py create mode 100644 app/models/overlay_feature.py create mode 100644 app/models/overlay_raw_event.py create mode 100644 app/models/overlay_registry.py create mode 100644 app/schemas/overlay.py create mode 100644 app/services/overlay/__init__.py create mode 100644 app/services/overlay/entity_resolver.py create mode 100644 app/services/overlay/feature_builder.py create mode 100644 app/services/overlay/finra_overlay_loader.py create mode 100644 app/services/overlay/google_trends_adapter.py create mode 100644 app/services/overlay/overlay_pipeline.py create mode 100644 app/services/overlay/overlay_scorer.py create mode 100644 app/services/overlay/scheduler.py create mode 100644 app/services/overlay/wikimedia_adapter.py create mode 100644 app/services/overlay/yahoo_rss_adapter.py create mode 100644 app/services/overlay/youtube_adapter.py create mode 100644 tests/test_overlay.py diff --git a/API_DOCUMENTATION.md b/API_DOCUMENTATION.md index 077e3bd..0244aa4 100644 --- a/API_DOCUMENTATION.md +++ b/API_DOCUMENTATION.md @@ -26,6 +26,8 @@ Currently no authentication required. API key support coming soon. - **SEC EDGAR**: Official company filings (10-K, 10-Q, N-PORT) - **Yahoo Finance**: Real-time price data and news +- **Alpaca**: Market price data (OHLCV + VWAP) via official REST API (optional API key) +- **FINRA**: RegSHO short sale volume data (public CDN, no API key required) - **NewsAPI**: Professional news aggregation - **Reddit API**: Social sentiment analysis @@ -189,6 +191,148 @@ Bulk price data for multiple tickers. --- +### Alpaca Market Data 🆕 + +#### `GET /alpaca/status` +Check Alpaca API key validity and connection health. + +**Response:** +```json +{ + "configured": true, + "connected": true, + "bars_returned": 1, + "base_url": "https://data.alpaca.markets" +} +``` + +#### `GET /alpaca/bars/{ticker}` +Fetch raw bars directly from Alpaca without DB storage. + +**Parameters:** +- `interval`: Bar interval — `1m`, `5m`, `15m`, `1h`, `1d`, `1w`, `1mo` (default: `1d`) +- `start_date`: Start date (YYYY-MM-DD) +- `end_date`: End date (YYYY-MM-DD) +- `limit`: Max bars to return (1-10000, default: 1000) + +**Response:** +```json +{ + "ticker": "AAPL", + "interval": "1d", + "count": 5, + "bars": [ + { + "timestamp": "2025-01-02T05:00:00Z", + "open": 248.93, + "high": 249.1, + "low": 241.82, + "close": 243.85, + "volume": 55740731, + "vwap": 244.339399, + "trade_count": 685448 + } + ] +} +``` + +#### `GET /alpaca/data/{ticker}` +Fetch OHLCV bars from Alpaca, store in DB, and return in PriceDataResponse format. + +**Parameters:** +- `interval`: Bar interval (default: `1d`) +- `start_date`: Start date (required, YYYY-MM-DD) +- `end_date`: End date (required, YYYY-MM-DD) +- `force_refresh`: Re-fetch even if data exists (default: false) + +**Response:** Same `PriceDataResponse` format as `/price/data/{ticker}` with `data_source: "ALPACA"`. + +#### `GET /alpaca/intraday/{ticker}` +Fetch intraday candles from Alpaca (not stored in DB). + +**Parameters:** +- `interval`: `1m`, `5m`, `15m`, `1h` (default: `1m`) +- `start_date`, `end_date`: Date range +- `limit`: Max candles (default: 1000) + +--- + +### FINRA Short Volume Data 🆕 + +#### `GET /finra/short-volume/{symbol}` +Query FINRA RegSHO short sale volume for a symbol. Auto-ingests missing data. + +**Parameters:** +- `days`: Number of days to look back (1-365, default: 30) +- `limit`: Max entries to return (1-1000, default: 100) + +**Response:** +```json +{ + "symbol": "AAPL", + "entries": [ + { + "date": "2025-03-11", + "symbol": "AAPL", + "short_volume": 3214614.5, + "short_exempt_volume": 29855.0, + "total_volume": 7471743.3, + "market": "B,Q,N", + "short_ratio": 0.430236 + } + ], + "total_count": 21, + "metadata": { + "days_requested": 30, + "start_date": "2025-02-10", + "end_date": "2025-03-12" + } +} +``` + +#### `GET /finra/short-ratio/{symbol}` +Return daily short_ratio (aggregated across markets) for the last N days. + +**Parameters:** +- `days`: Number of days (1-365, default: 60) + +**Response:** +```json +{ + "symbol": "AAPL", + "history": [ + { + "date": "2025-03-09", + "short_volume": 4040406.5, + "short_exempt_volume": 52868.5, + "total_volume": 11925537.0, + "short_ratio": 0.338803 + } + ], + "avg_short_ratio": 0.400623, + "metadata": { "days_requested": 5, "data_points": 3 } +} +``` + +#### `POST /finra/admin/ingest` +Download and ingest FINRA short volume file(s) for a specific date or date range. + +**Parameters:** +- `date`: Single date (YYYY-MM-DD) +- `start_date` + `end_date`: Date range +- `force`: Re-ingest even if data exists (default: false) + +**Response:** +```json +{ + "date": "2025-03-10", + "records_ingested": 10433, + "status": "completed" +} +``` + +--- + ### Stock Market Data 🆕 #### `GET /stocks/most-active` diff --git a/CHANGELOG.md b/CHANGELOG.md index 61c09f4..a066bcf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,42 @@ All notable changes to Stock Oracle API will be documented in this file. +## [3.0.0] - 2026-03-12 + +### Added +- **Alpaca Market Data Adapter**: Complete integration with Alpaca Market Data API v2 + - `GET /alpaca/status` — Connection status and API key validation + - `GET /alpaca/bars/{ticker}` — Raw bars from Alpaca (no DB) + - `GET /alpaca/data/{ticker}` — OHLCV price data with DB storage (PriceDataResponse format) + - `GET /alpaca/intraday/{ticker}` — Intraday candles (1m, 5m, 15m, 1h) + - Built-in rate limiting (200 req/min token bucket) with retry + exponential backoff + - VWAP (volume-weighted average price) and trade count per bar + - Auto-pagination for large date ranges + - Completely independent from Yahoo Finance — users choose their source + +- **FINRA Short Sale Volume Adapter**: RegSHO daily short volume data + - `GET /finra/short-volume/{symbol}` — Short volume data with auto-ingest + - `GET /finra/short-ratio/{symbol}` — Aggregated short ratio history + - `POST /finra/admin/ingest` — Manual ingest for single date or date range + - Public FINRA CDN data (no API key required) + - Pipe-delimited text parser for CNMS short volume files + - Multi-market data (NYSE TRF, NASDAQ TRF, NYSE) + - Auto-ingest on first query if data is missing + +- **New DB Model**: `FinraShortVolume` with `(symbol, date, market)` unique constraint +- **New DataSource Enum**: `DataSource.ALPACA` added to schema +- **Configuration**: `ALPACA_API_KEY`, `ALPACA_SECRET_KEY`, `ALPACA_BASE_URL` settings +- **Docker**: Alpaca env vars passed to API container + +### New Files +- `app/services/alpaca_client.py` — Alpaca HTTP client with httpx +- `app/services/alpaca_price_service.py` — Alpaca bars → PriceData conversion +- `app/api/v1/endpoints/alpaca.py` — Alpaca endpoints +- `app/models/finra_short_volume.py` — FINRA SQLAlchemy model +- `app/services/finra_short_volume_service.py` — FINRA fetch/parse/ingest service +- `app/schemas/finra.py` — FINRA Pydantic schemas +- `app/api/v1/endpoints/finra.py` — FINRA endpoints + ## [2.1.0] - 2025-08-10 ### Added diff --git a/README.md b/README.md index 95fd127..fcffe50 100644 --- a/README.md +++ b/README.md @@ -33,17 +33,33 @@ Stock Oracle is a comprehensive investment analysis API that leverages SEC EDGAR - **Error Handling**: Detailed error categorization and reporting - **Migration Support**: Database transfer capabilities +### Alpaca Market Data (NEW! 🆕) +- **Official REST API**: Alpaca Market Data v2 with proper authentication +- **OHLCV + VWAP**: Daily/intraday bars with volume-weighted average price +- **Independent Endpoints**: Separate from Yahoo Finance — users choose their source +- **DB Storage**: Alpaca bars stored with `data_source = "ALPACA"` for distinction +- **Rate Limiting**: Built-in token bucket (200 req/min) with retry + backoff + +### FINRA Short Volume Data (NEW! 🆕) +- **RegSHO Short Sale Volume**: Daily short volume from FINRA public CDN +- **No API Key Required**: Public data, free access +- **Auto-Ingest**: Automatically downloads missing data on first query +- **Short Ratio History**: Aggregated short_ratio trends over time +- **Bulk Ingest**: Date range ingest for backfilling historical data + ### Technical Stack - **Backend**: FastAPI, SQLAlchemy (async), Pydantic - **Frontend**: Next.js 15, React 18, TypeScript, TailwindCSS - **Database**: SQLite (development) / PostgreSQL (production) - **Caching**: Redis for performance optimization -- **Data Sources**: +- **Data Sources**: - SEC EDGAR via edgartools (ETF holdings, financial data) - Yahoo Finance via yfinance_plus (news & prices) - Yahoo Finance via intelligent curl_cffi scraping (market data) - Most Active Stocks (~170 stocks) - 52-Week Gainers (~1,350 stocks with rate limiting) + - Alpaca Market Data v2 (OHLCV bars, optional API key) + - FINRA RegSHO (short sale volume, public CDN) - NewsAPI (news articles) - Reddit API (social media sentiment) - **Deployment**: Docker with docker-compose @@ -366,6 +382,104 @@ curl "http://localhost:18001/api/v1/stocks/52-week-gainers?limit=1000&max_pages= - **Session Rotation**: Every 5 minutes - **Success Rate**: 99.5% even at scale +### Alpaca Market Data (NEW! 🆕) + +#### Check Connection +```bash +# Verify Alpaca API key validity +curl "http://localhost:18001/api/v1/alpaca/status" +``` + +#### Get Price Data (DB Storage) +```bash +# Daily bars with DB persistence (same PriceDataResponse format as /price) +curl "http://localhost:18001/api/v1/alpaca/data/AAPL?interval=1d&start_date=2025-01-01&end_date=2025-01-31" +``` + +#### Raw Bars (No DB) +```bash +# Fetch bars directly from Alpaca without DB storage +curl "http://localhost:18001/api/v1/alpaca/bars/AAPL?interval=1d&start_date=2025-01-02&end_date=2025-01-10" +``` + +#### Intraday Candles +```bash +# 1-hour candles from Alpaca +curl "http://localhost:18001/api/v1/alpaca/intraday/AAPL?interval=1h&start_date=2025-03-10&end_date=2025-03-11" + +# 5-minute candles +curl "http://localhost:18001/api/v1/alpaca/intraday/TSLA?interval=5m&start_date=2025-03-10&end_date=2025-03-10" +``` + +#### Key Features +- **Official REST API**: Proper authentication with API key/secret +- **VWAP Included**: Volume-weighted average price in every bar +- **Trade Count**: Number of trades per bar (`trade_count` field) +- **Rate Limited**: Built-in 200 req/min token bucket with auto-wait +- **Retry Logic**: 3x retry with exponential backoff on 429/5xx errors +- **Independent**: Completely separate from Yahoo Finance endpoints + +### FINRA Short Volume Data (NEW! 🆕) + +#### Get Short Volume +```bash +# Short volume for AAPL (last 30 days, auto-ingests missing data) +curl "http://localhost:18001/api/v1/finra/short-volume/AAPL?days=30" + +# With custom limit +curl "http://localhost:18001/api/v1/finra/short-volume/TSLA?days=60&limit=20" +``` + +#### Short Ratio History +```bash +# Aggregated short ratio over 60 days +curl "http://localhost:18001/api/v1/finra/short-ratio/AAPL?days=60" +``` + +#### Manual Ingest +```bash +# Ingest a single date +curl -X POST "http://localhost:18001/api/v1/finra/admin/ingest?date=2025-03-10" + +# Ingest a date range +curl -X POST "http://localhost:18001/api/v1/finra/admin/ingest?start_date=2025-03-01&end_date=2025-03-10" + +# Force re-ingest +curl -X POST "http://localhost:18001/api/v1/finra/admin/ingest?date=2025-03-10&force=true" +``` + +#### FINRA Response Format +```json +{ + "symbol": "AAPL", + "entries": [ + { + "date": "2025-03-11", + "symbol": "AAPL", + "short_volume": 3214614.5, + "short_exempt_volume": 29855.0, + "total_volume": 7471743.3, + "market": "B,Q,N", + "short_ratio": 0.430236 + } + ], + "total_count": 21, + "metadata": { + "days_requested": 30, + "start_date": "2025-02-10", + "end_date": "2025-03-12" + } +} +``` + +#### Key Features +- **No API Key Required**: Public FINRA CDN data +- **Auto-Ingest**: Missing data automatically downloaded on query +- **Multi-Market**: Data from B (NYSE TRF), Q (NASDAQ TRF), N (NYSE) markets +- **Short Ratio**: Pre-calculated `short_volume / total_volume` per record +- **Aggregated History**: `short-ratio` endpoint aggregates across all markets per day +- **Date Range Ingest**: Bulk backfill for historical data (weekdays only) + ### FRED Economic Data (NEW! 🏦) #### Universal FRED API Proxy (🚀 Recommended) @@ -620,6 +734,17 @@ curl "http://localhost:18001/api/v1/news/NVDA/social-only?days_back=3&max_social - `GET /api/v1/stocks/most-active` - Most actively traded stocks (optional limit parameter) - `GET /api/v1/stocks/52-week-gainers` - 52-week top gaining stocks with intelligent rate limiting +#### Alpaca Market Data (NEW! 🆕) +- `GET /api/v1/alpaca/status` - Alpaca connection status and API key validation +- `GET /api/v1/alpaca/bars/{ticker}` - Raw bars from Alpaca (no DB storage) +- `GET /api/v1/alpaca/data/{ticker}` - OHLCV price data via Alpaca (with DB storage) +- `GET /api/v1/alpaca/intraday/{ticker}` - Intraday candles from Alpaca + +#### FINRA Short Volume (NEW! 🆕) +- `GET /api/v1/finra/short-volume/{symbol}` - Short sale volume data (auto-ingests if missing) +- `GET /api/v1/finra/short-ratio/{symbol}` - Short ratio history (aggregated across markets) +- `POST /api/v1/finra/admin/ingest` - Manually ingest FINRA data for a date or range + #### FRED Economic Data (NEW! 🏦) - `GET /api/v1/fred/proxy/{endpoint:path}` - **Universal FRED API proxy with caching** (🚀 Recommended) - `GET /api/v1/fred/endpoints` - List all supported FRED API endpoints @@ -634,8 +759,8 @@ curl "http://localhost:18001/api/v1/news/NVDA/social-only?days_back=3&max_social - `POST /api/v1/financial/data` - Detailed financial data request - `POST /api/v1/financial/data/bulk` - Bulk financial data for multiple tickers -#### Price Data -- `GET /api/v1/price/data/{ticker}` - Simple price data with query parameters +#### Price Data (Yahoo Finance) +- `GET /api/v1/price/data/{ticker}` - Simple price data with query parameters - `POST /api/v1/price/data` - Detailed price data request - `POST /api/v1/price/data/bulk` - Bulk price data for multiple tickers @@ -737,6 +862,10 @@ DATABASE_URL=sqlite+aiosqlite:///./stock_oracle.db # SEC Data SEC_EMAIL=your@email.com # Required for SEC API access +# Alpaca Market Data (optional — leave blank to disable) +ALPACA_API_KEY=your_alpaca_api_key +ALPACA_SECRET_KEY=your_alpaca_secret_key + # Cache REDIS_URL=redis://localhost:16379/0 # Use redis://redis:6379/0 in Docker CACHE_TTL=3600 # Response cache TTL in seconds (default: 3600) diff --git a/alembic/env.py b/alembic/env.py index e24f909..2fb722c 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -13,6 +13,7 @@ from alembic import context from app.core.database import Base, engine from app.models import financial, error_log, request_log, fred_data, filing, finra_short_volume, alpaca_price from app.models import etf # ETF models +from app.models import overlay_registry, overlay_raw_event, overlay_feature # Phase 5 Overlay models # this is the Alembic Config object config = context.config diff --git a/app/api/v1/api.py b/app/api/v1/api.py index f1a64b3..a636112 100644 --- a/app/api/v1/api.py +++ b/app/api/v1/api.py @@ -3,7 +3,7 @@ API v1 router """ from fastapi import APIRouter -from app.api.v1.endpoints import financial, price, catalog, health, migration, database, error_logs, request_logs, news, etf, stocks, fred, filings, alpaca, finra +from app.api.v1.endpoints import financial, price, catalog, health, migration, database, error_logs, request_logs, news, etf, stocks, fred, filings, alpaca, finra, overlay api_router = APIRouter() @@ -23,4 +23,5 @@ api_router.include_router(database.router, prefix="/database", tags=["database"] api_router.include_router(error_logs.router, prefix="/admin/errors", tags=["error-logs"]) api_router.include_router(request_logs.router, prefix="/admin/requests", tags=["request-logs"]) api_router.include_router(alpaca.router, prefix="/alpaca", tags=["alpaca"]) -api_router.include_router(finra.router, prefix="/finra", tags=["finra"]) \ No newline at end of file +api_router.include_router(finra.router, prefix="/finra", tags=["finra"]) +api_router.include_router(overlay.router, prefix="/overlay", tags=["overlay"]) \ No newline at end of file diff --git a/app/api/v1/endpoints/overlay.py b/app/api/v1/endpoints/overlay.py new file mode 100644 index 0000000..433f206 --- /dev/null +++ b/app/api/v1/endpoints/overlay.py @@ -0,0 +1,665 @@ +""" +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_ + +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) + result = await db.execute( + select(OverlayFeatureRecord) + .where(OverlayFeatureRecord.as_of_ts >= cutoff) + .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-source status + sources = [] + for source_name in ["yahoo_rss", "wikimedia", "youtube", "google_trends", "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", + job_ids=[], + ) + + +@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 * e.channel_weight 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)}, + ) diff --git a/app/core/config.py b/app/core/config.py index cda15a1..a94f552 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -43,7 +43,7 @@ class Settings(BaseSettings): DATABASE_ECHO: bool = False # Redis - REDIS_URL: str = os.getenv("REDIS_URL", "redis://localhost:16379/0") + REDIS_URL: str = os.getenv("REDIS_URL", "redis://localhost:16380/0") CACHE_TTL: int = 3600 # 1 hour default # SEC Settings @@ -110,11 +110,17 @@ class Settings(BaseSettings): # Server API_PORT: int = int(os.getenv("API_PORT", "18000")) DB_PORT: int = int(os.getenv("DB_PORT", "15432")) - REDIS_PORT: int = int(os.getenv("REDIS_PORT", "16379")) + REDIS_PORT: int = int(os.getenv("REDIS_PORT", "16380")) # Migration MIGRATION_API_KEY: str = os.getenv("MIGRATION_API_KEY", "migration-key-change-in-production") ALLOW_MIGRATION: bool = os.getenv("ALLOW_MIGRATION", "True").lower() == "true" + + # Overlay (Phase 5) + YOUTUBE_API_KEY: str = os.getenv("YOUTUBE_API_KEY", "") + GOOGLE_TRENDS_ENABLED: bool = os.getenv("GOOGLE_TRENDS_ENABLED", "false").lower() == "true" + OVERLAY_ENABLED: bool = os.getenv("OVERLAY_ENABLED", "true").lower() == "true" + OVERLAY_STALE_HOURS: int = int(os.getenv("OVERLAY_STALE_HOURS", "8")) class Config: case_sensitive = True diff --git a/app/core/overlay_config.py b/app/core/overlay_config.py new file mode 100644 index 0000000..7095004 --- /dev/null +++ b/app/core/overlay_config.py @@ -0,0 +1,89 @@ +""" +Overlay feature configuration - source weights, thresholds, and feature flags. +""" + +from app.core.config import settings + +# --------------------------------------------------------------------------- +# Source weights (used for weighted-average scoring; normalized internally) +# --------------------------------------------------------------------------- +SOURCE_WEIGHTS = { + "yahoo": 0.30, + "youtube": 0.25, + "wikimedia": 0.20, + "finra": 0.15, + "google_trends": 0.10, +} + +# --------------------------------------------------------------------------- +# Overlay band thresholds (overlay_score 0~1) +# --------------------------------------------------------------------------- +BAND_THRESHOLDS = { + "frenzied": 0.80, + "loud": 0.60, + "supportive": 0.40, + "tepid": 0.20, + "silent": 0.0, +} + +# --------------------------------------------------------------------------- +# Confidence +# --------------------------------------------------------------------------- +# Each present source adds this much to overlay_confidence (max 1.0) +CONFIDENCE_PER_SOURCE = 0.20 + +# Min sources for non-trivial confidence +MIN_SOURCES_STRONG_CONFIDENCE = 3 +MIN_SOURCES_DEGRADED_MODE = 1 + +# --------------------------------------------------------------------------- +# z-score normalization +# --------------------------------------------------------------------------- +ZSCORE_WINDOW_DAYS = 30 + +# Winsorization clamps +WINSOR_LOWER = -3.0 +WINSOR_UPPER = 3.0 + +# --------------------------------------------------------------------------- +# Staleness +# --------------------------------------------------------------------------- +# Hours before a feature record is considered stale and on-demand rebuild triggers +FEATURE_STALE_HOURS: int = int(getattr(settings, "OVERLAY_STALE_HOURS", 8)) + +# On-demand pipeline timeout (seconds) - prevents blocking API requests +ONDEMAND_TIMEOUT_SECONDS = 25 + +# --------------------------------------------------------------------------- +# Batch scheduling (UTC) +# --------------------------------------------------------------------------- +SCHEDULE_RSS_UTC = "23:30" # 18:30 ET +SCHEDULE_WIKI_YOUTUBE_UTC = "01:00" # 20:00 ET (next UTC day) +SCHEDULE_FEATURE_BUILD_UTC = "01:30" +SCHEDULE_SCORING_UTC = "02:00" + +# --------------------------------------------------------------------------- +# Hint thresholds +# --------------------------------------------------------------------------- +HOLD_EXTENSION_EXTEND_THRESHOLD = 0.65 +HOLD_EXTENSION_TRIM_THRESHOLD = 0.30 +ADD_ON_ELIGIBILITY_THRESHOLD = 0.55 + +# --------------------------------------------------------------------------- +# Yahoo RSS feed URLs +# --------------------------------------------------------------------------- +YAHOO_RSS_FEEDS = [ + "https://finance.yahoo.com/rss/headline", +] + +# --------------------------------------------------------------------------- +# Seed symbols (Top 50 US equities) +# --------------------------------------------------------------------------- +TOP_50_SYMBOLS = [ + "AAPL", "MSFT", "NVDA", "AMZN", "GOOGL", "META", "TSLA", "BRK.B", + "JPM", "JNJ", "V", "UNH", "XOM", "PG", "MA", "HD", "CVX", "LLY", + "ABBV", "BAC", "KO", "PEP", "AVGO", "COST", "WMT", "MRK", "TMO", + "DIS", "ACN", "ABT", "VZ", "ADBE", "CRM", "NFLX", "CMCSA", "TXN", + "CSCO", "NKE", "NEE", "AMD", "DHR", "BMY", "QCOM", "T", "LOW", + "PM", "HON", "ORCL", "RTX", "UPS", +] diff --git a/app/main.py b/app/main.py index 19afc9d..265b942 100644 --- a/app/main.py +++ b/app/main.py @@ -13,6 +13,7 @@ from app.api.v1.api import api_router from app.core.database import engine, Base from app.middleware.error_logger import ErrorLoggingMiddleware from app.models import error_log, request_log, fred_data, filing, finra_short_volume, alpaca_price # Import to register models +from app.models import overlay_registry, overlay_raw_event, overlay_feature # Overlay models # Create database tables @asynccontextmanager @@ -20,8 +21,19 @@ async def lifespan(app: FastAPI): # Startup - ensure tables exist async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) + # Start overlay batch scheduler (graceful no-op if apscheduler not installed) + try: + from app.services.overlay.scheduler import start_scheduler + start_scheduler() + except Exception: + pass yield # Shutdown + try: + from app.services.overlay.scheduler import stop_scheduler + stop_scheduler() + except Exception: + pass await engine.dispose() # Create FastAPI app @@ -178,6 +190,22 @@ async def root_documentation():
  • POST /etf/admin/refresh-maps - Refresh CUSIP/CIK maps
    • +

    Attention Overlay NEW

    + +

    📊 Example Requests

    Financial Data

    @@ -253,8 +281,36 @@ news = client.get_news_social_data("AAPL", days_back=7)
  • ETF Holdings - Complete ETF portfolio analysis via N-PORT
  • Price Data - Historical OHLCV data from Yahoo Finance
  • Investment Metrics - P/E, ROE, debt ratios, growth metrics
    • +
  • Attention Overlay - Multi-source retail interest signal (news burst, Wikipedia views, YouTube mentions, FINRA crowding)
    • +

    Attention Overlay Response Example NEW

    + 
    GET /api/v1/overlay/AAPL
+{{
+  "symbol": "AAPL",
+  "as_of_ts": "2026-03-13T20:30:00Z",
+  "overlay_score": 0.72,          // 0~1 (higher = more retail attention)
+  "overlay_confidence": 0.80,     // fraction of sources with data
+  "overlay_band": "supportive",   // silent / tepid / supportive / loud / frenzied
+  "hold_extension_hint": "neutral", // extend / neutral / trim
+  "add_on_eligibility": false,
+  "features": {{
+    "headline_burst_z": 1.2,      // z-score vs 30-day window
+    "youtube_influence_z": 0.8,
+    "wiki_attention_z": null,
+    "theme_heat_z": null,
+    "crowding_stress_z": -0.3
+  }},
+  "source_presence": {{
+    "yahoo": true, "youtube": true,
+    "wikimedia": false, "google_trends": false, "finra": true
+  }},
+  "source_details": {{
+    "yahoo": {{"headline_count_6h": 5, "headline_count_24h": 12, "publisher_breadth_24h": 4}},
+    "finra": {{"short_volume_ratio": 0.42, "short_volume_spike_zscore": -0.3}}
+  }}
+}}
    +

    Alpaca Market Data NEW

    # Check Alpaca connection status
 curl "http://localhost:18001/api/v1/alpaca/status"
@@ -272,12 +328,40 @@ curl "http://localhost:18001/api/v1/finra/short-volume/AAPL?days=30"
 # Get short ratio history
 curl "http://localhost:18001/api/v1/finra/short-ratio/AAPL?days=60"
    +

    Attention Overlay NEW

    + 
    # Overlay score for a single symbol
+curl "http://localhost:18001/api/v1/overlay/AAPL"
+
+# Bulk scores (comma-separated, max 50)
+curl "http://localhost:18001/api/v1/overlay/bulk?symbols=AAPL,TSLA,NVDA,AMD,META"
+
+# Top movers by overlay score (last 24h)
+curl "http://localhost:18001/api/v1/overlay/top-movers?limit=10"
+
+# Recent headlines matched to AAPL
+curl "http://localhost:18001/api/v1/overlay/AAPL/headlines?hours=24"
+
+# Wikipedia pageview trend (last 30 days)
+curl "http://localhost:18001/api/v1/overlay/AAPL/wiki?days=30"
+
+# FINRA crowding metrics
+curl "http://localhost:18001/api/v1/overlay/AAPL/crowding"
+
+# Historical overlay scores (backtesting)
+curl "http://localhost:18001/api/v1/overlay/AAPL/history?days=90"
+
+# Manually trigger full pipeline
+curl -X POST "http://localhost:18001/api/v1/overlay/admin/trigger-pipeline"
    +

    🔧 Data Sources

    @@ -338,6 +422,14 @@ app.openapi_tags = [ { "name": "admin", "description": "Administrative endpoints (migration, etc.)" + }, + { + "name": "overlay", + "description": "Attention Overlay - retail investor interest, media diffusion, crowding signals" + }, + { + "name": "overlay-admin", + "description": "Overlay administrative endpoints (pipeline trigger, health, job log)" } ] diff --git a/app/models/__init__.py b/app/models/__init__.py index 9edd4ad..d9a0710 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -3,6 +3,9 @@ from app.models.etf import CusipMap, ETFCIKMap, ETFSeriesMap, ETFHoldingsSnapsho from app.models.filing import SECFiling from app.models.finra_short_volume import FinraShortVolume from app.models.alpaca_price import AlpacaPriceData +from app.models.overlay_registry import CompanyAlias, YouTubeChannelRegistry, WikiPageMap, ThemeTopicMap +from app.models.overlay_raw_event import OverlayHeadlineEvent, OverlayVideoEvent, OverlayWikiPageview, OverlayTrendObservation +from app.models.overlay_feature import OverlayFeatureRecord, OverlayJobLog __all__ = [ "Company", @@ -18,4 +21,15 @@ __all__ = [ "SECFiling", "FinraShortVolume", "AlpacaPriceData", + # Overlay + "CompanyAlias", + "YouTubeChannelRegistry", + "WikiPageMap", + "ThemeTopicMap", + "OverlayHeadlineEvent", + "OverlayVideoEvent", + "OverlayWikiPageview", + "OverlayTrendObservation", + "OverlayFeatureRecord", + "OverlayJobLog", ] \ No newline at end of file diff --git a/app/models/overlay_feature.py b/app/models/overlay_feature.py new file mode 100644 index 0000000..e70e9fe --- /dev/null +++ b/app/models/overlay_feature.py @@ -0,0 +1,84 @@ +""" +Overlay computed feature models - pre-computed scores served by the API +""" + +from sqlalchemy import Column, String, Float, Boolean, Index, UniqueConstraint, JSON, Integer, Text +from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP +from datetime import datetime, timezone +import uuid + +from app.core.database import Base + + +class OverlayFeatureRecord(Base): + __tablename__ = "overlay_feature_records" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + symbol = Column(String(10), nullable=False) + as_of_ts = Column(TIMESTAMP(timezone=True), nullable=False) + feature_version = Column(String(20), default="v1") + + # z-scores per source + headline_burst_z = Column(Float, nullable=True) + youtube_influence_z = Column(Float, nullable=True) + wiki_attention_z = Column(Float, nullable=True) + theme_heat_z = Column(Float, nullable=True) + crowding_stress_z = Column(Float, nullable=True) + + # headline detail + headline_count_6h = Column(Integer, default=0) + headline_count_24h = Column(Integer, default=0) + publisher_breadth_24h = Column(Integer, default=0) + + # youtube detail + youtube_mentions_24h = Column(Integer, default=0) + youtube_weighted_views_24h = Column(Float, default=0.0) + + # wiki detail + wiki_views_1d = Column(Integer, nullable=True) + wiki_views_7d_avg = Column(Float, nullable=True) + + # finra crowding detail + short_volume_ratio = Column(Float, nullable=True) + short_volume_spike_zscore = Column(Float, nullable=True) + + # final scores + overlay_score = Column(Float, nullable=False, default=0.0) + overlay_confidence = Column(Float, nullable=False, default=0.0) + overlay_band = Column(String(20), nullable=True) # silent/tepid/supportive/loud/frenzied + source_presence_mask = Column(JSON, default=dict) + + # hints + hold_extension_hint = Column(String(10), nullable=True) # extend/neutral/trim + add_on_eligibility = Column(Boolean, nullable=True) + + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + UniqueConstraint("symbol", "as_of_ts", "feature_version", name="uq_overlay_feature_record"), + Index("idx_overlay_feature_symbol", "symbol"), + Index("idx_overlay_feature_as_of_ts", "as_of_ts"), + Index("idx_overlay_feature_score", "overlay_score"), + ) + + +class OverlayJobLog(Base): + __tablename__ = "overlay_job_log" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + job_type = Column(String(50), nullable=False) # rss_collect / wiki_collect / yt_collect / feature_build / score + status = Column(String(20), nullable=False) # running / completed / failed / partial + started_at = Column(TIMESTAMP(timezone=True), nullable=False) + completed_at = Column(TIMESTAMP(timezone=True), nullable=True) + records_processed = Column(Integer, default=0) + error_message = Column(Text, nullable=True) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + Index("idx_overlay_job_log_type", "job_type"), + Index("idx_overlay_job_log_started_at", "started_at"), + ) diff --git a/app/models/overlay_raw_event.py b/app/models/overlay_raw_event.py new file mode 100644 index 0000000..95ed437 --- /dev/null +++ b/app/models/overlay_raw_event.py @@ -0,0 +1,89 @@ +""" +Overlay raw event models - time-series raw data from each source +""" + +from sqlalchemy import Column, String, Float, Integer, Index, UniqueConstraint, JSON, Text +from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP +from datetime import datetime, timezone +import uuid + +from app.core.database import Base + + +class OverlayHeadlineEvent(Base): + __tablename__ = "overlay_headline_events" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + article_guid = Column(String(500), unique=True, nullable=False) + title = Column(Text, nullable=False) + publisher = Column(String(255), nullable=True) + published_at = Column(TIMESTAMP(timezone=True), nullable=False) + matched_symbols = Column(JSON, default=list) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + Index("idx_headline_published_at", "published_at"), + ) + + +class OverlayVideoEvent(Base): + __tablename__ = "overlay_video_events" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + video_id = Column(String(100), unique=True, nullable=False) + channel_id = Column(String(100), nullable=False) + title = Column(Text, nullable=False) + view_count = Column(Integer, default=0) + comment_count = Column(Integer, default=0) + published_at = Column(TIMESTAMP(timezone=True), nullable=True) + matched_symbols = Column(JSON, default=list) + channel_weight = Column(Float, default=0.5) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + Index("idx_video_channel_id", "channel_id"), + Index("idx_video_published_at", "published_at"), + ) + + +class OverlayWikiPageview(Base): + __tablename__ = "overlay_wiki_pageviews" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + page_title = Column(String(500), nullable=False) + date = Column(TIMESTAMP(timezone=True), nullable=False) + project = Column(String(50), default="en.wikipedia") + views = Column(Integer, nullable=False) + mapped_symbol = Column(String(10), nullable=True, index=True) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + UniqueConstraint("page_title", "date", "project", name="uq_wiki_pageviews"), + Index("idx_wiki_pageviews_symbol", "mapped_symbol"), + Index("idx_wiki_pageviews_date", "date"), + ) + + +class OverlayTrendObservation(Base): + __tablename__ = "overlay_trend_observations" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + topic_id = Column(String(100), nullable=False) + observed_at = Column(TIMESTAMP(timezone=True), nullable=False) + geography = Column(String(10), default="US") + interest_value = Column(Integer, nullable=False) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + UniqueConstraint("topic_id", "observed_at", "geography", name="uq_trend_observations"), + Index("idx_trend_topic_id", "topic_id"), + Index("idx_trend_observed_at", "observed_at"), + ) diff --git a/app/models/overlay_registry.py b/app/models/overlay_registry.py new file mode 100644 index 0000000..eab432c --- /dev/null +++ b/app/models/overlay_registry.py @@ -0,0 +1,76 @@ +""" +Overlay Registry models - lookup tables for entity resolution +""" + +from sqlalchemy import Column, String, Float, Boolean, Index, UniqueConstraint, JSON +from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP +from datetime import datetime, timezone +import uuid + +from app.core.database import Base + + +class CompanyAlias(Base): + __tablename__ = "company_aliases" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + symbol = Column(String(10), nullable=False, index=True) + alias_type = Column(String(20), nullable=False) # canonical/short/alias/wiki + alias_value = Column(String(255), nullable=False) + confidence = Column(Float, default=1.0) + active = Column(Boolean, default=True) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + Index("idx_company_aliases_symbol", "symbol"), + Index("idx_company_aliases_value", "alias_value"), + ) + + +class YouTubeChannelRegistry(Base): + __tablename__ = "youtube_channel_registry" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + channel_id = Column(String(100), unique=True, nullable=False) + channel_title = Column(String(255), nullable=False) + category = Column(String(50), nullable=True) + channel_weight = Column(Float, default=0.5) # 0~1 + active = Column(Boolean, default=True) + watch_mode = Column(String(20), default="recent") # recent / all + symbol_focus_tags = Column(JSON, default=list) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + +class WikiPageMap(Base): + __tablename__ = "wiki_page_map" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + symbol = Column(String(10), nullable=False) + wiki_page_title = Column(String(500), nullable=False) + confidence = Column(Float, default=1.0) + active = Column(Boolean, default=True) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + UniqueConstraint("symbol", "wiki_page_title", name="uq_wiki_page_map"), + Index("idx_wiki_page_map_symbol", "symbol"), + ) + + +class ThemeTopicMap(Base): + __tablename__ = "theme_topic_map" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + topic_id = Column(String(100), nullable=False, unique=True) + topic_label = Column(String(255), nullable=False) + mapped_symbols = Column(JSON, default=list) # list of ticker strings + active = Column(Boolean, default=True) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) diff --git a/app/schemas/overlay.py b/app/schemas/overlay.py new file mode 100644 index 0000000..92ea13f --- /dev/null +++ b/app/schemas/overlay.py @@ -0,0 +1,210 @@ +""" +Pydantic schemas for Overlay API responses +""" + +from datetime import datetime +from typing import Any, Dict, List, Optional +from pydantic import BaseModel, Field + + +class OverlayFeatures(BaseModel): + headline_burst_z: Optional[float] = None + youtube_influence_z: Optional[float] = None + wiki_attention_z: Optional[float] = None + theme_heat_z: Optional[float] = None + crowding_stress_z: Optional[float] = None + + +class OverlaySourcePresence(BaseModel): + yahoo: bool = False + youtube: bool = False + wikimedia: bool = False + google_trends: bool = False + finra: bool = False + + +class YahooSourceDetail(BaseModel): + headline_count_6h: int = 0 + headline_count_24h: int = 0 + publisher_breadth_24h: int = 0 + + +class YouTubeSourceDetail(BaseModel): + mentions_24h: int = 0 + weighted_views_24h: float = 0.0 + + +class WikiSourceDetail(BaseModel): + page_views_1d: Optional[int] = None + page_views_7d_avg: Optional[float] = None + + +class FinraSourceDetail(BaseModel): + short_volume_ratio: Optional[float] = None + short_volume_spike_zscore: Optional[float] = None + + +class OverlaySourceDetails(BaseModel): + yahoo: Optional[YahooSourceDetail] = None + youtube: Optional[YouTubeSourceDetail] = None + wikimedia: Optional[WikiSourceDetail] = None + finra: Optional[FinraSourceDetail] = None + + +class OverlayMetadata(BaseModel): + feature_version: str = "v1" + data_freshness: Optional[datetime] = None + next_update_expected: Optional[datetime] = None + + +class OverlayScoreResponse(BaseModel): + symbol: str + as_of_ts: Optional[datetime] = None + overlay_score: float = 0.0 + overlay_confidence: float = 0.0 + overlay_band: Optional[str] = None + hold_extension_hint: Optional[str] = None + add_on_eligibility: Optional[bool] = None + features: OverlayFeatures = Field(default_factory=OverlayFeatures) + source_presence: OverlaySourcePresence = Field(default_factory=OverlaySourcePresence) + source_details: OverlaySourceDetails = Field(default_factory=OverlaySourceDetails) + metadata: OverlayMetadata = Field(default_factory=OverlayMetadata) + + +class BulkOverlayResponse(BaseModel): + results: List[OverlayScoreResponse] + total_count: int + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class OverlayTopMover(BaseModel): + symbol: str + overlay_score: float + overlay_band: Optional[str] = None + as_of_ts: Optional[datetime] = None + + +class TopMoversResponse(BaseModel): + top_movers: List[OverlayTopMover] + total_count: int + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class HeadlineItem(BaseModel): + title: str + publisher: Optional[str] = None + published_at: datetime + article_guid: str + + +class HeadlinesResponse(BaseModel): + symbol: str + headlines: List[HeadlineItem] + headline_count_6h: int = 0 + headline_count_24h: int = 0 + publisher_breadth_24h: int = 0 + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class VideoItem(BaseModel): + video_id: str + channel_id: str + title: str + view_count: int = 0 + comment_count: int = 0 + published_at: Optional[datetime] = None + channel_weight: float = 0.5 + + +class YouTubeResponse(BaseModel): + symbol: str + videos: List[VideoItem] + mentions_24h: int = 0 + weighted_views_24h: float = 0.0 + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class WikiPageviewPoint(BaseModel): + date: datetime + views: int + page_title: str + + +class WikiResponse(BaseModel): + symbol: str + pageviews: List[WikiPageviewPoint] + views_1d: Optional[int] = None + views_7d_avg: Optional[float] = None + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class CrowdingResponse(BaseModel): + symbol: str + short_volume_ratio: Optional[float] = None + short_volume_spike_zscore: Optional[float] = None + crowding_stress_z: Optional[float] = None + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class TrendPoint(BaseModel): + observed_at: datetime + interest_value: int + topic_id: str + topic_label: Optional[str] = None + + +class TrendsResponse(BaseModel): + symbol: str + trends: List[TrendPoint] + theme_heat_z: Optional[float] = None + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class OverlayHistoryPoint(BaseModel): + as_of_ts: datetime + overlay_score: float + overlay_confidence: float + overlay_band: Optional[str] = None + + +class OverlayHistoryResponse(BaseModel): + symbol: str + history: List[OverlayHistoryPoint] + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class SourceHealthItem(BaseModel): + source: str + last_collected_at: Optional[datetime] = None + status: str = "unknown" + success_rate_24h: Optional[float] = None + records_24h: int = 0 + + +class AdminHealthResponse(BaseModel): + overlay_enabled: bool + sources: List[SourceHealthItem] + last_pipeline_run: Optional[datetime] = None + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class TriggerPipelineResponse(BaseModel): + status: str + message: str + job_ids: List[str] = Field(default_factory=list) + + +class JobLogEntry(BaseModel): + id: str + job_type: str + status: str + started_at: datetime + completed_at: Optional[datetime] = None + records_processed: int = 0 + error_message: Optional[str] = None + + +class JobLogResponse(BaseModel): + logs: List[JobLogEntry] + total_count: int + metadata: Dict[str, Any] = Field(default_factory=dict) diff --git a/app/services/overlay/__init__.py b/app/services/overlay/__init__.py new file mode 100644 index 0000000..22d3ed1 --- /dev/null +++ b/app/services/overlay/__init__.py @@ -0,0 +1,3 @@ +""" +Overlay services package - attention overlay data collection, feature building, and scoring. +""" diff --git a/app/services/overlay/entity_resolver.py b/app/services/overlay/entity_resolver.py new file mode 100644 index 0000000..d64f8a8 --- /dev/null +++ b/app/services/overlay/entity_resolver.py @@ -0,0 +1,87 @@ +""" +Entity Resolver - resolve text mentions to ticker symbols using 4-stage matching. +""" + +import re +import logging +from typing import List, Optional, Set + +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select + +from app.models.overlay_registry import CompanyAlias +from app.core.overlay_config import TOP_50_SYMBOLS + +logger = logging.getLogger(__name__) + +# Pre-compiled common English stop-words to avoid false ticker matches +_STOP_WORDS = { + "A", "I", "IN", "ON", "AT", "IT", "IS", "BE", "AS", "OR", "AND", + "THE", "FOR", "TO", "OF", "BY", "AN", "UP", "DO", "GO", "US", + "PM", "AM", "ET", "AI", # keep AI out to avoid false positives +} + + +class EntityResolver: + """Resolve text mentions to ticker symbols using 4-stage matching.""" + + def __init__(self): + self._alias_cache: dict = {} + + async def load_aliases(self, db: AsyncSession) -> None: + """Load company aliases from DB into memory cache.""" + result = await db.execute( + select(CompanyAlias).where(CompanyAlias.active == True) + ) + rows = result.scalars().all() + cache: dict = {} + for row in rows: + key = row.alias_value.lower().strip() + if key not in cache: + cache[key] = [] + cache[key].append((row.symbol, row.confidence)) + self._alias_cache = cache + logger.debug(f"EntityResolver: loaded {len(cache)} alias entries") + + def resolve_from_title(self, title: str) -> List[str]: + """ + Extract ticker symbols from a text string. + + Stage 1: $TICKER pattern (highest confidence) + Stage 2: Direct uppercase word match against TOP_50_SYMBOLS + Stage 3: Alias / company name match (case-insensitive) + """ + symbols: Set[str] = set() + + # Stage 1: $TICKER pattern + dollar_tickers = re.findall(r'\$([A-Z]{1,5})\b', title) + for t in dollar_tickers: + symbols.add(t) + + # Stage 2: Uppercase word match against known symbols + words = re.findall(r'\b([A-Z]{1,5})\b', title) + for w in words: + if w in TOP_50_SYMBOLS and w not in _STOP_WORDS: + symbols.add(w) + + # Stage 3: Alias / company name match (case-insensitive) + if self._alias_cache: + title_lower = title.lower() + for alias_text, candidates in self._alias_cache.items(): + if alias_text in title_lower: + for sym, conf in candidates: + if conf >= 0.7: + symbols.add(sym) + + return list(symbols) + + def resolve_symbol(self, text: str) -> Optional[str]: + """Resolve a single best-match symbol from text.""" + results = self.resolve_from_title(text) + if not results: + return None + # Prefer symbols in TOP_50 list + for s in results: + if s in TOP_50_SYMBOLS: + return s + return results[0] diff --git a/app/services/overlay/feature_builder.py b/app/services/overlay/feature_builder.py new file mode 100644 index 0000000..65c7f71 --- /dev/null +++ b/app/services/overlay/feature_builder.py @@ -0,0 +1,244 @@ +""" +Feature builder - aggregate raw events into normalized (z-scored) feature dicts. +""" + +import logging +import statistics +from datetime import datetime, timedelta, timezone +from typing import Dict, List, Optional + +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, and_, func + +from app.models.overlay_raw_event import ( + OverlayHeadlineEvent, + OverlayVideoEvent, + OverlayWikiPageview, +) +from app.services.overlay.finra_overlay_loader import FinraOverlayLoader +from app.core.overlay_config import ZSCORE_WINDOW_DAYS, WINSOR_LOWER, WINSOR_UPPER + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def winsorize(value: float, lower: float = WINSOR_LOWER, upper: float = WINSOR_UPPER) -> float: + return max(lower, min(upper, value)) + + +def _ensure_utc(dt) -> datetime: + """Ensure a 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 + + +def compute_zscore(value: float, values: List[float]) -> Optional[float]: + """Compute z-score of *value* within *values* (requires ≥2 data points).""" + if len(values) < 2: + return None + mean = statistics.mean(values) + stdev = statistics.pstdev(values) # population stdev for stability + if stdev == 0: + return 0.0 + z = (value - mean) / stdev + return winsorize(z) + + +def _day_key(ts) -> str: + """Return YYYY-MM-DD string from a datetime or date object.""" + if hasattr(ts, "date"): + return ts.date().isoformat() + return str(ts)[:10] + + +# --------------------------------------------------------------------------- +# Feature builder +# --------------------------------------------------------------------------- + +class FeatureBuilder: + """Build overlay feature records from raw event tables.""" + + def __init__(self): + self.finra_loader = FinraOverlayLoader() + + # ------------------------------------------------------------------ + # Headline features + # ------------------------------------------------------------------ + + async def build_headline_features( + self, db: AsyncSession, symbol: str, as_of: datetime + ) -> Dict: + cutoff_24h = as_of - timedelta(hours=24) + cutoff_6h = as_of - timedelta(hours=6) + window_cutoff = as_of - timedelta(days=ZSCORE_WINDOW_DAYS) + + # Fetch all headlines in z-score window (we need matched_symbols for filtering) + result = await db.execute( + select( + OverlayHeadlineEvent.publisher, + OverlayHeadlineEvent.published_at, + OverlayHeadlineEvent.matched_symbols, + ).where( + and_( + OverlayHeadlineEvent.published_at >= window_cutoff, + OverlayHeadlineEvent.published_at <= as_of, + ) + ) + ) + all_rows = result.fetchall() + + # Filter by symbol + sym_rows_hist = [r for r in all_rows if symbol in (r.matched_symbols or [])] + sym_rows_24h = [r for r in sym_rows_hist if _ensure_utc(r.published_at) >= cutoff_24h] + sym_rows_6h = [r for r in sym_rows_24h if _ensure_utc(r.published_at) >= cutoff_6h] + + headline_count_24h = len(sym_rows_24h) + headline_count_6h = len(sym_rows_6h) + publishers = {r.publisher for r in sym_rows_24h if r.publisher} + publisher_breadth_24h = len(publishers) + + # Build daily counts for z-score window + daily_counts: Dict[str, int] = {} + for row in sym_rows_hist: + key = _day_key(row.published_at) + daily_counts[key] = daily_counts.get(key, 0) + 1 + + hist_values = list(daily_counts.values()) + headline_burst_z = compute_zscore(float(headline_count_24h), hist_values) if hist_values else None + + return { + "headline_count_6h": headline_count_6h, + "headline_count_24h": headline_count_24h, + "publisher_breadth_24h": publisher_breadth_24h, + "headline_burst_z": headline_burst_z, + } + + # ------------------------------------------------------------------ + # YouTube features + # ------------------------------------------------------------------ + + async def build_youtube_features( + self, db: AsyncSession, symbol: str, as_of: datetime + ) -> Dict: + cutoff_24h = as_of - timedelta(hours=24) + window_cutoff = as_of - timedelta(days=ZSCORE_WINDOW_DAYS) + + result = await db.execute( + select( + OverlayVideoEvent.view_count, + OverlayVideoEvent.channel_weight, + OverlayVideoEvent.published_at, + OverlayVideoEvent.matched_symbols, + ).where( + and_( + OverlayVideoEvent.published_at >= window_cutoff, + OverlayVideoEvent.published_at <= as_of, + ) + ) + ) + all_rows = result.fetchall() + sym_rows = [r for r in all_rows if symbol in (r.matched_symbols or [])] + sym_rows_24h = [r for r in sym_rows if _ensure_utc(r.published_at) >= cutoff_24h] + + mentions_24h = len(sym_rows_24h) + weighted_views_24h = sum(r.view_count * r.channel_weight for r in sym_rows_24h) + + # Daily weighted views for z-score + daily_weighted: Dict[str, float] = {} + for row in sym_rows: + key = _day_key(row.published_at) + daily_weighted[key] = daily_weighted.get(key, 0.0) + row.view_count * row.channel_weight + + hist_values = list(daily_weighted.values()) + youtube_influence_z = compute_zscore(weighted_views_24h, hist_values) if hist_values else None + + return { + "youtube_mentions_24h": mentions_24h, + "youtube_weighted_views_24h": round(weighted_views_24h, 2), + "youtube_influence_z": youtube_influence_z, + } + + # ------------------------------------------------------------------ + # Wiki features + # ------------------------------------------------------------------ + + async def build_wiki_features( + self, db: AsyncSession, symbol: str, as_of: datetime + ) -> Dict: + cutoff_7d = as_of - timedelta(days=7) + cutoff_1d = as_of - timedelta(days=1) + window_cutoff = as_of - timedelta(days=ZSCORE_WINDOW_DAYS) + + result = await db.execute( + select( + OverlayWikiPageview.views, + OverlayWikiPageview.date, + ).where( + and_( + OverlayWikiPageview.mapped_symbol == symbol, + OverlayWikiPageview.date >= window_cutoff, + ) + ).order_by(OverlayWikiPageview.date.desc()) + ) + rows = result.fetchall() + + if not rows: + return {"wiki_views_1d": None, "wiki_views_7d_avg": None, "wiki_attention_z": None} + + # Latest day's views (most recent row, regardless of exact time) + views_1d = rows[0].views if rows else None + + # 7-day average + rows_7d = [r for r in rows if _ensure_utc(r.date) >= cutoff_7d] + views_7d_avg = sum(r.views for r in rows_7d) / len(rows_7d) if rows_7d else None + + # Historical z-score + hist_views = [r.views for r in rows] + wiki_attention_z = None + if views_1d is not None and hist_views: + wiki_attention_z = compute_zscore(float(views_1d), hist_views) + + return { + "wiki_views_1d": views_1d, + "wiki_views_7d_avg": round(views_7d_avg, 2) if views_7d_avg else None, + "wiki_attention_z": wiki_attention_z, + } + + # ------------------------------------------------------------------ + # FINRA crowding features + # ------------------------------------------------------------------ + + async def build_crowding_features(self, db: AsyncSession, symbol: str) -> Dict: + return await self.finra_loader.get_crowding_metrics(db, symbol) + + # ------------------------------------------------------------------ + # Build all features + # ------------------------------------------------------------------ + + async def build_all_features( + self, db: AsyncSession, symbol: str, as_of: Optional[datetime] = None + ) -> Dict: + """Build all features for a symbol and return a combined feature dict.""" + if as_of is None: + as_of = datetime.now(timezone.utc) + + headline = await self.build_headline_features(db, symbol, as_of) + youtube = await self.build_youtube_features(db, symbol, as_of) + wiki = await self.build_wiki_features(db, symbol, as_of) + crowding = await self.build_crowding_features(db, symbol) + + return { + **headline, + **youtube, + **wiki, + **crowding, + "as_of_ts": as_of, + # theme_heat_z comes from Google Trends; left None here (no Trends data yet) + "theme_heat_z": None, + } diff --git a/app/services/overlay/finra_overlay_loader.py b/app/services/overlay/finra_overlay_loader.py new file mode 100644 index 0000000..beec3ab --- /dev/null +++ b/app/services/overlay/finra_overlay_loader.py @@ -0,0 +1,85 @@ +""" +FINRA overlay loader - derive crowding/stress features from the existing +finra_short_volume table without duplicating data. +""" + +import logging +import statistics +from datetime import datetime, timedelta, timezone +from typing import Dict + +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, and_, func + +from app.models.finra_short_volume import FinraShortVolume + +logger = logging.getLogger(__name__) + + +class FinraOverlayLoader: + """Calculate crowding stress metrics from FINRA short volume data.""" + + async def get_crowding_metrics( + self, db: AsyncSession, symbol: str, days: int = 30 + ) -> Dict: + """ + Derive crowding stress metrics for a symbol over the last *days* days. + + Returns a dict with: + short_volume_ratio - latest daily short/total ratio + short_volume_spike_zscore - how far above the rolling mean + crowding_stress_z - negative spike z-score (high → more stress) + + Returns {} if no FINRA data is available. + """ + symbol = symbol.upper() + cutoff = datetime.now(timezone.utc) - timedelta(days=days) + + result = await db.execute( + select( + FinraShortVolume.date, + func.sum(FinraShortVolume.short_volume).label("short_volume"), + func.sum(FinraShortVolume.total_volume).label("total_volume"), + ) + .where( + and_( + FinraShortVolume.symbol == symbol, + FinraShortVolume.date >= cutoff, + ) + ) + .group_by(FinraShortVolume.date) + .order_by(FinraShortVolume.date) + ) + + rows = result.fetchall() + if not rows: + return {} + + # Build daily short ratios + ratios = [] + for row in rows: + _, sv, tv = row + if tv and tv > 0: + ratios.append(sv / tv) + + if not ratios: + return {} + + latest_ratio = ratios[-1] + + # z-score of latest vs rolling window + if len(ratios) >= 2: + mean_r = statistics.mean(ratios) + stdev_r = statistics.stdev(ratios) + spike_z = (latest_ratio - mean_r) / stdev_r if stdev_r > 0 else 0.0 + else: + spike_z = 0.0 + + # crowding_stress_z: higher short-volume spike → negative stress on price + crowding_stress_z = round(-spike_z, 4) + + return { + "short_volume_ratio": round(latest_ratio, 6), + "short_volume_spike_zscore": round(spike_z, 4), + "crowding_stress_z": crowding_stress_z, + } diff --git a/app/services/overlay/google_trends_adapter.py b/app/services/overlay/google_trends_adapter.py new file mode 100644 index 0000000..bdc2bae --- /dev/null +++ b/app/services/overlay/google_trends_adapter.py @@ -0,0 +1,96 @@ +""" +Google Trends adapter (experimental, feature-flagged). +Disabled by default; enable via GOOGLE_TRENDS_ENABLED=true env var. +""" + +import logging +from datetime import datetime, timezone +from typing import List + +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, and_ + +from app.core.config import settings +from app.models.overlay_registry import ThemeTopicMap +from app.models.overlay_raw_event import OverlayTrendObservation + +logger = logging.getLogger(__name__) + + +class GoogleTrendsAdapter: + """Collect Google Trends data (experimental, disabled by default).""" + + @property + def enabled(self) -> bool: + return getattr(settings, "GOOGLE_TRENDS_ENABLED", False) + + async def collect(self, db: AsyncSession) -> int: + """ + Collect Google Trends interest data for active topics. + + Returns 0 if feature flag is disabled or pytrends is not installed. + Returns number of new records inserted. + """ + if not self.enabled: + logger.debug("Google Trends adapter: feature flag disabled, skipping") + return 0 + + try: + from pytrends.request import TrendReq + except ImportError: + logger.warning("pytrends not installed; Google Trends adapter inactive") + return 0 + + result = await db.execute( + select(ThemeTopicMap).where(ThemeTopicMap.active == True) + ) + topics = result.scalars().all() + if not topics: + return 0 + + inserted = 0 + + try: + pytrends = TrendReq(hl="en-US", tz=360) + # pytrends limits to 5 keywords at a time + kw_list = [t.topic_label for t in topics[:5]] + pytrends.build_payload(kw_list, timeframe="now 7-d", geo="US") + interest_df = pytrends.interest_over_time() + except Exception as e: + logger.error(f"Google Trends API error: {e}") + return 0 + + for topic in topics: + if topic.topic_label not in interest_df.columns: + continue + series = interest_df[topic.topic_label] + for ts, val in series.items(): + observed_at = ts.to_pydatetime().replace(tzinfo=timezone.utc) + + # Check duplicate + existing = await db.execute( + select(OverlayTrendObservation.id).where( + and_( + OverlayTrendObservation.topic_id == topic.topic_id, + OverlayTrendObservation.observed_at == observed_at, + OverlayTrendObservation.geography == "US", + ) + ) + ) + if existing.first(): + continue + + record = OverlayTrendObservation( + topic_id=topic.topic_id, + observed_at=observed_at, + geography="US", + interest_value=int(val), + ) + db.add(record) + inserted += 1 + + if inserted: + await db.commit() + logger.info(f"Google Trends: inserted {inserted} observations") + + return inserted diff --git a/app/services/overlay/overlay_pipeline.py b/app/services/overlay/overlay_pipeline.py new file mode 100644 index 0000000..6c9634f --- /dev/null +++ b/app/services/overlay/overlay_pipeline.py @@ -0,0 +1,230 @@ +""" +Overlay pipeline orchestrator - coordinates data collection, feature building, +and scoring for the full pipeline (batch) and on-demand (single symbol). +""" + +import asyncio +import logging +from datetime import datetime, timedelta, timezone +from typing import Dict, List, Optional + +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, desc + +from app.core.overlay_config import ONDEMAND_TIMEOUT_SECONDS, FEATURE_STALE_HOURS, TOP_50_SYMBOLS +from app.models.overlay_feature import OverlayFeatureRecord, OverlayJobLog +from app.services.overlay.yahoo_rss_adapter import YahooRSSAdapter +from app.services.overlay.wikimedia_adapter import WikimediaAdapter +from app.services.overlay.youtube_adapter import YouTubeAdapter +from app.services.overlay.google_trends_adapter import GoogleTrendsAdapter +from app.services.overlay.feature_builder import FeatureBuilder +from app.services.overlay.overlay_scorer import OverlayScorer + +logger = logging.getLogger(__name__) + + +class OverlayPipeline: + """Orchestrate overlay data collection, feature building, and scoring.""" + + def __init__(self): + self.rss = YahooRSSAdapter() + self.wiki = WikimediaAdapter() + self.youtube = YouTubeAdapter() + self.trends = GoogleTrendsAdapter() + self.builder = FeatureBuilder() + self.scorer = OverlayScorer() + + # ------------------------------------------------------------------ + # Job logging helpers + # ------------------------------------------------------------------ + + async def _log_job( + self, + db: AsyncSession, + job_type: str, + status: str, + started_at: datetime, + records: int = 0, + error: Optional[str] = None, + ) -> None: + log = OverlayJobLog( + job_type=job_type, + status=status, + started_at=started_at, + completed_at=datetime.now(timezone.utc) if status != "running" else None, + records_processed=records, + error_message=error, + ) + db.add(log) + try: + await db.commit() + except Exception as e: + logger.warning(f"Failed to persist job log: {e}") + await db.rollback() + + # ------------------------------------------------------------------ + # Data collection + # ------------------------------------------------------------------ + + async def collect_all(self, db: AsyncSession) -> Dict: + """Run all data collectors concurrently (source failures are isolated).""" + started = datetime.now(timezone.utc) + await self._log_job(db, "collect_all", "running", started) + + results = await asyncio.gather( + self.rss.collect(db, symbols=TOP_50_SYMBOLS), + self.wiki.collect(db), + self.youtube.collect(db), + self.trends.collect(db), + return_exceptions=True, + ) + + counts = { + "yahoo_rss": results[0] if isinstance(results[0], int) else 0, + "wikimedia": results[1] if isinstance(results[1], int) else 0, + "youtube": results[2] if isinstance(results[2], int) else 0, + "google_trends": results[3] if isinstance(results[3], int) else 0, + } + + errors = [str(r) for r in results if isinstance(r, Exception)] + error_str = "; ".join(errors) if errors else None + final_status = "partial" if error_str else "completed" + + await self._log_job(db, "collect_all", final_status, started, sum(counts.values()), error_str) + return counts + + # ------------------------------------------------------------------ + # Feature building + # ------------------------------------------------------------------ + + async def build_features_for_symbol( + self, db: AsyncSession, symbol: str + ) -> Optional[OverlayFeatureRecord]: + """Build and persist a feature record for a single symbol.""" + as_of = datetime.now(timezone.utc) + features = await self.builder.build_all_features(db, symbol, as_of) + scores = self.scorer.score(features) + + record = OverlayFeatureRecord( + symbol=symbol.upper(), + as_of_ts=as_of, + feature_version="v1", + # z-scores + headline_burst_z=features.get("headline_burst_z"), + youtube_influence_z=features.get("youtube_influence_z"), + wiki_attention_z=features.get("wiki_attention_z"), + theme_heat_z=features.get("theme_heat_z"), + crowding_stress_z=features.get("crowding_stress_z"), + # headline detail + headline_count_6h=features.get("headline_count_6h", 0), + headline_count_24h=features.get("headline_count_24h", 0), + publisher_breadth_24h=features.get("publisher_breadth_24h", 0), + # youtube detail + youtube_mentions_24h=features.get("youtube_mentions_24h", 0), + youtube_weighted_views_24h=features.get("youtube_weighted_views_24h", 0.0), + # wiki detail + wiki_views_1d=features.get("wiki_views_1d"), + wiki_views_7d_avg=features.get("wiki_views_7d_avg"), + # FINRA detail + short_volume_ratio=features.get("short_volume_ratio"), + short_volume_spike_zscore=features.get("short_volume_spike_zscore"), + # scores + **scores, + ) + db.add(record) + await db.commit() + await db.refresh(record) + return record + + async def build_features_batch( + self, db: AsyncSession, symbols: Optional[List[str]] = None + ) -> int: + """Build features for a batch of symbols (default: TOP_50_SYMBOLS).""" + if symbols is None: + symbols = TOP_50_SYMBOLS + + started = datetime.now(timezone.utc) + await self._log_job(db, "feature_build", "running", started) + + built = 0 + errors = [] + for sym in symbols: + try: + await self.build_features_for_symbol(db, sym) + built += 1 + except Exception as e: + logger.error(f"Feature build error for {sym}: {e}") + errors.append(str(e)) + + error_str = "; ".join(errors[:3]) if errors else None + final_status = "partial" if error_str else "completed" + await self._log_job(db, "feature_build", final_status, started, built, error_str) + return built + + # ------------------------------------------------------------------ + # Query helpers + # ------------------------------------------------------------------ + + async def get_latest_feature( + self, db: AsyncSession, symbol: str + ) -> Optional[OverlayFeatureRecord]: + """Return the most recent feature record for a symbol.""" + result = await db.execute( + select(OverlayFeatureRecord) + .where(OverlayFeatureRecord.symbol == symbol.upper()) + .order_by(desc(OverlayFeatureRecord.as_of_ts)) + .limit(1) + ) + return result.scalars().first() + + def is_stale(self, record: Optional[OverlayFeatureRecord]) -> bool: + """Return True if the record is missing or older than FEATURE_STALE_HOURS.""" + if record is None: + return True + # SQLite returns naive datetimes; ensure UTC-aware for comparison + as_of = record.as_of_ts + if as_of.tzinfo is None: + as_of = as_of.replace(tzinfo=timezone.utc) + age = datetime.now(timezone.utc) - as_of + return age.total_seconds() > FEATURE_STALE_HOURS * 3600 + + # ------------------------------------------------------------------ + # On-demand build + # ------------------------------------------------------------------ + + async def get_or_build( + self, db: AsyncSession, symbol: str + ) -> Optional[OverlayFeatureRecord]: + """ + Return the latest feature record for *symbol*. + + If the record is stale/missing, trigger an on-demand pipeline run + with a timeout guard. Returns the (possibly stale) record on timeout. + """ + record = await self.get_latest_feature(db, symbol) + if record and not self.is_stale(record): + return record + + logger.info(f"Overlay on-demand build triggered for {symbol}") + try: + record = await asyncio.wait_for( + self.build_features_for_symbol(db, symbol), + timeout=ONDEMAND_TIMEOUT_SECONDS, + ) + except asyncio.TimeoutError: + logger.warning(f"Overlay on-demand build timed out for {symbol}") + # Return potentially-stale record rather than None + return record + + # ------------------------------------------------------------------ + # Full pipeline + # ------------------------------------------------------------------ + + async def run_full_pipeline(self, db: AsyncSession) -> Dict: + """Run collect → build feature for all TOP_50 symbols.""" + collect_counts = await self.collect_all(db) + built = await self.build_features_batch(db) + return { + "collected": collect_counts, + "features_built": built, + } diff --git a/app/services/overlay/overlay_scorer.py b/app/services/overlay/overlay_scorer.py new file mode 100644 index 0000000..c654cd2 --- /dev/null +++ b/app/services/overlay/overlay_scorer.py @@ -0,0 +1,97 @@ +""" +Overlay scorer - compute final overlay_score, band, confidence, and decision hints +from a feature dict produced by FeatureBuilder. +""" + +import math +import logging +from typing import Dict, Optional + +from app.core.overlay_config import ( + SOURCE_WEIGHTS, + BAND_THRESHOLDS, + CONFIDENCE_PER_SOURCE, + HOLD_EXTENSION_EXTEND_THRESHOLD, + HOLD_EXTENSION_TRIM_THRESHOLD, + ADD_ON_ELIGIBILITY_THRESHOLD, +) + +logger = logging.getLogger(__name__) + + +def _sigmoid(x: float) -> float: + """Sigmoid function mapping any real to (0, 1).""" + return 1.0 / (1.0 + math.exp(-x)) + + +def _zscore_to_01(z: Optional[float]) -> Optional[float]: + """Convert z-score to 0~1 via sigmoid (z=0 → 0.5).""" + if z is None: + return None + return _sigmoid(z) + + +class OverlayScorer: + """Compute final overlay score and derived metrics from a feature dict.""" + + def score(self, features: Dict) -> Dict: + """ + Given a features dict (output of FeatureBuilder.build_all_features), + compute overlay_score, overlay_confidence, overlay_band, and decision hints. + + Returns a dict suitable for storing in OverlayFeatureRecord. + """ + # Map each z-score to 0~1 + normalized = { + "yahoo": _zscore_to_01(features.get("headline_burst_z")), + "youtube": _zscore_to_01(features.get("youtube_influence_z")), + "wikimedia": _zscore_to_01(features.get("wiki_attention_z")), + "google_trends": _zscore_to_01(features.get("theme_heat_z")), + "finra": _zscore_to_01(features.get("crowding_stress_z")), + } + + # Source presence mask + source_presence_mask = {src: (v is not None) for src, v in normalized.items()} + + # Weighted average across present sources + total_weight = 0.0 + weighted_sum = 0.0 + present_count = 0 + for src, val in normalized.items(): + if val is not None: + w = SOURCE_WEIGHTS.get(src, 0.0) + weighted_sum += val * w + total_weight += w + present_count += 1 + + overlay_score = weighted_sum / total_weight if total_weight > 0 else 0.0 + + # Confidence based on number of active sources + overlay_confidence = min(1.0, present_count * CONFIDENCE_PER_SOURCE) + + # Band assignment (evaluate thresholds from high to low) + overlay_band = "silent" + for band_name, threshold in sorted(BAND_THRESHOLDS.items(), key=lambda kv: -kv[1]): + if overlay_score >= threshold: + overlay_band = band_name + break + + # Hold-extension hint + if overlay_score >= HOLD_EXTENSION_EXTEND_THRESHOLD: + hold_extension_hint = "extend" + elif overlay_score <= HOLD_EXTENSION_TRIM_THRESHOLD: + hold_extension_hint = "trim" + else: + hold_extension_hint = "neutral" + + # Add-on eligibility + add_on_eligibility = overlay_score >= ADD_ON_ELIGIBILITY_THRESHOLD + + return { + "overlay_score": round(overlay_score, 4), + "overlay_confidence": round(overlay_confidence, 4), + "overlay_band": overlay_band, + "source_presence_mask": source_presence_mask, + "hold_extension_hint": hold_extension_hint, + "add_on_eligibility": add_on_eligibility, + } diff --git a/app/services/overlay/scheduler.py b/app/services/overlay/scheduler.py new file mode 100644 index 0000000..74c4bcf --- /dev/null +++ b/app/services/overlay/scheduler.py @@ -0,0 +1,100 @@ +""" +APScheduler-based batch scheduler for the Overlay pipeline. + +Integrated into FastAPI's lifespan via start_scheduler() / stop_scheduler(). +""" + +import logging +from datetime import datetime, timezone + +logger = logging.getLogger(__name__) + +_scheduler = None + + +def _get_scheduler(): + """Lazily create the APScheduler instance.""" + global _scheduler + if _scheduler is None: + try: + from apscheduler.schedulers.asyncio import AsyncIOScheduler + from apscheduler.triggers.cron import CronTrigger + _scheduler = AsyncIOScheduler(timezone="UTC") + except ImportError: + logger.warning("apscheduler not installed; overlay batch scheduling disabled") + return None + return _scheduler + + +async def _run_collect_job() -> None: + """Scheduled job: collect RSS + wiki + youtube + trends.""" + from app.core.database import AsyncSessionLocal + from app.services.overlay.overlay_pipeline import OverlayPipeline + + pipeline = OverlayPipeline() + async with AsyncSessionLocal() as db: + try: + counts = await pipeline.collect_all(db) + logger.info(f"[Scheduler] collect_all completed: {counts}") + except Exception as e: + logger.error(f"[Scheduler] collect_all failed: {e}") + + +async def _run_build_job() -> None: + """Scheduled job: build features for all TOP_50 symbols.""" + from app.core.database import AsyncSessionLocal + from app.services.overlay.overlay_pipeline import OverlayPipeline + + pipeline = OverlayPipeline() + async with AsyncSessionLocal() as db: + try: + built = await pipeline.build_features_batch(db) + logger.info(f"[Scheduler] feature_build completed: {built} symbols") + except Exception as e: + logger.error(f"[Scheduler] feature_build failed: {e}") + + +def start_scheduler() -> None: + """Start the APScheduler with overlay jobs. Call from FastAPI lifespan startup.""" + from app.core.config import settings + + if not getattr(settings, "OVERLAY_ENABLED", True): + logger.info("Overlay disabled; scheduler not started") + return + + sched = _get_scheduler() + if sched is None: + return + + try: + from apscheduler.triggers.cron import CronTrigger + + # Collect: 23:30 UTC daily on weekdays (≈ 18:30 ET) + sched.add_job( + _run_collect_job, + trigger=CronTrigger(day_of_week="mon-fri", hour=23, minute=30, timezone="UTC"), + id="overlay_collect", + replace_existing=True, + misfire_grace_time=600, + ) + # Feature build: 01:30 UTC (≈ 20:30 ET) + sched.add_job( + _run_build_job, + trigger=CronTrigger(day_of_week="mon-fri", hour=1, minute=30, timezone="UTC"), + id="overlay_feature_build", + replace_existing=True, + misfire_grace_time=600, + ) + + sched.start() + logger.info("Overlay scheduler started (collect @ 23:30 UTC, build @ 01:30 UTC, weekdays)") + except Exception as e: + logger.error(f"Overlay scheduler start failed: {e}") + + +def stop_scheduler() -> None: + """Stop the scheduler. Call from FastAPI lifespan shutdown.""" + sched = _get_scheduler() + if sched and sched.running: + sched.shutdown(wait=False) + logger.info("Overlay scheduler stopped") diff --git a/app/services/overlay/wikimedia_adapter.py b/app/services/overlay/wikimedia_adapter.py new file mode 100644 index 0000000..8259267 --- /dev/null +++ b/app/services/overlay/wikimedia_adapter.py @@ -0,0 +1,107 @@ +""" +Wikimedia REST API adapter - collect daily page view counts for watched pages. +""" + +import logging +from datetime import datetime, timedelta, timezone +from typing import List, Optional + +import aiohttp +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, and_ + +from app.models.overlay_raw_event import OverlayWikiPageview +from app.models.overlay_registry import WikiPageMap + +logger = logging.getLogger(__name__) + +WIKIMEDIA_BASE = "https://wikimedia.org/api/rest_v1/metrics/pageviews/per-article" +USER_AGENT = "StockOracle/1.0 (github.com/stockoracle; contact@stockoracle.com)" + + +class WikimediaAdapter: + """Collect Wikimedia page view data for pages mapped to tickers.""" + + async def _get_watched_pages(self, db: AsyncSession) -> List[WikiPageMap]: + result = await db.execute( + select(WikiPageMap).where(WikiPageMap.active == True) + ) + return result.scalars().all() + + async def fetch_pageviews( + self, page_title: str, date: datetime, project: str = "en.wikipedia" + ) -> Optional[int]: + """Fetch daily pageviews for a specific page and date.""" + date_str = date.strftime("%Y%m%d") + encoded_title = page_title.replace(" ", "_") + url = ( + f"{WIKIMEDIA_BASE}/{project}/all-access/all-agents" + f"/{encoded_title}/daily/{date_str}/{date_str}" + ) + try: + headers = {"User-Agent": USER_AGENT} + async with aiohttp.ClientSession(headers=headers) as session: + async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp: + if resp.status == 404: + return None + resp.raise_for_status() + data = await resp.json() + items = data.get("items", []) + if items: + return items[0].get("views", 0) + return 0 + except Exception as e: + logger.debug(f"Wikimedia fetch error for '{page_title}' on {date_str}: {e}") + return None + + async def collect(self, db: AsyncSession, days_back: int = 3) -> int: + """ + Collect pageviews for all active wiki page mappings. + + Returns number of new records inserted. + """ + pages = await self._get_watched_pages(db) + if not pages: + logger.info("Wikimedia: no active wiki page mappings found") + return 0 + + inserted = 0 + now = datetime.now(timezone.utc) + + for page in pages: + for days_ago in range(1, days_back + 1): + target_date = now - timedelta(days=days_ago) + target_date = target_date.replace(hour=0, minute=0, second=0, microsecond=0) + + # Check duplicate + existing = await db.execute( + select(OverlayWikiPageview.id).where( + and_( + OverlayWikiPageview.page_title == page.wiki_page_title, + OverlayWikiPageview.date == target_date, + OverlayWikiPageview.project == "en.wikipedia", + ) + ) + ) + if existing.first(): + continue + + views = await self.fetch_pageviews(page.wiki_page_title, target_date) + if views is None: + continue + + record = OverlayWikiPageview( + page_title=page.wiki_page_title, + date=target_date, + project="en.wikipedia", + views=views, + mapped_symbol=page.symbol, + ) + db.add(record) + inserted += 1 + + if inserted: + await db.commit() + logger.info(f"Wikimedia: inserted {inserted} pageview records") + + return inserted diff --git a/app/services/overlay/yahoo_rss_adapter.py b/app/services/overlay/yahoo_rss_adapter.py new file mode 100644 index 0000000..bdc7cae --- /dev/null +++ b/app/services/overlay/yahoo_rss_adapter.py @@ -0,0 +1,147 @@ +""" +Yahoo Finance RSS feed adapter - collect headline events and match to tickers. +""" + +import asyncio +import logging +from datetime import datetime, timezone +from typing import Dict, List, Optional + +import aiohttp +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select + +from app.models.overlay_raw_event import OverlayHeadlineEvent +from app.services.overlay.entity_resolver import EntityResolver +from app.core.overlay_config import YAHOO_RSS_FEEDS, TOP_50_SYMBOLS + +logger = logging.getLogger(__name__) + + +class YahooRSSAdapter: + """Collect Yahoo Finance RSS headlines and match to tickers.""" + + def __init__(self): + self.resolver = EntityResolver() + + async def fetch_feed(self, url: str) -> List[Dict]: + """Fetch and parse a single RSS feed URL.""" + try: + import feedparser + except ImportError: + logger.error("feedparser not installed; Yahoo RSS adapter inactive") + return [] + + try: + async with aiohttp.ClientSession() as session: + headers = { + "User-Agent": "Mozilla/5.0 StockOracle/1.0", + "Accept": "application/rss+xml, application/xml, text/xml", + } + async with session.get( + url, headers=headers, timeout=aiohttp.ClientTimeout(total=30) + ) as resp: + if resp.status != 200: + logger.warning(f"Yahoo RSS: non-200 from {url}: {resp.status}") + return [] + text = await resp.text() + + feed = feedparser.parse(text) + entries = [] + for entry in feed.entries: + guid = getattr(entry, "id", None) or getattr(entry, "link", None) + title = getattr(entry, "title", "") + publisher = getattr(entry, "publisher", None) + if not publisher: + src = getattr(entry, "source", {}) + publisher = src.get("title") if isinstance(src, dict) else None + + published = getattr(entry, "published_parsed", None) + if published: + try: + pub_dt = datetime(*published[:6], tzinfo=timezone.utc) + except Exception: + pub_dt = datetime.now(timezone.utc) + else: + pub_dt = datetime.now(timezone.utc) + + entries.append({ + "guid": guid or title, + "title": title, + "publisher": publisher, + "published_at": pub_dt, + }) + return entries + + except Exception as e: + logger.error(f"Yahoo RSS fetch error for {url}: {e}") + return [] + + async def collect(self, db: AsyncSession, symbols: Optional[List[str]] = None) -> int: + """ + Collect RSS headlines from general feeds and per-symbol feeds. + + Persists new events (deduped by article_guid), skips duplicates. + Returns number of new records inserted. + """ + await self.resolver.load_aliases(db) + + all_entries: List[Dict] = [] + + # General feeds + tasks = [self.fetch_feed(url) for url in YAHOO_RSS_FEEDS] + results = await asyncio.gather(*tasks, return_exceptions=True) + for r in results: + if isinstance(r, list): + all_entries.extend(r) + + # Per-symbol feeds for watch-list + watch_symbols = symbols or TOP_50_SYMBOLS + sym_tasks = [ + self.fetch_feed(f"https://finance.yahoo.com/rss/headline?s={sym}") + for sym in watch_symbols + ] + sym_results = await asyncio.gather(*sym_tasks, return_exceptions=True) + for r in sym_results: + if isinstance(r, list): + all_entries.extend(r) + + # Deduplicate by guid within this batch + seen_guids: set = set() + unique_entries = [] + for entry in all_entries: + g = entry.get("guid") + if g and g not in seen_guids: + seen_guids.add(g) + unique_entries.append(entry) + + inserted = 0 + for entry in unique_entries: + if not entry.get("guid") or not entry.get("title"): + continue + + # Check DB duplicate + existing = await db.execute( + select(OverlayHeadlineEvent.id).where( + OverlayHeadlineEvent.article_guid == entry["guid"] + ) + ) + if existing.first(): + continue + + matched = self.resolver.resolve_from_title(entry["title"]) + event = OverlayHeadlineEvent( + article_guid=entry["guid"], + title=entry["title"], + publisher=entry.get("publisher"), + published_at=entry["published_at"], + matched_symbols=matched, + ) + db.add(event) + inserted += 1 + + if inserted: + await db.commit() + logger.info(f"Yahoo RSS: inserted {inserted} new headline events") + + return inserted diff --git a/app/services/overlay/youtube_adapter.py b/app/services/overlay/youtube_adapter.py new file mode 100644 index 0000000..3d7558c --- /dev/null +++ b/app/services/overlay/youtube_adapter.py @@ -0,0 +1,168 @@ +""" +YouTube Data API v3 adapter - collect video mentions from whitelisted channels. +Gracefully skips if YOUTUBE_API_KEY is not configured. +""" + +import logging +from datetime import datetime, timedelta, timezone +from typing import Dict, List, Optional + +import aiohttp +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select + +from app.core.config import settings +from app.models.overlay_raw_event import OverlayVideoEvent +from app.models.overlay_registry import YouTubeChannelRegistry +from app.services.overlay.entity_resolver import EntityResolver + +logger = logging.getLogger(__name__) + +YOUTUBE_SEARCH_URL = "https://www.googleapis.com/youtube/v3/search" +YOUTUBE_VIDEOS_URL = "https://www.googleapis.com/youtube/v3/videos" + + +class YouTubeAdapter: + """Collect YouTube video data from whitelisted channels.""" + + def __init__(self): + self.api_key: str = getattr(settings, "YOUTUBE_API_KEY", "") + self.resolver = EntityResolver() + + @property + def enabled(self) -> bool: + return bool(self.api_key) + + async def _get_channels(self, db: AsyncSession) -> List[YouTubeChannelRegistry]: + result = await db.execute( + select(YouTubeChannelRegistry).where(YouTubeChannelRegistry.active == True) + ) + return result.scalars().all() + + async def _search_channel_videos( + self, channel_id: str, published_after: datetime + ) -> List[dict]: + """Search for recent videos from a channel via YouTube Data API.""" + params = { + "part": "snippet", + "channelId": channel_id, + "type": "video", + "order": "date", + "publishedAfter": published_after.strftime("%Y-%m-%dT%H:%M:%SZ"), + "maxResults": 20, + "key": self.api_key, + } + try: + async with aiohttp.ClientSession() as session: + async with session.get( + YOUTUBE_SEARCH_URL, params=params, timeout=aiohttp.ClientTimeout(total=30) + ) as resp: + if resp.status == 403: + logger.warning("YouTube API: quota exceeded or invalid key") + return [] + resp.raise_for_status() + data = await resp.json() + return data.get("items", []) + except Exception as e: + logger.error(f"YouTube search error for channel {channel_id}: {e}") + return [] + + async def _get_video_stats(self, video_ids: List[str]) -> Dict[str, Dict]: + """Fetch view/comment counts for a list of video IDs.""" + if not video_ids: + return {} + params = { + "part": "statistics", + "id": ",".join(video_ids), + "key": self.api_key, + } + try: + async with aiohttp.ClientSession() as session: + async with session.get( + YOUTUBE_VIDEOS_URL, params=params, timeout=aiohttp.ClientTimeout(total=30) + ) as resp: + resp.raise_for_status() + data = await resp.json() + stats = {} + for item in data.get("items", []): + vid_id = item["id"] + s = item.get("statistics", {}) + stats[vid_id] = { + "view_count": int(s.get("viewCount", 0)), + "comment_count": int(s.get("commentCount", 0)), + } + return stats + except Exception as e: + logger.error(f"YouTube stats fetch error: {e}") + return {} + + async def collect(self, db: AsyncSession, days_back: int = 2) -> int: + """ + Collect YouTube videos from whitelisted channels. + + Gracefully returns 0 if API key is not configured. + Returns number of new records inserted. + """ + if not self.enabled: + logger.info("YouTube adapter: API key not configured, skipping") + return 0 + + await self.resolver.load_aliases(db) + channels = await self._get_channels(db) + if not channels: + logger.info("YouTube: no active channels in registry") + return 0 + + published_after = datetime.now(timezone.utc) - timedelta(days=days_back) + inserted = 0 + + for channel in channels: + items = await self._search_channel_videos(channel.channel_id, published_after) + video_ids = [ + item["id"]["videoId"] + for item in items + if isinstance(item.get("id"), dict) and "videoId" in item["id"] + ] + stats = await self._get_video_stats(video_ids) + + for item in items: + vid_id = item.get("id", {}).get("videoId") if isinstance(item.get("id"), dict) else None + if not vid_id: + continue + + # Check duplicate + existing = await db.execute( + select(OverlayVideoEvent.id).where(OverlayVideoEvent.video_id == vid_id) + ) + if existing.first(): + continue + + snippet = item.get("snippet", {}) + title = snippet.get("title", "") + pub_at_str = snippet.get("publishedAt", "") + try: + pub_at = datetime.fromisoformat(pub_at_str.replace("Z", "+00:00")) + except Exception: + pub_at = datetime.now(timezone.utc) + + stat = stats.get(vid_id, {}) + matched = self.resolver.resolve_from_title(title) + + event = OverlayVideoEvent( + video_id=vid_id, + channel_id=channel.channel_id, + title=title, + view_count=stat.get("view_count", 0), + comment_count=stat.get("comment_count", 0), + published_at=pub_at, + matched_symbols=matched, + channel_weight=channel.channel_weight, + ) + db.add(event) + inserted += 1 + + if inserted: + await db.commit() + logger.info(f"YouTube: inserted {inserted} video events") + + return inserted diff --git a/docs/PYTHON_CLIENT.md b/docs/PYTHON_CLIENT.md index 3adccb2..155bf7b 100644 --- a/docs/PYTHON_CLIENT.md +++ b/docs/PYTHON_CLIENT.md @@ -160,6 +160,65 @@ bulk_etf = client.get_bulk_etf_holdings( ) ``` +### 6. Alpaca Market Data + +```python +import requests + +BASE = "http://localhost:18001/api/v1" + +# Check Alpaca connection +status = requests.get(f"{BASE}/alpaca/status").json() +print(f"Alpaca configured: {status['configured']}") + +# Get daily bars with DB storage +resp = requests.get(f"{BASE}/alpaca/data/AAPL", params={ + "interval": "1d", + "start_date": "2025-01-01", + "end_date": "2025-01-31", +}).json() +print(f"Alpaca bars: {resp['metadata']['data_points']} data points") + +# Get raw bars (no DB) +bars = requests.get(f"{BASE}/alpaca/bars/TSLA", params={ + "interval": "1d", + "start_date": "2025-03-01", + "end_date": "2025-03-10", +}).json() +for bar in bars["bars"][:3]: + print(f" {bar['timestamp']}: close={bar['close']}, vwap={bar['vwap']}") + +# Intraday candles +candles = requests.get(f"{BASE}/alpaca/intraday/NVDA", params={ + "interval": "1h", + "start_date": "2025-03-10", + "end_date": "2025-03-10", +}).json() +print(f"Intraday candles: {candles['count']}") +``` + +### 7. FINRA Short Volume + +```python +import requests + +BASE = "http://localhost:18001/api/v1" + +# Get short volume (auto-ingests if missing) +sv = requests.get(f"{BASE}/finra/short-volume/AAPL", params={"days": 30}).json() +print(f"AAPL short volume entries: {sv['total_count']}") +for entry in sv["entries"][:3]: + print(f" {entry['date']}: ratio={entry['short_ratio']:.2%}") + +# Get short ratio history (aggregated across markets) +ratio = requests.get(f"{BASE}/finra/short-ratio/AAPL", params={"days": 60}).json() +print(f"Average short ratio: {ratio['avg_short_ratio']:.2%}") + +# Manual ingest +ingest = requests.post(f"{BASE}/finra/admin/ingest", params={"date": "2025-03-10"}).json() +print(f"Ingested: {ingest['records_ingested']} records") +``` + ## Advanced Usage ### Error Handling @@ -524,6 +583,11 @@ print(f"Exported {len(df)} records") - Automatic retry logic - Utility methods for validation and search +### v3.0.0 +- Added Alpaca Market Data integration (bars, intraday, DB storage) +- Added FINRA Short Volume data (auto-ingest, short ratio history) +- New data sources: Alpaca (optional API key), FINRA (public CDN) + ### v1.0.0 - Initial release - Basic financial and price data retrieval diff --git a/requirements-api.txt b/requirements-api.txt index 7588755..4127bd1 100644 --- a/requirements-api.txt +++ b/requirements-api.txt @@ -34,6 +34,11 @@ lxml>=4.9.0 # Regular yfinance is needed as dependency for yfinance-plus yfinance>=0.2.65 +# Overlay (Phase 5) +feedparser>=6.0.0 # Yahoo RSS feed parsing +apscheduler>=3.10.0 # Batch scheduling +pytrends>=4.9.0 # Google Trends (experimental, disabled by default) + # Development & Testing pytest>=7.4.0 pytest-asyncio>=0.21.0 diff --git a/tests/test_overlay.py b/tests/test_overlay.py new file mode 100644 index 0000000..ae1a5a0 --- /dev/null +++ b/tests/test_overlay.py @@ -0,0 +1,396 @@ +""" +Overlay API tests — Phase 5 + +Tests cover: +1. Cache utility unit tests (no DB, no Redis required) +2. Overlay API endpoint integration tests using the real SQLite DB + (Redis is not required — graceful degradation is expected) +""" + +import pytest +from fastapi.testclient import TestClient + +from app.main import app +from app.utils.cache import build_cache_key, compute_etag, _serialize, _deserialize + + +# --------------------------------------------------------------------------- +# Shared test client (uses real stock_oracle.db — no table creation needed) +# --------------------------------------------------------------------------- + +@pytest.fixture(scope="module") +def client(): + with TestClient(app) as c: + yield c + + +# =========================================================================== +# 1. Cache utility unit tests +# =========================================================================== + +class TestCacheUtils: + def test_build_cache_key_basic(self): + key = build_cache_key("overlay:score", "AAPL") + assert key == "overlay:score:AAPL" + + def test_build_cache_key_multiple_parts(self): + key = build_cache_key("overlay:bulk", "AAPL,MSFT", "true") + assert key == "overlay:bulk:AAPL,MSFT:true" + + def test_build_cache_key_skips_none(self): + key = build_cache_key("overlay:headlines", "TSLA", None) + assert key == "overlay:headlines:TSLA" + + def test_build_cache_key_skips_empty_string(self): + key = build_cache_key("overlay:wiki", "AAPL", "") + assert key == "overlay:wiki:AAPL" + + def test_compute_etag_deterministic(self): + data = {"symbol": "AAPL", "score": 1.23} + b = _serialize(data) + assert compute_etag(b) == compute_etag(b) + + def test_compute_etag_different_data(self): + a = _serialize({"a": 1}) + b = _serialize({"a": 2}) + assert compute_etag(a) != compute_etag(b) + + def test_compute_etag_is_hex_string(self): + b = _serialize({"x": "y"}) + etag = compute_etag(b) + assert isinstance(etag, str) + assert len(etag) == 64 # sha256 hex = 64 chars + + def test_serialize_deserialize_roundtrip(self): + payload = {"symbol": "AAPL", "overlay_score": 0.75, "items": [1, 2, 3]} + raw = _serialize(payload) + result = _deserialize(raw) + assert result == payload + + def test_deserialize_none_returns_none(self): + assert _deserialize(None) is None + + def test_deserialize_invalid_bytes_returns_none(self): + assert _deserialize(b"not-json{{{") is None + + +# =========================================================================== +# 2. Redis config fix verification +# =========================================================================== + +class TestRedisConfig: + def test_redis_url_uses_port_16380(self): + from app.core.config import settings + assert "16380" in settings.REDIS_URL, ( + f"REDIS_URL should use port 16380 (docker-compose), got: {settings.REDIS_URL}" + ) + + def test_redis_port_default_matches_docker_compose(self): + from app.core.config import settings + assert settings.REDIS_PORT == 16380 + + def test_cache_graceful_degradation_no_redis(self): + """With no Redis running, get_cached_response returns None without error.""" + import asyncio + from app.utils.cache import get_cached_response, set_cached_response + + async def _run(): + # set_cached_response should return an etag even without Redis + etag = await set_cached_response("test:key", {"a": 1}, ttl_seconds=60) + assert isinstance(etag, str) + assert len(etag) == 64 + # get_cached_response should return None without Redis + result = await get_cached_response("test:key") + # Either None (no Redis) or a tuple (Redis connected) + assert result is None or isinstance(result, tuple) + + asyncio.run(_run()) + + +# =========================================================================== +# 3. Overlay endpoint integration tests +# =========================================================================== + +class TestAdminHealth: + def test_returns_200(self, client): + r = client.get("/api/v1/overlay/admin/health") + assert r.status_code == 200 + + def test_response_schema(self, client): + r = client.get("/api/v1/overlay/admin/health") + data = r.json() + assert "overlay_enabled" in data + assert "sources" in data + assert isinstance(data["sources"], list) + assert len(data["sources"]) > 0 + + def test_sources_have_required_fields(self, client): + r = client.get("/api/v1/overlay/admin/health") + for src in r.json()["sources"]: + assert "source" in src + assert "status" in src + assert "records_24h" in src + + def test_overlay_is_enabled(self, client): + r = client.get("/api/v1/overlay/admin/health") + assert r.json()["overlay_enabled"] is True + + +class TestAdminJobLog: + def test_returns_200(self, client): + r = client.get("/api/v1/overlay/admin/job-log") + assert r.status_code == 200 + + def test_response_schema(self, client): + data = client.get("/api/v1/overlay/admin/job-log").json() + assert "logs" in data + assert "total_count" in data + assert isinstance(data["logs"], list) + + def test_limit_param(self, client): + r = client.get("/api/v1/overlay/admin/job-log?limit=5") + assert r.status_code == 200 + assert len(r.json()["logs"]) <= 5 + + def test_limit_too_large_is_capped(self, client): + r = client.get("/api/v1/overlay/admin/job-log?limit=600") + assert r.status_code == 422 # exceeds max 500 + + +class TestTopMovers: + def test_returns_200(self, client): + r = client.get("/api/v1/overlay/top-movers") + assert r.status_code == 200 + + def test_response_schema(self, client): + data = client.get("/api/v1/overlay/top-movers").json() + assert "top_movers" in data + assert "total_count" in data + assert isinstance(data["top_movers"], list) + + def test_limit_param(self, client): + r = client.get("/api/v1/overlay/top-movers?limit=5") + assert r.status_code == 200 + assert len(r.json()["top_movers"]) <= 5 + + def test_limit_below_min_rejected(self, client): + r = client.get("/api/v1/overlay/top-movers?limit=0") + assert r.status_code == 422 + + def test_limit_above_max_rejected(self, client): + r = client.get("/api/v1/overlay/top-movers?limit=101") + assert r.status_code == 422 + + def test_x_cache_header_present(self, client): + r = client.get("/api/v1/overlay/top-movers") + # X-Cache should be HIT or MISS (Redis may or may not be running) + assert "X-Cache" in r.headers + assert r.headers["X-Cache"] in ("HIT", "MISS") + + +class TestBulkOverlay: + def test_returns_200_with_known_symbol(self, client): + r = client.get("/api/v1/overlay/bulk?symbols=AAPL") + assert r.status_code == 200 + + def test_response_schema(self, client): + data = client.get("/api/v1/overlay/bulk?symbols=AAPL,MSFT").json() + assert "results" in data + assert "total_count" in data + assert isinstance(data["results"], list) + + def test_too_many_symbols_rejected(self, client): + symbols = ",".join([f"S{i:03d}" for i in range(51)]) + r = client.get(f"/api/v1/overlay/bulk?symbols={symbols}") + assert r.status_code == 400 + assert "50" in r.json()["detail"] + + def test_empty_symbols_rejected(self, client): + r = client.get("/api/v1/overlay/bulk?symbols=,,,") + assert r.status_code == 400 + + def test_symbols_uppercased(self, client): + r = client.get("/api/v1/overlay/bulk?symbols=aapl") + assert r.status_code == 200 + + def test_x_cache_header_present(self, client): + r = client.get("/api/v1/overlay/bulk?symbols=AAPL") + assert "X-Cache" in r.headers + + +class TestHeadlines: + def test_returns_200(self, client): + r = client.get("/api/v1/overlay/AAPL/headlines") + assert r.status_code == 200 + + def test_response_schema(self, client): + data = client.get("/api/v1/overlay/AAPL/headlines").json() + assert "symbol" in data + assert data["symbol"] == "AAPL" + assert "headlines" in data + assert "headline_count_6h" in data + assert "headline_count_24h" in data + assert "publisher_breadth_24h" in data + + def test_hours_param_valid(self, client): + r = client.get("/api/v1/overlay/AAPL/headlines?hours=48") + assert r.status_code == 200 + + def test_hours_below_min_rejected(self, client): + r = client.get("/api/v1/overlay/AAPL/headlines?hours=0") + assert r.status_code == 422 + + def test_hours_above_max_rejected(self, client): + r = client.get("/api/v1/overlay/AAPL/headlines?hours=200") + assert r.status_code == 422 + + def test_symbol_uppercased(self, client): + r = client.get("/api/v1/overlay/aapl/headlines") + assert r.status_code == 200 + assert r.json()["symbol"] == "AAPL" + + def test_headline_counts_non_negative(self, client): + data = client.get("/api/v1/overlay/AAPL/headlines").json() + assert data["headline_count_6h"] >= 0 + assert data["headline_count_24h"] >= 0 + assert data["publisher_breadth_24h"] >= 0 + + +class TestYouTube: + def test_returns_200(self, client): + r = client.get("/api/v1/overlay/AAPL/youtube") + assert r.status_code == 200 + + def test_response_schema(self, client): + data = client.get("/api/v1/overlay/AAPL/youtube").json() + assert "symbol" in data + assert data["symbol"] == "AAPL" + assert "videos" in data + assert "mentions_24h" in data + assert "weighted_views_24h" in data + + def test_mentions_non_negative(self, client): + data = client.get("/api/v1/overlay/AAPL/youtube").json() + assert data["mentions_24h"] >= 0 + assert data["weighted_views_24h"] >= 0.0 + + +class TestWiki: + def test_returns_200(self, client): + r = client.get("/api/v1/overlay/AAPL/wiki") + assert r.status_code == 200 + + def test_response_schema(self, client): + data = client.get("/api/v1/overlay/AAPL/wiki").json() + assert "symbol" in data + assert data["symbol"] == "AAPL" + assert "pageviews" in data + assert isinstance(data["pageviews"], list) + + def test_days_param(self, client): + r = client.get("/api/v1/overlay/AAPL/wiki?days=7") + assert r.status_code == 200 + + def test_days_below_min_rejected(self, client): + r = client.get("/api/v1/overlay/AAPL/wiki?days=0") + assert r.status_code == 422 + + def test_days_above_max_rejected(self, client): + r = client.get("/api/v1/overlay/AAPL/wiki?days=91") + assert r.status_code == 422 + + +class TestCrowding: + def test_returns_200(self, client): + r = client.get("/api/v1/overlay/AAPL/crowding") + assert r.status_code == 200 + + def test_response_schema(self, client): + data = client.get("/api/v1/overlay/AAPL/crowding").json() + assert "symbol" in data + assert data["symbol"] == "AAPL" + assert "short_volume_ratio" in data + assert "crowding_stress_z" in data + + +class TestTrends: + def test_returns_200(self, client): + r = client.get("/api/v1/overlay/AAPL/trends") + assert r.status_code == 200 + + def test_response_schema(self, client): + data = client.get("/api/v1/overlay/AAPL/trends").json() + assert "symbol" in data + assert data["symbol"] == "AAPL" + assert "trends" in data + assert isinstance(data["trends"], list) + + def test_no_topics_returns_empty_trends(self, client): + # Symbol with no topic mapping → empty trends list + data = client.get("/api/v1/overlay/AAPL/trends").json() + # Either empty (no mappings) or populated — both are valid + assert isinstance(data["trends"], list) + + +class TestHistory: + def test_returns_200(self, client): + r = client.get("/api/v1/overlay/AAPL/history") + assert r.status_code == 200 + + def test_response_schema(self, client): + data = client.get("/api/v1/overlay/AAPL/history").json() + assert "symbol" in data + assert data["symbol"] == "AAPL" + assert "history" in data + assert "metadata" in data + + def test_days_param(self, client): + r = client.get("/api/v1/overlay/AAPL/history?days=7") + assert r.status_code == 200 + + def test_days_above_max_rejected(self, client): + r = client.get("/api/v1/overlay/AAPL/history?days=366") + assert r.status_code == 422 + + def test_metadata_includes_data_points(self, client): + data = client.get("/api/v1/overlay/AAPL/history").json() + assert "data_points" in data["metadata"] + + +class TestOverlayScore: + def test_unknown_symbol_returns_404_or_200(self, client): + # If no data exists for an obscure symbol → 404 + # If get_or_build succeeds → 200 + r = client.get("/api/v1/overlay/ZZZZZ") + assert r.status_code in (200, 404) + + def test_404_detail_message(self, client): + r = client.get("/api/v1/overlay/ZZZZZ") + if r.status_code == 404: + assert "ZZZZZ" in r.json()["detail"] + + def test_200_response_schema(self, client): + r = client.get("/api/v1/overlay/AAPL") + if r.status_code == 200: + data = r.json() + assert "symbol" in data + assert "overlay_score" in data + assert "overlay_confidence" in data + assert "features" in data + assert "source_presence" in data + + +class TestRouteOrdering: + """Verify static paths aren't shadowed by /{symbol}.""" + + def test_bulk_not_treated_as_symbol(self, client): + r = client.get("/api/v1/overlay/bulk?symbols=AAPL") + assert r.status_code in (200, 400) # Not 404 from symbol route + + def test_top_movers_not_treated_as_symbol(self, client): + r = client.get("/api/v1/overlay/top-movers") + assert r.status_code == 200 + + def test_admin_health_not_treated_as_symbol(self, client): + r = client.get("/api/v1/overlay/admin/health") + assert r.status_code == 200