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
+
+ GET /overlay/{{symbol}} - Overlay score + features + source details (0~1 score, band, hints)GET /overlay/bulk?symbols=AAPL,TSLA,NVDA - Bulk overlay scores (max 50 symbols)GET /overlay/top-movers - Symbols with highest overlay scores in last 24hGET /overlay/{{symbol}}/headlines - Recent news headlines matched to symbolGET /overlay/{{symbol}}/youtube - YouTube video mentions from investing channelsGET /overlay/{{symbol}}/wiki - Wikipedia pageview time seriesGET /overlay/{{symbol}}/crowding - FINRA short-sale crowding metricsGET /overlay/{{symbol}}/trends - Google Trends interest data (if enabled)GET /overlay/{{symbol}}/history - Historical overlay scores (backtesting)GET /overlay/admin/health - Data source health and last collection statusPOST /overlay/admin/trigger-pipeline - Manually trigger full data collection + scoringGET /overlay/admin/job-log - Pipeline job execution log
+
📊 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
- SEC EDGAR - Official company filings and ETF holdings
- - Yahoo Finance - Price data and financial news (via yfinance_plus)
+ - Yahoo Finance - Price data, financial news, and RSS headlines
- Alpaca - Market price data (OHLCV) with optional API key
- FINRA - RegSHO short sale volume data (public CDN, no API key)
+ - Wikimedia - Wikipedia daily page view counts per company
+ - YouTube Data API - Video mentions from curated investing channels
+ - Google Trends - Theme interest data (experimental, opt-in)
- NewsAPI - Professional news aggregation
- Reddit API - Social media sentiment from investing subreddits
@@ -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