SniperGold_ML/engine/certify.py

"""Source certification (Layer 1, spec 7).

A certificate records source identity (size, mtime_ns), the full-file
SHA-256 when actually computed, and format facts sniffed from the first
1 MiB / first 1000 parseable rows. The FULLY_VERIFIED status is written only
when the full hash was actually computed; the timeout path records
VERIFIED_WITH_LIMITATION with sha256_mode="not_computed" and never claims a
full hash (G-3). A size/mtime change detected mid-read is automatic FAILED.
"""

import os
import time

from . import versions
from .util import atomic_write_json, canonical_json, sha256_bytes
from .parse import split_lines, header_keyword
from .chunkmap import iter_line_spans

SNIFF_BYTES = 1024 * 1024
SNIFF_ROWS = 1000
TS_DOTTED_MS_RE = "dotted_ms"
TS_DOTTED_S_RE = "dotted_s"


def _classify_line_endings(data):
    crlf = lf = cr = 0
    i, n = 0, len(data)
    while i < n:
        b = data[i]
        if b == 0x0A:
            lf += 1
            i += 1
        elif b == 0x0D:
            if i + 1 < n and data[i + 1] == 0x0A:
                crlf += 1
                i += 2
            else:
                cr += 1
                i += 1
        else:
            i += 1
    total = crlf + lf + cr
    if total == 0:
        return "LF", {"crlf": 0, "lf": 0, "cr": 0}
    which = "MIXED"
    if crlf and not lf and not cr:
        which = "CRLF"
    elif lf and not crlf and not cr:
        which = "LF"
    elif cr and not crlf and not lf:
        which = "CR"
    return which, {"crlf": crlf, "lf": lf, "cr": cr}


def _detect_encoding(data):
    try:
        data.decode("ascii")
        return "ascii"
    except UnicodeDecodeError:
        try:
            data.decode("utf-8")
            return "utf8"
        except UnicodeDecodeError:
            return "other"


def _is_compact(date_field, time_field):
    import re
    return (re.match(r"^\d{8}$", date_field) is not None
            and re.match(r"^\d{2}:\d{2}:\d{2}$", time_field) is not None)


def _is_dotted_ms(f):
    import re
    return re.match(r"^\d{4}\.\d{2}\.\d{2} \d{2}:\d{2}:\d{2}\.\d{3}$", f) is not None


def _is_dotted_s(f):
    import re
    return re.match(r"^\d{4}\.\d{2}\.\d{2} \d{2}:\d{2}:\d{2}$", f) is not None


def _detect_format_rows(data):
    """Sniff up to SNIFF_ROWS parseable rows; return observed facts."""
    rows = 0
    field_count_hist = {}
    ts_formats = {}
    all_fields_comma = True
    has_header = False
    first_row_seen = False
    for line_bytes, _term in split_lines(data):
        if not line_bytes:
            continue
        if rows >= SNIFF_ROWS:
            break
        try:
            text = line_bytes.decode("utf-8")
        except UnicodeDecodeError:
            break
        fields = text.split(",")
        if len(fields) < 2:
            all_fields_comma = False
            continue
        field_count_hist[len(fields)] = field_count_hist.get(len(fields), 0) + 1
        if not first_row_seen:
            first_row_seen = True
            if header_keyword(fields[0]):
                has_header = True
                continue  # header row does not count as a data row
        ts_f = fields[0]
        if len(fields) >= 2 and _is_compact(fields[0], fields[1]):
            ts_formats["compact_date_time"] = ts_formats.get("compact_date_time", 0) + 1
        elif _is_dotted_ms(ts_f):
            ts_formats["dotted_ms"] = ts_formats.get("dotted_ms", 0) + 1
        elif _is_dotted_s(ts_f):
            ts_formats["dotted_s"] = ts_formats.get("dotted_s", 0) + 1
        else:
            ts_formats["unrecognized"] = ts_formats.get("unrecognized", 0) + 1
        rows += 1
    return {
        "rows_sniffed": rows,
        "field_count_hist": field_count_hist,
        "ts_formats": ts_formats,
        "first_field_comma_split": all_fields_comma,
        "has_header": has_header,
    }


def certify_source(source_path, output_path, certify_timeout_sec=3600,
                   block=8 * 1024 * 1024):
    """Produce the source certificate at ``output_path`` (atomic)."""
    st_pre = os.stat(source_path)
    size = st_pre.st_size
    mtime_ns = st_pre.st_mtime_ns
    started = time.time()
    hasher = __import__("hashlib").new("sha256")
    sniffed = b""
    rows = None
    fully_read = True
    status = "FULLY_VERIFIED"

    with open(source_path, "rb") as fh:
        while True:
            if time.time() - started > certify_timeout_sec:
                fully_read = False
                status = "VERIFIED_WITH_LIMITATION"
                break
            chunk = fh.read(block)
            if not chunk:
                break
            hasher.update(chunk)
            if len(sniffed) < SNIFF_BYTES:
                sniffed += chunk[: SNIFF_BYTES - len(sniffed)]
    elapsed = time.time() - started

    st_post = os.stat(source_path)
    if st_post.st_size != size or st_post.st_mtime_ns != mtime_ns:
        cert = _build_cert(source_path, None, size, mtime_ns, st_post,
                           "FAILED", "source_changed_during_certify", None)
        atomic_write_json(output_path, cert)
        return cert

    if status == "VERIFIED_WITH_LIMITATION":
        cert = _build_cert(source_path, None, size, mtime_ns, st_post,
                           "VERIFIED_WITH_LIMITATION", "timeout", None,
                           sniffed=sniffed)
        atomic_write_json(output_path, cert)
        return cert

    full_sha = hasher.hexdigest()
    cert = _build_cert(source_path, full_sha, size, mtime_ns, st_post,
                       "FULLY_VERIFIED", None, elapsed, sniffed=sniffed)
    atomic_write_json(output_path, cert)
    return cert


def _build_cert(source_path, full_sha, size, mtime_ns, st_post, status,
                limitation_reason, elapsed, sniffed=b""):
    le_class, le_ratio = _classify_line_endings(sniffed)
    enc = _detect_encoding(sniffed)
    facts = _detect_format_rows(sniffed)
    ts_formats = facts["ts_formats"]
    ts_detected = "unrecognized"
    if ts_formats.get("dotted_ms"):
        ts_detected = "dotted_ms"
    elif ts_formats.get("dotted_s"):
        ts_detected = "dotted_s"
    elif ts_formats.get("compact_date_time"):
        ts_detected = "compact_date_time"
    field_counts = facts["field_count_hist"]
    observed_fields = max(field_counts, key=field_counts.get) if field_counts else 0
    names = ["datetime", "bid", "ask"]
    if observed_fields == 4:
        names.append("vol")
    elif observed_fields == 6:
        names = ["date", "time", "bid", "ask", "last", "volume"]
    delim = "comma" if facts["first_field_comma_split"] or field_counts else "unrecognized"

    # Source-format decision (P3-DE-004 amendment). Two certified grammar
    # presets exist; the certificate pins grammar_id so init can cross-check
    # the configured grammar (fail-closed, never silent).
    source_format = "unrecognized"
    grammar_id = None
    if field_counts and observed_fields in (3, 4) and ts_detected in ("dotted_ms", "dotted_s"):
        source_format = "csv_tickstory_mt5"
        grammar_id = versions.GRAMMAR_ID_DOTTED
    elif field_counts and observed_fields == 6 and ts_detected == "compact_date_time":
        source_format = "csv_tickstory_mt5"
        grammar_id = versions.GRAMMAR_ID_TICKSTORY
    elif field_counts:
        source_format = "mismatch"

    cert = {
        "schema_version": versions.CERT_SCHEMA_VERSION,
        "certification_timestamp": _utc_now_iso(),
        "certification_status": status,
        "elapsed_sec": round(elapsed, 3) if elapsed is not None else None,
        "source_path": os.path.abspath(source_path),
        "source_size_bytes": size,
        "source_mtime_ns": mtime_ns,
        "sha256_full": full_sha,
        "sha256_mode": "full" if full_sha else "not_computed",
        "limitation_reason": limitation_reason,
        "source_format": source_format,
        "grammar_id": grammar_id,
        "line_ending": le_class,
        "line_ending_ratio": le_ratio,
        "line_ending_total": sum(le_ratio.values()),
        "encoding": enc,
        "timestamp_format_detected": ts_detected,
        "delimiter": delim,
        "has_header": bool(facts["has_header"]),
        "column_layout": {"fields": observed_fields, "names": names},
        "parser_version": versions.PARSER_VERSION,
        "engine_version": versions.ENGINE_VERSION,
    }
    return cert


def _utc_now_iso():
    import datetime
    return datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "+00:00")


def cert_matches_source(cert, source_path):
    """Resume guard: size+mtime recomputed vs certificate (spec 13.5 #2)."""
    st = os.stat(source_path)
    if cert.get("source_size_bytes") != st.st_size:
        return "source_size_changed"
    if cert.get("source_mtime_ns") != st.st_mtime_ns:
        return "source_mtime_changed"
    return None