FX-ML-Trading-Engine/Q Research/scripts/validate_dataset.py

#!/usr/bin/env python3
"""
Validate dataset quality (missing data, duplicate timestamps, gaps, outliers).

Example:
    python scripts/validate_dataset.py --path data/raw/EURUSD_H1.csv
"""

from __future__ import annotations

import argparse
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, Optional, Tuple

import numpy as np
import pandas as pd
from loguru import logger


TIME_COLUMNS = ["ts", "time", "timestamp", "datetime", "date"]
DEFAULT_MANIFEST = "data/_manifest.json"
REPORT_DIR = Path("results/data_quality")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Validate dataset quality.")
    parser.add_argument("--path", required=True, help="Dataset file path (relative or absolute).")
    parser.add_argument(
        "--manifest",
        default=DEFAULT_MANIFEST,
        help="Optional manifest JSON to enrich report.",
    )
    parser.add_argument(
        "--output",
        default=None,
        help="Optional explicit path for JSON report.",
    )
    parser.add_argument(
        "--max-outlier-z",
        type=float,
        default=5.0,
        help="Z-score threshold to flag numeric outliers.",
    )
    return parser.parse_args()


def detect_time_column(df: pd.DataFrame) -> Optional[str]:
    for col in TIME_COLUMNS:
        if col in df.columns:
            return col
    for col in df.columns:
        if pd.api.types.is_datetime64_any_dtype(df[col]):
            return col
    return None


def infer_interval_seconds(series: pd.Series) -> Optional[float]:
    ts = pd.to_datetime(series, utc=True, errors="coerce").dropna()
    if ts.size < 3:
        return None
    diffs = np.diff(ts.values.astype("datetime64[ns]").astype(np.int64) // 1_000_000_000)
    diffs = diffs[diffs > 0]
    if diffs.size == 0:
        return None
    return float(np.median(diffs))


def load_manifest_entry(manifest_path: Path, dataset_path: Path) -> Optional[Dict]:
    if not manifest_path.exists():
        return None
    try:
        with manifest_path.open("r", encoding="utf-8") as fh:
            manifest = json.load(fh)
    except Exception as exc:
        logger.warning(f"Failed to parse manifest {manifest_path}: {exc}")
        return None
    rel = str(dataset_path)
    rel_alt = str(dataset_path.resolve().relative_to(Path.cwd()))
    for entry in manifest.get("files", []):
        if entry.get("path") in (rel, rel_alt):
            return entry
    return None


def summarize_numeric_outliers(df: pd.DataFrame, threshold: float) -> Dict[str, int]:
    outliers: Dict[str, int] = {}
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    if numeric_cols.empty:
        return outliers
    for col in numeric_cols:
        series = df[col].dropna()
        if series.empty:
            continue
        zscores = np.abs((series - series.mean()) / (series.std(ddof=1) or 1))
        count = int((zscores > threshold).sum())
        if count:
            outliers[col] = count
    return outliers


def compute_report(path: Path, manifest_entry: Optional[Dict], z_threshold: float) -> Dict:
    df = pd.read_csv(path)
    total_rows = int(df.shape[0])
    null_counts = df.isna().sum().to_dict()

    time_col = detect_time_column(df)
    duplicates = 0
    gap_ratio = 0.0
    inferred_interval = None
    if time_col:
        ts = pd.to_datetime(df[time_col], utc=True, errors="coerce")
        duplicates = int(ts.duplicated().sum())
        valid_ts = ts.dropna().sort_values()
        inferred_interval = infer_interval_seconds(valid_ts)
        if inferred_interval:
            diffs = np.diff(valid_ts.values.astype("datetime64[ns]").astype(np.int64) // 1_000_000_000)
            expected = inferred_interval
            gaps = diffs[diffs > expected * 1.5]
            gap_ratio = float(len(gaps) / max(len(diffs), 1))
    outliers = summarize_numeric_outliers(df, z_threshold)

    messages = []
    severity = "pass"
    if total_rows == 0:
        severity = "error"
        messages.append("Dataset is empty.")
    if duplicates > 0:
        severity = "error"
        messages.append(f"Found {duplicates} duplicate timestamps in column '{time_col}'.")
    if gap_ratio > 0.01:
        severity = "error"
        messages.append(f"Timestamp gaps exceed 1% of intervals (ratio={gap_ratio:.4f}).")
    elif gap_ratio > 0:
        severity = "warn"
        messages.append(f"Non-zero timestamp gaps detected (ratio={gap_ratio:.4f}).")
    null_ratio = max((count / total_rows) if total_rows else 0 for count in null_counts.values() or [0])
    if null_ratio > 0.05:
        severity = "error"
        messages.append(f"Columns contain >5% nulls (max ratio={null_ratio:.4f}).")
    elif null_ratio > 0:
        severity = "warn"
        messages.append(f"Columns contain nulls (max ratio={null_ratio:.4f}).")
    if outliers:
        if severity == "pass":
            severity = "warn"
        messages.append(f"Detected numeric outliers (threshold z>{z_threshold}).")

    report = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "dataset_path": str(path),
        "total_rows": total_rows,
        "time_column": time_col,
        "inferred_interval_seconds": inferred_interval,
        "duplicate_timestamps": duplicates,
        "gap_ratio": gap_ratio,
        "null_counts": null_counts,
        "numeric_outliers": outliers,
        "severity": severity,
        "messages": messages,
    }
    if manifest_entry:
        report["manifest"] = {
            "path": manifest_entry.get("path"),
            "sha256": manifest_entry.get("sha256"),
            "rows": manifest_entry.get("rows"),
            "time_start": manifest_entry.get("time_start"),
            "time_end": manifest_entry.get("time_end"),
        }
    return report


def save_report(report: Dict, output_path: Optional[Path], dataset_path: Path) -> Path:
    if output_path is None:
        REPORT_DIR.mkdir(parents=True, exist_ok=True)
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        name = dataset_path.stem
        output_path = REPORT_DIR / f"{ts}_{name}.json"
    else:
        output_path = output_path.resolve()
        output_path.parent.mkdir(parents=True, exist_ok=True)

    with output_path.open("w", encoding="utf-8") as fh:
        json.dump(report, fh, indent=2, ensure_ascii=False)
    return output_path


def main() -> None:
    args = parse_args()
    dataset_path = Path(args.path).expanduser().resolve()
    if not dataset_path.exists():
        raise FileNotFoundError(f"Dataset not found: {dataset_path}")

    manifest_entry = None
    if args.manifest:
        manifest_entry = load_manifest_entry(Path(args.manifest), dataset_path)

    report = compute_report(dataset_path, manifest_entry, args.max_outlier_z)
    output_path = save_report(report, Path(args.output) if args.output else None, dataset_path)
    logger.info(f"Validation severity: {report['severity']}")
    logger.info(f"Report written to {output_path}")


if __name__ == "__main__":
    main()