diff --git a/app/api/v1/endpoints/attention.py b/app/api/v1/endpoints/attention.py
index f30bb31..9736f4c 100644
--- a/app/api/v1/endpoints/attention.py
+++ b/app/api/v1/endpoints/attention.py
@@ -15,25 +15,27 @@ Admin endpoints:
 
 import asyncio
 import logging
-from datetime import date, timedelta
+from datetime import date, datetime, timedelta, timezone
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.database import get_db
-from app.utils.cache import with_cache
+from app.utils.cache import build_cache_key, get_cached_response, set_cached_response, with_cache
 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,
+    EntityOverrideResponse,
     EntityResolveResponse,
     EventAttentionResponse,
     NewsFeatures,
     WikiFeatures,
 )
 from app.services.attention.entity_resolver import resolve_entity
+from app.services.attention.entity_resolver import _build_gdelt_query, _normalize_name
 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
@@ -46,6 +48,22 @@ router = APIRouter()
 _EVENT_SEMAPHORE = asyncio.Semaphore(8)
 _EVENT_SEMAPHORE_WAIT = 10
 
+# Re-resolve a present-but-unresolved entity at most once per this window.
+# Bounds the Wikipedia query rate when fithia2 re-runs its bulk upcoming-
+# earnings scan (at semaphore=8) against still-NULL rows — prevents the 429
+# storm that originally left 1423/1697 mappings broken.
+_RESOLVE_RETRY_WINDOW = timedelta(hours=1)
+
+
+def _resolve_retry_due(updated_at) -> bool:
+    """True if enough time has elapsed since the last resolution attempt to
+    retry a still-unresolved entity. A missing timestamp counts as due."""
+    if updated_at is None:
+        return True
+    if updated_at.tzinfo is None:
+        updated_at = updated_at.replace(tzinfo=timezone.utc)
+    return datetime.now(timezone.utc) - updated_at >= _RESOLVE_RETRY_WINDOW
+
 
 def _entity_to_info(entity: CompanyEntityMap) -> EntityInfo:
     return EntityInfo(
@@ -226,6 +244,62 @@ async def admin_collect_gdelt(
     )
 
 
+@router.post(
+    "/admin/entity/{ticker}/override",
+    response_model=EntityOverrideResponse,
+    summary="Manually set wiki_title for a ticker (override automatic resolver)",
+    description=(
+        "Directly sets the Wikipedia article title for a ticker, bypassing the automatic resolver. "
+        "Sets `is_manual_override=True` so the resolver will never overwrite this mapping.\n\n"
+        "Use when the resolver persistently picks the wrong article "
+        "(e.g. a lawsuit page or an acquired subsidiary instead of the company itself).\n\n"
+        "**Example**: `POST /admin/entity/CSCO/override?wiki_title=Cisco%20Systems`"
+    ),
+    tags=["attention-admin"],
+)
+async def admin_override_entity(
+    ticker: str,
+    wiki_title: str = Query(..., description="Exact Wikipedia article title to use"),
+    db: AsyncSession = Depends(get_db),
+) -> EntityOverrideResponse:
+    ticker = ticker.upper()
+
+    entity_result = await db.execute(
+        select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker)
+    )
+    entity = entity_result.scalars().first()
+    if not entity:
+        raise HTTPException(
+            status_code=404,
+            detail=f"No entity mapping found for {ticker}. POST /admin/resolve/{ticker} first.",
+        )
+
+    canonical_name = entity.canonical_name
+    aliases = entity.aliases_json or []
+    gdelt_query = _build_gdelt_query(canonical_name, aliases)
+
+    from sqlalchemy import update as sa_update
+    await db.execute(
+        sa_update(CompanyEntityMap)
+        .where(CompanyEntityMap.ticker == ticker)
+        .values(
+            wiki_title=wiki_title,
+            gdelt_query=gdelt_query,
+            is_manual_override=True,
+            resolver_confidence=1.0,
+        )
+    )
+    await db.commit()
+    logger.info("Manual override applied: %s → wiki_title=%r", ticker, wiki_title)
+
+    return EntityOverrideResponse(
+        ticker=ticker,
+        wiki_title=wiki_title,
+        gdelt_query=gdelt_query,
+        message=f"Manual override set: {ticker} → {wiki_title!r}. is_manual_override=True.",
+    )
+
+
 # ===========================================================================
 # Parameterized entity route (before /event/ to avoid shadowing)
 # ===========================================================================
@@ -273,6 +347,10 @@ async def get_entity(
 # ===========================================================================
 
 
+_EVENT_CACHE_TTL_COMPLETE = 3600   # full response with wiki data → 1 h
+_EVENT_CACHE_TTL_INCOMPLETE = 300  # wiki_views still null → 5 min, retry sooner
+
+
 @router.get(
     "/event/{ticker}",
     response_model=EventAttentionResponse,
@@ -306,7 +384,6 @@ async def get_entity(
         500: {"description": "Feature materialization or collection error"},
     },
 )
-@with_cache(namespace="attention:event", ttl=3600, key_params=["ticker", "event_date"])
 async def get_event_attention(
     ticker: str,
     event_date: date = Query(..., description="Event date in YYYY-MM-DD format"),
@@ -314,6 +391,17 @@ async def get_event_attention(
     db: AsyncSession = Depends(get_db),
 ) -> EventAttentionResponse:
     ticker = ticker.upper()
+    cache_key = build_cache_key("attention:event", ticker, str(event_date))
+
+    cached = await get_cached_response(cache_key)
+    if cached:
+        cached_body, etag = cached
+        if response is not None:
+            response.headers["X-Cache"] = "HIT"
+            response.headers["Cache-Control"] = f"public, max-age={_EVENT_CACHE_TTL_COMPLETE}"
+            response.headers["ETag"] = etag
+            response.headers["X-Data-Source"] = "redis-cache"
+        return cached_body
 
     try:
         await asyncio.wait_for(_EVENT_SEMAPHORE.acquire(), timeout=_EVENT_SEMAPHORE_WAIT)
@@ -323,10 +411,22 @@ async def get_event_attention(
             detail="서버가 바빠서 요청을 처리할 수 없습니다. 잠시 후 다시 시도하세요.",
         )
     try:
-        return await _get_event_attention_impl(ticker, event_date, db)
+        result = await _get_event_attention_impl(ticker, event_date, db)
     finally:
         _EVENT_SEMAPHORE.release()
 
+    # Incomplete responses (wiki_views still null) cache for only 5 min so they
+    # re-trigger collection sooner rather than serving stale nulls for up to 1 h.
+    wiki_complete = result.wiki.views is not None
+    ttl = _EVENT_CACHE_TTL_COMPLETE if wiki_complete else _EVENT_CACHE_TTL_INCOMPLETE
+    etag = await set_cached_response(cache_key, result.model_dump(), ttl_seconds=ttl)
+    if response is not None:
+        response.headers["X-Cache"] = "MISS"
+        response.headers["Cache-Control"] = f"public, max-age={ttl}"
+        response.headers["ETag"] = etag
+
+    return result
+
 
 async def _get_event_attention_impl(
     ticker: str,
@@ -348,6 +448,23 @@ async def _get_event_attention_impl(
         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}")
+    elif (
+        not entity.wiki_title
+        and (entity.resolver_confidence or 0.0) < 0.5
+        and not entity.is_manual_override
+        and _resolve_retry_due(entity.updated_at)
+    ):
+        # Present-but-unresolved row — e.g. a transient Wikipedia 429 during a
+        # prior bulk run persisted wiki_title=NULL. Retry resolution, but at
+        # most once per _RESOLVE_RETRY_WINDOW so a bulk scan over still-NULL
+        # tickers can't re-create the 429 storm. Best-effort: on failure keep
+        # serving the (empty) row rather than 500-ing. The resolver's
+        # no-downgrade guard ensures a failed retry can't worsen the row.
+        logger.info("Entity %s unresolved — attempting on-demand re-resolve", ticker)
+        try:
+            entity = await resolve_entity(db, ticker)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("On-demand re-resolve failed for %s: %s", ticker, exc)
 
     # 2. Check if features already exist in DB
     features_result = await db.execute(
@@ -358,12 +475,19 @@ async def _get_event_attention_impl(
     )
     features = features_result.scalars().first()
 
-    if features is None:
-        # 3. On-demand collection + materialization
+    needs_wiki = entity.wiki_title and (
+        features is None
+        or features.wiki_views is None
+        or features.wiki_zscore_20d is None
+    )
+    if features is None or needs_wiki:
+        # 3. On-demand collection + materialization.
+        # Re-triggers when features are missing OR wiki_views / wiki_zscore_20d is null:
+        # - wiki_views null: wiki_title was unset at collection time, or API failed
+        # - wiki_zscore_20d null: insufficient lookback data when previously materialized
         # 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)
+        logger.info("No features (or incomplete wiki) for %s on %s — collecting wiki on-demand", ticker, event_date)
 
         if entity.wiki_title:
             try:
diff --git a/app/schemas/attention.py b/app/schemas/attention.py
index 89e1ff3..32fbb26 100644
--- a/app/schemas/attention.py
+++ b/app/schemas/attention.py
@@ -153,3 +153,10 @@ class CollectionStatusResponse(BaseModel):
     records_collected: int
     date_range: Dict[str, Any] = Field(default_factory=dict)
     status: str
+
+
+class EntityOverrideResponse(BaseModel):
+    ticker: str
+    wiki_title: str
+    gdelt_query: Optional[str]
+    message: str
diff --git a/app/services/attention/entity_resolver.py b/app/services/attention/entity_resolver.py
index bbe83a5..2b64da7 100644
--- a/app/services/attention/entity_resolver.py
+++ b/app/services/attention/entity_resolver.py
@@ -11,10 +11,11 @@ Resolution pipeline:
 
 import logging
 import re
+from datetime import datetime, timezone
 from typing import Optional
 
 import httpx
-from sqlalchemy import select, update
+from sqlalchemy import or_, select, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -26,13 +27,28 @@ 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"L\.P\.?|LP|"  # Limited Partnership (e.g. "Enterprise Products Partners L.P.")
+    r"Group|Co\.?|Compan(?:y|ies)|Technologies|Technology|International|Industries|"
     r"Pharmaceuticals?|Therapeutics?|Sciences?|Bancorp|Financial|Holding|"
     r"Acquisition|Acquisitions|Capital|Partners|Trust|"
+    r"Nv|N\.V\.?|Plc\.?|p\.l\.c\.?|"  # Dutch (N.V.) / British (Plc / p.l.c.)
+    r"S\.A\.?B?(?:\s+de\s+C\.V\.)?|"  # Spanish (S.A., S.A.B., S.A.B. de C.V.)
+    r"Aktiengesellschaft|GmbH|AG|SE|"  # German/Swiss/European
     r"Com)\s*$",  # "Com" catches SEC-style ".com" artifacts (e.g. "AMAZON COM")
     re.IGNORECASE,
 )
 
+# SEC filing artifact: state-of-incorporation suffix after slash
+# e.g. "Costco Wholesale Corp /New", "Wells Fargo & Company/Mn", "Applied Materials Inc /DE",
+# "Canadian Imperial Bank Of Commerce /Can/"
+_SEC_NEW_PATTERN = re.compile(r"\s*/\s*(?:New|[A-Za-z]{2,4})\s*$", re.IGNORECASE)
+
+# Danish/Norwegian corporate designation: "NOVO NORDISK A/S" → "NOVO NORDISK"
+_AS_PATTERN = re.compile(r"\bA/S\s*$", re.IGNORECASE)
+
+# Belgian SA/NV: "Anheuser-Busch InBev SA/NV" → strip "/NV" first, then "SA" via suffix
+_SA_NV_PATTERN = re.compile(r"\s*SA/NV\s*$", re.IGNORECASE)
+
 _COMPANY_KEYWORDS = {
     "company", "corporation", "inc", "corp", "ltd", "llc", "holdings",
     "stock", "shares", "nasdaq", "nyse", "ticker", "finance", "financial",
@@ -47,7 +63,30 @@ def _normalize_name(raw_name: str) -> tuple[str, list[str]]:
     Preserves the original and intermediate forms as aliases.
     """
     aliases = []
-    current = raw_name.strip()
+    # Normalize path separators: some DB records use backslash (e.g. "US BANCORP \DE\")
+    current = raw_name.strip().replace("\\", "/").rstrip("/").strip()
+
+    # Belgian "SA/NV" corporate designation (e.g. "Anheuser-Busch InBev SA/NV")
+    stripped_sanv = _SA_NV_PATTERN.sub("", current).strip()
+    if stripped_sanv and stripped_sanv != current:
+        aliases.append(current)
+        current = stripped_sanv
+
+    # Strip Danish/Norwegian "A/S" corporate designation (e.g. "NOVO NORDISK A/S")
+    stripped_as = _AS_PATTERN.sub("", current).strip()
+    if stripped_as and stripped_as != current:
+        aliases.append(current)
+        current = stripped_as
+
+    # Strip SEC reincorporation artifact "/New" (and state/province codes like "/DE", "/Can")
+    stripped_new = _SEC_NEW_PATTERN.sub("", current).strip()
+    if stripped_new and stripped_new != current:
+        aliases.append(current)
+        current = stripped_new
+
+    # Normalize SEC dot-com artifact: "Amazon.Com" / "Amazon.com" → "Amazon Com"
+    # so the iterative loop can strip "Com" as a regular suffix.
+    current = re.sub(r"\.com\b", " Com", current, flags=re.IGNORECASE)
 
     for _ in range(5):  # max 5 iterations to avoid infinite loops
         stripped = _SUFFIX_PATTERN.sub("", current).strip().rstrip(",").strip()
@@ -56,7 +95,9 @@ def _normalize_name(raw_name: str) -> tuple[str, list[str]]:
         aliases.append(current)
         current = stripped
 
-    canonical = current
+    # Strip trailing punctuation artifacts left by suffix removal (e.g. "&" from
+    # "JPMorgan Chase & Co" → strip "Co" → "JPMorgan Chase &").
+    canonical = current.rstrip(" &/,").strip()
     # 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())
@@ -154,7 +195,11 @@ def _score_wiki_result(result: dict, canonical_name: str, aliases: list[str]) ->
     """
     raw_title = result.get("title", "")
     title = raw_title.lower()
-    snippet = result.get("snippet", "").lower()
+    # Wikipedia snippets contain <span class="searchmatch"> HTML — strip before matching.
+    # Replace each tag with a space then collapse runs so "Costco</span> <span>Wholesale"
+    # becomes "Costco Wholesale" rather than "Costco  Wholesale" (breaking string match).
+    raw_snippet = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", result.get("snippet", ""))).strip()
+    snippet = raw_snippet.lower()
     combined = title + " " + snippet
 
     # Penalize obvious non-company pages immediately
@@ -165,31 +210,56 @@ def _score_wiki_result(result: dict, canonical_name: str, aliases: list[str]) ->
     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
+    # Penalize legal case titles: "X v. Y" format (e.g. "FSF v. Cisco Systems, Inc.")
+    if re.search(r'\w v\. \w', raw_title):
+        return 0.05
 
     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.")
+    # Exact title match — checked BEFORE the finance-signal filter so that
+    # valid company pages whose snippet focuses on technical/product details
+    # (e.g. TSMC → fabs, JPMorgan → banking operations) still score 0.95.
     raw_title_stripped = raw_title.strip()
     for name in all_names:
         if raw_title_stripped.lower() == name.lower():
-            return 0.95  # exact match
+            return 0.95
+
+    # Space-collapsed match: handles merged brand names like "ExxonMobil" vs "Exxon Mobil"
+    title_no_space = raw_title_stripped.lower().replace(" ", "")
+    for name in all_names:
+        name_no_space = name.lower().replace(" ", "")
+        if len(name_no_space) > 4 and title_no_space == name_no_space:
+            return 0.90
+
+    # 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
+
+    # Hyphen-normalized forms: "COCA COLA" matches "The Coca-Cola Company"
+    # because "coca cola" is in "the coca cola company" after replacing hyphens with spaces.
+    title_norm = title.replace("-", " ")
+    canonical_norm = canonical_lower.replace("-", " ")
+    all_names_norm = [n.replace("-", " ") for n in all_names_lower]
 
     # Check if title starts with canonical name
-    title_starts_with_canonical = title.startswith(canonical_lower)
-    title_contains_canonical = canonical_lower in title
+    title_starts_with_canonical = title_norm.startswith(canonical_norm)
+    title_contains_canonical = canonical_norm in title_norm
 
     # 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)
+    title_starts_with_alias = any(title_norm.startswith(n) for n in all_names_norm)
+    title_contains_alias = any(n in title_norm for n in all_names_norm)
 
     if not (title_contains_canonical or title_contains_alias):
+        # Snippet-contains fallback: handles acronym titles (e.g. "TSMC" article whose
+        # snippet reads "Taiwan Semiconductor Manufacturing Company Limited (TSMC)...")
+        # and short-title articles (e.g. "Costco" snippet contains "Costco Wholesale").
+        snippet_norm = snippet.replace("-", " ")
+        if canonical_norm in snippet_norm or any(n in snippet_norm for n in all_names_norm):
+            return 0.6
         return 0.0
 
     # Check for pages that are ABOUT the company (vs. lists, histories, etc.)
@@ -207,11 +277,13 @@ def _score_wiki_result(result: dict, canonical_name: str, aliases: list[str]) ->
     else:
         base = 0.0
 
-    # Reward company/finance keywords in title or snippet
+    # Reward company/finance keywords in title or snippet.
+    # Cap at 0.89 so title-starts-with matches never outrank exact-title (0.95)
+    # or space-collapsed exact matches (0.90), regardless of keyword density.
     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)
+    return min(base + keyword_score * 0.3, 0.89)
 
 
 async def _resolve_wiki(
@@ -311,41 +383,60 @@ async def resolve_entity(
         ticker, wiki_title, confidence,
     )
 
-    # Upsert into company_entity_map
-    stmt = (
-        pg_insert(CompanyEntityMap)
-        .values(
-            ticker=ticker,
+    # Upsert into company_entity_map.
+    #
+    # No-downgrade guard: a transient Wikipedia failure (429/network) is
+    # swallowed by _search_wikipedia → returns [] → (wiki_title=None,
+    # confidence=0.0). Without this guard, re-running resolution while
+    # rate-limited would overwrite a previously-good mapping with NULL —
+    # which is exactly how the 2026-03-17 bulk run left 1423/1697 rows
+    # broken. The upsert therefore only updates when the new result is
+    # itself good (wiki_title not NULL) OR the existing row was already
+    # unresolved (wiki_title NULL). Manual overrides are never touched.
+    insert_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,
+    )
+    stmt = insert_stmt.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,
-            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)
-    )
+            # Explicit: ORM `onupdate` does NOT fire for INSERT...ON CONFLICT,
+            # so stamp it here. The on-demand re-resolve recency guard in the
+            # /event endpoint relies on this reflecting the last attempt.
+            updated_at=datetime.now(timezone.utc),
+        ),
+        where=(
+            (CompanyEntityMap.is_manual_override == False)  # noqa: E712
+            & or_(
+                insert_stmt.excluded.wiki_title.isnot(None),
+                CompanyEntityMap.wiki_title.is_(None),
+            )
+        ),
+    ).returning(CompanyEntityMap)
 
-    result = await db.execute(stmt)
+    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()
+    # Read back via fresh SELECT with populate_existing=True.
+    # After commit, SQLAlchemy expires identity-map entries but does NOT evict them.
+    # A plain SELECT in the same session can return the expired (stale) cached object
+    # instead of reading the committed DB state. populate_existing forces the ORM to
+    # overwrite the identity-map entry with the fresh DB row.
+    fresh_result = await db.execute(
+        select(CompanyEntityMap)
+        .where(CompanyEntityMap.ticker == ticker)
+        .execution_options(populate_existing=True)
+    )
+    row = fresh_result.scalars().first()
 
     return row
diff --git a/app/services/attention/feature_materializer.py b/app/services/attention/feature_materializer.py
index a0d1fb1..c6e36d4 100644
--- a/app/services/attention/feature_materializer.py
+++ b/app/services/attention/feature_materializer.py
@@ -169,19 +169,20 @@ async def materialize_features(
         .returning(AttentionFeaturesDaily)
     )
 
-    result = await db.execute(stmt)
+    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,
-            )
+    # Always re-fetch with populate_existing=True to avoid SQLAlchemy identity-map
+    # returning stale pre-upsert values (same pattern as resolve_entity).
+    row_result = await db.execute(
+        select(AttentionFeaturesDaily)
+        .where(
+            AttentionFeaturesDaily.ticker == ticker,
+            AttentionFeaturesDaily.date == event_date,
         )
-        row = row_result.scalars().first()
+        .execution_options(populate_existing=True)
+    )
+    row = row_result.scalars().first()
 
     logger.info(
         "Materialized features for %s on %s: wiki=%s gdelt_1d=%d gdelt_3d=%d",