# -*- 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()}