PIPE Data Pipelines

import numpy as np
import pandas as pd

class DataPipeline:
    """Simple data pipeline with validation."""
    
    def __init__(self, max_gap_bars=5, spike_threshold=5.0):
        self.max_gap_bars = max_gap_bars
        self.spike_threshold = spike_threshold
    
    def validate(self, df: pd.DataFrame) -> dict:
        """Validate OHLCV data quality."""
        issues = []
        
        # Check for missing bars
        if hasattr(df.index, 'freq') and df.index.freq:
            expected = pd.date_range(df.index[0], df.index[-1], freq=df.index.freq)
            missing = len(expected) - len(df)
            if missing > 0:
                issues.append(f"[WARN] {missing} missing bars detected")
        
        # Check for negative prices
        if (df[['open', 'high', 'low', 'close']] <= 0).any().any():
            issues.append("[ALERT] Negative or zero prices found")
        
        # Check OHLC consistency
        if (df['high'] < df['low']).any():
            issues.append("[ALERT] High < Low detected")
        
        # Check for spikes
        returns = df['close'].pct_change()
        spike_mask = returns.abs() > self.spike_threshold * returns.std()
        if spike_mask.any():
            issues.append(f"[WARN] {spike_mask.sum()} price spikes detected")
        
        # Check volume
        if (df['volume'] == 0).mean() > 0.1:
            issues.append(f"[WARN] {(df['volume']==0).mean():.0%} zero-volume bars")
        
        status = "[PASS] CLEAN" if not issues else "[WARN] ISSUES FOUND"
        print(f"Data validation: {status}")
        for issue in issues:
            print(f"  {issue}")
        
        return {'clean': len(issues) == 0, 'issues': issues}