SniperGold_ML/ml/p3/p3_s251_external_ingest/s251_ingest.py

# -*- coding: utf-8 -*-
"""P3-S25.1 STREAMING INGEST ORCHESTRATOR (chunk manager + hashing + resume).

Deterministic line-safe chunked reader of the external Tickstory CSV:

  FILE -> byte/line-safe chunk -> parser -> UTC epoch -> chunk validation ->
  M15 aggregation -> M30 aggregation -> atomic checkpoint -> next chunk

Chunk raw hash = SHA-256 of the exact raw byte span of the chunk
[byte_start, byte_end); parsed hash = SHA-256 of canonical parsed records.
Rows are appended to output files ONLY after the chunk is fully processed, so
an interrupted chunk is reprocessed deterministically and no duplicates are
possible.
"""
import hashlib
import os
import time

import numpy as np

import s251_config as CFG
import s251_parse as P
import s251_aggregate as AGG
import s251_checkpoint as CKPT


def sha256_file(path, blk=1 << 24):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for b in iter(lambda: f.read(blk), b""):
            h.update(b)
    return h.hexdigest()


def _append_rows(fh, rows):
    for r in rows:
        fh.write(b"%d,%.5f,%.5f,%.5f,%.5f,%d\n" % (
            r[0], r[1], r[2], r[3], r[4], r[5]))


class IngestRun(object):
    def __init__(self, source, out_m15, out_m30, manifest_csv,
                 chunk_rows=None, run_id=None, resume=False):
        self.source = source
        self.out_m15 = out_m15
        self.out_m30 = out_m30
        self.manifest_csv = manifest_csv
        self.chunk_rows = chunk_rows or CFG.default_chunk_rows()
        self.run_id = run_id or ("p3s251_%d" % int(time.time()))
        self.resume = resume
        self.cp = None
        self.history = []

    # ------------------------------------------------------------------
    def begin(self):
        stat = {
            "path": self.source,
            "size_bytes": os.path.getsize(self.source),
            "mtime_unix": float(os.path.getmtime(self.source)),
        }
        if self.resume and CKPT.exists() and os.path.exists(self.out_m15):
            cp = CKPT.load()
            ok, why = CKPT.source_matches(cp)
            if not ok:
                raise RuntimeError(
                    "RESUME REFUSED: %s (new run identity required)" % why)
            # full-source SHA check before trusting resume state
            if cp.data.get("source_sha256") != sha256_file(self.source):
                raise RuntimeError("RESUME REFUSED: source SHA-256 changed")
            self.cp = cp.data
            self.run_id = cp.data["run_id"]
            self.chunk_rows = cp.data["chunk_size"]
        else:
            stat["sha256"] = sha256_file(self.source)
            self.cp = CKPT.empty_checkpoint(stat, self.run_id, self.chunk_rows)
            self._write_headers()
        return self.cp

    def _write_headers(self):
        with open(self.out_m15, "wb") as f:
            f.write(b"timestamp_utc,open,high,low,close,tick_count\n")
        with open(self.out_m30, "wb") as f:
            f.write(b"timestamp_utc,open,high,low,close,n_m15\n")
        with open(self.manifest_csv, "w", encoding="utf-8") as f:
            f.write("chunk_id,byte_start,byte_end,row_count,first_timestamp,"
                    "last_timestamp,sha256_raw,sha256_parsed\n")

    # ------------------------------------------------------------------
    def open_blocks(self):
        """Open source and return an iterator of raw byte blocks starting at
        the resume offset."""
        f = open(self.source, "rb")
        if self.cp is not None:
            f.seek(self.cp["next_byte_start"])
        return self._file_blocks(f)

    @staticmethod
    def _file_blocks(f, blk=1 << 26):
        try:
            while True:
                b = f.read(blk)
                if not b:
                    break
                yield b
        finally:
            f.close()

    def run(self, max_chunks=None, verbose=True, progress_every=5):
        if self.cp is None:
            self.begin()
        cp = self.cp
        fh15 = open(self.out_m15, "ab")
        fh30 = open(self.out_m30, "ab")
        manf = open(self.manifest_csv, "a", encoding="utf-8")
        try:
            blk_iter = self.open_blocks()
            buf = b""
            chunk_idx = cp["last_completed_chunk"] + 1
            chunk_start = cp["next_byte_start"]
            lines = []
            n_lines = 0
            chunk_bytes = 0
            raw_hasher = hashlib.sha256()
            parser = P.ChunkParser()
            processed_since_checkpoint = 0
            chunk_count = 0
            t0 = time.time()
            last_progress = 0
            for ln, has_nl in self._line_stream(blk_iter, buf):
                lines.append(ln)
                n_lines += 1
                chunk_bytes += len(ln) + (1 if has_nl else 0)
                raw_hasher.update(ln + (b"\n" if has_nl else b""))
                if n_lines >= self.chunk_rows:
                    self._finish_chunk(lines, chunk_start, chunk_bytes,
                                       raw_hasher.hexdigest(), chunk_idx,
                                       parser, fh15, fh30, manf)
                    chunk_idx += 1
                    chunk_start += chunk_bytes
                    lines = []
                    n_lines = 0
                    chunk_bytes = 0
                    raw_hasher = hashlib.sha256()
                    parser = P.ChunkParser()   # fresh per chunk
                    chunk_count += 1
                    if max_chunks and chunk_count >= max_chunks:
                        break
                    if verbose and (chunk_count % progress_every) == 0:
                        el = time.time() - t0
                        rate = (cp["rows_processed"] / el) if el else 0
                        print("chunk %d rows %d rate %.0f/s el %.1fs" % (
                            chunk_count, cp["rows_processed"], rate, el))
            if lines:
                self._finish_chunk(lines, chunk_start, chunk_bytes,
                                   raw_hasher.hexdigest(), chunk_idx,
                                   parser, fh15, fh30, manf)
                chunk_idx += 1
            cp["updated_utc"] = CKPT.now_utc()
            CKPT.save(CKPT.Checkpoint(cp))
            return {"chunks_completed": chunk_idx, "cp": cp}
        finally:
            fh15.close()
            fh30.close()
            manf.close()

    @staticmethod
    def _line_stream(blk_iter, buf):
        for blk in blk_iter:
            data = buf + blk
            parts = data.split(b"\n")
            buf = parts.pop()
            for ln in parts:
                yield ln, True
        if buf:
            yield buf, False

    def _finish_chunk(self, lines, start, nbytes, raw_sha, idx, parser,
                      fh15, fh30, manf):
        res = parser.process(lines)
        ts = res["ts"]
        first_ts = int(min(ts)) if ts else None
        last_ts = int(max(ts)) if ts else None
        if ts:
            order = np.argsort(np.asarray(ts, dtype=np.int64), kind="stable")
            ticks = [(int(ts[i]), res["bid"][i]) for i in order]
        else:
            ticks = []
        out = AGG.process_chunk(ticks, self.cp.get("partial_carry_m15"),
                                self.cp.get("partial_carry_m30"))
        _append_rows(fh15, out["m15_rows"])
        _append_rows(fh30, out["m30_rows"])
        fh15.flush()
        fh30.flush()
        manf.write("%d,%d,%d,%d,%s,%s,%s,%s\n" % (
            idx, start, start + nbytes, res["valid_lines"],
            first_ts if first_ts is not None else "",
            last_ts if last_ts is not None else "",
            raw_sha, res["sha256_parsed"]))
        manf.flush()
        self.history.append({
            "chunk_id": idx, "byte_start": start, "byte_end": start + nbytes,
            "row_count": res["valid_lines"], "first_timestamp": first_ts,
            "last_timestamp": last_ts, "sha256_raw": raw_sha,
            "sha256_parsed": res["sha256_parsed"]})
        for k in res["malformed"]:
            self.cp["malformed_counts"][k] += res["malformed"][k]
        self.cp["last_completed_chunk"] = idx
        self.cp["current_chunk"] = idx
        self.cp["byte_offset"] = start + nbytes
        self.cp["next_byte_start"] = start + nbytes
        self.cp["rows_processed"] += res["valid_lines"]
        self.cp["m15_rows_emitted"] += len(out["m15_rows"])
        self.cp["m30_rows_emitted"] += len(out["m30_rows"])
        self.cp["partial_carry_m15"] = out["m15_carry"]
        self.cp["partial_carry_m30"] = out["m30_carry"]
        if last_ts is not None:
            self.cp["last_timestamp"] = last_ts
        self.cp["updated_utc"] = CKPT.now_utc()
        CKPT.save(CKPT.Checkpoint(self.cp))


def finalize(cp, source_sha, source_size, source_mtime, elap_sec):
    m15_carry = cp.get("partial_carry_m15")
    m30_carry = cp.get("partial_carry_m30")
    return {
        "run_id": cp.get("run_id"),
        "source_path": cp.get("source_path"),
        "source_sha256_full": source_sha,
        "source_size_bytes": source_size,
        "source_mtime_unix": source_mtime,
        "parser_version": cp.get("parser_version"),
        "schema_version": cp.get("schema_version"),
        "algorithm_version": cp.get("algorithm_version"),
        "timestamp_basis": cp.get("timestamp_basis"),
        "tz_offset_seconds": cp.get("tz_offset_seconds"),
        "chunk_size": cp.get("chunk_size"),
        "chunks_completed": cp.get("last_completed_chunk") + 1,
        "rows_processed": cp.get("rows_processed"),
        "malformed_counts": cp.get("malformed_counts"),
        "m15_rows_emitted": cp.get("m15_rows_emitted"),
        "m30_rows_emitted": cp.get("m30_rows_emitted"),
        "last_completed_chunk": cp.get("last_completed_chunk"),
        "byte_offset": cp.get("next_byte_start"),
        "last_timestamp": cp.get("last_timestamp"),
        "final_incomplete_m15": AGG.bundle_m15(m15_carry),
        "final_incomplete_m30": AGG.bundle_m30(m30_carry),
        "elapsed_sec": float(elap_sec),
    }