diff --git a/AGENTS.md b/AGENTS.md
index 336afde..514d659 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -76,7 +76,7 @@ review → `/code-review` + `/security-review`; everything → `/using-superpowe
 | `fundamentals/`, `technical/`, `sixty_seven/` | The three AI-assisted subsystems. |
 | `ipo/` | IPO domain: SEBI filing ingestion, verified content-addressed document cache, manual extraction records, deterministic ratio engine, immutable score/recommendation history, factor derivation + hard caution flags (`scoring/`), read-only dashboard builder, quarantined SerpAPI enrichment (`sources/enrichment.py`), and the fail-closed AI extraction agent (`agents/`, `documents/table_extractor.py`, `documents/section_classifier.py`) (IPO-001…010). |
 | `jobs/` | Headless CLIs (daily scan, forward-return computation, candle cache repair, IPO filing ingestion, and the idempotent IPO scan/download/enrich/extract/score pipeline). |
-| `admin/`, `auth/`, `notifications/`, `data_quality/` | Config overrides, OIDC gate, alerts, candle-quality receipts. |
+| `admin/`, `auth/`, `notifications/`, `data_quality/` | Config overrides, OIDC gate, alerts, candle-quality receipts, and the OBS-004 universe mapping-health baseline. |
 | `screener_registry.py`, `scanner_base.py`, `indicators.py`, `daily_data_loader.py`, `universe_*` | Screener framework, indicators, candle cache, universe management. |
 
 ---
@@ -253,6 +253,7 @@ allowlist gate. Full details and the **accepted residual risks**:
 - **Observability / audit / config:** [observability](docs/architecture/components/observability.md) ·
   [audit-log](docs/architecture/components/audit-log.md) ·
   [obs-003 design](docs/architecture/obs-003-audit-log.md) ·
+  [obs-004 universe health](docs/architecture/obs-004-universe-health-alerts.md) ·
   [configuration](docs/architecture/components/configuration.md)
 - **Screener framework:** [screener-framework](docs/architecture/components/screener-framework.md) ·
   [screener-catalog](docs/architecture/components/screener-catalog.md) ·
diff --git a/app.py b/app.py
index 4eaa4b0..6b189f3 100644
--- a/app.py
+++ b/app.py
@@ -287,6 +287,8 @@ def prefetch_data_assets() -> None:
         display_name = UNIVERSE_CONFIG.get(key, {}).get("display_name", key)
         print(f"[prefetch]   {display_name:<25} -> {path}", flush=True)
 
+    _log_universe_health()
+
     # Computing the union AFTER the refresh guarantees we see the freshest
     # mapped rows. If no universes loaded, we still let Streamlit boot.
     union = union_of_mapped_universes()
@@ -500,6 +502,28 @@ def _inject_css() -> None:
     st.markdown(_CUSTOM_CSS, unsafe_allow_html=True)
 
 
+def _log_universe_health() -> None:
+    """Emit the OBS-004 per-universe mapping counts. Best-effort, log only.
+
+    Beginner note:
+    This deliberately does NOT record a snapshot, unlike the daily job. Whoever
+    writes the baseline defines what "last time" means, so if this morning
+    prefetch also persisted, a symbol that dropped out here would already be part
+    of the baseline by the time the evening job ran - and the alert would never
+    fire. The prefetch's job is only to make the counts visible in the log; the
+    daily job owns the comparison.
+    """
+    try:
+        from backend.data_quality.universe_health import (
+            collect_universe_health,
+            log_universe_health,
+        )
+
+        log_universe_health(collect_universe_health())
+    except Exception:  # noqa: BLE001 - never let a health log break the prefetch
+        logger.warning("universe health logging failed during prefetch", exc_info=True)
+
+
 def refresh_universes_and_invalidate() -> dict[str, Path]:
     """Refresh universe CSVs, then clear every cache that reads them.
 
diff --git a/backend/data_quality/universe_health.py b/backend/data_quality/universe_health.py
new file mode 100644
index 0000000..aaac5b8
--- /dev/null
+++ b/backend/data_quality/universe_health.py
@@ -0,0 +1,292 @@
+"""OBS-004 - notice when a universe silently stops being scannable.
+
+The problem this solves
+-----------------------
+When Dhan's instrument master stops listing a symbol (a merger, a delisting, a
+ticker change), :func:`backend.universe_builder.refresh_universe_files` writes it
+into the universe CSV as ``mapping_status='missing_security_id'`` and
+:func:`backend.universe_loader.mapped_only` then filters it out of every scan.
+That is the correct behaviour - we cannot fetch candles for a security id we do
+not have - but until OBS-004 nothing *said* so outside the interactive Streamlit
+sidebar. ``universe_status()`` had exactly one non-test caller
+(``ui/status_panel.py``), the headless daily job emitted no mapping signal, and
+``backend/notifications/`` never mentioned ``mapping_status``. A universe could
+therefore shrink for weeks without anyone noticing; ~3% of the Hemant Good 200
+list was already unscanned when this module was written.
+
+The shape of the answer
+-----------------------
+Three deliberately separate pieces, because they have different requirements:
+
+* :func:`collect_universe_health` is **pure** - it reads the CSVs and returns
+  counts. Anything can call it, including the Streamlit prefetch.
+* :func:`detect_mapping_regressions` is **pure** - it compares today's counts to
+  a previous set and returns only the universes that got worse.
+* :func:`check_universe_health` is the **stateful** one: it needs a database
+  session because detecting "worse than last time" requires a durable baseline,
+  and the Render daily-scan cron runs on an ephemeral filesystem with no disk.
+
+Beginner note on why only the alerting path persists:
+Whoever writes the baseline decides what "last time" means. If the morning
+Streamlit prefetch also recorded a snapshot, a symbol that dropped out at 09:00
+would already be part of the baseline by the time the evening job ran, and the
+alert would never fire. So the prefetch calls the pure helpers for logging only,
+and ``check_universe_health`` - used by the daily job - owns the baseline.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from backend.config import UNIVERSE_DIR
+from backend.observability import (
+    EVENT_UNIVERSE_HEALTH_CHECKED,
+    EVENT_UNIVERSE_MAPPING_REGRESSED,
+    log_event,
+)
+
+if TYPE_CHECKING:  # pragma: no cover - import-cycle break for type checking only
+    from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+#: How many unmapped symbol names we are willing to store and report per
+#: universe. A universe whose CSV went badly wrong should not be able to write an
+#: unbounded blob into the database or a multi-page message into a Telegram
+#: alert; past this point the count alone tells the story.
+MAX_REPORTED_SYMBOLS = 25
+
+
+@dataclass(frozen=True)
+class UniverseHealth:
+    """Mapping health for one universe at one point in time."""
+
+    universe_key: str
+    total_rows: int
+    mapped_rows: int
+    unmapped_symbols: tuple[str, ...]
+
+    @property
+    def unmapped_rows(self) -> int:
+        """Rows the scanner cannot fetch, derived from the two counts.
+
+        Beginner note: derived rather than stored so it can never disagree with
+        ``total_rows``/``mapped_rows``. The database column is written from this
+        property, so the persisted row is internally consistent too.
+        """
+        return max(self.total_rows - self.mapped_rows, 0)
+
+
+@dataclass(frozen=True)
+class MappingRegression:
+    """One universe that has more unmapped symbols than it did last check."""
+
+    universe_key: str
+    previous_unmapped: int
+    current_unmapped: int
+    newly_unmapped: tuple[str, ...]
+
+    def describe(self) -> str:
+        """Return a one-line, alert-ready summary naming what changed."""
+        delta = self.current_unmapped - self.previous_unmapped
+        detail = ", ".join(self.newly_unmapped) if self.newly_unmapped else "symbols not named"
+        return (
+            f"{self.universe_key}: {self.previous_unmapped} -> {self.current_unmapped} "
+            f"unmapped (+{delta}); {detail}"
+        )
+
+
+@dataclass(frozen=True)
+class UniverseHealthReport:
+    """What one check found: every universe's counts, plus any regressions."""
+
+    snapshots: tuple[UniverseHealth, ...] = ()
+    regressions: tuple[MappingRegression, ...] = ()
+
+
+def _unmapped_symbols(frame: Any) -> tuple[str, ...]:
+    """Return the sorted, capped symbols in ``frame`` that cannot be fetched.
+
+    Beginner note: ``universe_status()`` gives us counts but not names, and a
+    count alone makes for a useless alert ("one more symbol is missing" - which
+    one?). This re-reads the same frame for the names. Sorted so the stored list
+    is stable and two runs are directly comparable.
+    """
+    if "symbol" not in frame.columns:
+        return ()
+    if "mapping_status" in frame.columns:
+        unmapped = frame.loc[
+            ~frame["mapping_status"].astype(str).str.lower().eq("mapped")
+        ]
+    elif "security_id" in frame.columns:
+        unmapped = frame.loc[frame["security_id"].astype(str).str.strip().eq("")]
+    else:
+        return ()
+    symbols = sorted({str(value).strip() for value in unmapped["symbol"] if str(value).strip()})
+    return tuple(symbols[:MAX_REPORTED_SYMBOLS])
+
+
+def collect_universe_health(
+    universe_dir: Path | str = UNIVERSE_DIR,
+) -> tuple[UniverseHealth, ...]:
+    """Read every universe CSV and return its mapping health. Never raises.
+
+    Reuses :func:`backend.universe_loader.universe_status` for the counts rather
+    than re-deriving them, so the numbers here and the numbers in the Streamlit
+    status panel can never disagree. A universe whose CSV is missing or
+    unreadable yields a zero row instead of an exception - a health check that
+    can take the daily job down would be worse than the problem it reports.
+    """
+    # Imported here rather than at module scope: universe_loader pulls in pandas
+    # and the universe registry, and this module is imported by the storage-aware
+    # job path. Keeping it local matches the lazy-import convention used to break
+    # cycles elsewhere in backend/.
+    import pandas as pd
+
+    from backend.universe_builder import UNIVERSE_CONFIG, universe_file_path
+    from backend.universe_loader import universe_status
+
+    results: list[UniverseHealth] = []
+    for universe_key in UNIVERSE_CONFIG:
+        status = universe_status(universe_key, universe_dir)
+        total_rows = int(status.get("rows", 0) or 0)
+        mapped_rows = int(status.get("mapped_rows", 0) or 0)
+
+        symbols: tuple[str, ...] = ()
+        # Only worth re-reading the file when it exists, parsed cleanly, and
+        # actually has something unmapped to name.
+        if status.get("exists") and not status.get("error") and total_rows > mapped_rows:
+            try:
+                frame = pd.read_csv(
+                    universe_file_path(universe_key, universe_dir), dtype=str
+                ).fillna("")
+                symbols = _unmapped_symbols(frame)
+            except Exception:  # noqa: BLE001 - a health check must never break the caller
+                logger.warning(
+                    "could not read unmapped symbols for universe %s", universe_key, exc_info=True
+                )
+
+        results.append(
+            UniverseHealth(
+                universe_key=universe_key,
+                total_rows=total_rows,
+                mapped_rows=mapped_rows,
+                unmapped_symbols=symbols,
+            )
+        )
+    return tuple(results)
+
+
+def log_universe_health(snapshots: Sequence[UniverseHealth]) -> None:
+    """Emit one structured event per universe so the counts are searchable.
+
+    This runs on every check, not just when something is wrong: an operator
+    asking "was Good 200 already down two names last Tuesday?" needs the routine
+    receipts, not only the alarms.
+    """
+    for snapshot in snapshots:
+        log_event(
+            logger,
+            EVENT_UNIVERSE_HEALTH_CHECKED,
+            universe_key=snapshot.universe_key,
+            rows=snapshot.total_rows,
+            mapped=snapshot.mapped_rows,
+            unmapped=snapshot.unmapped_rows,
+        )
+
+
+def detect_mapping_regressions(
+    current: Sequence[UniverseHealth],
+    previous: Mapping[str, Any],
+) -> tuple[MappingRegression, ...]:
+    """Return the universes whose unmapped count grew since ``previous``.
+
+    ``previous`` maps a universe key to its last persisted snapshot row (anything
+    exposing ``unmapped_rows`` and ``unmapped_symbols_json``).
+
+    Two deliberate rules:
+
+    * **A universe with no previous row never regresses.** The first check has no
+      baseline, so treating "absent" as zero would alert on every pre-existing
+      unmapped symbol - exactly the noise that makes people mute alerts.
+    * **Only an increase counts.** A universe sitting at a steady three unmapped
+      symbols is already-known damage and stays quiet; recovery (the count going
+      down) is good news and is not an alert either.
+    """
+    regressions: list[MappingRegression] = []
+    for snapshot in current:
+        baseline = previous.get(snapshot.universe_key)
+        if baseline is None:
+            continue
+        previous_unmapped = int(getattr(baseline, "unmapped_rows", 0) or 0)
+        if snapshot.unmapped_rows <= previous_unmapped:
+            continue
+
+        stored = getattr(baseline, "unmapped_symbols_json", None) or {}
+        known = {str(value) for value in stored.get("symbols", [])}
+        newly = tuple(symbol for symbol in snapshot.unmapped_symbols if symbol not in known)
+        regressions.append(
+            MappingRegression(
+                universe_key=snapshot.universe_key,
+                previous_unmapped=previous_unmapped,
+                current_unmapped=snapshot.unmapped_rows,
+                newly_unmapped=newly,
+            )
+        )
+    return tuple(regressions)
+
+
+def check_universe_health(
+    session: Session,
+    *,
+    universe_dir: Path | str = UNIVERSE_DIR,
+) -> UniverseHealthReport:
+    """Collect, log, compare against the stored baseline, then record today's.
+
+    The caller owns the transaction (REFACTOR-002): this adds rows and flushes,
+    but never commits.
+
+    Ordering matters. The baseline is read *before* today's snapshot is written,
+    otherwise every run would compare against itself and nothing would ever
+    regress. Writing afterwards is also what makes the alert fire exactly once -
+    the next run's baseline already contains the drop-out.
+    """
+    # Local import keeps the repository boundary one-directional and avoids a
+    # module-level cycle between data_quality and storage.
+    from backend.storage import repository
+
+    snapshots = collect_universe_health(universe_dir)
+    log_universe_health(snapshots)
+
+    previous = repository.get_latest_universe_health_snapshots(session)
+    regressions = detect_mapping_regressions(snapshots, previous)
+
+    for regression in regressions:
+        log_event(
+            logger,
+            EVENT_UNIVERSE_MAPPING_REGRESSED,
+            level=logging.WARNING,
+            universe_key=regression.universe_key,
+            previous_unmapped=regression.previous_unmapped,
+            current_unmapped=regression.current_unmapped,
+            newly_unmapped=list(regression.newly_unmapped),
+        )
+
+    repository.record_universe_health_snapshots(
+        session,
+        [
+            {
+                "universe_key": snapshot.universe_key,
+                "total_rows": snapshot.total_rows,
+                "mapped_rows": snapshot.mapped_rows,
+                "unmapped_rows": snapshot.unmapped_rows,
+                "unmapped_symbols": list(snapshot.unmapped_symbols),
+            }
+            for snapshot in snapshots
+        ],
+    )
+    return UniverseHealthReport(snapshots=snapshots, regressions=regressions)
diff --git a/backend/jobs/run_daily_scan.py b/backend/jobs/run_daily_scan.py
index aa2cb16..655c5d7 100644
--- a/backend/jobs/run_daily_scan.py
+++ b/backend/jobs/run_daily_scan.py
@@ -145,6 +145,10 @@ class DailyScanSummary:
     """
 
     outcomes: list[DailyScanOutcome]
+    # OBS-004: one line per universe that lost mapped symbols since the previous
+    # run. Defaulted so every existing construction (including the pre-scan
+    # failure paths below) keeps working unchanged.
+    universe_warnings: tuple[str, ...] = ()
 
     @property
     def exit_code(self) -> int:
@@ -257,6 +261,11 @@ def run_daily_scan(
         flush=True,
     )
 
+    # OBS-004: check mapping health BEFORE scanning, so an operator reading the
+    # log sees "this run covered a universe that just lost two symbols" rather
+    # than discovering it after the results look thin.
+    universe_warnings = _check_universe_health(session_factory, out)
+
     outcomes: list[DailyScanOutcome] = []
     for entry in enabled_entries:
         definition = registry.get(entry.screener_key)
@@ -287,7 +296,7 @@ def run_daily_scan(
         outcomes.append(outcome)
         _print_outcome(out, outcome)
 
-    summary = DailyScanSummary(outcomes=outcomes)
+    summary = DailyScanSummary(outcomes=outcomes, universe_warnings=universe_warnings)
     if summary.exit_code:
         print("[daily-scan] Finished with fatal failure(s).", file=out, flush=True)
     else:
@@ -295,6 +304,39 @@ def run_daily_scan(
     return summary
 
 
+def _check_universe_health(
+    session_factory: SessionFactory, out: TextIO
+) -> tuple[str, ...]:
+    """Best-effort OBS-004 mapping-health check; never affects the exit code.
+
+    Returns one human-readable line per universe that lost mapped symbols since
+    the previous recorded check, ready to paste into the daily alert.
+
+    Beginner note:
+    This owns the durable baseline (it writes today's snapshot), which is why the
+    Streamlit prefetch only *logs* health and does not record it - if two writers
+    moved the baseline, a symbol that dropped out in the morning would already be
+    "known" by the evening and the alert would never fire.
+
+    Wrapped in a broad except on purpose: a universe CSV that will not parse is a
+    reason to warn, never a reason to skip the night's scan.
+    """
+    try:
+        from backend.data_quality.universe_health import check_universe_health
+
+        with session_factory() as session:
+            report = check_universe_health(session)
+            session.commit()
+    except Exception:  # noqa: BLE001 - a health check must never fail the job
+        logger.warning("universe health check failed", exc_info=True)
+        return ()
+
+    warnings = tuple(regression.describe() for regression in report.regressions)
+    for warning in warnings:
+        print(f"[daily-scan] Universe mapping regressed - {warning}", file=out, flush=True)
+    return warnings
+
+
 def _send_scan_notification(summary: DailyScanSummary) -> None:
     """Best-effort ALERT-001 summary/alert; never affects the job's exit code.
 
diff --git a/backend/notifications/render.py b/backend/notifications/render.py
index 88c646b..d1af57c 100644
--- a/backend/notifications/render.py
+++ b/backend/notifications/render.py
@@ -61,6 +61,12 @@ def _body_lines(report: DailyScanReport) -> list[str]:
             f"  - {line.screener_key}{universe}: {line.status}, "
             f"{line.shortlisted} shortlisted{detail}"
         )
+    # OBS-004: a universe that quietly stopped being scannable is a warning about
+    # the integrity of THIS scan, so it sits above the results and is included at
+    # the summary-only content level too.
+    if report.universe_warnings:
+        lines += ["", "Universe warnings:"]
+        lines += [f"  - {warning}" for warning in report.universe_warnings]
     # ALERT-002: summary-only alerts stop here (status + counts); full alerts add
     # the per-stock results list below.
     if report.include_results:
diff --git a/backend/notifications/report.py b/backend/notifications/report.py
index 93a9408..e6174c1 100644
--- a/backend/notifications/report.py
+++ b/backend/notifications/report.py
@@ -79,6 +79,10 @@ class DailyScanReport:
     # renderer omits the per-stock results block. Defaults True (full) so existing
     # constructions keep the ALERT-001 behaviour.
     include_results: bool = True
+    # OBS-004: one line per universe that lost mapped symbols since the previous
+    # run. Rendered even at the summary-only content level - a shrinking universe
+    # is a data-integrity warning about the scan itself, not a per-stock result.
+    universe_warnings: tuple[str, ...] = ()
 
 
 def _status_label(outcome: DailyScanOutcome) -> str:
@@ -216,4 +220,7 @@ def build_daily_scan_report(
         top_results=top_results,
         app_url=settings.app_url,
         include_results=include_results,
+        # OBS-004: getattr keeps this tolerant of a DailyScanSummary built by
+        # older code (or a test fake) that predates the field.
+        universe_warnings=tuple(getattr(summary, "universe_warnings", ()) or ()),
     )
diff --git a/backend/observability/__init__.py b/backend/observability/__init__.py
index 15e9771..a27d95b 100644
--- a/backend/observability/__init__.py
+++ b/backend/observability/__init__.py
@@ -112,6 +112,13 @@
 EVENT_CANDLE_CACHE_REPAIR_COMPLETED = "candle_cache_repair_completed"
 EVENT_CANDLE_CACHE_SYMBOL_REPAIRED = "candle_cache_symbol_repaired"
 EVENT_CANDLE_CACHE_REPAIR_FAILED = "candle_cache_repair_failed"
+# OBS-004 universe mapping-health events. `_checked` is the routine per-universe
+# receipt (rows / mapped / unmapped) emitted on every check so the counts are
+# searchable even when nothing changed; `_regressed` fires only when a universe
+# has MORE unmapped symbols than the last recorded check - i.e. a stock silently
+# left the scannable set. Symbols and counts only, never prices.
+EVENT_UNIVERSE_HEALTH_CHECKED = "universe_health_checked"
+EVENT_UNIVERSE_MAPPING_REGRESSED = "universe_mapping_regressed"
 EVENT_AUTH_DENIED = "auth_denied"
 EVENT_DATA_REFRESH_STARTED = "data_refresh_started"
 EVENT_DATA_REFRESH_COMPLETED = "data_refresh_completed"
@@ -197,6 +204,8 @@
     "EVENT_SCAN_SCORING_FAILED",
     "EVENT_SCAN_STARTED",
     "EVENT_SYMBOL_SCAN_FAILED",
+    "EVENT_UNIVERSE_HEALTH_CHECKED",
+    "EVENT_UNIVERSE_MAPPING_REGRESSED",
     "ExceptionInfo",
     "JsonEventFormatter",
     "TextEventFormatter",
diff --git a/backend/storage/__init__.py b/backend/storage/__init__.py
index b994963..2f960d5 100644
--- a/backend/storage/__init__.py
+++ b/backend/storage/__init__.py
@@ -51,6 +51,7 @@
     ScanRun,
     ScanStatus,
     SignalForwardReturn,
+    UniverseHealthSnapshot,
     UserRole,
 )
 from backend.storage.repository import (
@@ -112,6 +113,7 @@
     "ScanStatus",
     "SessionLocal",
     "SignalForwardReturn",
+    "UniverseHealthSnapshot",
     "UserRole",
     "count_scan_results_for_runs",
     "count_user_role_admins",
diff --git a/backend/storage/models.py b/backend/storage/models.py
index 60d1520..1bc8b6b 100644
--- a/backend/storage/models.py
+++ b/backend/storage/models.py
@@ -1795,6 +1795,89 @@ def __repr__(self) -> str:
         )
 
 
+class UniverseHealthSnapshot(Base):
+    """OBS-004 - one row per universe per health check (the mapping baseline).
+
+    A symbol that Dhan's instrument master no longer contains is written into the
+    universe CSV as ``mapping_status='missing_security_id'`` and then filtered out
+    of every scan by ``mapped_only()``. Nothing outside the interactive Streamlit
+    sidebar ever reported that, so a universe could quietly shrink for weeks -
+    ~3% of the Hemant Good 200 list was already unscanned when OBS-004 was
+    written, and two more names dropped out mid-audit without a peep.
+
+    Beginner note:
+    This table exists purely so the daily job can answer *"is this worse than last
+    time?"*. Detecting an increase needs a previous number to compare against, and
+    the Render cron runs on an ephemeral filesystem with no disk - the shared
+    Postgres is the only thing that survives between runs. Hence a table rather
+    than a file next to the CSVs.
+
+    Rows are an append-only history: the check reads the newest row per universe
+    and then writes today's. Keeping the history (rather than upserting one row
+    per universe) means an operator can see *when* a symbol dropped out, which is
+    usually the question that follows the alert.
+
+    ``unmapped_symbols_json`` holds a bounded, sorted list of the symbol strings
+    that are currently unmapped, so the alert can name what changed instead of
+    just reporting a count. Symbols only - no prices, no credentials.
+    """
+
+    __tablename__ = "universe_health_snapshots"
+
+    # Same BigInt/SQLite-Integer variant as every other surrogate key here.
+    id: Mapped[int] = mapped_column(BigIntPrimaryKey, primary_key=True)
+
+    # Indexed together with universe_key: every read is "newest row for universe X".
+    captured_at: Mapped[dt.datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=lambda: dt.datetime.now(dt.UTC),
+        index=True,
+        comment="UTC time this universe was inspected",
+    )
+
+    universe_key: Mapped[str] = mapped_column(
+        String(100),
+        nullable=False,
+        index=True,
+        comment="Universe registry key, e.g. 'hemant_good_200'",
+    )
+
+    total_rows: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="Rows in the universe CSV"
+    )
+    mapped_rows: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="Rows the scanner can actually fetch"
+    )
+    # Stored rather than derived so a reader (and the comparison) never has to
+    # trust that total_rows and mapped_rows came from the same read.
+    unmapped_rows: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="Rows with no usable Dhan security_id"
+    )
+
+    # Bounded, sorted list of the unmapped symbols. Nullable so a universe whose
+    # CSV could not be read still leaves its header row behind as evidence.
+    unmapped_symbols_json: Mapped[dict[str, Any] | None] = mapped_column(
+        JSON, nullable=True, comment="Capped, sorted list of currently unmapped symbols"
+    )
+
+    __table_args__ = (
+        # The comparison query is "newest row for this universe", so lead with
+        # universe_key and let captured_at order within it.
+        Index(
+            "ix_universe_health_snapshots_key_captured",
+            "universe_key",
+            "captured_at",
+        ),
+    )
+
+    def __repr__(self) -> str:
+        return (
+            f"UniverseHealthSnapshot(universe_key={self.universe_key!r}, "
+            f"unmapped_rows={self.unmapped_rows!r})"
+        )
+
+
 # ============================================================================
 # NEXT: SCAN-002 (owner: Codex) — implement the database layer on top of this
 # schema. This file gives you the tables; SCAN-002 gives the app a way to talk
diff --git a/backend/storage/repository.py b/backend/storage/repository.py
index c75cae2..ead527f 100644
--- a/backend/storage/repository.py
+++ b/backend/storage/repository.py
@@ -34,6 +34,7 @@
     ScanRun,
     ScanStatus,
     SignalForwardReturn,
+    UniverseHealthSnapshot,
     UserRole,
 )
 
@@ -832,6 +833,66 @@ def get_latest_candle_repair_run(session: Session) -> CandleRepairRun | None:
     return session.scalars(stmt).first()
 
 
+def record_universe_health_snapshots(
+    session: Session,
+    snapshots: Sequence[Mapping[str, Any]],
+) -> list[UniverseHealthSnapshot]:
+    """Append one OBS-004 mapping-health row per universe and return them.
+
+    Beginner note:
+    Rows are appended, never updated. The comparison only ever reads the newest
+    row per universe, and keeping the history means an operator can answer "when
+    did this symbol drop out?" - which is the question that always follows the
+    alert. ``flush`` assigns the ids without ending the caller's transaction
+    (the caller owns it, per REFACTOR-002).
+
+    Each mapping needs ``universe_key``, ``total_rows``, ``mapped_rows`` and
+    ``unmapped_rows``; ``unmapped_symbols`` is optional and stored as JSON.
+    """
+    rows = [
+        UniverseHealthSnapshot(
+            universe_key=str(snapshot["universe_key"]),
+            total_rows=int(snapshot.get("total_rows", 0)),
+            mapped_rows=int(snapshot.get("mapped_rows", 0)),
+            unmapped_rows=int(snapshot.get("unmapped_rows", 0)),
+            unmapped_symbols_json=(
+                {"symbols": list(snapshot["unmapped_symbols"])}
+                if snapshot.get("unmapped_symbols") is not None
+                else None
+            ),
+        )
+        for snapshot in snapshots
+    ]
+    session.add_all(rows)
+    session.flush()
+    return rows
+
+
+def get_latest_universe_health_snapshots(
+    session: Session,
+) -> dict[str, UniverseHealthSnapshot]:
+    """Return the newest mapping-health row per universe, keyed by universe key.
+
+    Beginner note:
+    This is the baseline the daily job compares today's counts against. The rows
+    are read newest-first and the first one seen per universe wins, so a universe
+    with a long history still yields exactly one row. The primary-key tie-breaker
+    keeps the order deterministic when two checks land in the same millisecond.
+
+    An empty result means the check has never run - the caller must treat that as
+    "no baseline", not as "zero unmapped", or the very first run would alert on
+    every pre-existing unmapped symbol.
+    """
+    stmt = select(UniverseHealthSnapshot).order_by(
+        UniverseHealthSnapshot.captured_at.desc(),
+        UniverseHealthSnapshot.id.desc(),
+    )
+    latest: dict[str, UniverseHealthSnapshot] = {}
+    for row in session.scalars(stmt):
+        latest.setdefault(row.universe_key, row)
+    return latest
+
+
 def get_recent_audit_logs(
     session: Session,
     limit: int = 100,
diff --git a/docs/architecture/README.md b/docs/architecture/README.md
index f8954fa..bee5529 100644
--- a/docs/architecture/README.md
+++ b/docs/architecture/README.md
@@ -85,6 +85,7 @@ and dashboard freshness/provenance.
 - **[scan-run-persistence.md](scan-run-persistence.md)** — SCAN-001 scan-run persistence schema (the column-by-column rationale the Storage LLD links to).
 - **[scan-002-handoff.md](scan-002-handoff.md)** — SCAN-002 database-layer implementation handoff brief.
 - **[obs-003-audit-log.md](obs-003-audit-log.md)** — OBS-003 audit log + runtime-config schema, recorder design, and the seven tracked events.
+- **[obs-004-universe-health-alerts.md](obs-004-universe-health-alerts.md)** — OBS-004 universe mapping-health baseline: why the alert needs a database table, the two quiet rules (no baseline never regresses; only an increase counts), and why only the daily job owns the baseline.
 - **[valid-001-forward-return-validation.md](valid-001-forward-return-validation.md)** — VALID-001 methodology, no-lookahead rules, and schema rationale.
 - **[valid-002-handoff.md](valid-002-handoff.md)** — VALID-002 build brief plus resolved implementation decisions.
 - **[rank-001-final-scoring-model.md](rank-001-final-scoring-model.md)** — RANK-001 scoring methodology: the four v1 components, normalization, weighting, score ranges, missing-data behaviour, and the no-hidden-reasons invariant (no schema/migration).
diff --git a/docs/architecture/components/data-quality.md b/docs/architecture/components/data-quality.md
index f05af5f..623be08 100644
--- a/docs/architecture/components/data-quality.md
+++ b/docs/architecture/components/data-quality.md
@@ -3,9 +3,9 @@
 | | |
 |---|---|
 | **Component** | Reusable OHLCV candle-quality validation + scan-time quarantine, receipt, health surfacing, and prefetch-time repair |
-| **Source** | [`backend/data_quality/candles.py`](../../../backend/data_quality/candles.py), [`repair.py`](../../../backend/data_quality/repair.py), [`cache_repair.py`](../../../backend/data_quality/cache_repair.py), [`backend/data_quality/__init__.py`](../../../backend/data_quality/__init__.py); integrated in [`daily_data_loader.py`](../../../backend/daily_data_loader.py), [`scanning/service.py`](../../../backend/scanning/service.py), [`jobs/repair_candle_cache.py`](../../../backend/jobs/repair_candle_cache.py), [`health.py`](../../../backend/health.py), [`ui/health_page.py`](../../../ui/health_page.py) |
+| **Source** | [`backend/data_quality/candles.py`](../../../backend/data_quality/candles.py), [`repair.py`](../../../backend/data_quality/repair.py), [`cache_repair.py`](../../../backend/data_quality/cache_repair.py), [`universe_health.py`](../../../backend/data_quality/universe_health.py), [`backend/data_quality/__init__.py`](../../../backend/data_quality/__init__.py); integrated in [`daily_data_loader.py`](../../../backend/daily_data_loader.py), [`scanning/service.py`](../../../backend/scanning/service.py), [`jobs/repair_candle_cache.py`](../../../backend/jobs/repair_candle_cache.py), [`health.py`](../../../backend/health.py), [`ui/health_page.py`](../../../ui/health_page.py) |
 | **Layer** | Foundation checker + repair planner (pure, no I/O) + boundary integration in the data/scan layers |
-| **Status** | Stable (DATA-001A checker · DATA-001B integration · DATA-002 repair) |
+| **Status** | Stable (DATA-001A checker · DATA-001B integration · DATA-002 repair · OBS-004 universe mapping health) |
 | **Related** | [HLD](../high-level-design.md) · [data-002 repair design](../data-002-candle-cache-repair.md) · [data-acquisition.md](data-acquisition.md) · [scan-service-and-provenance.md](scan-service-and-provenance.md) · [storage-persistence.md](storage-persistence.md) · [health-monitoring.md](health-monitoring.md) · [observability.md](observability.md) · [security.md](security.md) |
 
 ## 1. Purpose & responsibilities
@@ -20,7 +20,7 @@ repair pass then runs at the end of the `python app.py` prefetch, so the app
 starts against the cleanest cache we can produce rather than silently dropping
 corrupt symbols from every scan.
 
-**Three parts:**
+**Four parts:**
 - **DATA-001A — checker** (`candles.py`): `validate_candles(...)` → an immutable
   `CandleQualityReport` of `DataQualityFinding`s with stable codes/severities.
   Pure, dependency-light (stdlib + pandas), never mutates the caller's frame, no
@@ -33,6 +33,15 @@ corrupt symbols from every scan.
   only the vendor can answer, **re-validates its own work**, and writes the cache
   back atomically. Full rationale in
   [data-002-candle-cache-repair.md](../data-002-candle-cache-repair.md).
+- **OBS-004 — universe mapping health** (`universe_health.py`): a different
+  question from the three above. They ask *"is this symbol's candle data sound?"*;
+  this asks *"is this symbol still in the scannable set at all?"*. A symbol that
+  leaves Dhan's instrument master is filtered out by `mapped_only()` and was
+  previously invisible to every headless path. The check compares today's
+  per-universe unmapped count against a persisted baseline
+  (`universe_health_snapshots`) and alerts only on an *increase*. Full rationale,
+  including why the prefetch logs but never records, in
+  [obs-004-universe-health-alerts.md](../obs-004-universe-health-alerts.md).
 
 ### Repair, in one paragraph
 
diff --git a/docs/architecture/obs-004-universe-health-alerts.md b/docs/architecture/obs-004-universe-health-alerts.md
new file mode 100644
index 0000000..fc30b4b
--- /dev/null
+++ b/docs/architecture/obs-004-universe-health-alerts.md
@@ -0,0 +1,128 @@
+# OBS-004 — Universe mapping-health alerts
+
+| | |
+|---|---|
+| **Ticket** | OBS-004 (issue [#119](https://github.com/DoRmAmMu1997/Streamlit-Scanner-App/issues/119)) |
+| **Source** | [`backend/data_quality/universe_health.py`](../../backend/data_quality/universe_health.py) · [`backend/storage/models.py`](../../backend/storage/models.py) (`UniverseHealthSnapshot`) · [`backend/jobs/run_daily_scan.py`](../../backend/jobs/run_daily_scan.py) · [`backend/notifications/`](../../backend/notifications/) · [`app.py`](../../app.py) |
+| **Migration** | `20260904obs004_create_universe_health_snapshots` |
+| **Status** | Shipped |
+| **Related** | [data-quality.md](components/data-quality.md) · [universe-management.md](components/universe-management.md) · [observability.md](components/observability.md) · [notifications.md](components/notifications.md) · [storage-persistence.md](components/storage-persistence.md) · [audit-2026-06.md](audit-2026-06.md) |
+
+## 1. The problem
+
+When Dhan's instrument master stops listing a symbol — a merger, a delisting, a
+ticker change — `refresh_universe_files()` writes it into the universe CSV as
+`mapping_status='missing_security_id'`, and `mapped_only()` then filters it out
+of every scan. That behaviour is correct: we cannot fetch candles for a security
+id we do not have.
+
+What was missing is that **nothing said so**. Before OBS-004:
+
+- `universe_status()` / `all_universe_statuses()` had exactly one non-test
+  caller, `ui/status_panel.py` — the interactive Streamlit sidebar.
+- `backend/jobs/run_daily_scan.py` emitted no mapping signal at all.
+- `backend/notifications/` never referenced `mapping_status`.
+
+So in production — where the Render cron runs `refresh_universe_files()` and then
+`run_daily_scan` with no human watching a sidebar — a universe could quietly
+shrink indefinitely. It already had: **~3% of the Hemant Good 200 list was
+unscanned** when this was found, and two more names (`JBCHEPHARM`, `GUJGASLTD`)
+dropped out mid-audit without a sound.
+
+Worth stating clearly, because it shapes the design: **the drop-outs were not a
+bug.** Both symbols were genuinely absent from the 2026-08-24 instrument master
+by symbol *and* company name, and the snapshot itself was intact (213,213 rows).
+This is a real-world vendor event that the system should *report*, not prevent.
+
+## 2. Design
+
+### 2.1 Three pieces, split by requirement
+
+| Piece | Kind | Why separate |
+|---|---|---|
+| `collect_universe_health()` | pure | Reads CSVs, returns counts + names. Anything may call it. |
+| `detect_mapping_regressions()` | pure | Compares today to a baseline. Trivially testable, no I/O. |
+| `check_universe_health(session)` | stateful | Needs a database session, because "worse than last time" needs a durable baseline. |
+
+### 2.2 Why a database table and not a file
+
+The Render daily-scan cron runs on an **ephemeral filesystem with no disk**
+(`render.yaml` attaches the disk to the web service only). Anything written next
+to the universe CSVs is gone before the next run, so a file-based baseline would
+mean the alert could never fire in the one environment that needs it. The shared
+Postgres is the only state that survives between runs.
+
+`universe_health_snapshots` is **append-only** rather than one upserted row per
+universe. The question that always follows "GUJGASLTD dropped out" is "when?",
+and keeping the history answers it for free. The read path only ever wants the
+newest row per universe, which `ix_universe_health_snapshots_key_captured`
+serves directly.
+
+### 2.3 The two quiet rules
+
+Both exist to keep the alert credible enough that nobody mutes the channel:
+
+1. **A universe with no previous row never regresses.** The first check has no
+   baseline. Treating "absent" as zero would alert on every pre-existing unmapped
+   symbol on first run.
+2. **Only an increase counts.** A universe sitting at a steady three unmapped
+   symbols is already-known damage and stays silent. Recovery (the count going
+   down) is good news, not an alert.
+
+### 2.4 Only the alerting path owns the baseline
+
+This is the subtle one. Whoever *writes* the baseline defines what "last time"
+means. If the morning Streamlit prefetch also recorded a snapshot, a symbol that
+dropped out at 09:00 would already be part of the baseline by the time the
+evening job ran — and the alert would never fire on any machine where both run.
+
+So:
+
+- **`app.py` prefetch** calls `collect_universe_health()` + `log_universe_health()`
+  — observability only, no persistence.
+- **`run_daily_scan`** calls `check_universe_health()` — collect, log, compare,
+  **then** record. Reading the baseline before writing today's snapshot is what
+  makes the alert fire exactly once.
+
+### 2.5 Bounded by construction
+
+`MAX_REPORTED_SYMBOLS = 25` caps both the stored JSON and the alert text. A
+universe whose CSV went badly wrong must not be able to write an unbounded blob
+into Postgres or a multi-page message into Telegram; past that point the count
+alone tells the story. Only symbol strings and counts are stored — never prices,
+never credentials — and the alert text still goes through `redact_text` like
+every other notification.
+
+## 3. Failure posture
+
+Every entry point is wrapped: `_check_universe_health()` in the daily job catches
+broadly and returns `()`, and the prefetch helper does the same. A universe CSV
+that will not parse is a reason to warn, never a reason to skip the night's scan.
+The health check can therefore never change the job's exit code — the same
+contract ALERT-001 gives notifications.
+
+## 4. What an operator sees
+
+Structured log events on every run (`universe_health_checked`, one per universe,
+carrying `rows` / `mapped` / `unmapped`), a `universe_mapping_regressed` warning
+when something got worse, a printed line in the job's stdout, and a
+**Universe warnings** block in the daily alert:
+
+```
+Universe warnings:
+  - hemant_good_200: 6 -> 8 unmapped (+2); GUJGASLTD, JBCHEPHARM
+```
+
+That block renders even at the ALERT-002 summary-only content level. A shrinking
+universe is a warning about the integrity of *this scan*, not a per-stock result,
+so it is not suppressed alongside the results list.
+
+## 5. Alternatives considered
+
+| Option | Why not |
+|---|---|
+| Alert on an absolute threshold ("more than 5 unmapped") | Every universe would need its own tuned number, and a slow drift below the threshold stays invisible — which is the exact failure being fixed. |
+| Store the baseline in `app_config` | That table is admin *config overrides*, read wholesale by `apply_config_overrides()`. Injecting job state into it risks a key being applied as a setting. |
+| Upsert one row per universe | Cheaper, but throws away the "when did this happen?" answer for no real saving — these rows are tiny and written once a day. |
+| Fail the scan when a universe shrinks | Wrong severity. A vendor delisting is normal; the scan over the remaining symbols is still valid and useful. |
+| Diff the CSVs in git | Only works on a developer machine. The production cron has no checkout and no disk. |
diff --git a/migrations/versions/20260904obs004_create_universe_health_snapshots.py b/migrations/versions/20260904obs004_create_universe_health_snapshots.py
new file mode 100644
index 0000000..be9db41
--- /dev/null
+++ b/migrations/versions/20260904obs004_create_universe_health_snapshots.py
@@ -0,0 +1,91 @@
+"""Create the universe mapping-health baseline table for OBS-004.
+
+Adds ``universe_health_snapshots``: one row per universe per health check (see
+``backend/data_quality/universe_health.py`` and ``backend/jobs/run_daily_scan.py``).
+
+Why a table and not a file: the check has to answer *"is this worse than last
+time?"*, which needs a previous number to compare against. The Render daily-scan
+cron deliberately runs on an **ephemeral filesystem with no disk**, so anything
+written next to the universe CSVs is gone before the next run. The shared
+Postgres is the only state that survives between runs.
+
+Why append-only rather than one upserted row per universe: the question that
+always follows "GUJGASLTD dropped out" is "when?". Keeping the history answers it
+for free, and the read path only ever wants the newest row per universe - which
+``ix_universe_health_snapshots_key_captured`` serves directly.
+
+``unmapped_symbols_json`` is nullable so a universe whose CSV could not be read
+still leaves its header row behind as evidence. The drift test in
+``tests/test_scan_storage_migrations.py`` keeps this in sync with the ORM model.
+
+Revision ID: 20260904obs004
+Revises: 20260820ipo011
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# Alembic reads these module globals to order migrations: ``revision`` is this
+# script's id and ``down_revision`` is the one it must run after.
+revision = "20260904obs004"
+down_revision = "20260820ipo011"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "universe_health_snapshots",
+        # BigInteger with a SQLite Integer variant, matching every other
+        # surrogate key in this schema so autoincrement behaves the same on both.
+        sa.Column(
+            "id",
+            sa.BigInteger().with_variant(sa.Integer, "sqlite"),
+            primary_key=True,
+            nullable=False,
+        ),
+        sa.Column("captured_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("universe_key", sa.String(length=100), nullable=False),
+        sa.Column("total_rows", sa.Integer(), nullable=False),
+        sa.Column("mapped_rows", sa.Integer(), nullable=False),
+        sa.Column("unmapped_rows", sa.Integer(), nullable=False),
+        # sa.JSON maps to JSON on Postgres and JSON-encoded TEXT on SQLite, so the
+        # same code path serves tests and production.
+        sa.Column("unmapped_symbols_json", sa.JSON(), nullable=True),
+    )
+    # Single-column indexes mirror the ORM's ``index=True`` columns.
+    op.create_index(
+        "ix_universe_health_snapshots_captured_at",
+        "universe_health_snapshots",
+        ["captured_at"],
+    )
+    op.create_index(
+        "ix_universe_health_snapshots_universe_key",
+        "universe_health_snapshots",
+        ["universe_key"],
+    )
+    # The composite index the comparison actually uses: newest row per universe.
+    op.create_index(
+        "ix_universe_health_snapshots_key_captured",
+        "universe_health_snapshots",
+        ["universe_key", "captured_at"],
+    )
+
+
+def downgrade() -> None:
+    """Drop the indexes before the table so the schema unwinds cleanly."""
+    op.drop_index(
+        "ix_universe_health_snapshots_key_captured",
+        table_name="universe_health_snapshots",
+    )
+    op.drop_index(
+        "ix_universe_health_snapshots_universe_key",
+        table_name="universe_health_snapshots",
+    )
+    op.drop_index(
+        "ix_universe_health_snapshots_captured_at",
+        table_name="universe_health_snapshots",
+    )
+    op.drop_table("universe_health_snapshots")
diff --git a/tests/test_scan_storage_migrations.py b/tests/test_scan_storage_migrations.py
index e3ee890..0469355 100644
--- a/tests/test_scan_storage_migrations.py
+++ b/tests/test_scan_storage_migrations.py
@@ -108,6 +108,7 @@ def test_alembic_upgrade_and_downgrade_use_temp_sqlite(monkeypatch, tmp_path: Pa
         "scan_runs",
         "scan_results",
         "signal_forward_returns",
+        "universe_health_snapshots",
         "user_roles",
     }
     assert {index["name"] for index in inspector.get_indexes("audit_logs")} >= {
@@ -777,6 +778,7 @@ def test_ensure_database_schema_creates_tables_and_short_circuits(monkeypatch, t
         "scan_runs",
         "scan_results",
         "signal_forward_returns",
+        "universe_health_snapshots",
         "user_roles",
     }
     engine.dispose()
diff --git a/tests/test_universe_health.py b/tests/test_universe_health.py
new file mode 100644
index 0000000..56aa854
--- /dev/null
+++ b/tests/test_universe_health.py
@@ -0,0 +1,394 @@
+"""Tests for the OBS-004 universe mapping-health check.
+
+These use the shared in-memory ``db_session`` fixture and tiny hand-written
+universe CSVs in ``tmp_path``, so nothing touches Dhan, the network, or the
+developer's real database.
+
+The behaviour under test is a *comparison against a stored baseline*, so most of
+these tests run the check twice and assert on what changed between the two runs.
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+
+import pandas as pd
+import pytest
+
+from backend.data_quality.universe_health import (
+    MAX_REPORTED_SYMBOLS,
+    MappingRegression,
+    UniverseHealth,
+    check_universe_health,
+    collect_universe_health,
+    detect_mapping_regressions,
+    log_universe_health,
+)
+from backend.storage import repository
+
+
+def _write_universe(directory, universe_key, rows):
+    """Write a minimal universe CSV of ``(symbol, mapping_status)`` pairs."""
+    frame = pd.DataFrame(
+        [
+            {
+                "universe": universe_key,
+                "symbol": symbol,
+                "security_id": "1234" if status == "mapped" else "",
+                "mapping_status": status,
+            }
+            for symbol, status in rows
+        ]
+    )
+    frame.to_csv(directory / f"{universe_key}.csv", index=False)
+
+
+@pytest.fixture
+def universe_dir(tmp_path, monkeypatch):
+    """A universe directory holding exactly one registered universe key."""
+    from backend import universe_builder
+
+    # Restrict the registry so the test does not depend on how many universes the
+    # real UNIVERSE_CONFIG happens to hold today.
+    monkeypatch.setattr(
+        universe_builder,
+        "UNIVERSE_CONFIG",
+        {"nifty_100": {"file_name": "nifty_100.csv", "display_name": "NIFTY 100"}},
+    )
+    return tmp_path
+
+
+def test_collect_reports_counts_and_names_the_unmapped_symbols(universe_dir):
+    _write_universe(
+        universe_dir,
+        "nifty_100",
+        [("RELIANCE", "mapped"), ("TCS", "mapped"), ("GUJGASLTD", "missing_security_id")],
+    )
+
+    (health,) = collect_universe_health(universe_dir)
+
+    assert health.universe_key == "nifty_100"
+    assert health.total_rows == 3
+    assert health.mapped_rows == 2
+    assert health.unmapped_rows == 1
+    # A count alone makes a useless alert; the names are what an operator acts on.
+    assert health.unmapped_symbols == ("GUJGASLTD",)
+
+
+def test_collect_survives_a_missing_universe_file(universe_dir):
+    """A health check must never be the reason the daily job dies."""
+    (health,) = collect_universe_health(universe_dir)
+
+    assert health.total_rows == 0
+    assert health.mapped_rows == 0
+    assert health.unmapped_symbols == ()
+
+
+def test_collect_caps_the_reported_symbol_list(universe_dir):
+    """A badly broken CSV must not write an unbounded blob or a huge alert."""
+    rows = [(f"SYM{index:03d}", "missing_security_id") for index in range(MAX_REPORTED_SYMBOLS + 10)]
+    _write_universe(universe_dir, "nifty_100", rows)
+
+    (health,) = collect_universe_health(universe_dir)
+
+    assert health.unmapped_rows == MAX_REPORTED_SYMBOLS + 10
+    assert len(health.unmapped_symbols) == MAX_REPORTED_SYMBOLS
+
+
+def test_no_baseline_never_regresses():
+    """The first check has nothing to compare against, so it must stay quiet.
+
+    Treating "absent" as zero would alert on every pre-existing unmapped symbol
+    on first run - exactly the noise that makes people mute an alert channel.
+    """
+    current = [
+        UniverseHealth(
+            universe_key="nifty_100",
+            total_rows=10,
+            mapped_rows=7,
+            unmapped_symbols=("A", "B", "C"),
+        )
+    ]
+
+    assert detect_mapping_regressions(current, {}) == ()
+
+
+def test_steady_state_and_recovery_do_not_regress():
+    """Already-known damage stays quiet, and getting better is not an alert."""
+
+    class _Baseline:
+        unmapped_rows = 3
+        unmapped_symbols_json = {"symbols": ["A", "B", "C"]}
+
+    steady = [
+        UniverseHealth(
+            universe_key="nifty_100", total_rows=10, mapped_rows=7, unmapped_symbols=("A", "B", "C")
+        )
+    ]
+    recovered = [
+        UniverseHealth(
+            universe_key="nifty_100", total_rows=10, mapped_rows=9, unmapped_symbols=("A",)
+        )
+    ]
+
+    assert detect_mapping_regressions(steady, {"nifty_100": _Baseline()}) == ()
+    assert detect_mapping_regressions(recovered, {"nifty_100": _Baseline()}) == ()
+
+
+def test_regression_names_only_the_newly_unmapped_symbols():
+    class _Baseline:
+        unmapped_rows = 1
+        unmapped_symbols_json = {"symbols": ["A"]}
+
+    current = [
+        UniverseHealth(
+            universe_key="nifty_100", total_rows=10, mapped_rows=8, unmapped_symbols=("A", "GUJGASLTD")
+        )
+    ]
+
+    (regression,) = detect_mapping_regressions(current, {"nifty_100": _Baseline()})
+
+    assert regression.previous_unmapped == 1
+    assert regression.current_unmapped == 2
+    # "A" was already known; only the new drop-out is worth naming.
+    assert regression.newly_unmapped == ("GUJGASLTD",)
+
+
+def test_describe_is_a_single_actionable_line():
+    regression = MappingRegression(
+        universe_key="hemant_good_200",
+        previous_unmapped=6,
+        current_unmapped=8,
+        newly_unmapped=("GUJGASLTD", "JBCHEPHARM"),
+    )
+
+    assert regression.describe() == (
+        "hemant_good_200: 6 -> 8 unmapped (+2); GUJGASLTD, JBCHEPHARM"
+    )
+
+
+def test_log_universe_health_emits_one_event_per_universe(caplog):
+    snapshots = [
+        UniverseHealth(
+            universe_key="nifty_100", total_rows=10, mapped_rows=8, unmapped_symbols=("A", "B")
+        )
+    ]
+
+    with caplog.at_level(logging.INFO):
+        log_universe_health(snapshots)
+
+    # log_event stashes the key/value detail on the record as `structured_fields`
+    # (see backend/observability), which is how the other suites read it back.
+    fields = [
+        getattr(record, "structured_fields", {})
+        for record in caplog.records
+        if getattr(record, "event", None) == "universe_health_checked"
+    ]
+    assert len(fields) == 1
+    assert fields[0] == {
+        "universe_key": "nifty_100",
+        "rows": 10,
+        "mapped": 8,
+        "unmapped": 2,
+    }
+
+
+def test_check_records_a_baseline_and_stays_quiet_on_the_first_run(db_session, universe_dir):
+    _write_universe(
+        universe_dir, "nifty_100", [("RELIANCE", "mapped"), ("GUJGASLTD", "missing_security_id")]
+    )
+
+    report = check_universe_health(db_session, universe_dir=universe_dir)
+
+    assert report.regressions == ()
+    stored = repository.get_latest_universe_health_snapshots(db_session)
+    assert stored["nifty_100"].unmapped_rows == 1
+    assert stored["nifty_100"].unmapped_symbols_json == {"symbols": ["GUJGASLTD"]}
+
+
+def test_check_alerts_exactly_once_when_a_symbol_drops_out(db_session, universe_dir, caplog):
+    """The whole point of OBS-004: one alert on the run where it happens."""
+    _write_universe(universe_dir, "nifty_100", [("RELIANCE", "mapped"), ("TCS", "mapped")])
+    assert check_universe_health(db_session, universe_dir=universe_dir).regressions == ()
+
+    # TCS leaves Dhan's master overnight.
+    _write_universe(
+        universe_dir, "nifty_100", [("RELIANCE", "mapped"), ("TCS", "missing_security_id")]
+    )
+    with caplog.at_level(logging.WARNING):
+        second = check_universe_health(db_session, universe_dir=universe_dir)
+
+    (regression,) = second.regressions
+    assert regression.newly_unmapped == ("TCS",)
+    assert any(
+        getattr(record, "event", None) == "universe_mapping_regressed"
+        for record in caplog.records
+    )
+
+    # Third run: nothing further changed, so the alert must NOT repeat. This is
+    # the assertion that proves the baseline write actually happened.
+    third = check_universe_health(db_session, universe_dir=universe_dir)
+    assert third.regressions == ()
+
+
+def test_check_reads_the_baseline_before_writing_todays_snapshot(db_session, universe_dir):
+    """Ordering guard: comparing after the write would make regression impossible."""
+    _write_universe(universe_dir, "nifty_100", [("RELIANCE", "mapped"), ("TCS", "mapped")])
+    check_universe_health(db_session, universe_dir=universe_dir)
+    _write_universe(
+        universe_dir, "nifty_100", [("RELIANCE", "mapped"), ("TCS", "missing_security_id")]
+    )
+
+    report = check_universe_health(db_session, universe_dir=universe_dir)
+
+    # Two checks, two history rows retained (append-only), and a real regression.
+    assert len(report.regressions) == 1
+    rows = db_session.query(repository.UniverseHealthSnapshot).all()
+    assert len(rows) == 2
+
+
+# ---------------------------------------------------------------------------
+# Wiring: the daily job, the alert, and the Streamlit prefetch.
+# ---------------------------------------------------------------------------
+
+
+def test_daily_job_surfaces_regressions_without_ever_failing(
+    session_factory, universe_dir, monkeypatch, capsys
+):
+    """The job prints and returns warnings, and a broken check stays non-fatal."""
+    from backend.jobs import run_daily_scan as job
+
+    _write_universe(universe_dir, "nifty_100", [("RELIANCE", "mapped"), ("TCS", "mapped")])
+    monkeypatch.setattr(
+        "backend.config.UNIVERSE_DIR", universe_dir, raising=False
+    )
+
+    import backend.data_quality.universe_health as health_module
+
+    monkeypatch.setattr(
+        health_module,
+        "check_universe_health",
+        lambda session, **_: health_module.UniverseHealthReport(
+            regressions=(
+                health_module.MappingRegression(
+                    universe_key="nifty_100",
+                    previous_unmapped=0,
+                    current_unmapped=1,
+                    newly_unmapped=("TCS",),
+                ),
+            )
+        ),
+    )
+
+    warnings = job._check_universe_health(session_factory, sys.stdout)
+
+    assert warnings == ("nifty_100: 0 -> 1 unmapped (+1); TCS",)
+    assert "Universe mapping regressed" in capsys.readouterr().out
+
+
+def test_daily_job_swallows_a_broken_health_check(monkeypatch):
+    """A universe CSV that will not parse must never take the night's scan down."""
+    from backend.jobs import run_daily_scan as job
+
+    def _explode():
+        raise RuntimeError("database unavailable")
+
+    assert job._check_universe_health(_explode, sys.stdout) == ()
+
+
+def test_summary_defaults_keep_existing_constructions_working():
+    """Every pre-OBS-004 DailyScanSummary(...) call still has to work."""
+    from backend.jobs.run_daily_scan import DailyScanSummary
+
+    assert DailyScanSummary(outcomes=[]).universe_warnings == ()
+
+
+def test_alert_renders_universe_warnings_even_in_summary_only_mode():
+    """A shrinking universe is a warning about the scan, not a per-stock result."""
+    from backend.notifications.render import render_telegram
+    from backend.notifications.report import DailyScanReport
+
+    report = DailyScanReport(
+        ok=True,
+        screeners=(),
+        total_symbols_scanned=100,
+        total_shortlisted=3,
+        failed_count=0,
+        failed_symbols_or_findings=0,
+        top_results=(),
+        app_url="",
+        # ALERT-002 summary-only: the results block is suppressed, but the
+        # integrity warning must still reach the operator.
+        include_results=False,
+        universe_warnings=("hemant_good_200: 6 -> 8 unmapped (+2); GUJGASLTD",),
+    )
+
+    text = render_telegram(report)
+
+    assert "Universe warnings:" in text
+    assert "hemant_good_200: 6 -> 8 unmapped (+2); GUJGASLTD" in text
+    assert "Top results:" not in text
+
+
+def test_alert_omits_the_warning_block_when_everything_is_healthy():
+    from backend.notifications.render import render_telegram
+    from backend.notifications.report import DailyScanReport
+
+    report = DailyScanReport(
+        ok=True,
+        screeners=(),
+        total_symbols_scanned=100,
+        total_shortlisted=0,
+        failed_count=0,
+        failed_symbols_or_findings=0,
+        top_results=(),
+        app_url="",
+    )
+
+    assert "Universe warnings:" not in render_telegram(report)
+
+
+def test_prefetch_logs_health_without_recording_a_baseline(monkeypatch, caplog):
+    """The prefetch must not move the baseline, or the evening alert never fires."""
+    import app as app_module
+
+    recorded: list[str] = []
+    monkeypatch.setattr(
+        "backend.data_quality.universe_health.collect_universe_health",
+        lambda *_args, **_kwargs: (
+            UniverseHealth(
+                universe_key="nifty_100",
+                total_rows=5,
+                mapped_rows=4,
+                unmapped_symbols=("TCS",),
+            ),
+        ),
+    )
+    monkeypatch.setattr(
+        "backend.storage.repository.record_universe_health_snapshots",
+        lambda *args, **kwargs: recorded.append("written"),
+    )
+
+    with caplog.at_level(logging.INFO):
+        app_module._log_universe_health()
+
+    assert recorded == []
+    assert any(
+        getattr(record, "event", None) == "universe_health_checked"
+        for record in caplog.records
+    )
+
+
+def test_prefetch_health_logging_is_best_effort(monkeypatch):
+    """A failure here must not stop the prefetch from launching Streamlit."""
+    import app as app_module
+
+    def _explode(*_args, **_kwargs):
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr(
+        "backend.data_quality.universe_health.collect_universe_health", _explode
+    )
+
+    app_module._log_universe_health()  # must not raise