"""Inspect raw legacy snapshots or audit the supported real-history dataset. Inspection never turns status/gate acceptance or missing outcomes into labels. Training is provided by Build-MarketModel.ps1 and Train_MarketModel.mq5 only. """ from __future__ import annotations import argparse from collections import Counter, defaultdict import csv from datetime import datetime import json import math import os from pathlib import Path from typing import Any COMMON_FILES_DIR = Path(os.environ.get("APPDATA", "")) / "MetaQuotes" / "Terminal" / "Common" / "Files" SNAPSHOT_PATTERN = "feature_batch_*.txt" MARKET_BASE_FEATURES = ( "mm_signed_logret1", "mm_signed_logret3", "mm_signed_logret12", "mm_signed_zscore12", "mm_return_std12", "mm_range", "mm_signed_body", "mm_spread", "mm_hour_sin", "mm_hour_cos", "mm_direction", "mm_atr14", ) ACTIVE_FEATURES = ("active_symbol_flag", "active_ret1", "active_ret3", "active_zscore") RETIRED_TRAINING_MESSAGE = ( "Legacy snapshot-label training is retired. Gate/status acceptance is not " "profitability, and absent future outcomes must stay unknown. Run " "ML/Build-MarketModel.ps1 (or snapshot_train.py) for the supported real-history " "universal ONNX model with causal shared features, train-only scaling, and " "chronological horizon-purged evaluation." ) def _read_text_lines_with_fallback(path: str | Path) -> list[str]: data = Path(path).read_bytes() if data.startswith((b"\xff\xfe", b"\xfe\xff")): return data.decode("utf-16").splitlines() candidates = ("utf-16-le", "utf-16-be") if b"\x00" in data else ("utf-8-sig", "cp1252") for encoding in candidates: try: text = data.decode(encoding) except UnicodeDecodeError: continue if "\x00" not in text: return text.splitlines() raise ValueError(f"Snapshot has unsupported or malformed encoding: {path}") def parse_snapshot_file(path: str | Path) -> dict[str, str]: record: dict[str, str] = {} for raw in _read_text_lines_with_fallback(path): line = raw.strip() if not line: continue if ":" not in line: raise ValueError(f"Malformed snapshot key/value line in {path}") key, value = line.split(":", 1) key = key.strip().lower() if not key or key in record: raise ValueError(f"Empty or duplicate snapshot field {key!r} in {path}") record[key] = value.strip() record["__source_file"] = Path(path).name record["__source_path"] = str(Path(path).resolve()) return record def list_snapshot_files(snapshot_dir: str | Path | None = None, pattern: str = SNAPSHOT_PATTERN) -> list[str]: return [str(path) for path in sorted(Path(snapshot_dir or COMMON_FILES_DIR).glob(pattern)) if path.is_file()] def inspect_snapshots(snapshot_dir: str | Path | None = None, limit: int | None = None) -> dict[str, Any]: files = list_snapshot_files(snapshot_dir) if limit is not None: if limit <= 0: raise ValueError("limit must be positive") files = files[-limit:] records = [parse_snapshot_file(path) for path in files] return { "records": len(records), "symbols": sorted({record.get("symbol", "unknown") for record in records}), "status_counts_for_diagnostics_only": dict(Counter(record.get("status", "unknown") for record in records)), "training_supported": False, "training_path": "ML/Build-MarketModel.ps1", "reason": RETIRED_TRAINING_MESSAGE, } def audit_market_dataset(path: str | Path, train_end: str | None = None, validation_end: str | None = None, test_end: str | None = None, model_data: str | Path | None = None) -> dict[str, Any]: """Verify observed labels, paired timestamps and purged chronological splits. The CSV is sparse: twelve causal base features plus the active symbol's four interactions. It is an audit, not an alternative trainer or imputer. All timestamps use the broker-native MQL bar-time coordinate consistently. """ # Artifact metadata is authoritative unless a caller explicitly overrides # a boundary. Historical fixture defaults apply only without a manifest. metadata_path = Path(model_data) if model_data is not None else Path(path).resolve().with_name("model_data.json") metadata = {} if model_data is not None or metadata_path.is_file(): with metadata_path.open("r", encoding="utf-8-sig") as handle: metadata = json.load(handle) if not isinstance(metadata, dict): raise ValueError("model_data.json must contain a metadata object") keys = ("training_end", "validation_end", "test_end") fallback = ("2025.07.01", "2025.10.01", "2026.01.01") resolved = [] for explicit, key, default in zip((train_end, validation_end, test_end), keys, fallback): if explicit is not None: value = explicit elif metadata: if not isinstance(metadata.get(key), str) or not metadata[key]: raise ValueError(f"Model metadata is missing a valid {key} split boundary") value = metadata[key] else: value = default resolved.append(value) boundaries = [datetime.strptime(value, "%Y.%m.%d") for value in resolved] if not boundaries[0] < boundaries[1] < boundaries[2]: raise ValueError("Split boundaries must increase chronologically") required = {"decision_time", "target_end", "split", "symbol", "direction", "label", "net_log_return", "entry_spread_points", "exit_spread_points"} counts: Counter[str] = Counter() classes: dict[str, Counter[int]] = defaultdict(Counter) groups: dict[tuple[str, datetime], set[int]] = defaultdict(set) timestamp_splits: dict[datetime, str] = {} symbols: set[str] = set() rows = 0 with Path(path).open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) fields = set(reader.fieldnames or ()) if not required <= fields: raise ValueError(f"Market dataset missing columns: {sorted(required - fields)}") feature_fields = [field for field in fields if field.startswith("mm_") or field.startswith("active_")] if not set(MARKET_BASE_FEATURES + ACTIVE_FEATURES) <= fields or set(field for field in feature_fields if field.startswith("mm_")) != set(MARKET_BASE_FEATURES): raise ValueError("Market dataset must include the twelve shared causal base features") for line, row in enumerate(reader, start=2): try: decision = datetime.strptime(row["decision_time"], "%Y.%m.%d %H:%M") target_end = datetime.strptime(row["target_end"], "%Y.%m.%d %H:%M") split = row["split"] symbol = row["symbol"].strip() direction = int(row["direction"]) label = int(row["label"]) net_return = float(row["net_log_return"]) numbers = [float(row[field]) for field in feature_fields] spreads = [float(row[field]) for field in ("entry_spread_points", "exit_spread_points")] except (KeyError, TypeError, ValueError) as error: raise ValueError(f"Unknown/malformed observed value on line {line}: {error}") from error if not symbol or direction not in (0, 1) or label not in (0, 1): raise ValueError(f"Invalid symbol/direction/label on line {line}") if not all(math.isfinite(value) for value in [net_return, *numbers, *spreads]) or min(spreads) < 0: raise ValueError(f"Nonfinite feature/outcome or negative spread on line {line}") if label != int(net_return > 0): raise ValueError(f"Label is inconsistent with observed spread-adjusted return on line {line}") direction_value = 1.0 if direction == 0 else -1.0 if float(row["mm_direction"]) != direction_value or float(row["active_symbol_flag"]) != 1.0: raise ValueError(f"Direction or active-symbol feature contract violated on line {line}") volatility = float(row["mm_return_std12"]) if volatility < 0.0: raise ValueError(f"Negative return volatility on line {line}") expected = (float(row["mm_signed_logret1"]) / max(volatility, 1e-8), float(row["mm_signed_logret3"]) / max(volatility, 1e-8), float(row["mm_signed_zscore12"])) actual = tuple(float(row[name]) for name in ACTIVE_FEATURES[1:]) if any(not math.isclose(left, right, rel_tol=1e-9, abs_tol=1e-9) for left, right in zip(actual, expected)): raise ValueError(f"Sparse symbol interactions mismatch causal base features on line {line}") if target_end <= decision: raise ValueError(f"Future outcome must end after decision on line {line}") if split == "train": valid = decision < boundaries[0] and target_end < boundaries[0] elif split == "validation": valid = boundaries[0] <= decision < boundaries[1] and target_end < boundaries[1] elif split == "test": valid = boundaries[1] <= decision < boundaries[2] and target_end < boundaries[2] else: valid = False if not valid: raise ValueError(f"Unknown split or outcome crossing a split boundary on line {line}") if decision in timestamp_splits and timestamp_splits[decision] != split: raise ValueError(f"A decision timestamp is split across partitions on line {line}") timestamp_splits[decision] = split group = groups[(symbol, decision)] if direction in group: raise ValueError(f"Duplicate symbol/time/direction on line {line}") group.add(direction) counts[split] += 1 classes[split][label] += 1 symbols.add(symbol) rows += 1 if not rows: raise ValueError("No real-history rows to audit") if any(directions != {0, 1} for directions in groups.values()): raise ValueError("Each symbol/decision group must retain both BUY and SELL observations") if any(set(classes[split]) != {0, 1} for split in ("train", "validation", "test")): raise ValueError("Each chronological split must contain observed positive and negative outcomes") return {"rows": rows, "symbols": sorted(symbols), "split_counts": dict(counts), "class_counts": {key: dict(value) for key, value in classes.items()}, "split_boundaries": dict(zip(keys, resolved)), "model_data": str(metadata_path.resolve()) if metadata else None, "audit_passed": True} # Old import callers fail clearly instead of changing return types silently. def label_from_status(*args: Any, **kwargs: Any) -> None: raise RuntimeError(RETIRED_TRAINING_MESSAGE) def labels_from_future_prices(*args: Any, **kwargs: Any) -> None: raise RuntimeError(RETIRED_TRAINING_MESSAGE) def load_snapshot_dataframe(*args: Any, **kwargs: Any) -> None: raise RuntimeError(RETIRED_TRAINING_MESSAGE) def build_feature_matrix(*args: Any, **kwargs: Any) -> None: raise RuntimeError(RETIRED_TRAINING_MESSAGE) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) source = parser.add_mutually_exclusive_group() source.add_argument("--snapshot_dir") source.add_argument("--market-csv") parser.add_argument("--limit", type=int) parser.add_argument("--model-data", help="Model metadata JSON; defaults to model_data.json beside --market-csv") parser.add_argument("--train-end", help="Override the model metadata training boundary") parser.add_argument("--validation-end", help="Override the model metadata validation boundary") parser.add_argument("--test-end", help="Override the model metadata test boundary") args = parser.parse_args(argv) try: report = audit_market_dataset(args.market_csv, args.train_end, args.validation_end, args.test_end, args.model_data) if args.market_csv else inspect_snapshots(args.snapshot_dir, args.limit) print(json.dumps(report, indent=2)) except (OSError, ValueError) as error: parser.exit(1, f"Dataset audit failed: {error}\n") return 0 if __name__ == "__main__": raise SystemExit(main())