archive/testing/validate_forecast.py

#!/usr/bin/env python3
"""Validate forecast results"""
import sys
sys.stdout.reconfigure(encoding='utf-8', errors='replace')

import polars as pl
from pathlib import Path

# Find the most recent forecast file in Windows temp directory
temp_dir = Path(r"C:\Users\evgue\AppData\Local\Temp\gradio")
forecast_files = list(temp_dir.glob("**/forecast_*.parquet"))

if not forecast_files:
    print("[ERROR] No forecast files found", flush=True)
    sys.exit(1)

# Get the most recent file
latest_forecast = max(forecast_files, key=lambda p: p.stat().st_mtime)
print(f"Examining: {latest_forecast.name}", flush=True)
print(f"Full path: {latest_forecast}", flush=True)

# Load and examine the forecast
df = pl.read_parquet(latest_forecast)

print(f"\n[OK] Forecast loaded successfully", flush=True)
print(f"Shape: {df.shape} (rows x columns)", flush=True)
print(f"\nColumns: {df.columns}", flush=True)
print(f"\nData types:\n{df.dtypes}", flush=True)

# Check for expected structure
print(f"\n--- Validation ---", flush=True)
assert 'timestamp' in df.columns, "Missing timestamp column"
print("[OK] timestamp column present", flush=True)

# Check for forecast columns (median, q10, q90)
forecast_cols = [c for c in df.columns if c != 'timestamp']
print(f"[OK] Found {len(forecast_cols)} forecast columns", flush=True)

# Check number of rows (should be 168 for 7 days)
expected_rows = 168  # 7 days * 24 hours
print(f"[OK] Rows: {len(df)} (expected: {expected_rows})", flush=True)

# Display first few rows
print(f"\n--- First 5 rows ---", flush=True)
print(df.head(5))

# Display summary statistics
print(f"\n--- Summary Statistics ---", flush=True)
print(df.select([c for c in df.columns if c != 'timestamp']).describe())

print(f"\n[SUCCESS] Smoke test validation complete!", flush=True)