SniperGold_ML/engine/config.py

"""Strict engine configuration (spec 6.3).

Every identity-affecting field participates in ``config_sha256``; operational
fields are recorded in the snapshot but excluded from the hash. The exact
field classification is fixed in v1 and documented in the implementation map.
"""

import os

from . import versions
from .util import canonical_json_sha256

# Identity-affecting fields (participate in config_sha256). Changing any of
# these changes the reproducibility identity of produced data.
IDENTITY_FIELDS = (
    "source_path",
    "source_grammar",   # amended P3-DE-004: grammar preset binding (identity)
    "source_tz_offset_minutes",
    "has_volume",
    "timeframes",
    "chunk_bytes_nominal",
    "workload_bytes_nominal",
)

# Operational fields (in snapshot only, never in config_sha256).
OPERATIONAL_FIELDS = (
    "output_root",
    "workers_requested",
    "checkpoint_every_workload",
    "retry_limit",
    "memory_limits_mb",
    "disk_floor_bytes",
    "certify_timeout_sec",
    "reference_now_ms",          # test override of certify time; operational
)

ALL_FIELDS = sorted(set(IDENTITY_FIELDS) | set(OPERATIONAL_FIELDS))


class ConfigError(Exception):
    pass


def _check(cond, msg):
    if not cond:
        raise ConfigError(msg)


def default_config(source_path, output_root=None, source_tz_offset_minutes=0,
                   **overrides):
    """Produce a fully-populated config dict with v1 defaults.

    ``source_tz_offset_minutes`` is an explicit, call-site established value
    (G-6: never silently invented; production value must be established at
    source initialization in the pilot phase). Synthetic fixtures use 0 (UTC).

    ``source_grammar`` (amended P3-DE-004) defaults to the dotted synthetic/
    corpus preset; the production preset ``tickstory_mt5`` is set explicitly
    for the real six-column Tickstory MT5 source and is validated against the
    certificate ``grammar_id`` at init (fail-closed; never silent).
    """
    cfg = {
        "source_path": os.path.abspath(source_path),
        "source_grammar": versions.GRAMMAR_DOTTED,
        "source_tz_offset_minutes": int(source_tz_offset_minutes),
        "has_volume": False,
        "timeframes": list(versions.DEFAULT_TIMEFRAMES),
        "chunk_bytes_nominal": versions.CHUNK_BYTES_NOMINAL,
        "workload_bytes_nominal": versions.WORKLOAD_BYTES_NOMINAL,
        "output_root": output_root,
        "workers_requested": versions.WORKERS_REQUESTED_DEFAULT,
        "checkpoint_every_workload": True,
        "retry_limit": 2,
        "memory_limits_mb": {"parent": 512, "worker": 1024},
        "disk_floor_bytes": 4 * 2 ** 30,
        "certify_timeout_sec": 3600,
        "reference_now_ms": None,
    }
    for k, v in overrides.items():
        if k not in ALL_FIELDS:
            raise ConfigError("unknown config field %r" % k)
        cfg[k] = v
    validate_config(cfg)
    return cfg


def validate_config(cfg):
    _check(isinstance(cfg, dict), "config must be a JSON object")
    _check(isinstance(cfg.get("source_path"), str) and os.path.isabs(cfg["source_path"]),
           "source_path must be an absolute path")
    grammar = cfg.get("source_grammar", versions.GRAMMAR_DOTTED)
    _check(grammar in versions.SUPPORTED_GRAMMARS,
           "source_grammar must be one of %s" % (versions.SUPPORTED_GRAMMARS,))
    _check(isinstance(cfg.get("source_tz_offset_minutes"), int),
           "source_tz_offset_minutes must be an integer")
    _check(isinstance(cfg.get("has_volume"), bool), "has_volume must be bool")
    _check(not (grammar == versions.GRAMMAR_TICKSTORY_MT5 and not cfg.get("has_volume")),
           "grammar tickstory_mt5 requires has_volume=true (mandatory volume column)")
    tfs = cfg.get("timeframes")
    _check(isinstance(tfs, list) and len(tfs) >= 1, "timeframes must be a non-empty list")
    _check(all(tf in versions.TIMEFRAMES for tf in tfs), "timeframe not in v1 set M1/M5/M15/M30/H1")
    _check(len(set(tfs)) == len(tfs), "duplicate timeframe in timeframes")
    _check(isinstance(cfg.get("chunk_bytes_nominal"), int) and cfg["chunk_bytes_nominal"] >= 1,
           "chunk_bytes_nominal must be a positive integer")
    _check(isinstance(cfg.get("workload_bytes_nominal"), int)
           and cfg["workload_bytes_nominal"] >= cfg["chunk_bytes_nominal"],
           "workload_bytes_nominal must be >= chunk_bytes_nominal")
    _check(isinstance(cfg.get("workers_requested"), int) and cfg["workers_requested"] >= 1,
           "workers_requested must be >= 1")
    _check(isinstance(cfg.get("retry_limit"), int) and cfg["retry_limit"] >= 1,
           "retry_limit must be >= 1")
    for f in ("output_root",):
        v = cfg.get(f)
        _check(v is None or (isinstance(v, str) and os.path.isabs(v)),
               "%s must be an absolute path or null" % f)
    if "timeframes" in cfg:
        cfg["timeframes"] = [tf for tf in versions.DEFAULT_TIMEFRAMES if tf in tfs]


def config_sha256(cfg):
    """Hash over the identity-affecting subset only (spec 6.3)."""
    id_subset = {k: cfg[k] for k in IDENTITY_FIELDS if k in cfg}
    return canonical_json_sha256(id_subset)


def config_snapshot(cfg):
    """Full config snapshot recorded in manifests (identity + operational)."""
    return {k: cfg[k] for k in ALL_FIELDS if k in cfg}