Spaces:
Running on Zero
Running on Zero
| """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.""" | |
| 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 | |
| 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() | |