From 8c9abe3e31025cbe9f63c4f9773c0783ec1986b0 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Tue, 17 Mar 2026 11:48:24 -0700 Subject: [PATCH] =?UTF-8?q?feat(attention):=20Attention=20=EC=84=9C?= =?UTF-8?q?=EB=B8=8C=EC=8B=9C=EC=8A=A4=ED=85=9C=20=EC=B6=94=EA=B0=80=20?= =?UTF-8?q?=E2=80=94=20=EC=9D=B4=EB=B2=A4=ED=8A=B8=20=EC=A4=91=EC=8B=AC=20?= =?UTF-8?q?Wikipedia/GDELT=20=EA=B4=80=EC=8B=AC=EB=8F=84=20=ED=94=BC?= =?UTF-8?q?=EC=B2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 주요 기능 - Entity resolver: ticker → canonical name → Wikipedia 매칭 - SEC company_tickers.json fallback으로 placeholder name 자동 수정 - all-caps SEC 이름 title-case 변환, "Com" suffix 처리 - Wikipedia 페이지뷰 수집 + spike_10d / zscore_20d 피처 계산 - GDELT V2 DOC API 뉴스 기사 수집 (2017-01-01 이후) ## GDELT rate limit 제약 강제 - /event/{ticker} 온디맨드 GDELT 수집 제거 (IP ban 방지) - 프로세스 전역 asyncio.Lock + 10초 최소 간격 강제 - 429 시 exponential backoff (30→60→120s) - news.gdelt_status 필드로 클라이언트에 수집 상태 명시 ('collected' | 'not_collected' | 'not_available') ## API - GET /api/v1/attention/event/{ticker}?event_date=YYYY-MM-DD - GET /api/v1/attention/entity/{ticker} - POST /api/v1/attention/admin/resolve/{ticker} - POST /api/v1/attention/admin/collect/wiki/{ticker} - POST /api/v1/attention/admin/collect/gdelt/{ticker} ← scheduler 전용 Co-Authored-By: Claude Sonnet 4.6 --- alembic/env.py | 1 + ...c3d4e5f6_add_attention_subsystem_tables.py | 107 ++++++ app/api/v1/api.py | 5 +- app/api/v1/endpoints/attention.py | 353 +++++++++++++++++ app/models/attention.py | 114 ++++++ app/schemas/attention.py | 64 ++++ app/services/attention/__init__.py | 3 + app/services/attention/entity_resolver.py | 351 +++++++++++++++++ .../attention/feature_materializer.py | 190 +++++++++ app/services/attention/gdelt_collector.py | 232 +++++++++++ app/services/attention/wiki_collector.py | 143 +++++++ app/utils/cache.py | 22 ++ tests/test_attention.py | 359 ++++++++++++++++++ 13 files changed, 1942 insertions(+), 2 deletions(-) create mode 100644 alembic/versions/a1b2c3d4e5f6_add_attention_subsystem_tables.py create mode 100644 app/api/v1/endpoints/attention.py create mode 100644 app/models/attention.py create mode 100644 app/schemas/attention.py create mode 100644 app/services/attention/__init__.py create mode 100644 app/services/attention/entity_resolver.py create mode 100644 app/services/attention/feature_materializer.py create mode 100644 app/services/attention/gdelt_collector.py create mode 100644 app/services/attention/wiki_collector.py create mode 100644 tests/test_attention.py diff --git a/alembic/env.py b/alembic/env.py index 2fb722c..64b8cfd 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -14,6 +14,7 @@ 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 +from app.models import attention # Attention subsystem (event-centric backtest-friendly) # this is the Alembic Config object config = context.config diff --git a/alembic/versions/a1b2c3d4e5f6_add_attention_subsystem_tables.py b/alembic/versions/a1b2c3d4e5f6_add_attention_subsystem_tables.py new file mode 100644 index 0000000..3660a4d --- /dev/null +++ b/alembic/versions/a1b2c3d4e5f6_add_attention_subsystem_tables.py @@ -0,0 +1,107 @@ +"""add attention subsystem tables + +Revision ID: a1b2c3d4e5f6 +Revises: 5c3d0565afcc +Create Date: 2026-03-17 + +Adds four tables for the event-centric attention subsystem: + - company_entity_map + - wiki_pageviews_daily + - gdelt_article_raw + - attention_features_daily +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "a1b2c3d4e5f6" +down_revision: Union[str, Sequence[str], None] = "5c3d0565afcc" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + conn = op.get_bind() + + # company_entity_map + if not conn.dialect.has_table(conn, "company_entity_map"): + op.create_table( + "company_entity_map", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column("ticker", sa.String(10), nullable=False), + sa.Column("canonical_name", sa.String(500), nullable=False), + sa.Column("wiki_title", sa.String(500), nullable=True), + sa.Column("gdelt_query", sa.String(1000), nullable=True), + sa.Column("aliases_json", postgresql.JSON(astext_type=sa.Text()), nullable=True), + sa.Column("resolver_confidence", sa.Float(), nullable=True), + sa.Column("is_manual_override", sa.Boolean(), nullable=True), + sa.Column("created_at", postgresql.TIMESTAMP(timezone=True), nullable=True), + sa.Column("updated_at", postgresql.TIMESTAMP(timezone=True), nullable=True), + sa.UniqueConstraint("ticker", name="company_entity_map_ticker_key"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("idx_company_entity_map_ticker", "company_entity_map", ["ticker"]) + + # wiki_pageviews_daily + if not conn.dialect.has_table(conn, "wiki_pageviews_daily"): + op.create_table( + "wiki_pageviews_daily", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column("wiki_title", sa.String(500), nullable=False), + sa.Column("date", sa.Date(), nullable=False), + sa.Column("views", sa.Integer(), nullable=False), + sa.Column("created_at", postgresql.TIMESTAMP(timezone=True), nullable=True), + sa.UniqueConstraint("wiki_title", "date", name="uq_wiki_pageviews_daily"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("idx_wiki_pageviews_daily_title_date", "wiki_pageviews_daily", ["wiki_title", "date"]) + + # gdelt_article_raw + if not conn.dialect.has_table(conn, "gdelt_article_raw"): + op.create_table( + "gdelt_article_raw", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column("url", sa.String(2000), nullable=False), + sa.Column("title", sa.Text(), nullable=True), + sa.Column("domain", sa.String(255), nullable=True), + sa.Column("published_at", postgresql.TIMESTAMP(timezone=True), nullable=True), + sa.Column("sourcecountry", sa.String(10), nullable=True), + sa.Column("matched_ticker", sa.String(10), nullable=False), + sa.Column("match_method", sa.String(50), nullable=True), + sa.Column("match_confidence", sa.Float(), nullable=True), + sa.Column("created_at", postgresql.TIMESTAMP(timezone=True), nullable=True), + sa.UniqueConstraint("url", name="gdelt_article_raw_url_key"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("idx_gdelt_article_raw_ticker", "gdelt_article_raw", ["matched_ticker"]) + op.create_index("idx_gdelt_article_raw_published_at", "gdelt_article_raw", ["published_at"]) + + # attention_features_daily + if not conn.dialect.has_table(conn, "attention_features_daily"): + op.create_table( + "attention_features_daily", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column("ticker", sa.String(10), nullable=False), + sa.Column("date", sa.Date(), nullable=False), + sa.Column("wiki_views", sa.Integer(), nullable=True), + sa.Column("wiki_spike_10d", sa.Float(), nullable=True), + sa.Column("wiki_zscore_20d", sa.Float(), nullable=True), + sa.Column("gdelt_article_count_1d", sa.Integer(), nullable=True), + sa.Column("gdelt_article_count_3d", sa.Integer(), nullable=True), + sa.Column("gdelt_unique_domains_3d", sa.Integer(), nullable=True), + sa.Column("gdelt_us_article_count_3d", sa.Integer(), nullable=True), + sa.Column("created_at", postgresql.TIMESTAMP(timezone=True), nullable=True), + sa.UniqueConstraint("ticker", "date", name="uq_attention_features_daily"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("idx_attention_features_daily_ticker_date", "attention_features_daily", ["ticker", "date"]) + + +def downgrade() -> None: + op.drop_table("attention_features_daily") + op.drop_table("gdelt_article_raw") + op.drop_table("wiki_pageviews_daily") + op.drop_table("company_entity_map") diff --git a/app/api/v1/api.py b/app/api/v1/api.py index 3af1572..c45997e 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, overlay, screener +from app.api.v1.endpoints import financial, price, catalog, health, migration, database, error_logs, request_logs, news, etf, stocks, fred, filings, alpaca, finra, overlay, screener, attention api_router = APIRouter() @@ -25,4 +25,5 @@ api_router.include_router(request_logs.router, prefix="/admin/requests", tags=[" api_router.include_router(alpaca.router, prefix="/alpaca", tags=["alpaca"]) api_router.include_router(finra.router, prefix="/finra", tags=["finra"]) api_router.include_router(overlay.router, prefix="/overlay", tags=["overlay"]) -api_router.include_router(screener.router, prefix="/screener", tags=["screener"]) \ No newline at end of file +api_router.include_router(screener.router, prefix="/screener", tags=["screener"]) +api_router.include_router(attention.router, prefix="/attention", tags=["attention"]) \ No newline at end of file diff --git a/app/api/v1/endpoints/attention.py b/app/api/v1/endpoints/attention.py new file mode 100644 index 0000000..f44ec50 --- /dev/null +++ b/app/api/v1/endpoints/attention.py @@ -0,0 +1,353 @@ +""" +Attention API endpoints — event-centric attention data for backtesting. + +Static routes are registered before parameterized routes to avoid path shadowing. + +Main endpoint: + GET /api/v1/attention/event/{ticker}?event_date=YYYY-MM-DD + +Admin endpoints: + GET /api/v1/attention/entity/{ticker} + POST /api/v1/attention/admin/resolve/{ticker} + POST /api/v1/attention/admin/collect/wiki/{ticker}?event_date=... + POST /api/v1/attention/admin/collect/gdelt/{ticker}?event_date=... +""" + +import logging +from datetime import date, timedelta + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.database import get_db +from app.models.attention import AttentionFeaturesDaily, CompanyEntityMap, GdeltArticleRaw +from app.services.attention.gdelt_collector import GDELT_EARLIEST_DATE +from app.schemas.attention import ( + CollectionStatusResponse, + EntityInfo, + EntityResolveResponse, + EventAttentionResponse, + NewsFeatures, + WikiFeatures, +) +from app.services.attention.entity_resolver import resolve_entity +from app.services.attention.feature_materializer import materialize_features +from app.services.attention.gdelt_collector import collect_gdelt_articles +from app.services.attention.wiki_collector import collect_wiki_pageviews + +logger = logging.getLogger(__name__) +router = APIRouter() + + +def _entity_to_info(entity: CompanyEntityMap) -> EntityInfo: + return EntityInfo( + ticker=entity.ticker, + canonical_name=entity.canonical_name, + wiki_title=entity.wiki_title, + gdelt_query=entity.gdelt_query, + aliases=entity.aliases_json or [], + resolver_confidence=entity.resolver_confidence or 0.0, + is_manual_override=entity.is_manual_override or False, + ) + + +def _features_to_response( + ticker: str, + event_date: date, + entity: CompanyEntityMap, + features: AttentionFeaturesDaily, + gdelt_status: str = "not_collected", +) -> EventAttentionResponse: + # Compute baseline_10d from spike (inverse: baseline = views / spike) + baseline_10d = None + if features.wiki_spike_10d and features.wiki_views and features.wiki_spike_10d > 0: + baseline_10d = features.wiki_views / features.wiki_spike_10d + + return EventAttentionResponse( + ticker=ticker, + event_date=event_date, + entity=_entity_to_info(entity), + wiki=WikiFeatures( + views=features.wiki_views, + baseline_10d=round(baseline_10d, 2) if baseline_10d else None, + spike_10d=round(features.wiki_spike_10d, 4) if features.wiki_spike_10d is not None else None, + zscore_20d=round(features.wiki_zscore_20d, 4) if features.wiki_zscore_20d is not None else None, + ), + news=NewsFeatures( + article_count_1d=features.gdelt_article_count_1d or 0, + article_count_3d=features.gdelt_article_count_3d or 0, + unique_domains_3d=features.gdelt_unique_domains_3d or 0, + us_article_count_3d=features.gdelt_us_article_count_3d or 0, + gdelt_status=gdelt_status, + ), + metadata={ + "wiki_title": entity.wiki_title, + "resolver_confidence": entity.resolver_confidence, + }, + ) + + +# =========================================================================== +# Admin / utility routes (static — must come before parameterized routes) +# =========================================================================== + + +@router.post( + "/admin/resolve/{ticker}", + response_model=EntityResolveResponse, + summary="Resolve ticker → canonical entity", + description=( + "Maps a ticker symbol to a canonical company entity by looking up the company name, " + "normalizing it, and validating against Wikipedia. Stores the result (canonical name, " + "wiki_title, gdelt_query) in `company_entity_map`.\n\n" + "Skips re-resolution if `is_manual_override` is set. " + "If the company name in the DB is a placeholder (e.g. 'AMZN Corporation'), " + "falls back to SEC company_tickers.json to fetch the real name and updates the DB." + ), + tags=["attention-admin"], +) +async def admin_resolve_entity( + ticker: str, + db: AsyncSession = Depends(get_db), +) -> EntityResolveResponse: + ticker = ticker.upper() + + # Check for existing manual override + existing_result = await db.execute( + select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker) + ) + existing = existing_result.scalars().first() + if existing and existing.is_manual_override: + return EntityResolveResponse( + ticker=ticker, + entity=_entity_to_info(existing), + status="manual_override_skipped", + message="Manual override is active — resolution skipped.", + ) + + try: + entity = await resolve_entity(db, ticker) + status = "resolved" + message = f"Entity resolved: wiki_title={entity.wiki_title!r} confidence={entity.resolver_confidence:.2f}" + except ValueError as exc: + raise HTTPException(status_code=404, detail=str(exc)) + except Exception as exc: + logger.error("Entity resolution failed for %s: %s", ticker, exc) + raise HTTPException(status_code=500, detail=f"Resolution failed: {exc}") + + return EntityResolveResponse( + ticker=ticker, + entity=_entity_to_info(entity), + status=status, + message=message, + ) + + +@router.post( + "/admin/collect/wiki/{ticker}", + response_model=CollectionStatusResponse, + summary="Collect Wikipedia pageviews for an event date", + description=( + "Fetches daily Wikipedia pageview counts for the ticker's canonical wiki_title, " + "covering `event_date` and enough lookback days (≥20) to compute spike and z-score. " + "Safe to call on-demand — Wikipedia API has no meaningful rate limit for this use.\n\n" + "Requires entity resolution to have been run first (`wiki_title` must be set)." + ), + tags=["attention-admin"], +) +async def admin_collect_wiki( + ticker: str, + event_date: date = Query(..., description="Event date in YYYY-MM-DD format"), + db: AsyncSession = Depends(get_db), +) -> CollectionStatusResponse: + ticker = ticker.upper() + try: + count = await collect_wiki_pageviews(db, ticker, event_date) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) + except Exception as exc: + logger.error("Wiki collection failed for %s: %s", ticker, exc) + raise HTTPException(status_code=500, detail=f"Wiki collection failed: {exc}") + + return CollectionStatusResponse( + ticker=ticker, + source="wiki", + records_collected=count, + date_range={"event_date": str(event_date)}, + status="success", + ) + + +@router.post( + "/admin/collect/gdelt/{ticker}", + response_model=CollectionStatusResponse, + summary="Collect GDELT news articles for an event date", + description=( + "Fetches news articles from GDELT V2 DOC API for the window `event_date ± 1 day`.\n\n" + "**Coverage**: 2017-01-01 onwards. Requests for earlier dates return 0 immediately.\n\n" + "**Rate limit**: GDELT enforces a global per-IP quota. This endpoint is protected by " + "a process-wide lock (10s minimum interval) and retries with exponential backoff " + "(30s → 60s → 120s) on 429 responses.\n\n" + "⚠️ **Call this endpoint from a scheduler only** — never trigger it in response to " + "user requests. Concurrent or rapid calls will exhaust the IP quota and cause " + "temporary bans. The main `/event/{ticker}` endpoint intentionally does NOT collect " + "GDELT on-demand for this reason." + ), + tags=["attention-admin"], +) +async def admin_collect_gdelt( + ticker: str, + event_date: date = Query(..., description="Event date in YYYY-MM-DD format"), + db: AsyncSession = Depends(get_db), +) -> CollectionStatusResponse: + ticker = ticker.upper() + try: + count = await collect_gdelt_articles(db, ticker, event_date) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) + except Exception as exc: + logger.error("GDELT collection failed for %s: %s", ticker, exc) + raise HTTPException(status_code=500, detail=f"GDELT collection failed: {exc}") + + return CollectionStatusResponse( + ticker=ticker, + source="gdelt", + records_collected=count, + date_range={"event_date": str(event_date)}, + status="success", + ) + + +# =========================================================================== +# Parameterized entity route (before /event/ to avoid shadowing) +# =========================================================================== + + +@router.get( + "/entity/{ticker}", + response_model=EntityResolveResponse, + summary="Get entity mapping for a ticker", + description=( + "Returns the stored entity mapping for a ticker: canonical name, Wikipedia title, " + "GDELT query string, and resolver confidence score.\n\n" + "Returns 404 if no mapping exists — run `POST /admin/resolve/{ticker}` first." + ), +) +async def get_entity( + ticker: str, + db: AsyncSession = Depends(get_db), +) -> EntityResolveResponse: + ticker = ticker.upper() + result = await db.execute( + select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker) + ) + entity = result.scalars().first() + if not entity: + raise HTTPException( + status_code=404, + detail=f"No entity mapping found for {ticker}. POST /admin/resolve/{ticker} to create one.", + ) + return EntityResolveResponse( + ticker=ticker, + entity=_entity_to_info(entity), + status="exists", + message="Entity mapping retrieved from database.", + ) + + +# =========================================================================== +# Main event attention endpoint +# =========================================================================== + + +@router.get( + "/event/{ticker}", + response_model=EventAttentionResponse, + summary="Get attention features for a ticker on an event date", + description=( + "Returns Wikipedia pageview spike/z-score and GDELT news article counts " + "for the given ticker centered on `event_date`.\n\n" + "**Wikipedia data** is collected on-demand if not in DB. " + "**GDELT data** is never collected on-demand — it must be pre-populated via " + "`POST /admin/collect/gdelt/{ticker}` (scheduler). " + "Check `news.gdelt_status` in the response to understand data availability:\n\n" + "- `collected` — GDELT was collected; counts are accurate (0 means genuinely no articles)\n" + "- `not_collected` — scheduler has not run for this date yet\n" + "- `not_available` — event date is before GDELT V2 coverage (2017-01-01)\n\n" + "If no entity mapping exists, resolution runs automatically before collection." + ), +) +async def get_event_attention( + ticker: str, + event_date: date = Query(..., description="Event date in YYYY-MM-DD format"), + db: AsyncSession = Depends(get_db), +) -> EventAttentionResponse: + ticker = ticker.upper() + + # 1. Get or resolve entity + entity_result = await db.execute( + select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker) + ) + entity = entity_result.scalars().first() + + if not entity: + logger.info("No entity mapping for %s — auto-resolving", ticker) + try: + entity = await resolve_entity(db, ticker) + except ValueError as exc: + raise HTTPException(status_code=404, detail=str(exc)) + except Exception as exc: + logger.error("Auto entity resolution failed for %s: %s", ticker, exc) + raise HTTPException(status_code=500, detail=f"Entity resolution failed: {exc}") + + # 2. Check if features already exist in DB + features_result = await db.execute( + select(AttentionFeaturesDaily).where( + AttentionFeaturesDaily.ticker == ticker, + AttentionFeaturesDaily.date == event_date, + ) + ) + features = features_result.scalars().first() + + if features is None: + # 3. On-demand collection + materialization + # NOTE: GDELT is intentionally excluded here — it must be collected via + # the scheduler (POST /admin/collect/gdelt/{ticker}) to avoid IP rate bans. + # This endpoint only collects Wikipedia data on-demand. + logger.info("No features for %s on %s — collecting wiki on-demand", ticker, event_date) + + if entity.wiki_title: + try: + await collect_wiki_pageviews(db, ticker, event_date) + except Exception as exc: + logger.warning("Wiki collection failed for %s: %s", ticker, exc) + + try: + features = await materialize_features(db, ticker, event_date) + except Exception as exc: + logger.error("Feature materialization failed for %s: %s", ticker, exc) + raise HTTPException(status_code=500, detail=f"Feature materialization failed: {exc}") + + # Re-fetch entity in case it was updated during resolution + entity_result2 = await db.execute( + select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker) + ) + entity = entity_result2.scalars().first() + + # Determine GDELT collection status for the client + if event_date < GDELT_EARLIEST_DATE: + gdelt_status = "not_available" + else: + window_start = event_date - timedelta(days=1) + window_end = event_date + timedelta(days=1) + collected_count = await db.execute( + select(func.count()).where( + GdeltArticleRaw.matched_ticker == ticker, + func.date(GdeltArticleRaw.published_at) >= window_start, + func.date(GdeltArticleRaw.published_at) <= window_end, + ) + ) + gdelt_status = "collected" if (collected_count.scalar() or 0) > 0 else "not_collected" + + return _features_to_response(ticker, event_date, entity, features, gdelt_status) diff --git a/app/models/attention.py b/app/models/attention.py new file mode 100644 index 0000000..4f32893 --- /dev/null +++ b/app/models/attention.py @@ -0,0 +1,114 @@ +""" +Attention subsystem models — event-centric attention data for backtest-friendly queries. + +Completely separate from the overlay subsystem (which is real-time monitoring). +This subsystem is designed for fithia2 backtester to query attention data +relative to events (earnings, etc.). +""" + +from sqlalchemy import Column, String, Float, Boolean, Integer, Text, Date, Index, UniqueConstraint, JSON +from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP +from datetime import datetime, timezone +import uuid + +from app.core.database import Base + + +class CompanyEntityMap(Base): + """Ticker → canonical company entity mapping with Wikipedia and GDELT resolution.""" + + __tablename__ = "company_entity_map" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + ticker = Column(String(10), unique=True, nullable=False, index=True) + canonical_name = Column(String(500), nullable=False) + wiki_title = Column(String(500), nullable=True) + gdelt_query = Column(String(1000), nullable=True) + aliases_json = Column(JSON, default=list) + resolver_confidence = Column(Float, default=0.0) + is_manual_override = Column(Boolean, default=False) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + updated_at = Column( + TIMESTAMP(timezone=True), + default=lambda: datetime.now(timezone.utc), + onupdate=lambda: datetime.now(timezone.utc), + ) + + __table_args__ = ( + Index("idx_company_entity_map_ticker", "ticker"), + ) + + +class WikiPageviewsDaily(Base): + """Daily Wikipedia pageview counts per article title.""" + + __tablename__ = "wiki_pageviews_daily" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + wiki_title = Column(String(500), nullable=False) + date = Column(Date, nullable=False) + views = Column(Integer, nullable=False) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + UniqueConstraint("wiki_title", "date", name="uq_wiki_pageviews_daily"), + Index("idx_wiki_pageviews_daily_title_date", "wiki_title", "date"), + ) + + +class GdeltArticleRaw(Base): + """Raw GDELT article records matched to a ticker.""" + + __tablename__ = "gdelt_article_raw" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + url = Column(String(2000), unique=True, nullable=False) + title = Column(Text, nullable=True) + domain = Column(String(255), nullable=True) + published_at = Column(TIMESTAMP(timezone=True), nullable=True) + sourcecountry = Column(String(10), nullable=True) + matched_ticker = Column(String(10), nullable=False, index=True) + match_method = Column(String(50), nullable=True) + match_confidence = Column(Float, default=1.0) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + Index("idx_gdelt_article_raw_ticker", "matched_ticker"), + Index("idx_gdelt_article_raw_published_at", "published_at"), + ) + + +class AttentionFeaturesDaily(Base): + """Derived daily attention features per ticker — materialized from wiki + gdelt raw data.""" + + __tablename__ = "attention_features_daily" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + ticker = Column(String(10), nullable=False) + date = Column(Date, nullable=False) + + # Wikipedia features + wiki_views = Column(Integer, nullable=True) + wiki_spike_10d = Column(Float, nullable=True) + wiki_zscore_20d = Column(Float, nullable=True) + + # GDELT / news features + gdelt_article_count_1d = Column(Integer, default=0) + gdelt_article_count_3d = Column(Integer, default=0) + gdelt_unique_domains_3d = Column(Integer, default=0) + gdelt_us_article_count_3d = Column(Integer, default=0) + + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + UniqueConstraint("ticker", "date", name="uq_attention_features_daily"), + Index("idx_attention_features_daily_ticker_date", "ticker", "date"), + ) diff --git a/app/schemas/attention.py b/app/schemas/attention.py new file mode 100644 index 0000000..b80a8e2 --- /dev/null +++ b/app/schemas/attention.py @@ -0,0 +1,64 @@ +""" +Pydantic v2 schemas for the Attention subsystem API. +""" + +from datetime import date, datetime +from typing import Any, Dict, List, Optional +from pydantic import BaseModel, Field + + +class EntityInfo(BaseModel): + ticker: str + canonical_name: str = Field(description="Normalized company name with legal suffixes stripped (e.g. 'Apple')") + wiki_title: Optional[str] = Field(default=None, description="Matched Wikipedia article title; null if unresolved") + gdelt_query: Optional[str] = Field(default=None, description="GDELT DOC API query string (quoted OR phrases)") + aliases: List[str] = Field(default_factory=list, description="Intermediate forms used during name normalization") + resolver_confidence: float = Field(default=0.0, description="Wikipedia match confidence [0, 1]") + is_manual_override: bool = Field(default=False, description="If true, automated re-resolution is skipped") + + +class WikiFeatures(BaseModel): + views: Optional[int] = Field(default=None, description="Wikipedia pageviews on the event date") + baseline_10d: Optional[float] = Field(default=None, description="Median pageviews over the prior 10 days") + spike_10d: Optional[float] = Field(default=None, description="views / baseline_10d; >1 means above-average attention") + zscore_20d: Optional[float] = Field(default=None, description="Z-score vs prior 20-day mean/stdev; null if stdev=0") + + +class NewsFeatures(BaseModel): + article_count_1d: int = Field(default=0, description="GDELT articles published on the event date") + article_count_3d: int = Field(default=0, description="GDELT articles in the event_date ± 1 day window") + unique_domains_3d: int = Field(default=0, description="Distinct publisher domains in the 3-day window") + us_article_count_3d: int = Field(default=0, description="US-sourced articles in the 3-day window") + gdelt_status: str = Field( + default="not_collected", + description=( + "GDELT data availability for this event date. " + "'collected' — scheduler has run; counts are accurate (0 means genuinely no articles). " + "'not_collected' — scheduler has not run yet; POST /admin/collect/gdelt/{ticker}?event_date=... to populate. " + "'not_available' — event date is before GDELT V2 coverage start (2017-01-01)." + ), + ) + + +class EventAttentionResponse(BaseModel): + ticker: str + event_date: date + entity: EntityInfo + wiki: WikiFeatures + news: NewsFeatures + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class EntityResolveResponse(BaseModel): + ticker: str + entity: EntityInfo + status: str # "resolved", "already_exists", "failed", "manual_override_skipped" + message: str + + +class CollectionStatusResponse(BaseModel): + ticker: str + source: str # "wiki" or "gdelt" + records_collected: int + date_range: Dict[str, Any] = Field(default_factory=dict) + status: str diff --git a/app/services/attention/__init__.py b/app/services/attention/__init__.py new file mode 100644 index 0000000..baacbe1 --- /dev/null +++ b/app/services/attention/__init__.py @@ -0,0 +1,3 @@ +""" +Attention subsystem services — event-centric attention data collection and feature materialization. +""" diff --git a/app/services/attention/entity_resolver.py b/app/services/attention/entity_resolver.py new file mode 100644 index 0000000..bbe83a5 --- /dev/null +++ b/app/services/attention/entity_resolver.py @@ -0,0 +1,351 @@ +""" +Entity Resolver — maps ticker symbols to canonical company entities. + +Resolution pipeline: +1. Look up company name from `companies` table +2. Normalize name (strip legal suffixes) → canonical_name + aliases +3. Validate against Wikipedia Search API → best wiki_title + confidence +4. Build GDELT query from canonical_name + aliases +5. Upsert into company_entity_map (respects is_manual_override flag) +""" + +import logging +import re +from typing import Optional + +import httpx +from sqlalchemy import select, update +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.attention import CompanyEntityMap +from app.models.financial import Company + +logger = logging.getLogger(__name__) + +# Legal suffixes to strip for canonical name derivation +_SUFFIX_PATTERN = re.compile( + r",?\s+\b(Inc\.?|Corp\.?|Corporation|Holdings?|Ltd\.?|Limited|LLC|L\.L\.C\.|" + r"Group|Co\.?|Company|Technologies|Technology|International|Industries|" + r"Pharmaceuticals?|Therapeutics?|Sciences?|Bancorp|Financial|Holding|" + r"Acquisition|Acquisitions|Capital|Partners|Trust|" + r"Com)\s*$", # "Com" catches SEC-style ".com" artifacts (e.g. "AMAZON COM") + re.IGNORECASE, +) + +_COMPANY_KEYWORDS = { + "company", "corporation", "inc", "corp", "ltd", "llc", "holdings", + "stock", "shares", "nasdaq", "nyse", "ticker", "finance", "financial", + "investor", "business", "enterprise", "industries", +} + + +def _normalize_name(raw_name: str) -> tuple[str, list[str]]: + """Return (canonical_name, aliases). + + Strips common legal suffixes iteratively to produce a short canonical name. + Preserves the original and intermediate forms as aliases. + """ + aliases = [] + current = raw_name.strip() + + for _ in range(5): # max 5 iterations to avoid infinite loops + stripped = _SUFFIX_PATTERN.sub("", current).strip().rstrip(",").strip() + if stripped == current or not stripped: + break + aliases.append(current) + current = stripped + + canonical = current + # Also add the fully original name if not already captured + if raw_name.strip() != canonical and raw_name.strip() not in aliases: + aliases.insert(0, raw_name.strip()) + + # Deduplicate while preserving order + seen = set() + unique_aliases = [] + for a in aliases: + if a not in seen and a != canonical: + seen.add(a) + unique_aliases.append(a) + + return canonical, unique_aliases + + +_WIKI_HEADERS = { + "User-Agent": "StockOracleBot/1.0 (attention-subsystem; contact@stockoracle.internal)" +} + +_SEC_HEADERS = { + "User-Agent": "Stock Oracle contact@stockoracle.internal", + "Accept": "application/json", +} + +# Module-level cache: populated on first SEC fetch, maps ticker.upper() → company title +_SEC_TICKER_MAP: dict[str, str] = {} + + +def _is_placeholder_name(ticker: str, canonical_name: str) -> bool: + """Detect if the company name is just the ticker symbol repeated.""" + return canonical_name.upper() == ticker.upper() + + +async def _fetch_sec_company_name(ticker: str) -> Optional[str]: + """Fetch real company name from SEC company_tickers.json. + + Caches the full ticker→name mapping on first call (~13K entries). + Returns the SEC `title` field for the given ticker, or None if not found. + """ + global _SEC_TICKER_MAP + if not _SEC_TICKER_MAP: + url = "https://www.sec.gov/files/company_tickers.json" + try: + async with httpx.AsyncClient(timeout=15.0, headers=_SEC_HEADERS) as client: + resp = await client.get(url) + resp.raise_for_status() + data = resp.json() + _SEC_TICKER_MAP = { + v["ticker"].upper(): v["title"] + for v in data.values() + if "ticker" in v and "title" in v + } + logger.info("Loaded %d tickers from SEC company_tickers.json", len(_SEC_TICKER_MAP)) + except Exception as exc: + logger.warning("Failed to fetch SEC company_tickers.json: %s", exc) + return None + + title = _SEC_TICKER_MAP.get(ticker.upper()) + if title and title == title.upper() and not title.isnumeric(): + # SEC stores many names in ALL-CAPS (e.g. "AMAZON COM INC") — title-case them + title = title.title() + return title + + +async def _search_wikipedia(query: str) -> list[dict]: + """Call Wikipedia Search API and return raw results list.""" + url = "https://en.wikipedia.org/w/api.php" + params = { + "action": "query", + "list": "search", + "srsearch": query, + "srlimit": 5, + "format": "json", + } + try: + async with httpx.AsyncClient(timeout=10.0, headers=_WIKI_HEADERS) as client: + resp = await client.get(url, params=params) + resp.raise_for_status() + data = resp.json() + return data.get("query", {}).get("search", []) + except Exception as exc: + logger.warning("Wikipedia search failed for %r: %s", query, exc) + return [] + + +def _score_wiki_result(result: dict, canonical_name: str, aliases: list[str]) -> float: + """Score a Wikipedia search result [0, 1] for company relevance. + + Scoring strategy: + - Direct title match (title == canonical or alias): 0.9+ + - Title starts with canonical: 0.7+ + - Title contains canonical: 0.5 base + - No canonical in title: 0.0 + - Non-company signals (album, film, etc.): capped at 0.1 + """ + raw_title = result.get("title", "") + title = raw_title.lower() + snippet = result.get("snippet", "").lower() + combined = title + " " + snippet + + # Penalize obvious non-company pages immediately + non_company_signals = ["album", "film", "television", "televisión", "song", "band", + "musician", "footballer", "athlete", "politician", "novel", + "book", "discography", "filmography", "radio station", + "broadcasting", "tv channel", "telev"] + if any(sig in combined for sig in non_company_signals): + return 0.1 + + # Require snippet to have at least one finance keyword for non-exact-match titles + finance_signals = ["company", "corporation", "stock", "nasdaq", "nyse", "shares", + "investor", "business", "enterprise", "holdings", "inc."] + if not any(sig in combined for sig in finance_signals): + return 0.15 + + canonical_lower = canonical_name.lower() + all_names = [canonical_name] + aliases + all_names_lower = [n.lower() for n in all_names] + + # Check for exact title match (e.g. "Apple Inc." == alias "Apple Inc.") + raw_title_stripped = raw_title.strip() + for name in all_names: + if raw_title_stripped.lower() == name.lower(): + return 0.95 # exact match + + # Check if title starts with canonical name + title_starts_with_canonical = title.startswith(canonical_lower) + title_contains_canonical = canonical_lower in title + + # Also check aliases + title_starts_with_alias = any(title.startswith(n) for n in all_names_lower) + title_contains_alias = any(n in title for n in all_names_lower) + + if not (title_contains_canonical or title_contains_alias): + return 0.0 + + # Check for pages that are ABOUT the company (vs. lists, histories, etc.) + # Penalize titles that suggest subsidiary/list/history pages + secondary_patterns = ["list of", "history of", "acquisition", "merger", + "subsidiary", "criticism", "controversy"] + if any(pat in title for pat in secondary_patterns): + return 0.25 + + # Base score + if title_starts_with_canonical or title_starts_with_alias: + base = 0.7 + elif title_contains_canonical or title_contains_alias: + base = 0.5 + else: + base = 0.0 + + # Reward company/finance keywords in title or snippet + keyword_hits = sum(1 for kw in _COMPANY_KEYWORDS if kw in combined) + keyword_score = min(keyword_hits / 3, 1.0) + + return min(base + keyword_score * 0.3, 1.0) + + +async def _resolve_wiki( + canonical_name: str, + aliases: list[str], +) -> tuple[Optional[str], float]: + """Return (wiki_title, confidence). wiki_title is None if confidence < 0.5.""" + # Try multiple search strategies, prefer more specific queries first + search_queries = [] + # Try full original name first (e.g. "Apple Inc.") + for alias in aliases[:2]: + search_queries.append(alias) + # Then canonical + search_queries.append(f"{canonical_name} company") + + best_title: Optional[str] = None + best_score = 0.0 + + for query in search_queries: + results = await _search_wikipedia(query) + for result in results: + score = _score_wiki_result(result, canonical_name, aliases) + if score > best_score: + best_score = score + best_title = result["title"] + if best_score >= 0.85: + break # good enough — no need to try more queries + + if best_score < 0.5: + return None, best_score + + return best_title, best_score + + +def _build_gdelt_query(canonical_name: str, aliases: list[str]) -> str: + """Build a GDELT DOC API query string using quoted phrase OR logic.""" + terms = [canonical_name] + [a for a in aliases if a != canonical_name] + quoted = [f'"{t}"' for t in terms[:4]] # cap at 4 terms + return " OR ".join(quoted) + + +async def resolve_entity( + db: AsyncSession, + ticker: str, +) -> CompanyEntityMap: + """Resolve ticker → entity and upsert into company_entity_map. + + Returns the upserted CompanyEntityMap row. + Raises ValueError if the ticker cannot be resolved. + """ + ticker = ticker.upper() + + # Check if manual override already exists — skip re-resolution + existing_result = await db.execute( + select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker) + ) + existing = existing_result.scalars().first() + if existing and existing.is_manual_override: + logger.info("Skipping resolution for %s: manual override active", ticker) + return existing + + # Look up company name from companies table + company_result = await db.execute( + select(Company).where(Company.ticker == ticker) + ) + company = company_result.scalars().first() + if not company: + raise ValueError(f"Ticker {ticker!r} not found in companies table") + + raw_name = company.name + canonical_name, aliases = _normalize_name(raw_name) + + # Detect placeholder names like "AMZN Corporation" → canonical becomes "AMZN" + if _is_placeholder_name(ticker, canonical_name): + logger.info( + "Detected placeholder name for %s (%r) — querying SEC for real name", + ticker, raw_name, + ) + sec_name = await _fetch_sec_company_name(ticker) + if sec_name: + logger.info("SEC fallback for %s: %r → %r", ticker, raw_name, sec_name) + # Update the DB so next resolve skips this path + await db.execute( + update(Company).where(Company.ticker == ticker).values(name=sec_name) + ) + canonical_name, aliases = _normalize_name(sec_name) + else: + logger.warning("SEC fallback found no name for %s, proceeding with placeholder", ticker) + + logger.info("Resolving entity for %s: canonical=%r aliases=%r", ticker, canonical_name, aliases) + + wiki_title, confidence = await _resolve_wiki(canonical_name, aliases) + gdelt_query = _build_gdelt_query(canonical_name, aliases) + + logger.info( + "Entity resolved: %s → wiki_title=%r confidence=%.2f", + ticker, wiki_title, confidence, + ) + + # Upsert into company_entity_map + stmt = ( + pg_insert(CompanyEntityMap) + .values( + ticker=ticker, + canonical_name=canonical_name, + wiki_title=wiki_title, + gdelt_query=gdelt_query, + aliases_json=aliases, + resolver_confidence=confidence, + is_manual_override=False, + ) + .on_conflict_do_update( + index_elements=["ticker"], + set_=dict( + canonical_name=canonical_name, + wiki_title=wiki_title, + gdelt_query=gdelt_query, + aliases_json=aliases, + resolver_confidence=confidence, + ), + where=CompanyEntityMap.is_manual_override == False, # noqa: E712 + ) + .returning(CompanyEntityMap) + ) + + result = await db.execute(stmt) + await db.commit() + + row = result.scalars().first() + if row is None: + # Manual override prevented update — return existing + existing_result2 = await db.execute( + select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker) + ) + row = existing_result2.scalars().first() + + return row diff --git a/app/services/attention/feature_materializer.py b/app/services/attention/feature_materializer.py new file mode 100644 index 0000000..a0d1fb1 --- /dev/null +++ b/app/services/attention/feature_materializer.py @@ -0,0 +1,190 @@ +""" +Feature Materializer — computes derived attention features from wiki + gdelt raw data. + +Materializes into attention_features_daily for the given ticker + event_date. +""" + +import logging +import statistics +from datetime import date, timedelta +from typing import Optional + +from sqlalchemy import select, func +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.attention import ( + AttentionFeaturesDaily, + CompanyEntityMap, + GdeltArticleRaw, + WikiPageviewsDaily, +) + +logger = logging.getLogger(__name__) + + +async def _get_wiki_views_series( + db: AsyncSession, + wiki_title: str, + target_date: date, + lookback: int, +) -> list[int]: + """Return list of view counts for [target_date - lookback, target_date - 1].""" + start = target_date - timedelta(days=lookback) + end = target_date - timedelta(days=1) + result = await db.execute( + select(WikiPageviewsDaily.views) + .where( + WikiPageviewsDaily.wiki_title == wiki_title, + WikiPageviewsDaily.date >= start, + WikiPageviewsDaily.date <= end, + ) + .order_by(WikiPageviewsDaily.date) + ) + return list(result.scalars().all()) + + +async def materialize_features( + db: AsyncSession, + ticker: str, + event_date: date, +) -> AttentionFeaturesDaily: + """Compute and upsert daily attention features for ticker on event_date. + + Returns the upserted AttentionFeaturesDaily row. + """ + ticker = ticker.upper() + + entity_result = await db.execute( + select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker) + ) + entity = entity_result.scalars().first() + wiki_title = entity.wiki_title if entity else None + + # --- Wiki features --- + wiki_views: Optional[int] = None + wiki_spike_10d: Optional[float] = None + wiki_zscore_20d: Optional[float] = None + + if wiki_title: + # Fetch today's views + today_result = await db.execute( + select(WikiPageviewsDaily.views).where( + WikiPageviewsDaily.wiki_title == wiki_title, + WikiPageviewsDaily.date == event_date, + ) + ) + wiki_views = today_result.scalars().first() + + if wiki_views is not None: + # Spike: views / median of previous 10 days + prev_10 = await _get_wiki_views_series(db, wiki_title, event_date, 10) + if prev_10: + median_10 = statistics.median(prev_10) + wiki_spike_10d = wiki_views / median_10 if median_10 > 0 else None + + # Z-score over previous 20 days + prev_20 = await _get_wiki_views_series(db, wiki_title, event_date, 20) + if len(prev_20) >= 3: + mean_20 = statistics.mean(prev_20) + stdev_20 = statistics.stdev(prev_20) + if stdev_20 > 0: + wiki_zscore_20d = (wiki_views - mean_20) / stdev_20 + + # --- GDELT features --- + # 1-day: articles published on event_date only + day_start = event_date + day_end = event_date + + count_1d_result = await db.execute( + select(func.count()).where( + GdeltArticleRaw.matched_ticker == ticker, + func.date(GdeltArticleRaw.published_at) >= day_start, + func.date(GdeltArticleRaw.published_at) <= day_end, + ) + ) + gdelt_article_count_1d = count_1d_result.scalar() or 0 + + # 3-day window: event_date ± 1 + window_start = event_date - timedelta(days=1) + window_end = event_date + timedelta(days=1) + + count_3d_result = await db.execute( + select(func.count()).where( + GdeltArticleRaw.matched_ticker == ticker, + func.date(GdeltArticleRaw.published_at) >= window_start, + func.date(GdeltArticleRaw.published_at) <= window_end, + ) + ) + gdelt_article_count_3d = count_3d_result.scalar() or 0 + + # Unique domains in 3-day window + domains_result = await db.execute( + select(func.count(func.distinct(GdeltArticleRaw.domain))).where( + GdeltArticleRaw.matched_ticker == ticker, + GdeltArticleRaw.domain.isnot(None), + func.date(GdeltArticleRaw.published_at) >= window_start, + func.date(GdeltArticleRaw.published_at) <= window_end, + ) + ) + gdelt_unique_domains_3d = domains_result.scalar() or 0 + + # US articles in 3-day window + us_result = await db.execute( + select(func.count()).where( + GdeltArticleRaw.matched_ticker == ticker, + GdeltArticleRaw.sourcecountry == "US", + func.date(GdeltArticleRaw.published_at) >= window_start, + func.date(GdeltArticleRaw.published_at) <= window_end, + ) + ) + gdelt_us_article_count_3d = us_result.scalar() or 0 + + # Upsert + stmt = ( + pg_insert(AttentionFeaturesDaily) + .values( + ticker=ticker, + date=event_date, + wiki_views=wiki_views, + wiki_spike_10d=wiki_spike_10d, + wiki_zscore_20d=wiki_zscore_20d, + gdelt_article_count_1d=gdelt_article_count_1d, + gdelt_article_count_3d=gdelt_article_count_3d, + gdelt_unique_domains_3d=gdelt_unique_domains_3d, + gdelt_us_article_count_3d=gdelt_us_article_count_3d, + ) + .on_conflict_do_update( + constraint="uq_attention_features_daily", + set_=dict( + wiki_views=wiki_views, + wiki_spike_10d=wiki_spike_10d, + wiki_zscore_20d=wiki_zscore_20d, + gdelt_article_count_1d=gdelt_article_count_1d, + gdelt_article_count_3d=gdelt_article_count_3d, + gdelt_unique_domains_3d=gdelt_unique_domains_3d, + gdelt_us_article_count_3d=gdelt_us_article_count_3d, + ), + ) + .returning(AttentionFeaturesDaily) + ) + + result = await db.execute(stmt) + await db.commit() + + row = result.scalars().first() + if row is None: + # Fetch after upsert + row_result = await db.execute( + select(AttentionFeaturesDaily).where( + AttentionFeaturesDaily.ticker == ticker, + AttentionFeaturesDaily.date == event_date, + ) + ) + row = row_result.scalars().first() + + logger.info( + "Materialized features for %s on %s: wiki=%s gdelt_1d=%d gdelt_3d=%d", + ticker, event_date, wiki_views, gdelt_article_count_1d, gdelt_article_count_3d, + ) + return row diff --git a/app/services/attention/gdelt_collector.py b/app/services/attention/gdelt_collector.py new file mode 100644 index 0000000..ae5b3e7 --- /dev/null +++ b/app/services/attention/gdelt_collector.py @@ -0,0 +1,232 @@ +""" +GDELT Collector — event-centric GDELT news article collection. + +Collects articles for window: event_date - 1d to event_date + 1d. +Deduplicates by URL before inserting. + +Historical coverage: GDELT V2 DOC API covers from 2017-01-01 onwards. + +Rate limit: GDELT uses a global per-IP quota shared across all users worldwide. +- _MIN_INTERVAL: process-wide minimum gap between calls (enforced via asyncio.Lock) +- On 429: exponential backoff — 30s, 60s, 120s (up to _MAX_RETRIES attempts) +- GDELT collection must NEVER be triggered on-demand from user-facing endpoints. + Use the scheduler (POST /admin/collect/gdelt/{ticker}) only. +""" + +import asyncio +import logging +from datetime import date, datetime, timedelta, timezone +from typing import Optional +from urllib.parse import urlparse + +import httpx +from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.attention import CompanyEntityMap, GdeltArticleRaw + +logger = logging.getLogger(__name__) + +_GDELT_DOC_API = "https://api.gdeltproject.org/api/v2/doc/doc" +_MAX_RECORDS = 250 +_MIN_INTERVAL = 10.0 # minimum seconds between any two GDELT calls (process-wide) +_RETRY_BACKOFF = [30, 60, 120] # wait times on successive 429s +_MAX_RETRIES = len(_RETRY_BACKOFF) + +# Earliest date covered by GDELT V2 DOC API +GDELT_EARLIEST_DATE = date(2017, 1, 1) + +# Process-wide rate guard — enforced inside collect_gdelt_articles() +_gdelt_lock = asyncio.Lock() +_last_gdelt_call_ts: float = 0.0 + + +def _date_to_gdelt_ts(d: date, end_of_day: bool = False) -> str: + """Format date as GDELT datetime string YYYYMMDDHHMMSS.""" + if end_of_day: + return d.strftime("%Y%m%d235959") + return d.strftime("%Y%m%d000000") + + +async def _do_gdelt_request(params: dict, ticker: str) -> Optional[dict]: + """Fire the GDELT HTTP request with retry/backoff on 429. Returns parsed JSON or None.""" + for attempt in range(_MAX_RETRIES + 1): + try: + async with httpx.AsyncClient(timeout=30.0) as client: + resp = await client.get(_GDELT_DOC_API, params=params) + + if resp.status_code == 429: + if attempt < _MAX_RETRIES: + wait = _RETRY_BACKOFF[attempt] + logger.warning( + "GDELT rate limit (429) for %s — retry %d/%d in %ds", + ticker, attempt + 1, _MAX_RETRIES, wait, + ) + await asyncio.sleep(wait) + continue + else: + logger.error( + "GDELT rate limit persists after %d retries for %s — giving up", + _MAX_RETRIES, ticker, + ) + return None + + resp.raise_for_status() + + if not resp.content.strip(): + logger.warning("GDELT returned empty response for %s", ticker) + return None + try: + return resp.json() + except Exception: + logger.warning("GDELT returned non-JSON for %s: %r", ticker, resp.text[:200]) + return None + + except httpx.HTTPStatusError as exc: + logger.error("GDELT API HTTP error %s: %s", exc.response.status_code, exc) + raise + except Exception as exc: + logger.error("GDELT API request failed: %s", exc) + raise + + return None + + +def _extract_domain(url: str) -> Optional[str]: + try: + return urlparse(url).netloc or None + except Exception: + return None + + +async def collect_gdelt_articles( + db: AsyncSession, + ticker: str, + event_date: date, +) -> int: + """Collect GDELT articles for event_date ± 1 day. + + Returns number of new records inserted. + Raises ValueError if no entity mapping or gdelt_query exists. + """ + global _last_gdelt_call_ts + + ticker = ticker.upper() + + if event_date < GDELT_EARLIEST_DATE: + logger.warning( + "GDELT data not available before %s (requested %s) — skipping %s", + GDELT_EARLIEST_DATE, event_date, ticker, + ) + return 0 + + result = await db.execute( + select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker) + ) + entity = result.scalars().first() + if not entity: + raise ValueError(f"No entity mapping found for {ticker!r}. Run entity resolution first.") + if not entity.gdelt_query: + raise ValueError(f"Entity {ticker!r} has no gdelt_query.") + + gdelt_query = entity.gdelt_query + start_date = event_date - timedelta(days=1) + end_date = event_date + timedelta(days=1) + + start_ts = _date_to_gdelt_ts(start_date, end_of_day=False) + end_ts = _date_to_gdelt_ts(end_date, end_of_day=True) + + logger.info( + "Fetching GDELT articles for %s query=%r window=%s→%s", + ticker, gdelt_query, start_ts, end_ts, + ) + + params = { + "query": gdelt_query, + "mode": "ArtList", + "maxrecords": str(_MAX_RECORDS), + "format": "json", + "startdatetime": start_ts, + "enddatetime": end_ts, + } + + # Global process-wide rate guard: at most 1 call per _MIN_INTERVAL seconds. + # Uses a lock so concurrent callers queue up rather than both firing. + async with _gdelt_lock: + import time as _time + elapsed = _time.monotonic() - _last_gdelt_call_ts + wait = max(0.0, _MIN_INTERVAL - elapsed) + if wait > 0: + logger.debug("GDELT rate guard: sleeping %.1fs before request", wait) + await asyncio.sleep(wait) + + data = None + try: + data = await _do_gdelt_request(params, ticker) + finally: + _last_gdelt_call_ts = _time.monotonic() + + if data is None: + return 0 + + articles = data.get("articles", []) if isinstance(data, dict) else [] + if not articles: + logger.info("No GDELT articles returned for %s", ticker) + return 0 + + # Collect existing URLs to deduplicate + urls = [a.get("url", "") for a in articles if a.get("url")] + existing_result = await db.execute( + select(GdeltArticleRaw.url).where(GdeltArticleRaw.url.in_(urls)) + ) + existing_urls = set(existing_result.scalars().all()) + + rows_to_insert = [] + for article in articles: + url = article.get("url", "") + if not url or url in existing_urls: + continue + + # Parse published_at from seendate or socialimage timestamp + published_at: Optional[datetime] = None + seen_date = article.get("seendate", "") + if seen_date and len(seen_date) >= 14: + try: + published_at = datetime( + int(seen_date[0:4]), + int(seen_date[4:6]), + int(seen_date[6:8]), + int(seen_date[8:10]), + int(seen_date[10:12]), + int(seen_date[12:14]), + tzinfo=timezone.utc, + ) + except ValueError: + pass + + rows_to_insert.append({ + "url": url, + "title": article.get("title"), + "domain": _extract_domain(url), + "published_at": published_at, + "sourcecountry": article.get("sourcecountry"), + "matched_ticker": ticker, + "match_method": "gdelt_query", + "match_confidence": 1.0, + }) + + if not rows_to_insert: + logger.info("No new GDELT articles for %s (all duplicates)", ticker) + return 0 + + stmt = ( + pg_insert(GdeltArticleRaw) + .values(rows_to_insert) + .on_conflict_do_nothing(index_elements=["url"]) + ) + await db.execute(stmt) + await db.commit() + + logger.info("Inserted %d GDELT article records for %s", len(rows_to_insert), ticker) + return len(rows_to_insert) diff --git a/app/services/attention/wiki_collector.py b/app/services/attention/wiki_collector.py new file mode 100644 index 0000000..e1fdeb7 --- /dev/null +++ b/app/services/attention/wiki_collector.py @@ -0,0 +1,143 @@ +""" +Wiki Collector — event-centric Wikipedia pageview collection. + +Collects daily pageview counts for the event window: event_date - 20d to event_date + 2d. +Only fetches dates not already in the database. +Uses Wikimedia REST API range endpoint for efficiency. +""" + +import logging +from datetime import date, timedelta + +import httpx +from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.attention import CompanyEntityMap, WikiPageviewsDaily + +logger = logging.getLogger(__name__) + +_WIKI_BASE = "https://wikimedia.org/api/rest_v1/metrics/pageviews/per-article" +_WIKI_HEADERS = { + "User-Agent": "StockOracleBot/1.0 (attention-subsystem; contact@stockoracle.internal)" +} +_LOOKBACK_DAYS = 20 +_LOOKAHEAD_DAYS = 2 + + +def _date_range(start: date, end: date) -> list[date]: + """Return list of dates from start to end inclusive.""" + dates = [] + current = start + while current <= end: + dates.append(current) + current += timedelta(days=1) + return dates + + +async def collect_wiki_pageviews( + db: AsyncSession, + ticker: str, + event_date: date, +) -> int: + """Collect Wikipedia pageviews for the event window. + + Returns the number of new records inserted. + Raises ValueError if the entity has no wiki_title. + """ + ticker = ticker.upper() + + # Look up wiki_title from entity map + result = await db.execute( + select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker) + ) + entity = result.scalars().first() + if not entity: + raise ValueError(f"No entity mapping found for {ticker!r}. Run entity resolution first.") + if not entity.wiki_title: + raise ValueError(f"Entity {ticker!r} has no wiki_title — resolution confidence was too low.") + + wiki_title = entity.wiki_title + start_date = event_date - timedelta(days=_LOOKBACK_DAYS) + end_date = event_date + timedelta(days=_LOOKAHEAD_DAYS) + all_dates = _date_range(start_date, end_date) + + # Find which dates are already in DB + existing_result = await db.execute( + select(WikiPageviewsDaily.date).where( + WikiPageviewsDaily.wiki_title == wiki_title, + WikiPageviewsDaily.date.in_(all_dates), + ) + ) + existing_dates = set(existing_result.scalars().all()) + missing_dates = [d for d in all_dates if d not in existing_dates] + + if not missing_dates: + logger.info("All wiki pageviews already collected for %s event=%s", ticker, event_date) + return 0 + + # Fetch range from Wikimedia REST API + fetch_start = min(missing_dates) + fetch_end = max(missing_dates) + start_str = fetch_start.strftime("%Y%m%d") + end_str = fetch_end.strftime("%Y%m%d") + + encoded_title = wiki_title.replace(" ", "_") + url = ( + f"{_WIKI_BASE}/en.wikipedia/all-access/all-agents" + f"/{encoded_title}/daily/{start_str}00/{end_str}00" + ) + + logger.info("Fetching wiki pageviews for %r (%s → %s)", wiki_title, start_str, end_str) + + try: + async with httpx.AsyncClient(timeout=15.0, headers=_WIKI_HEADERS) as client: + resp = await client.get(url) + resp.raise_for_status() + data = resp.json() + except httpx.HTTPStatusError as exc: + if exc.response.status_code == 404: + logger.warning("Wikipedia page not found: %r", wiki_title) + return 0 + raise + except Exception as exc: + logger.error("Wiki API request failed: %s", exc) + raise + + items = data.get("items", []) + if not items: + logger.info("No pageview data returned for %r", wiki_title) + return 0 + + # Parse and insert missing dates only + missing_set = set(missing_dates) + rows_to_insert = [] + for item in items: + timestamp_str = item.get("timestamp", "") + if len(timestamp_str) >= 8: + d = date( + int(timestamp_str[0:4]), + int(timestamp_str[4:6]), + int(timestamp_str[6:8]), + ) + if d in missing_set: + rows_to_insert.append({ + "wiki_title": wiki_title, + "date": d, + "views": item.get("views", 0), + }) + + if not rows_to_insert: + return 0 + + stmt = ( + pg_insert(WikiPageviewsDaily) + .values(rows_to_insert) + .on_conflict_do_nothing(constraint="uq_wiki_pageviews_daily") + ) + await db.execute(stmt) + await db.commit() + + logger.info("Inserted %d wiki pageview records for %s", len(rows_to_insert), ticker) + return len(rows_to_insert) diff --git a/app/utils/cache.py b/app/utils/cache.py index 2f3e96a..c5dce35 100644 --- a/app/utils/cache.py +++ b/app/utils/cache.py @@ -136,6 +136,28 @@ async def set_cached_response(key: str, body: Dict[str, Any], ttl_seconds: Optio return etag +async def get_negative_cached(key: str) -> bool: + """Return True if this key is in the negative cache (known 404).""" + client = await get_redis() + if client is None: + return False + try: + return (await client.exists(key)) > 0 + except Exception: + return False + + +async def set_negative_cached(key: str, ttl: int = 3600) -> None: + """Store a negative cache entry (known 404) with given TTL in seconds.""" + client = await get_redis() + if client is None: + return + try: + await client.set(key, b"1", ex=ttl) + except Exception as e: + logger.debug("Negative cache set failed: %s", e) + + def with_cache(namespace: str, ttl: int = None, key_params: list[str] = None): """ FastAPI endpoint caching decorator. diff --git a/tests/test_attention.py b/tests/test_attention.py new file mode 100644 index 0000000..cf01303 --- /dev/null +++ b/tests/test_attention.py @@ -0,0 +1,359 @@ +""" +Tests for the Attention subsystem. + +Covers: +- Entity resolver name normalization +- Wikipedia scoring logic +- GDELT query building +- Feature materializer calculations +- API endpoint routing and response schemas +""" + +import pytest +from datetime import date +from unittest.mock import AsyncMock, MagicMock, patch + +from app.services.attention.entity_resolver import ( + _normalize_name, + _score_wiki_result, + _build_gdelt_query, + _is_placeholder_name, + _fetch_sec_company_name, +) +from app.schemas.attention import ( + EntityInfo, + WikiFeatures, + NewsFeatures, + EventAttentionResponse, + EntityResolveResponse, + CollectionStatusResponse, +) + + +# --------------------------------------------------------------------------- +# Entity Resolver: _normalize_name +# --------------------------------------------------------------------------- + +class TestNormalizeName: + def test_strips_inc(self): + canonical, aliases = _normalize_name("Apple Inc.") + assert canonical == "Apple" + assert "Apple Inc." in aliases + + def test_strips_corp(self): + canonical, aliases = _normalize_name("Microsoft Corp") + assert canonical == "Microsoft" + + def test_strips_holdings(self): + canonical, aliases = _normalize_name("Ondas Holdings Inc.") + assert canonical == "Ondas" + assert "Ondas Holdings" in aliases or "Ondas Holdings Inc." in aliases + + def test_strips_multiple_suffixes(self): + canonical, aliases = _normalize_name("SomeCompany Holdings Ltd.") + assert canonical == "SomeCompany" + + def test_no_suffix(self): + canonical, aliases = _normalize_name("Tesla") + assert canonical == "Tesla" + assert aliases == [] + + def test_preserves_original_in_aliases(self): + canonical, aliases = _normalize_name("Alphabet Inc.") + assert "Alphabet Inc." in aliases + + def test_strips_technologies(self): + canonical, aliases = _normalize_name("Palantir Technologies Inc.") + assert canonical == "Palantir" + + def test_strips_pharmaceuticals(self): + canonical, aliases = _normalize_name("Pfizer Pharmaceuticals Inc.") + assert canonical == "Pfizer" + + def test_comma_handling(self): + canonical, aliases = _normalize_name("Berkshire Hathaway, Inc.") + assert canonical == "Berkshire Hathaway" + + +# --------------------------------------------------------------------------- +# Entity Resolver: _is_placeholder_name +# --------------------------------------------------------------------------- + +class TestIsPlaceholderName: + def test_ticker_equals_canonical(self): + assert _is_placeholder_name("AMZN", "AMZN") is True + + def test_case_insensitive(self): + assert _is_placeholder_name("GOOGL", "googl") is True + + def test_real_name_not_placeholder(self): + assert _is_placeholder_name("AAPL", "Apple") is False + + def test_partial_ticker_not_placeholder(self): + assert _is_placeholder_name("META", "Meta Platforms") is False + + def test_empty_canonical(self): + assert _is_placeholder_name("TSLA", "") is False + + +# --------------------------------------------------------------------------- +# Entity Resolver: _fetch_sec_company_name +# --------------------------------------------------------------------------- + +class TestFetchSecCompanyName: + @pytest.mark.asyncio + async def test_returns_name_for_known_ticker(self): + import app.services.attention.entity_resolver as er + mock_data = { + "0": {"cik_str": 1018724, "ticker": "AMZN", "title": "AMAZON COM INC"}, + "1": {"cik_str": 320193, "ticker": "AAPL", "title": "Apple Inc."}, + } + with patch.object(er, "_SEC_TICKER_MAP", {}): + with patch("httpx.AsyncClient") as mock_client_cls: + mock_resp = MagicMock() + mock_resp.raise_for_status = MagicMock() + mock_resp.json.return_value = mock_data + mock_client_cls.return_value.__aenter__ = AsyncMock( + return_value=MagicMock(get=AsyncMock(return_value=mock_resp)) + ) + mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False) + result = await er._fetch_sec_company_name("AMZN") + assert result == "Amazon Com Inc" + + @pytest.mark.asyncio + async def test_returns_none_for_unknown_ticker(self): + import app.services.attention.entity_resolver as er + # Pre-populate cache with known tickers only + with patch.object(er, "_SEC_TICKER_MAP", {"AAPL": "Apple Inc."}): + result = await er._fetch_sec_company_name("ZZZZZ") + assert result is None + + @pytest.mark.asyncio + async def test_uses_cache_on_second_call(self): + import app.services.attention.entity_resolver as er + with patch.object(er, "_SEC_TICKER_MAP", {"NVDA": "NVIDIA CORP"}): + result = await er._fetch_sec_company_name("NVDA") + assert result == "Nvidia Corp" + + @pytest.mark.asyncio + async def test_returns_none_on_http_error(self): + import app.services.attention.entity_resolver as er + with patch.object(er, "_SEC_TICKER_MAP", {}): + with patch("httpx.AsyncClient") as mock_client_cls: + mock_client_cls.return_value.__aenter__ = AsyncMock( + return_value=MagicMock( + get=AsyncMock(side_effect=Exception("network error")) + ) + ) + mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False) + result = await er._fetch_sec_company_name("AMZN") + assert result is None + + +# --------------------------------------------------------------------------- +# Entity Resolver: _score_wiki_result +# --------------------------------------------------------------------------- + +class TestScoreWikiResult: + def test_perfect_match_company_keyword(self): + result = { + "title": "Apple Inc.", + "snippet": "American multinational technology corporation and company stock", + } + score = _score_wiki_result(result, "Apple", []) + assert score > 0.5 + + def test_no_name_in_title_penalized(self): + result = { + "title": "Some Random Album", + "snippet": "Music album released in 2005", + } + score = _score_wiki_result(result, "Apple", []) + # Returns 0.1 (penalized for album signal), not 0.0 + assert score <= 0.15 + + def test_penalizes_album(self): + result = { + "title": "Ondas (album)", + "snippet": "Ondas is a music album by some band", + } + score = _score_wiki_result(result, "Ondas", ["Ondas Holdings"]) + assert score <= 0.15 + + def test_alias_match_in_title(self): + result = { + "title": "Ondas Holdings", + "snippet": "American company stock nasdaq finance", + } + score = _score_wiki_result(result, "Ondas", ["Ondas Holdings"]) + assert score > 0.5 + + def test_film_penalized(self): + result = { + "title": "Terns (film)", + "snippet": "A 2003 film directed by someone", + } + score = _score_wiki_result(result, "Terns", []) + assert score <= 0.15 + + +# --------------------------------------------------------------------------- +# Entity Resolver: _build_gdelt_query +# --------------------------------------------------------------------------- + +class TestBuildGdeltQuery: + def test_single_term(self): + query = _build_gdelt_query("Apple", []) + assert query == '"Apple"' + + def test_multiple_terms_joined_with_or(self): + query = _build_gdelt_query("Apple", ["Apple Inc."]) + assert '"Apple"' in query + assert '"Apple Inc."' in query + assert " OR " in query + + def test_caps_at_four_terms(self): + query = _build_gdelt_query("Tesla", ["Tesla Inc.", "Tesla Motors", "Tesla Corp", "Extra"]) + parts = query.split(" OR ") + assert len(parts) == 4 + + def test_no_duplicate_canonical(self): + query = _build_gdelt_query("Palantir", ["Palantir Technologies Inc.", "Palantir Technologies"]) + # canonical should appear only once + assert query.count('"Palantir"') == 1 + + +# --------------------------------------------------------------------------- +# Schemas +# --------------------------------------------------------------------------- + +class TestSchemas: + def test_entity_info_defaults(self): + info = EntityInfo(ticker="AAPL", canonical_name="Apple") + assert info.aliases == [] + assert info.resolver_confidence == 0.0 + assert info.is_manual_override is False + assert info.wiki_title is None + + def test_wiki_features_all_none(self): + wf = WikiFeatures() + assert wf.views is None + assert wf.spike_10d is None + assert wf.zscore_20d is None + + def test_news_features_defaults(self): + nf = NewsFeatures() + assert nf.article_count_1d == 0 + assert nf.article_count_3d == 0 + assert nf.unique_domains_3d == 0 + assert nf.us_article_count_3d == 0 + assert nf.gdelt_status == "not_collected" + + def test_news_features_gdelt_status_values(self): + assert NewsFeatures(gdelt_status="collected").gdelt_status == "collected" + assert NewsFeatures(gdelt_status="not_available").gdelt_status == "not_available" + + def test_event_attention_response(self): + resp = EventAttentionResponse( + ticker="AAPL", + event_date=date(2026, 2, 6), + entity=EntityInfo(ticker="AAPL", canonical_name="Apple"), + wiki=WikiFeatures(views=10000, spike_10d=2.5, zscore_20d=1.8), + news=NewsFeatures(article_count_1d=5, article_count_3d=12), + ) + assert resp.ticker == "AAPL" + assert resp.wiki.spike_10d == 2.5 + assert resp.news.article_count_3d == 12 + + def test_collection_status_response(self): + resp = CollectionStatusResponse( + ticker="ONDS", + source="wiki", + records_collected=22, + date_range={"event_date": "2026-02-11"}, + status="success", + ) + assert resp.records_collected == 22 + assert resp.source == "wiki" + + def test_entity_resolve_response(self): + resp = EntityResolveResponse( + ticker="ONDS", + entity=EntityInfo( + ticker="ONDS", + canonical_name="Ondas", + wiki_title="Ondas Holdings", + resolver_confidence=0.75, + ), + status="resolved", + message="Entity resolved", + ) + assert resp.status == "resolved" + assert resp.entity.wiki_title == "Ondas Holdings" + + +# --------------------------------------------------------------------------- +# Feature Materializer: stats calculations (unit-tested inline) +# --------------------------------------------------------------------------- + +class TestStatCalculations: + """Test the statistical computation logic without DB.""" + + def test_spike_calculation(self): + import statistics + wiki_views = 15000 + prev_10 = [5000, 6000, 4500, 5500, 7000, 6500, 4000, 5000, 6000, 5500] + median_10 = statistics.median(prev_10) + spike = wiki_views / median_10 + assert spike == pytest.approx(15000 / 5500, rel=1e-3) + + def test_zscore_calculation(self): + import statistics + wiki_views = 15000 + prev_20 = [5000] * 20 + mean_20 = statistics.mean(prev_20) + stdev_20 = statistics.stdev(prev_20) + # All same values → stdev=0, no zscore + assert stdev_20 == 0.0 + + def test_zscore_with_variance(self): + import statistics + wiki_views = 15000 + prev_20 = [4000, 5000, 6000, 4500, 5500, 7000, 3500, 4000, 5000, 6000, + 4000, 5000, 6000, 4500, 5500, 7000, 3500, 4000, 5000, 6000] + mean_20 = statistics.mean(prev_20) + stdev_20 = statistics.stdev(prev_20) + zscore = (wiki_views - mean_20) / stdev_20 + assert zscore > 0 # 15000 is well above mean ~5000 + + +# --------------------------------------------------------------------------- +# API endpoint integration (mock DB) +# --------------------------------------------------------------------------- + +class TestAttentionEndpoints: + """Lightweight route-level tests using FastAPI TestClient with mocked DB.""" + + @pytest.fixture + def client(self): + from fastapi import FastAPI + from app.api.v1.endpoints.attention import router + app = FastAPI() + app.include_router(router, prefix="/attention") + + from fastapi.testclient import TestClient + return TestClient(app) + + def test_entity_not_found_returns_404(self, client): + # Full integration tests require a live DB — this is a placeholder. + # Verified manually via: curl -X POST http://localhost:18001/api/v1/attention/admin/resolve/AAPL + pass + + def test_routes_registered(self): + from app.api.v1.endpoints.attention import router + paths = [r.path for r in router.routes] + assert "/event/{ticker}" in paths + assert "/entity/{ticker}" in paths + assert "/admin/resolve/{ticker}" in paths + assert "/admin/collect/wiki/{ticker}" in paths + assert "/admin/collect/gdelt/{ticker}" in paths