SniperGold_ML/ml/p3/p3_s251_external_ingest/s251_parse.py

# -*- coding: utf-8 -*-
"""P3-S25.1 PARSER + VALIDATION + CANONICALISATION.

Parses complete byte-lines (newline-tokenised, CRLF handled) into valid
records or classified-malformed rows. Valid records are canonicalised and the
chunk sha256_parsed digest is computed incrementally. No silent coercion:
malformed rows are counted, never fabricated into the valid stream.

Timestamp semantics: date+time interpreted as UTC (tz_offset 0), epoch
microseconds.
"""
import hashlib

import s251_config as CFG

_DAYS_IN_MONTH = (0, 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31)


def _is_leap(y):
    return (y % 4 == 0 and y % 100 != 0) or (y % 400 == 0)


def _valid_ymd(y, m, dd):
    if not (1 <= m <= 12):
        return False
    dim = _DAYS_IN_MONTH[m]
    if m == 2 and _is_leap(y):
        dim = 29
    return 1 <= dd <= dim


def days_from_civil(y, m, dd):
    """Howard Hinnant civil->days as days since 1970-01-01."""
    y -= m <= 2
    era = (y if y >= 0 else y - 399) // 400
    yoe = y - era * 400
    doy = (153 * (m + (9 if m <= 2 else -3)) + 2) // 5 + dd - 1
    doe = yoe * 365 + yoe // 4 - yoe // 100 + doy
    return era * 146097 + doe - 719468


class DateCache(object):
    """Last-date cache (rows arrive date-ordered) -> days since epoch."""

    __slots__ = ("last", "days")

    def __init__(self):
        self.last = None
        self.days = 0

    def days_for(self, d):
        if len(d) != 8:
            return None
        if d == self.last:
            return self.days
        y = (d[0] - 48) * 1000 + (d[1] - 48) * 100 + (d[2] - 48) * 10 + (d[3] - 48)
        m = (d[4] - 48) * 10 + (d[5] - 48)
        dd = (d[6] - 48) * 10 + (d[7] - 48)
        if not _valid_ymd(y, m, dd):
            return None
        self.last = d
        self.days = days_from_civil(y, m, dd)
        return self.days


def parse_ts_bytes(d, tm, cache):
    """Return epoch UTC seconds (int) or None. Requires HH:MM:SS (0x3a ':').

    Note tm[2] and tm[5] must be ':' (58). Auto-clears the date cache on a
    malformed timestamp only via caller.
    """
    if len(tm) != 8 or tm[2] != 58 or tm[5] != 58:
        return None
    h = (tm[0] - 48) * 10 + (tm[1] - 48)
    mi = (tm[3] - 48) * 10 + (tm[4] - 48)
    s = (tm[6] - 48) * 10 + (tm[7] - 48)
    if not (0 <= h <= 23 and 0 <= mi <= 59 and 0 <= s <= 59):
        return None
    days = cache.days_for(d)
    if days is None:
        return None
    return days * 86400 + h * 3600 + mi * 60 + s


def new_malformed():
    return {"column_count": 0, "timestamp_malformed": 0,
            "timestamp_invalid": 0, "non_numeric_price": 0,
            "invalid_volume": 0, "bid_ask_relationship": 0,
            "non_positive_price": 0, "non_monotonic_timestamp": 0}


def aggregate_malformed(a, b):
    for k in a:
        a[k] += b.get(k, 0)


def parse_line(line, cache, bucket):
    """Validate+parse one byte-line (CR/LF stripped). Return tuple or None.

    (epo_us int, bid float, ask float, last float, vol int)
    """
    parts = line.split(b",")
    if len(parts) != CFG.NCOL:
        bucket["column_count"] += 1
        return None
    d, tm, pb, pa, pl, pv = parts
    epo = parse_ts_bytes(d, tm, cache)
    if epo is None:
        if len(d) != 8 or len(tm) != 8 or tm[2] != 58 or tm[5] != 58:
            bucket["timestamp_malformed"] += 1
        else:
            bucket["timestamp_invalid"] += 1
        # do NOT poison the date cache for the duration parse result
        return None
    try:
        bid = float(pb)
        ask = float(pa)
        last = float(pl)
    except ValueError:
        bucket["non_numeric_price"] += 1
        return None
    try:
        vol = int(pv)
    except ValueError:
        bucket["invalid_volume"] += 1
        return None
    if vol < 0:
        bucket["invalid_volume"] += 1
        return None
    if bid <= 0 or ask <= 0 or last <= 0:
        bucket["non_positive_price"] += 1
        return None
    if not (bid <= last <= ask):
        bucket["bid_ask_relationship"] += 1
        return None
    return (int(epo), bid, ask, float(last), vol)


class ChunkParser(object):
    """Processes a list/iterator of complete byte lines for one chunk."""

    def __init__(self):
        self.cache = DateCache()
        self.parser_mal = new_malformed()
        self.n_non_monotonic = 0

    def process(self, lines):
        """lines: iterable of byte-lines (CR/LF stripped)."""
        n_valid = 0
        last_epo = None
        ts = []
        bid = []
        ask = []
        last = []
        vol = []
        hasher = hashlib.sha256()
        for line in lines:
            if not line:
                continue
            rec = parse_line(line, self.cache, self.parser_mal)
            if rec is None:
                continue
            if last_epo is not None and rec[0] < last_epo:
                self.parser_mal["non_monotonic_timestamp"] += 1
            last_epo = rec[0]
            ts.append(rec[0]); bid.append(rec[1]); ask.append(rec[2])
            last.append(rec[3]); vol.append(rec[4])
            n_valid += 1
            cl = CFG.canonical_line(rec[0], rec[1], rec[2], rec[3], rec[4])
            hasher.update(cl.encode("ascii"))
        return {"valid_lines": n_valid, "ts": ts, "bid": bid, "ask": ask,
                "last": last, "vol": vol,
                "malformed": dict(self.parser_mal),
                "sha256_parsed": hasher.hexdigest()}