forked from chiki2bum2/SniperGold_ML
321 lines
No EOL
13 KiB
Python
321 lines
No EOL
13 KiB
Python
"""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
|
|
|