From 9d77c68186c2bc46fe13272fe234e39f17bb65ef Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Fri, 4 Sep 2026 20:05:08 +0530 Subject: [PATCH] feat(OBS-004): alert when a universe silently loses mapped symbols When Dhan's instrument master stops listing a symbol, refresh_universe_files() marks it `missing_security_id` and mapped_only() filters it out of every scan. That is correct - we cannot fetch candles for a security id we do not have - but 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. So in production, where the Render cron refreshes universes and scans with no human watching a sidebar, a universe could shrink indefinitely. It already had: ~3% of the Hemant Good 200 list was unscanned when this was found, and two more names dropped out mid-audit without a sound. Both were genuine vendor events - absent from the 2026-08-24 master by symbol AND company name, with the snapshot itself intact - so the system should report them, not prevent them. Three pieces, split by requirement: collect_universe_health() and detect_mapping_regressions() are pure and trivially testable; check_universe_health() is the stateful one because "worse than last time" needs a durable baseline. The baseline is a database table, not a file, because the Render daily-scan cron runs on an ephemeral filesystem with no disk - a file-based baseline could never fire the alert in the one environment that needs it. Rows are append-only: the question after "GUJGASLTD dropped out" is always "when?". Two rules keep the alert credible enough that nobody mutes the channel. A universe with no previous row never regresses, so the first run does not alert on every pre-existing unmapped symbol. And only an increase counts, so steady-state damage stays quiet and recovery is not an alert either. Only the alerting path owns the baseline. Whoever writes it defines "last time", so if the morning prefetch also recorded, a symbol dropping out at 09:00 would already be baseline by the evening and the alert would never fire. The prefetch therefore logs health without persisting; the daily job reads the baseline, compares, and only then records today's snapshot - which is what makes the alert fire exactly once. Bounded by construction: MAX_REPORTED_SYMBOLS caps both the stored JSON and the alert text, and only symbols and counts are stored, never prices. Every entry point catches broadly and degrades to no-alert: a universe CSV that will not parse is a reason to warn, never a reason to skip the night's scan. Ships the Alembic migration in the same commit and updates both hardcoded table-name sets in tests/test_scan_storage_migrations.py, per AGENTS.md section 7. Design rationale, including the alternatives rejected, in docs/architecture/obs-004-universe-health-alerts.md. Closes #119 Co-Authored-By: Claude Opus 5 --- AGENTS.md | 3 +- app.py | 24 ++ backend/data_quality/universe_health.py | 292 +++++++++++++ backend/jobs/run_daily_scan.py | 44 +- backend/notifications/render.py | 6 + backend/notifications/report.py | 7 + backend/observability/__init__.py | 9 + backend/storage/__init__.py | 2 + backend/storage/models.py | 83 ++++ backend/storage/repository.py | 61 +++ docs/architecture/README.md | 1 + docs/architecture/components/data-quality.md | 15 +- .../obs-004-universe-health-alerts.md | 128 ++++++ ...obs004_create_universe_health_snapshots.py | 91 ++++ tests/test_scan_storage_migrations.py | 2 + tests/test_universe_health.py | 394 ++++++++++++++++++ 16 files changed, 1157 insertions(+), 5 deletions(-) create mode 100644 backend/data_quality/universe_health.py create mode 100644 docs/architecture/obs-004-universe-health-alerts.md create mode 100644 migrations/versions/20260904obs004_create_universe_health_snapshots.py create mode 100644 tests/test_universe_health.py 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