SniperGold_ML/engine/util.py

"""Shared deterministic utilities: hashing, atomic writes, integer math.

All hashing uses a fixed canonical JSON serialization (sorted keys, compact
separators, ``\\n`` line ending, no trailing whitespace) per spec 17.3.
PYTHONHASHSEED is pinned to 0 at entry points; nothing here iterates sets or
dicts in a serialization path.
"""

import hashlib
import io
import json
import os
import re

HASH_BLOCK = 8 * 1024 * 1024  # 8 MiB streaming buffers (spec 7.3)

# ``_PRICE_TOKEN`` matches a plain decimal token with 0..6 fractional digits.
_PRICE_TOKEN = re.compile(r"^[0-9]+(?:\.[0-9]{1,6})?$")
_PRICE_TOKEN_PRECISE = re.compile(r"^[0-9]+(?:\.[0-9]+)?$")
_VOL_TOKEN = re.compile(r"^[0-9]+$")


class DeterminismError(Exception):
    """Raised for producer/verifier identity mismatches (material)."""


# ---------------------------------------------------------------------------
# Hashing
# ---------------------------------------------------------------------------

def sha256_bytes(data):
    m = hashlib.sha256()
    if isinstance(data, str):
        data = data.encode("utf-8")
    m.update(data)
    return m.hexdigest()


def sha256_file(path, block=HASH_BLOCK):
    """Full streaming SHA-256 of a file. Returns hex digest."""
    m = hashlib.sha256()
    with open(path, "rb") as fh:
        while True:
            chunk = fh.read(block)
            if not chunk:
                break
            m.update(chunk)
    return m.hexdigest()


def canonical_json(obj):
    """Deterministic JSON text: sorted keys, compact separators, one LF at
    the end, no other trailing whitespace."""
    return json.dumps(obj, sort_keys=True, separators=(",", ":"),
                      ensure_ascii=False) + "\n"


def canonical_json_sha256(obj):
    return sha256_bytes(canonical_json(obj))


def hash_chain_lines(lines):
    """Hash a sequence of journal lines, each line hash over its own bytes.
    Returns (line_hashes, tail_hash); tail_hash is the last line hash or
    SHA-256 of empty when no lines exist (used as prev for seq 0)."""
    hashes = []
    for line in lines:
        hashes.append(sha256_bytes(line))
    if hashes:
        tail = hashes[-1]
    else:
        tail = sha256_bytes(b"")
    return hashes, tail


# ---------------------------------------------------------------------------
# Atomic write (spec 13.1: tmp + fsync + rename + dir fsync)
# ---------------------------------------------------------------------------

def _try_fsync_dir(path):
    try:
        dfd = os.open(path, os.O_RDONLY)
    except OSError:
        return
    try:
        os.fsync(dfd)
    except OSError:
        pass
    finally:
        os.close(dfd)


def atomic_write_bytes(path, data):
    """Atomically write ``data`` (bytes or str) to ``path`` via tmp+fsync+rename."""
    if isinstance(data, str):
        data = data.encode("utf-8")
    d = os.path.dirname(os.path.abspath(path))
    if d:
        os.makedirs(d, exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "wb") as fh:
        fh.write(data)
        fh.flush()
        os.fsync(fh.fileno())
    os.replace(tmp, path)
    _try_fsync_dir(os.path.dirname(os.path.abspath(path)) or ".")


def atomic_write_json(path, obj):
    atomic_write_bytes(path, canonical_json(obj))


# ---------------------------------------------------------------------------
# Integer timestamp math (spec 8.1) - pure integer, no floats
# ---------------------------------------------------------------------------

def days_from_civil(y, m, d):
    """Days since 1970-01-01 for a proleptic Gregorian date (pure integers)."""
    y -= m <= 2
    era = (y if y >= 0 else y - 399) // 400
    yoe = y - era * 400  # [0, 399]
    doy = (153 * (m + (9 if m <= 2 else -3)) + 2) // 5 + d - 1
    doe = yoe * 365 + yoe // 4 - yoe // 100 + doy
    return era * 146097 + doe - 719468


def wallclock_to_epoch_ms(year, mon, day, hour, minute, sec, ms):
    """Epoch milliseconds (UTC) from validated wall-clock fields."""
    days = days_from_civil(year, mon, day)
    return ((days * 86400 + hour * 3600 + minute * 60 + sec) * 1000) + ms


TS_FIELD_RE = re.compile(
    r"^(\d{4})\.(\d{2})\.(\d{2})[ ](\d{2}):(\d{2}):(\d{2})(?:\.(\d{3}))?$")


def parse_timestamp_dotted(text):
    """Parse ``YYYY.MM.DD HH:MM:SS[.mmm]`` -> epoch ms UTC.

    Returns (ts_ms, None) on success or (None, error_message) on
    MALFORMED_TIMESTAMP_PARSE. Seconds field allows 00..59 only.
    """
    m = TS_FIELD_RE.match(text)
    if not m:
        return None, "timestamp shape not recognized"
    y, mo, d, hh, mi, ss = (int(g) for g in m.group(1, 2, 3, 4, 5, 6))
    ms = int(m.group(7)) if m.group(7) else 0
    if not (1 <= mo <= 12 and 1 <= d <= 31):
        return None, "month/day out of range"
    if not (0 <= hh <= 23 and 0 <= mi <= 59 and 0 <= ss <= 59 and 0 <= ms <= 999):
        return None, "time field out of range"
    return wallclock_to_epoch_ms(y, mo, d, hh, mi, ss, ms), None


DATE8_RE = re.compile(r"^(\d{4})(\d{2})(\d{2})$")
TIME_HMS_RE = re.compile(r"^(\d{2}):(\d{2}):(\d{2})$")


def parse_compact_datetime(date_text, time_text):
    """Parse the G_TICKSTORY_MT5 timestamp pair -> epoch ms UTC.

    Date ``YYYYMMDD`` (exactly 8 digits) + time ``HH:MM:SS`` (seconds only,
    milliseconds are NOT part of the grammar; a period in the time token is a
    MALFORMED_TIMESTAMP_PARSE). Seconds allow 00..59. Returns (ts_ms, None)
    on success or (None, error_message) on MALFORMED_TIMESTAMP_PARSE.
    Window/offset handling is applied by the caller (spec 8.1).
    """
    m = DATE8_RE.match(date_text)
    if not m:
        return None, "date shape not recognized (expected YYYYMMDD)"
    y, mo, d = (int(g) for g in m.group(1, 2, 3))
    mt = TIME_HMS_RE.match(time_text)
    if not mt:
        return None, "time shape not recognized (expected HH:MM:SS)"
    hh, mi, ss = (int(g) for g in mt.group(1, 2, 3))
    if not (1 <= mo <= 12 and 1 <= d <= 31):
        return None, "month/day out of range"
    if not (0 <= hh <= 23 and 0 <= mi <= 59 and 0 <= ss <= 59):
        return None, "time field out of range"
    return wallclock_to_epoch_ms(y, mo, d, hh, mi, ss, 0), None


def apply_tz_offset(ts_ms, source_tz_offset_minutes):
    """Canonical ts_ms in UTC from wall-clock epoch ms and fixed offset."""
    return ts_ms - source_tz_offset_minutes * 60_000


def in_ts_valid_window(ts_ms, certify_time_ms):
    """True when 2000-01-01T00:00:00Z <= ts_ms <= certify_time + 7 days."""
    return ts_ms >= 946_684_800_000 and ts_ms <= certify_time_ms + 7 * 86_400_000


# ---------------------------------------------------------------------------
# Decimal price parsing (spec 8.2) - exact, no floats
# ---------------------------------------------------------------------------

def parse_price_to_micro(text, scale=1_000_000):
    """Parse a decimal price token into scaled integer micro-units.

    Returns (value, None) on success.
    On malformed input returns (None, "MALFORMED_PRICE_PARSE") for non-numeric
    tokens (including leading signs / extra characters) and
    (None, "MALFORMED_PRICE_PRECISION") when fractional digits exceed the scale.

    Deterministic grammar: ``[0-9]+ ( . [0-9]{1,6} )?`` for scale 1e6.
    """
    if _PRICE_TOKEN.match(text):
        return _token_to_int(text, scale)
    if _PRICE_TOKEN_PRECISE.match(text):
        # numeric token with too many fractional digits -> precision class
        return None, "MALFORMED_PRICE_PRECISION"
    return None, "MALFORMED_PRICE_PARSE"


def _token_to_int(text, scale):
    if "." in text:
        whole, frac = text.split(".", 1)
        digits = len(frac)
        value = int(whole) * scale + int(frac) * (scale // (10 ** digits))
        return value, None
    return int(text) * scale, None


def parse_volume_token(text):
    """Integer volume token. (value, None) or (None, 'MALFORMED_VOLUME')."""
    if _VOL_TOKEN.match(text):
        return int(text), None
    return None, "MALFORMED_VOLUME"


# ---------------------------------------------------------------------------
# Path safety (spec 25.2)
# ---------------------------------------------------------------------------

def resolve_under_root(path, root):
    """Return an absolute resolved path that is strictly under ``root``.

    Rejects relative traversal, symlink/junction escapes and absolute paths
    outside the root. Raises ValueError otherwise.
    """
    root_abs = os.path.abspath(root)
    p_abs = os.path.abspath(path)
    common = os.path.commonpath([root_abs, p_abs])
    if common != root_abs:
        raise ValueError("path %r escapes configured root %r" % (path, root_abs))
    # Resolve real path when the target exists; guard symlink escapes.
    try:
        real = os.path.realpath(p_abs)
        real_common = os.path.commonpath([root_abs, real])
        if real_common != root_abs:
            raise ValueError("path %r resolves outside root %r via link" % (path, root_abs))
    except OSError:
        pass
    return p_abs


def read_binary_range(path, byte_start, byte_end):
    """Read ``[byte_start, byte_end)`` from a file (read-only access)."""
    with open(path, "rb") as fh:
        fh.seek(byte_start)
        return fh.read(byte_end - byte_start)