SniperGold_ML/engine/parse.py

"""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