data_quality.py

"""
Data quality assessment for market data.
Important for quant research - shows data is realistic.
"""

import pandas as pd
import numpy as np
from typing import Dict, Any

def assess_data_quality(df: pd.DataFrame) -> Dict[str, Any]:
    """
    Comprehensive data quality assessment.
    Returns metrics that quant firms care about.
    """
    assessment = {}
    
    # Handle timestamp - could be in column or index
    if 'timestamp' in df.columns:
        timestamps = df['timestamp']
    else:
        timestamps = df.index
    
    # Basic statistics
    assessment['total_ticks'] = len(df)
    assessment['time_span'] = str(timestamps.max() - timestamps.min())
    assessment['unique_timestamps'] = timestamps.nunique() if hasattr(timestamps, 'nunique') else len(timestamps.unique())
    
    # Price dynamics
    mid_prices = (df['bid'] + df['ask']) / 2
    returns = mid_prices.pct_change().dropna()
    
    assessment['price_stats'] = {
        'mean_price': float(mid_prices.mean()),
        'price_volatility': float(returns.std() * np.sqrt(252 * 390 * 60)),  # Annualized
        'max_return': float(returns.max()),
        'min_return': float(returns.min()),
        'skewness': float(returns.skew()),
        'kurtosis': float(returns.kurtosis())
    }
    
    # Spread analysis
    spreads = df['ask'] - df['bid']
    spread_bps = spreads / mid_prices * 10000
    
    assessment['spread_stats'] = {
        'mean_spread_bps': float(spread_bps.mean()),
        'median_spread_bps': float(spread_bps.median()),
        'spread_volatility': float(spread_bps.std()),
        'tight_spread_pct': float((spread_bps < 5).mean() * 100),  # % under 5 bps
        'wide_spread_pct': float((spread_bps > 20).mean() * 100)   # % over 20 bps
    }
    
    # Microstructure analysis
    assessment['microstructure'] = {
        'mean_bid_size': float(df['bid_size'].mean()),
        'mean_ask_size': float(df['ask_size'].mean()),
        'size_imbalance': float((df['bid_size'] - df['ask_size']).mean()),
        'tick_frequency_per_minute': float(len(df) / ((timestamps.max() - timestamps.min()).total_seconds() / 60))
    }
    
    # Data quality flags
    assessment['quality_flags'] = {
        'negative_spreads': int((spreads < 0).sum()),
        'zero_sizes': int(((df['bid_size'] == 0) | (df['ask_size'] == 0)).sum()),
        'extreme_spreads': int((spread_bps > 100).sum()),  # > 100 bps
        'missing_timestamps': int(timestamps.isna().sum()) if hasattr(timestamps, 'isna') else 0,
        'duplicate_timestamps': int(len(df) - assessment['unique_timestamps'])
    }
    
    # Overall quality score (0-100)
    quality_score = 100
    if assessment['quality_flags']['negative_spreads'] > 0:
        quality_score -= 20
    if assessment['quality_flags']['zero_sizes'] > len(df) * 0.01:  # >1% zero sizes
        quality_score -= 10
    if assessment['quality_flags']['extreme_spreads'] > len(df) * 0.05:  # >5% extreme
        quality_score -= 15
    if assessment['spread_stats']['mean_spread_bps'] < 0.1:  # Unrealistically tight
        quality_score -= 10
    
    assessment['overall_quality_score'] = max(0, quality_score)
    
    return assessment


def print_data_quality_report(df: pd.DataFrame):
    """Print a formatted data quality report."""
    assessment = assess_data_quality(df)
    
    print("=" * 60)
    print("📊 MARKET DATA QUALITY ASSESSMENT")
    print("=" * 60)
    
    print(f"📈 Dataset Overview:")
    print(f"  • Total ticks: {assessment['total_ticks']:,}")
    print(f"  • Time span: {assessment['time_span']}")
    print(f"  • Tick frequency: {assessment['microstructure']['tick_frequency_per_minute']:.1f}/min")
    
    print(f"\n💰 Price Dynamics:")
    stats = assessment['price_stats']
    print(f"  • Mean price: ${stats['mean_price']:.2f}")
    print(f"  • Annualized volatility: {stats['price_volatility']:.1%}")
    print(f"  • Return skew: {stats['skewness']:.2f}")
    print(f"  • Return kurtosis: {stats['kurtosis']:.2f}")
    
    print(f"\n📏 Spread Analysis:")
    spreads = assessment['spread_stats']
    print(f"  • Mean spread: {spreads['mean_spread_bps']:.1f} bps")
    print(f"  • Median spread: {spreads['median_spread_bps']:.1f} bps")
    print(f"  • Tight spreads (<5 bps): {spreads['tight_spread_pct']:.1f}%")
    print(f"  • Wide spreads (>20 bps): {spreads['wide_spread_pct']:.1f}%")
    
    print(f"\n🏗️ Market Microstructure:")
    micro = assessment['microstructure']
    print(f"  • Average bid size: {micro['mean_bid_size']:.0f}")
    print(f"  • Average ask size: {micro['mean_ask_size']:.0f}")
    print(f"  • Size imbalance: {micro['size_imbalance']:.0f}")
    
    print(f"\n⚠️ Quality Flags:")
    flags = assessment['quality_flags']
    print(f"  • Negative spreads: {flags['negative_spreads']}")
    print(f"  • Zero sizes: {flags['zero_sizes']}")
    print(f"  • Extreme spreads: {flags['extreme_spreads']}")
    print(f"  • Duplicate timestamps: {flags['duplicate_timestamps']}")
    
    score = assessment['overall_quality_score']
    print(f"\n🎯 Overall Quality Score: {score}/100", end="")
    if score >= 90:
        print(" ✅ Excellent")
    elif score >= 75:
        print(" ⚠️ Good")
    elif score >= 60:
        print(" ⚠️ Fair") 
    else:
        print(" ❌ Poor")
    
    print("=" * 60)