"""Streaming inspection for the NSE 15-minute ZIP archive. This module intentionally reads the CSV member directly from the ZIP. It never extracts the original archive and keeps only bounded per-symbol accounting state. """ from __future__ import annotations import csv import hashlib import json import math import zipfile from collections import Counter from collections.abc import Callable from dataclasses import asdict, dataclass, field from datetime import datetime from pathlib import Path from typing import TextIO EXPECTED_COLUMNS = ("date", "open", "high", "low", "close", "volume", "symbol") TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S" class ArchiveAuditError(ValueError): """Raised when an archive cannot be audited safely.""" @dataclass(slots=True) class SymbolAudit: """Bounded quality counters for one raw source symbol.""" rows: int = 0 first_timestamp: str | None = None last_timestamp: str | None = None zero_volume_rows: int = 0 missing_or_invalid_rows: int = 0 invalid_ohlc_rows: int = 0 duplicate_timestamps: int = 0 timestamp_regressions: int = 0 intraday_unexpected_gaps: int = 0 bars_by_weekday: Counter[str] = field(default_factory=Counter) _last_timestamp: datetime | None = None def serializable(self) -> dict[str, object]: payload = asdict(self) payload.pop("_last_timestamp", None) payload["bars_by_weekday"] = dict(sorted(self.bars_by_weekday.items())) return payload @dataclass(frozen=True, slots=True) class ArchiveAudit: """Serializable report used as a reproducible raw-dataset descriptor.""" archive_path: str archive_bytes: int archive_sha256: str member_name: str member_bytes: int columns: tuple[str, ...] rows: int symbols: int first_timestamp: str | None last_timestamp: str | None zero_volume_rows: int missing_or_invalid_rows: int invalid_ohlc_rows: int duplicate_timestamps: int timestamp_regressions: int intraday_unexpected_gaps: int source_symbols_reopened: int symbol_stats: dict[str, dict[str, object]] def to_json(self) -> str: return json.dumps(asdict(self), indent=2, sort_keys=True) def _hash_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _select_csv_member(archive: zipfile.ZipFile) -> zipfile.ZipInfo: files = [entry for entry in archive.infolist() if not entry.is_dir()] csv_members = [entry for entry in files if entry.filename.lower().endswith(".csv")] if len(csv_members) != 1: raise ArchiveAuditError( f"Expected exactly one CSV member; found {len(csv_members)} in {archive.filename}." ) return csv_members[0] def _parse_row( row: dict[str, str | None], ) -> tuple[datetime, float, float, float, float, float, str]: try: timestamp = datetime.strptime(row["date"] or "", TIMESTAMP_FORMAT) open_price = float(row["open"] or "") high = float(row["high"] or "") low = float(row["low"] or "") close = float(row["close"] or "") volume = float(row["volume"] or "") symbol = (row["symbol"] or "").strip() except (KeyError, TypeError, ValueError) as exc: raise ArchiveAuditError("Row has missing or unparsable required fields.") from exc if not symbol: raise ArchiveAuditError("Row has blank symbol.") return timestamp, open_price, high, low, close, volume, symbol def _is_valid_ohlcv( open_price: float, high: float, low: float, close: float, volume: float ) -> bool: values = (open_price, high, low, close, volume) if not all(math.isfinite(value) for value in values): return False if min(open_price, high, low, close) <= 0 or volume < 0: return False return high >= max(open_price, close) and low <= min(open_price, close) and high >= low def _write_progress(callback: Callable[[str], None] | None, rows: int) -> None: if callback is not None and rows % 1_000_000 == 0: callback(f"Audited {rows:,} rows.") def audit_archive( archive_path: Path, *, progress: Callable[[str], None] | None = None, ) -> ArchiveAudit: """Audit one raw archive with bounded memory. The source timestamps have no timezone designator. This function preserves them as naïve values and does not infer or convert a timezone. """ if not archive_path.is_file(): raise ArchiveAuditError(f"Archive does not exist: {archive_path}") archive_hash = _hash_file(archive_path) symbol_stats: dict[str, SymbolAudit] = {} source_symbols_closed: set[str] = set() current_source_symbol: str | None = None source_symbols_reopened = 0 rows = 0 first_timestamp: datetime | None = None last_timestamp: datetime | None = None totals: Counter[str] = Counter() with zipfile.ZipFile(archive_path) as archive: member = _select_csv_member(archive) with archive.open(member) as raw_stream, open_text_stream(raw_stream) as text_stream: reader = csv.DictReader(text_stream) if tuple(reader.fieldnames or ()) != EXPECTED_COLUMNS: raise ArchiveAuditError( f"Unexpected CSV schema {reader.fieldnames!r}; expected {EXPECTED_COLUMNS!r}." ) for row in reader: rows += 1 try: timestamp, open_price, high, low, close, volume, symbol = _parse_row(row) except ArchiveAuditError: totals["missing_or_invalid_rows"] += 1 _write_progress(progress, rows) continue if current_source_symbol is not None and symbol != current_source_symbol: source_symbols_closed.add(current_source_symbol) if symbol in source_symbols_closed: source_symbols_reopened += 1 current_source_symbol = symbol elif current_source_symbol is None: current_source_symbol = symbol stats = symbol_stats.setdefault(symbol, SymbolAudit()) stats.rows += 1 stats.first_timestamp = stats.first_timestamp or timestamp.isoformat(sep=" ") stats.last_timestamp = timestamp.isoformat(sep=" ") stats.bars_by_weekday[timestamp.strftime("%A")] += 1 if not _is_valid_ohlcv(open_price, high, low, close, volume): stats.invalid_ohlc_rows += 1 totals["invalid_ohlc_rows"] += 1 if volume == 0: stats.zero_volume_rows += 1 totals["zero_volume_rows"] += 1 if stats._last_timestamp is not None: delta_seconds = (timestamp - stats._last_timestamp).total_seconds() if delta_seconds == 0: stats.duplicate_timestamps += 1 totals["duplicate_timestamps"] += 1 elif delta_seconds < 0: stats.timestamp_regressions += 1 totals["timestamp_regressions"] += 1 elif ( timestamp.date() == stats._last_timestamp.date() and delta_seconds != 15 * 60 ): stats.intraday_unexpected_gaps += 1 totals["intraday_unexpected_gaps"] += 1 stats._last_timestamp = timestamp first_timestamp = min(first_timestamp, timestamp) if first_timestamp else timestamp last_timestamp = max(last_timestamp, timestamp) if last_timestamp else timestamp _write_progress(progress, rows) report = ArchiveAudit( archive_path=str(archive_path.resolve()), archive_bytes=archive_path.stat().st_size, archive_sha256=archive_hash, member_name=member.filename, member_bytes=member.file_size, columns=EXPECTED_COLUMNS, rows=rows, symbols=len(symbol_stats), first_timestamp=first_timestamp.isoformat(sep=" ") if first_timestamp else None, last_timestamp=last_timestamp.isoformat(sep=" ") if last_timestamp else None, zero_volume_rows=totals["zero_volume_rows"], missing_or_invalid_rows=totals["missing_or_invalid_rows"], invalid_ohlc_rows=totals["invalid_ohlc_rows"], duplicate_timestamps=totals["duplicate_timestamps"], timestamp_regressions=totals["timestamp_regressions"], intraday_unexpected_gaps=totals["intraday_unexpected_gaps"], source_symbols_reopened=source_symbols_reopened, symbol_stats={ symbol: stats.serializable() for symbol, stats in sorted(symbol_stats.items()) }, ) if progress is not None: progress(f"Completed audit of {rows:,} rows and {len(symbol_stats):,} symbols.") return report class open_text_stream: """Context manager for ZIP byte streams decoded as UTF-8 CSV text.""" def __init__(self, binary_stream: object) -> None: from io import TextIOWrapper self._stream: TextIO = TextIOWrapper(binary_stream, encoding="utf-8", newline="") def __enter__(self) -> TextIO: return self._stream def __exit__(self, *_: object) -> None: self._stream.close()