"""Producer parser: raw chunk bytes -> canonical ticks + malformed sidecars. Deterministic CSV interpretation per spec 8.1–8.3. Integer arithmetic only. This module is one of two independent parsing implementations that exist in the engine (the other is engine/verify/vparse.py); they must agree on every canonical row. """ from . import versions from .util import ( parse_timestamp_dotted, parse_compact_datetime, apply_tz_offset, in_ts_valid_window, parse_price_to_micro, parse_volume_token, sha256_bytes, ) MALFORMED_EMPTY_LINE = "MALFORMED_EMPTY_LINE" MALFORMED_ENCODING = "MALFORMED_ENCODING" MALFORMED_FIELD_COUNT = "MALFORMED_FIELD_COUNT" MALFORMED_TIMESTAMP_PARSE = "MALFORMED_TIMESTAMP_PARSE" MALFORMED_TIMESTAMP_RANGE = "MALFORMED_TIMESTAMP_RANGE" MALFORMED_PRICE_PARSE = "MALFORMED_PRICE_PARSE" MALFORMED_PRICE_PRECISION = "MALFORMED_PRICE_PRECISION" MALFORMED_PRICE_NONPOSITIVE = "MALFORMED_PRICE_NONPOSITIVE" MALFORMED_BID_ASK_RELATION = "MALFORMED_BID_ASK_RELATION" MALFORMED_VOLUME = "MALFORMED_VOLUME" MALFORMED_DUPLICATE = "MALFORMED_DUPLICATE" MALFORMED_NON_MONOTONIC = "MALFORMED_NON_MONOTONIC" MALFORMED_NON_MONOTONIC_BOUNDARY = "MALFORMED_NON_MONOTONIC_BOUNDARY" ALL_MALFORMED_CLASSES = [ MALFORMED_EMPTY_LINE, MALFORMED_ENCODING, MALFORMED_FIELD_COUNT, MALFORMED_TIMESTAMP_PARSE, MALFORMED_TIMESTAMP_RANGE, MALFORMED_PRICE_PARSE, MALFORMED_PRICE_PRECISION, MALFORMED_PRICE_NONPOSITIVE, MALFORMED_BID_ASK_RELATION, MALFORMED_VOLUME, MALFORMED_DUPLICATE, MALFORMED_NON_MONOTONIC, MALFORMED_NON_MONOTONIC_BOUNDARY, ] _HEADER_RE_PREFIX = ("datetime", "date", "time") def header_keyword(first_field): """Header detection rule (spec 8.1): first field starts with datetime|date|time followed by a non-digit or end of string, case-insensitive.""" low = first_field.lower() for kw in _HEADER_RE_PREFIX: if low.startswith(kw): rest = low[len(kw):] if rest == "" or not rest[0].isdigit(): return True return False def split_lines(data): """Deterministic byte-level line splitter. Yields (line_bytes_without_terminator, terminator_len). Terminators are recognized in order: b"\\r\\n" (2), b"\\n" (1), b"\\r" (1). A trailing unterminated line yields terminator_len 0. """ start = 0 n = len(data) i = 0 while i < n: b = data[i] if b == 0x0A: # \n yield data[start:i], 1 i += 1 start = i elif b == 0x0D: # \r j = i + 1 if j < n and data[j] == 0x0A: # \r\n yield data[start:i], 2 i = j + 1 else: yield data[start:i], 1 i = j start = i else: i += 1 if start < n: yield data[start:], 0 def empty_malformed_counter(): return {cls: 0 for cls in ALL_MALFORMED_CLASSES} def parse_chunk(data, *, chunk_index, byte_start, global_line_start, cfg, expect_header, certify_time_ms): """Parse one chunk's byte range into canonical records + malformed sidecars. Returns (records, malformed, counters, flags) where: records: list of (ts_ms, bid_u, ask_u, spread_u, vol, src_line, rec_ord) malformed: list of sidecar dicts per malformed row counters: dict class -> count flags: dict with keys has_header (bool), rows_parsed (int) ``global_line_start`` is the 0-based raw line offset of this chunk's first line; ``byte_start`` is the 0-based file offset. cert_time is the UTC reference used for the timestamp validity window. """ grammar = cfg.get("source_grammar", versions.GRAMMAR_DOTTED) if grammar == versions.GRAMMAR_TICKSTORY_MT5: return _parse_chunk_tickstory( data, chunk_index=chunk_index, byte_start=byte_start, global_line_start=global_line_start, cfg=cfg, expect_header=expect_header, certify_time_ms=certify_time_ms) counters = empty_malformed_counter() malformed = [] records = [] rec_ord = 0 prev_ts = None seen = set() duplicate_cap = versions.DUPLICATE_TRACKER_CAP expect_header = bool(expect_header) has_header = False rows_parsed = 0 global_line = global_line_start # 0-based; first raw line -> +1 below expected_fields = 4 if cfg["has_volume"] else 3 def reject(cls, reason, raw_bytes): counters[cls] += 1 malformed.append({ "chunk_index": chunk_index, "byte_start": byte_start, "global_line_no": global_line + 1, "class": cls, "raw_sha256": sha256_bytes(raw_bytes), "reason": reason, }) for line_bytes, _term in split_lines(data): global_line += 1 # Blank line -> MALFORMED_EMPTY_LINE (expected class), no parse. if not line_bytes: counters[MALFORMED_EMPTY_LINE] += 1 continue rows_parsed += 1 # UTF-8 decode (BOM was stripped upstream for chunk 0). try: text = line_bytes.decode("utf-8") except UnicodeDecodeError: reject(MALFORMED_ENCODING, "undecodable utf-8 bytes", line_bytes) continue if expect_header and not has_header: first_field = text.split(",", 1)[0] if header_keyword(first_field): has_header = True continue fields = text.split(",") if len(fields) != expected_fields: reject(MALFORMED_FIELD_COUNT, "field count %d expected %d" % (len(fields), expected_fields), line_bytes) continue t_field = fields[0] ts_wall, err = parse_timestamp_dotted(t_field) if err is not None: reject(MALFORMED_TIMESTAMP_PARSE, err, line_bytes) continue ts_ms = apply_tz_offset(ts_wall, cfg["source_tz_offset_minutes"]) if not in_ts_valid_window(ts_ms, certify_time_ms): reject(MALFORMED_TIMESTAMP_RANGE, "ts outside validity window", line_bytes) continue bid_u, perr = parse_price_to_micro(fields[1], versions.PRICE_SCALE) if perr in (MALFORMED_PRICE_PARSE, MALFORMED_PRICE_PRECISION): reject(perr, "bid token invalid: %r" % fields[1], line_bytes) continue ask_u, aerr = parse_price_to_micro(fields[2], versions.PRICE_SCALE) if aerr in (MALFORMED_PRICE_PARSE, MALFORMED_PRICE_PRECISION): reject(aerr, "ask token invalid: %r" % fields[2], line_bytes) continue if bid_u <= 0 or ask_u <= 0: reject(MALFORMED_PRICE_NONPOSITIVE, "bid/ask must be > 0", line_bytes) continue if ask_u < bid_u: reject(MALFORMED_BID_ASK_RELATION, "ask < bid", line_bytes) continue if cfg["has_volume"]: vol, verr = parse_volume_token(fields[3]) if verr is not None: reject(MALFORMED_VOLUME, "volume token invalid: %r" % fields[3], line_bytes) continue else: vol = 1 spread_u = ask_u - bid_u key = (ts_ms, bid_u, ask_u, vol) if key in seen: reject(MALFORMED_DUPLICATE, "exact duplicate within chunk", line_bytes) continue if len(seen) < duplicate_cap: seen.add(key) if prev_ts is not None and ts_ms < prev_ts: reject(MALFORMED_NON_MONOTONIC, "ts regression within chunk", line_bytes) continue prev_ts = ts_ms records.append((ts_ms, bid_u, ask_u, spread_u, vol, global_line, rec_ord)) rec_ord += 1 flags = {"has_header": has_header, "rows_parsed": rows_parsed} return records, malformed, counters, flags def _parse_chunk_tickstory(data, *, chunk_index, byte_start, global_line_start, cfg, expect_header, certify_time_ms): """G_TICKSTORY_MT5 six-column parse (P3-DATA-ENGINE-004 spec 8.1/8.2/8.6). Grammar (derived from the authoritative pilot evidence): exactly six comma-separated fields ``YYYYMMDD,HH:MM:SS,bid,ask,last,volume``; a line with any other field count is MALFORMED_FIELD_COUNT. - date ``YYYYMMDD`` + time ``HH:MM:SS`` (seconds only) -> integer epoch ms UTC via deterministic civil math, then tz-offset applied - bid / ask / last: decimal prices to micro-units (PRICE_SCALE); their token grammar is identical to the dotted preset - bid_u > 0, ask_u > 0 (MALFORMED_PRICE_NONPOSITIVE), ask_u >= bid_u (MALFORMED_BID_ASK_RELATION; mandatory) - volume: non-negative integer token; 0 is VALID (observed in the real source); negative/non-integer -> MALFORMED_VOLUME ``last`` is validated here but is deliberately NOT part of CTS_V1; it is returned in ``flags["last_u"]`` (aligned with records in rec_ord order) for the source-preservation layer (spec 8.6). """ counters = empty_malformed_counter() malformed = [] records = [] last_values = [] rec_ord = 0 prev_ts = None seen = set() duplicate_cap = versions.DUPLICATE_TRACKER_CAP expect_header = bool(expect_header) has_header = False rows_parsed = 0 global_line = global_line_start def reject(cls, reason, raw_bytes): counters[cls] += 1 malformed.append({ "chunk_index": chunk_index, "byte_start": byte_start, "global_line_no": global_line + 1, "class": cls, "raw_sha256": sha256_bytes(raw_bytes), "reason": reason, }) for line_bytes, _term in split_lines(data): global_line += 1 if not line_bytes: counters[MALFORMED_EMPTY_LINE] += 1 continue rows_parsed += 1 try: text = line_bytes.decode("utf-8") except UnicodeDecodeError: reject(MALFORMED_ENCODING, "undecodable utf-8 bytes", line_bytes) continue if expect_header and not has_header: first_field = text.split(",", 1)[0] if header_keyword(first_field): has_header = True continue fields = text.split(",") if len(fields) != 6: reject(MALFORMED_FIELD_COUNT, "field count %d expected 6" % len(fields), line_bytes) continue ts_wall, err = parse_compact_datetime(fields[0], fields[1]) if err is not None: reject(MALFORMED_TIMESTAMP_PARSE, err, line_bytes) continue ts_ms = apply_tz_offset(ts_wall, cfg["source_tz_offset_minutes"]) if not in_ts_valid_window(ts_ms, certify_time_ms): reject(MALFORMED_TIMESTAMP_RANGE, "ts outside validity window", line_bytes) continue bid_u, perr = parse_price_to_micro(fields[2], versions.PRICE_SCALE) if perr in (MALFORMED_PRICE_PARSE, MALFORMED_PRICE_PRECISION): reject(perr, "bid token invalid: %r" % fields[2], line_bytes) continue ask_u, aerr = parse_price_to_micro(fields[3], versions.PRICE_SCALE) if aerr in (MALFORMED_PRICE_PARSE, MALFORMED_PRICE_PRECISION): reject(aerr, "ask token invalid: %r" % fields[3], line_bytes) continue last_u, lerr = parse_price_to_micro(fields[4], versions.PRICE_SCALE) if lerr in (MALFORMED_PRICE_PARSE, MALFORMED_PRICE_PRECISION): reject(lerr, "last token invalid: %r" % fields[4], line_bytes) continue if bid_u <= 0 or ask_u <= 0 or last_u <= 0: reject(MALFORMED_PRICE_NONPOSITIVE, "bid/ask/last must be > 0", line_bytes) continue if ask_u < bid_u: reject(MALFORMED_BID_ASK_RELATION, "ask < bid", line_bytes) continue vol, verr = parse_volume_token(fields[5]) if verr is not None: reject(MALFORMED_VOLUME, "volume token invalid: %r" % fields[5], line_bytes) continue spread_u = ask_u - bid_u key = (ts_ms, bid_u, ask_u, vol) if key in seen: reject(MALFORMED_DUPLICATE, "exact duplicate within chunk", line_bytes) continue if len(seen) < duplicate_cap: seen.add(key) if prev_ts is not None and ts_ms < prev_ts: reject(MALFORMED_NON_MONOTONIC, "ts regression within chunk", line_bytes) continue prev_ts = ts_ms records.append((ts_ms, bid_u, ask_u, spread_u, vol, global_line, rec_ord)) last_values.append(last_u) rec_ord += 1 flags = {"has_header": has_header, "rows_parsed": rows_parsed, "last_u": last_values} return records, malformed, counters, flags