tool_validate_market_data.py

import os
import polars as pl
from datetime import datetime, timedelta

REQUIRED_COLUMNS = [
    'timestamp', 'open', 'high', 'low', 'close', 'volume',
    'quote_asset_volume', 'number_of_trades',
    'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume'
]

# Map intervals to their timedelta for continuity checks
INTERVAL_DELTAS = {
    '15m': timedelta(minutes=15),
    '30m': timedelta(minutes=30),
    '1h': timedelta(hours=1),
    '2h': timedelta(hours=2),
    '4h': timedelta(hours=4),
    '6h': timedelta(hours=6),
    '8h': timedelta(hours=8),
    '12h': timedelta(hours=12),
}


def validate_file(filepath, interval):
    """
    Universal validation function for all intervals.
    Checks for missing columns, nulls, duplicates, and continuity.
    """
    try:
        df = pl.read_csv(filepath)
        filename = os.path.basename(filepath)

        # Check if all required columns exist
        missing_cols = [col for col in REQUIRED_COLUMNS if col not in df.columns]
        if missing_cols:
            return f"{filename}: ❌ Missing columns: {missing_cols}"

        # Parse timestamps with consistent precision
        df = df.with_columns([
            pl.col("timestamp")
            .str.strptime(pl.Datetime, "%Y-%m-%d %H:%M:%S", strict=False)
            .cast(pl.Datetime("us"))  # Force microsecond precision
            .alias("timestamp")
        ])

        # Check for malformed timestamps
        if df["timestamp"].null_count() > 0:
            return f"{filename}: ❌ Malformed timestamps"

        # Sort by timestamp
        df = df.sort("timestamp")

        # Check for nulls in any data columns
        null_counts = df.select(pl.exclude("timestamp").null_count())
        total_nulls = null_counts.to_numpy().sum()
        if total_nulls > 0:
            return f"{filename}: ❌ Contains {total_nulls} nulls in data columns"

        # Check for duplicate timestamps
        if df["timestamp"].is_duplicated().any():
            duplicate_count = df["timestamp"].is_duplicated().sum()
            return f"{filename}: ❌ {duplicate_count} duplicate timestamps found"

        # Check data validity (prices should be positive, volumes non-negative)
        price_cols = ['open', 'high', 'low', 'close']
        volume_cols = ['volume', 'quote_asset_volume', 'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume']

        # Check for non-positive prices
        for col in price_cols:
            if (df[col] <= 0).any():
                return f"{filename}: ❌ Non-positive values found in {col}"

        # Check for negative volumes
        for col in volume_cols:
            if (df[col] < 0).any():
                return f"{filename}: ❌ Negative values found in {col}"

        # Check OHLC relationships (high >= low, open/close between high/low)
        if (df["high"] < df["low"]).any():
            return f"{filename}: ❌ High < Low detected"

        if (df["open"] > df["high"]).any() or (df["open"] < df["low"]).any():
            return f"{filename}: ❌ Open price outside High/Low range"

        if (df["close"] > df["high"]).any() or (df["close"] < df["low"]).any():
            return f"{filename}: ❌ Close price outside High/Low range"

        # Check continuity - ensure no missing timestamps
        if interval in INTERVAL_DELTAS:
            delta = INTERVAL_DELTAS[interval]
            ts_list = df["timestamp"].to_list()

            if len(ts_list) > 1:
                # Generate expected timeline
                start_time = ts_list[0]
                end_time = ts_list[-1]

                expected = []
                current = start_time
                while current <= end_time:
                    expected.append(current)
                    current += delta

                actual_set = set(ts_list)
                expected_set = set(expected)
                missing = sorted(expected_set - actual_set)

                if missing:
                    sample = [d.strftime("%Y-%m-%d %H:%M") for d in missing[:3]]
                    return f"{filename}: ❌ Missing {len(missing)} timestamps (examples: {sample})"

        # Check for reasonable data ranges
        row_count = len(df)
        if row_count == 0:
            return f"{filename}: ❌ Empty file"
        elif row_count < 10:
            return f"{filename}: ⚠️ Very small dataset ({row_count} rows)"

        # All checks passed
        date_range = f"{df['timestamp'].min().strftime('%Y-%m-%d')} to {df['timestamp'].max().strftime('%Y-%m-%d')}"
        return f"{filename}: ✅ Passed ({row_count} rows, {date_range})"

    except Exception as e:
        return f"{os.path.basename(filepath)}: ❌ Exception — {str(e)}"


# Individual checker functions for each interval
def check_15m_file(filepath):
    return validate_file(filepath, '15m')


def check_30m_file(filepath):
    return validate_file(filepath, '30m')


def check_1h_file(filepath):
    return validate_file(filepath, '1h')


def check_2h_file(filepath):
    return validate_file(filepath, '2h')


def check_4h_file(filepath):
    return validate_file(filepath, '4h')


def check_6h_file(filepath):
    return validate_file(filepath, '6h')


def check_8h_file(filepath):
    return validate_file(filepath, '8h')


def check_12h_file(filepath):
    return validate_file(filepath, '12h')


CHECKERS = {
    '15m': check_15m_file,
    '30m': check_30m_file,
    '1h': check_1h_file,
    '2h': check_2h_file,
    '4h': check_4h_file,
    '6h': check_6h_file,
    '8h': check_8h_file,
    '12h': check_12h_file,
}


def scan_directory(base_dir):
    """
    Scan all interval directories and validate CSV files.
    """
    results = []
    summary = {
        'total_files': 0,
        'passed': 0,
        'failed': 0,
        'warnings': 0,
        'by_interval': {}
    }

    for interval, checker in CHECKERS.items():
        folder = os.path.join(base_dir, interval)
        if not os.path.exists(folder):
            continue

        interval_stats = {'total': 0, 'passed': 0, 'failed': 0, 'warnings': 0}

        print(f"\n--- Validating {interval} files ---")

        csv_files = [f for f in os.listdir(folder) if f.endswith(".csv")]
        csv_files.sort()

        for filename in csv_files:
            filepath = os.path.join(folder, filename)
            result = checker(filepath)
            if result:
                results.append(result)
                print(result)

                # Update statistics
                summary['total_files'] += 1
                interval_stats['total'] += 1

                if "✅ Passed" in result:
                    summary['passed'] += 1
                    interval_stats['passed'] += 1
                elif "⚠️" in result:
                    summary['warnings'] += 1
                    interval_stats['warnings'] += 1
                else:
                    summary['failed'] += 1
                    interval_stats['failed'] += 1

        summary['by_interval'][interval] = interval_stats
        print(
            f"   {interval} summary: {interval_stats['passed']} passed, {interval_stats['failed']} failed, {interval_stats['warnings']} warnings")

    return results, summary


def print_summary(summary):
    """
    Print a detailed summary of validation results.
    """
    print("\n" + "=" * 60)
    print("VALIDATION SUMMARY")
    print("=" * 60)

    print(f"Total files validated: {summary['total_files']}")
    print(f"✅ Passed: {summary['passed']}")
    print(f"❌ Failed: {summary['failed']}")
    print(f"⚠️ Warnings: {summary['warnings']}")

    if summary['failed'] > 0:
        print(f"\n🚨 {summary['failed']} files have issues that need attention!")
    elif summary['warnings'] > 0:
        print(f"\n⚠️ All files passed validation, but {summary['warnings']} have warnings.")
    else:
        print(f"\n🎉 All {summary['passed']} files passed validation!")

    print("\nBy interval:")
    for interval, stats in summary['by_interval'].items():
        if stats['total'] > 0:
            success_rate = (stats['passed'] / stats['total']) * 100
            print(f"  {interval}: {stats['passed']}/{stats['total']} passed ({success_rate:.1f}%)")


if __name__ == "__main__":
    base_directory = "datasets/raw/full_scope"

    print("🔍 Starting comprehensive data validation...")
    print(f"📁 Scanning directory: {base_directory}")

    results, summary = scan_directory(base_directory)
    print_summary(summary)