robinhood/backend/check_duplicates.py

#!/usr/bin/env python3
"""
Check for duplicates and data integrity in the generated historical gold price data.
"""

import os
import pandas as pd
import pyarrow.parquet as pq
from collections import Counter
from datetime import datetime

def check_data_integrity():
    """Check for duplicates and gaps in the historical data."""
    
    # Find all parquet files
    data_dir = "/Users/user/Downloads/gold-trading-simulator/data/parquet/live/XAUUSD/1m"
    all_timestamps = []
    all_records = []
    
    print("🔍 Scanning historical gold price data for duplicates and integrity issues...")
    print("=" * 80)
    
    # Walk through all date directories
    file_count = 0
    for root, dirs, files in os.walk(data_dir):
        for file in files:
            if file.endswith('.parquet'):
                file_path = os.path.join(root, file)
                try:
                    # Read parquet file
                    table = pq.read_table(file_path)
                    df = table.to_pandas()
                    
                    file_count += 1
                    if file_count <= 5:  # Show progress for first few files
                        print(f"📁 Processing {os.path.basename(root)}: {len(df)} records")
                    
                    # Collect timestamps
                    timestamps = df['time'].tolist()
                    all_timestamps.extend(timestamps)
                    
                    # Collect full records for duplicate detection
                    for _, row in df.iterrows():
                        record_tuple = (row['time'], row['open'], row['high'], row['low'], row['close'], row['volume'])
                        all_records.append(record_tuple)
                        
                except Exception as e:
                    print(f"❌ Error reading {file_path}: {e}")
    
    print(f"📊 Processed {file_count} parquet files")
    print(f"📊 Total records found: {len(all_records):,}")
    print(f"📊 Total timestamps found: {len(all_timestamps):,}")
    
    print("\n" + "="*80)
    print("🕐 DUPLICATE TIMESTAMP ANALYSIS")
    print("="*80)
    
    # Check for duplicate timestamps
    timestamp_counts = Counter(all_timestamps)
    duplicate_timestamps = {ts: count for ts, count in timestamp_counts.items() if count > 1}
    
    print(f"Unique timestamps: {len(timestamp_counts):,}")
    print(f"Total timestamps: {len(all_timestamps):,}")
    print(f"Duplicate timestamps found: {len(duplicate_timestamps):,}")
    
    if duplicate_timestamps:
        print("\n⚠️ DUPLICATE TIMESTAMPS DETECTED:")
        for i, (ts, count) in enumerate(list(duplicate_timestamps.items())[:10]):
            dt = datetime.utcfromtimestamp(ts)
            print(f"  • Timestamp {ts} ({dt}): appears {count} times")
            if i >= 9:
                print(f"  ... and {len(duplicate_timestamps) - 10} more")
                break
    else:
        print("✅ No duplicate timestamps found!")
    
    print("\n" + "="*80)
    print("📋 DUPLICATE RECORD ANALYSIS")
    print("="*80)
    
    # Check for duplicate full records
    record_counts = Counter(all_records)
    duplicate_records = {record: count for record, count in record_counts.items() if count > 1}
    
    print(f"Unique records: {len(record_counts):,}")
    print(f"Total records: {len(all_records):,}")
    print(f"Duplicate records found: {len(duplicate_records):,}")
    
    if duplicate_records:
        print("\n⚠️ DUPLICATE RECORDS DETECTED:")
        for i, (record, count) in enumerate(list(duplicate_records.items())[:5]):
            ts, open_price, high, low, close, volume = record
            dt = datetime.utcfromtimestamp(ts)
            print(f"  • {dt}: O={open_price}, H={high}, L={low}, C={close}, V={volume} (appears {count} times)")
            if i >= 4:
                print(f"  ... and {len(duplicate_records) - 5} more")
                break
    else:
        print("✅ No duplicate records found!")
    
    print("\n" + "="*80)
    print("⏰ TIMESTAMP CONTINUITY ANALYSIS")
    print("="*80)
    
    # Check timestamp continuity and gaps
    sorted_timestamps = sorted(all_timestamps)
    
    print(f"Earliest timestamp: {datetime.utcfromtimestamp(sorted_timestamps[0])}")
    print(f"Latest timestamp: {datetime.utcfromtimestamp(sorted_timestamps[-1])}")
    
    # Check for gaps (should be 60 seconds apart for 1-minute data)
    gaps_found = []
    large_gaps = []
    
    for i in range(1, len(sorted_timestamps)):
        time_diff = sorted_timestamps[i] - sorted_timestamps[i-1]
        
        # Expected is 60 seconds (1 minute)
        if time_diff != 60:
            gap_info = {
                'prev_time': datetime.utcfromtimestamp(sorted_timestamps[i-1]),
                'curr_time': datetime.utcfromtimestamp(sorted_timestamps[i]),
                'gap_seconds': time_diff,
                'gap_minutes': time_diff // 60
            }
            gaps_found.append(gap_info)
            
            # Track large gaps (more than 1 day)
            if time_diff > 86400:  # 24 hours
                large_gaps.append(gap_info)
    
    print(f"\nGaps analysis:")
    print(f"Total gaps found: {len(gaps_found):,}")
    print(f"Large gaps (>24h): {len(large_gaps):,}")
    
    if large_gaps:
        print(f"\n📅 Large gaps (expected for weekends):")
        for i, gap in enumerate(large_gaps[:10]):
            days = gap['gap_minutes'] // (60 * 24)
            hours = (gap['gap_minutes'] % (60 * 24)) // 60
            print(f"  • {gap['prev_time']} -> {gap['curr_time']} ({days}d {hours}h)")
            if i >= 9:
                print(f"  ... and {len(large_gaps) - 10} more")
                break
    
    # Check weekday patterns for gaps
    weekend_gaps = 0
    weekday_gaps = 0
    
    for gap in gaps_found:
        # Check if gap starts on Friday or Saturday (weekend gap)
        prev_weekday = gap['prev_time'].weekday()  # Monday=0, Sunday=6
        if prev_weekday >= 4:  # Friday or Saturday
            weekend_gaps += 1
        else:
            weekday_gaps += 1
    
    print(f"\nGap patterns:")
    print(f"Weekend gaps (Fri-Mon): {weekend_gaps:,}")
    print(f"Weekday gaps: {weekday_gaps:,}")
    
    if weekday_gaps > 0:
        print(f"\n⚠️ Unexpected weekday gaps:")
        weekday_gap_count = 0
        for gap in gaps_found:
            prev_weekday = gap['prev_time'].weekday()
            if prev_weekday < 4 and weekday_gap_count < 5:  # Not Friday/Saturday
                print(f"  • {gap['prev_time']} -> {gap['curr_time']} ({gap['gap_minutes']} minutes)")
                weekday_gap_count += 1
        if weekday_gaps > 5:
            print(f"  ... and {weekday_gaps - 5} more weekday gaps")
    
    print("\n" + "="*80)
    print("📈 DATA QUALITY SUMMARY")
    print("="*80)
    
    # Summary
    issues = []
    if duplicate_timestamps:
        issues.append(f"{len(duplicate_timestamps)} duplicate timestamps")
    if duplicate_records:
        issues.append(f"{len(duplicate_records)} duplicate records")
    if weekday_gaps > 0:
        issues.append(f"{weekday_gaps} unexpected weekday gaps")
    
    if issues:
        print("⚠️ Issues found:")
        for issue in issues:
            print(f"  • {issue}")
    else:
        print("✅ Data quality check PASSED!")
        print("  • No duplicate timestamps")
        print("  • No duplicate records")
        print("  • Weekend gaps are expected (market closed)")
        print(f"  • Total records: {len(all_records):,}")
        print(f"  • Date range: {datetime.utcfromtimestamp(sorted_timestamps[0]).date()} to {datetime.utcfromtimestamp(sorted_timestamps[-1]).date()}")

if __name__ == "__main__":
    check_data_integrity()