Fix Mismatched Gold Data Feeds in 20 Minutes - Spot vs GC Futures

import pandas as pd
import pytz
from datetime import datetime

# Personal note: Learned this after my first merge created duplicate timestamps
def load_spot_gold(filepath):
    """Load spot gold (XAU/USD) data - trades nearly 24/5"""
    df = pd.read_csv(filepath)
    
    # Spot gold typically comes in EST from most providers
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df['timestamp'] = df['timestamp'].dt.tz_localize('US/Eastern').dt.tz_convert('UTC')
    
    # Resample to 1-minute bars
    df = df.set_index('timestamp')
    df = df[['price']].resample('1min').agg({
        'price': 'last'  # Use last price in each minute
    }).dropna()
    
    df.columns = ['spot_price']
    return df

def load_gc_futures(filepath):
    """Load COMEX GC futures - trades 23 hours with specific breaks"""
    df = pd.read_csv(filepath)
    
    # Watch out: GC data usually comes in CST (COMEX time)
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df['timestamp'] = df['timestamp'].dt.tz_localize('US/Central').dt.tz_convert('UTC')
    
    df = df.set_index('timestamp')
    df = df[['contract', 'price', 'volume']].resample('1min').agg({
        'contract': 'last',
        'price': 'last',
        'volume': 'sum'
    }).dropna()
    
    df.columns = ['futures_contract', 'futures_price', 'futures_volume']
    return df

# Load your data
spot_df = load_spot_gold('xauusd_data.csv')
futures_df = load_gc_futures('gc_futures_data.csv')

print(f"Spot data: {len(spot_df)} rows from {spot_df.index[0]} to {spot_df.index[-1]}")
print(f"Futures data: {len(futures_df)} rows")

def identify_roll_dates(futures_df):
    """Detect when the active contract changes"""
    # Personal note: I initially missed rolls happening mid-day
    contract_changes = futures_df['futures_contract'] != futures_df['futures_contract'].shift(1)
    roll_dates = futures_df[contract_changes].index
    
    print(f"\nDetected {len(roll_dates)} contract rolls:")
    for i, date in enumerate(roll_dates[:5]):  # Show first 5
        old_contract = futures_df.loc[:date, 'futures_contract'].iloc[-2] if i > 0 else 'N/A'
        new_contract = futures_df.loc[date, 'futures_contract']
        print(f"  {date.strftime('%Y-%m-%d %H:%M')} UTC: {old_contract} â†' {new_contract}")
    
    return roll_dates

def create_continuous_series(futures_df, method='ratio'):
    """
    Adjust prices across roll dates for continuity
    method: 'ratio' or 'difference' (ratio works better for gold)
    """
    df = futures_df.copy()
    df['continuous_price'] = df['futures_price']
    
    roll_dates = identify_roll_dates(df)
    
    # Watch out: Apply adjustments backward from most recent
    for roll_date in reversed(roll_dates[1:]):  # Skip first roll
        # Get prices at roll
        old_price = df.loc[:roll_date, 'futures_price'].iloc[-2]
        new_price = df.loc[roll_date, 'futures_price']
        
        if method == 'ratio':
            adjustment = new_price / old_price
            # Adjust all historical prices
            df.loc[:roll_date, 'continuous_price'] *= adjustment
        else:  # difference
            adjustment = new_price - old_price
            df.loc[:roll_date, 'continuous_price'] += adjustment
    
    return df

# Apply roll adjustments
futures_df = create_continuous_series(futures_df, method='ratio')
print(f"\nContinuous series created. Price range: ${futures_df['continuous_price'].min():.2f} - ${futures_df['continuous_price'].max():.2f}")

def merge_gold_data(spot_df, futures_df):
    """
    Merge spot and futures with forward-fill for futures gaps
    (spot trades more hours than futures)
    """
    # Outer join to keep all spot timestamps
    merged = spot_df.join(futures_df, how='outer')
    
    # Forward-fill futures data during trading breaks
    # Personal note: Backfill would be wrong - it leaks future info
    merged[['futures_contract', 'continuous_price', 'futures_volume']] = \
        merged[['futures_contract', 'continuous_price', 'futures_volume']].ffill(limit=60)
    
    # Only keep rows where we have both spot and futures
    merged = merged.dropna(subset=['spot_price', 'continuous_price'])
    
    return merged

def calculate_basis(merged_df):
    """Calculate basis spread and percentage"""
    df = merged_df.copy()
    
    # Basis = Futures - Spot (positive when futures trade at premium)
    df['basis_raw'] = df['continuous_price'] - df['spot_price']
    df['basis_pct'] = (df['basis_raw'] / df['spot_price']) * 100
    
    # Watch out: Large basis swings might indicate data errors
    large_moves = df['basis_pct'].abs() > 2.0  # Flag if basis > 2%
    if large_moves.any():
        print(f"\nWarning: {large_moves.sum()} timestamps with basis > 2%")
        print("Check for data quality issues or extreme market conditions")
    
    return df

# Merge and calculate
merged_df = merge_gold_data(spot_df, futures_df)
final_df = calculate_basis(merged_df)

print(f"\nMerged dataset: {len(final_df)} aligned timestamps")
print(f"Date range: {final_df.index[0].strftime('%Y-%m-%d')} to {final_df.index[-1].strftime('%Y-%m-%d')}")
print(f"\nBasis statistics:")
print(f"  Mean: ${final_df['basis_raw'].mean():.2f} ({final_df['basis_pct'].mean():.3f}%)")
print(f"  Std Dev: ${final_df['basis_raw'].std():.2f}")
print(f"  Min/Max: ${final_df['basis_raw'].min():.2f} / ${final_df['basis_raw'].max():.2f}")

# Save for further analysis
final_df.to_csv('merged_gold_data.csv')
print("\nSaved to merged_gold_data.csv")

def validate_merged_data(df):
    """Run quality checks on merged dataset"""
    issues = []
    
    # Check 1: Price reasonableness
    if df['spot_price'].min() < 1000 or df['spot_price'].max() > 5000:
        issues.append(f"Spot prices outside expected range: ${df['spot_price'].min():.2f} - ${df['spot_price'].max():.2f}")
    
    # Check 2: Basis magnitude
    extreme_basis = (df['basis_pct'].abs() > 1.0).sum()
    if extreme_basis > len(df) * 0.05:  # More than 5% of data
        issues.append(f"{extreme_basis} rows ({extreme_basis/len(df)*100:.1f}%) with basis > 1%")
    
    # Check 3: Data gaps
    time_diffs = df.index.to_series().diff()
    large_gaps = time_diffs > pd.Timedelta(hours=2)
    if large_gaps.any():
        issues.append(f"{large_gaps.sum()} gaps > 2 hours detected")
        print("\nLarge gaps found at:")
        print(df[large_gaps].index[:3])  # Show first 3
    
    # Check 4: Duplicate timestamps
    if df.index.duplicated().any():
        issues.append(f"{df.index.duplicated().sum()} duplicate timestamps")
    
    # Check 5: Volume sanity (futures should have some volume)
    low_volume = (df['futures_volume'] < 10).sum()
    if low_volume > len(df) * 0.3:
        issues.append(f"{low_volume} rows ({low_volume/len(df)*100:.1f}%) with very low futures volume")
    
    # Report
    if issues:
        print("\n⚠️  Data Quality Issues Found:")
        for issue in issues:
            print(f"  - {issue}")
        return False
    else:
        print("\n✓ All validation checks passed")
        return True

# Run validation
is_valid = validate_merged_data(final_df)

# Quick visual check
print("\nSample of merged data:")
print(final_df[['spot_price', 'continuous_price', 'basis_raw', 'basis_pct', 'futures_contract']].head(10))