Upload run_pipeline.py

run_pipeline.py

@@ -0,0 +1,177 @@
+"""
+Main execution script for the Robust Recommender System Pipeline.
+Demonstrates: streaming data, incremental training, error compensation,
+data drift detection, model drift detection, and robust output.
+"""
+import numpy as np
+import torch
+import json
+import time
+from recommender_pipeline.data_stream import StreamingDataSource
+from recommender_pipeline.pipeline import RobustRecommenderPipeline
+
+
+def main():
+    print("=" * 70)
+    print("ROBUST RECOMMENDER SYSTEM PIPELINE")
+    print("Features: Streaming | Error Compensation | Data Drift | Model Drift")
+    print("=" * 70)
+    
+    # --- 1. Create streaming data source with occasional drift ---
+    print("\n[1] Initializing streaming data source...")
+    stream = StreamingDataSource(
+        dataset_name="reczoo/Movielens1M_m1",
+        batch_size=128,
+        shuffle=True,
+        synthetic=True,  # Use synthetic data for reliable demo
+        n_users=1000,
+        n_items=500,
+        drift_probability=0.08,  # 8% chance of drift per batch
+        random_state=42
+    )
+    
+    # --- 2. Initialize pipeline ---
+    print("\n[2] Initializing robust recommender pipeline...")
+    device = 'cuda' if torch.cuda.is_available() else 'cpu'
+    print(f"Using device: {device}")
+    
+    pipeline = RobustRecommenderPipeline(
+        n_users=None,  # Auto-infer
+        n_items=None,
+        embedding_dim=64,
+        hidden_dims=[128, 64, 32],
+        learning_rate=1e-3,
+        device=device,
+        error_memory_size=2000,
+        k_neighbors=30,
+        lambda_comp=1.0,
+        data_drift_window=1000,
+        model_drift_reference=500,
+        model_drift_current=500,
+        degradation_threshold=0.05,
+        batch_size=128
+    )
+    
+    # --- 3. Run pipeline on stream ---
+    print("\n[3] Running pipeline on data stream...")
+    print("-" * 70)
+    
+    pipeline.initialize(stream)
+    
+    # Process 50 batches (simulate real-time streaming)
+    results = pipeline.run(stream, max_batches=50)
+    
+    print("-" * 70)
+    
+    # --- 4. Display summary ---
+    print("\n[4] Pipeline Summary")
+    print("=" * 70)
+    
+    summary = pipeline.get_summary()
+    print(json.dumps(summary, indent=2, default=str))
+    
+    # --- 5. Detailed drift analysis ---
+    print("\n[5] Drift Analysis")
+    print("=" * 70)
+    
+    data_drift_history = list(pipeline.data_detector.drift_history)
+    model_drift_history = list(pipeline.model_detector.metric_history)
+    
+    print(f"Total data drift checks: {len(data_drift_history)}")
+    print(f"Data drift events: {sum(1 for d in data_drift_history if d.get('drift_detected', False))}")
+    print(f"Total model drift checks: {len(model_drift_history)}")
+    print(f"Model drift events: {sum(1 for d in model_drift_history if d.get('drift_detected', False))}")
+    
+    # Show recent alerts
+    alerts = list(pipeline.drift_monitor.alerts)
+    if alerts:
+        print(f"\nRecent Alerts ({len(alerts)} total):")
+        for alert in alerts[-5:]:
+            print(f"  [{alert['alert_level'].upper()}] {alert['recommendation'][:100]}...")
+    else:
+        print("\nNo drift alerts triggered.")
+    
+    # --- 6. Performance comparison: base vs compensated predictions ---
+    print("\n[6] Prediction Quality: Base vs Error-Compensated")
+    print("=" * 70)
+    
+    base_preds = []
+    comp_preds = []
+    labels = []
+    
+    for r in results:
+        base_preds.extend(r['predictions']['base'])
+        comp_preds.extend(r['predictions']['compensated'])
+        if 'label' in r['data'].columns:
+            labels.extend(r['data']['label'].values)
+        elif 'rating' in r['data'].columns:
+            labels.extend((r['data']['rating'].values >= 3.5).astype(int))
+    
+    if labels:
+        from sklearn.metrics import roc_auc_score, log_loss
+        
+        base_preds = np.array(base_preds)
+        comp_preds = np.array(comp_preds)
+        labels = np.array(labels)
+        
+        try:
+            base_auc = roc_auc_score(labels, base_preds)
+            comp_auc = roc_auc_score(labels, comp_preds)
+            print(f"Base Model AUC:      {base_auc:.4f}")
+            print(f"Compensated AUC:     {comp_auc:.4f}")
+            print(f"Improvement:         {(comp_auc - base_auc):.4f}")
+        except Exception as e:
+            print(f"AUC calculation skipped: {e}")
+        
+        try:
+            base_ll = log_loss(labels, np.clip(base_preds, 1e-6, 1 - 1e-6))
+            comp_ll = log_loss(labels, np.clip(comp_preds, 1e-6, 1 - 1e-6))
+            print(f"Base Model LogLoss:  {base_ll:.4f}")
+            print(f"Compensated LogLoss: {comp_ll:.4f}")
+        except Exception as e:
+            print(f"LogLoss calculation skipped: {e}")
+    
+    # --- 7. Recommendation Output Quality ---
+    print("\n[7] Recommendation Output Quality")
+    print("=" * 70)
+    
+    if pipeline.recommender_output:
+        pop_dist = pipeline.recommender_output.get_popularity_distribution()
+        print(f"Items recommended:    {pop_dist.get('unique_items_recommended', 0)}")
+        print(f"Coverage ratio:         {pop_dist.get('coverage_ratio', 0):.4f}")
+        print(f"Rec entropy:            {pop_dist.get('popularity_entropy', 0):.4f}")
+    
+    if pipeline.output_buffer:
+        output_metrics = pipeline.output_buffer.get_quality_metrics()
+        print(f"Output buffer size:     {output_metrics.get('buffer_size', 0)}")
+        print(f"Avg latency (ms):       {output_metrics.get('avg_latency_ms', 0):.2f}")
+        print(f"Avg confidence:         {output_metrics.get('avg_confidence', 0):.4f}")
+        print(f"Fallback rate:          {output_metrics.get('fallback_rate', 0):.4f}")
+    
+    # Show a sample recommendation
+    if results and results[-1].get('recommendations'):
+        print(f"\nSample recommendation from last batch:")
+        sample = results[-1]['recommendations'][0]
+        print(f"  User {sample['user_id']}: {[r['item_id'] for r in sample['recommendations'][:5]]}")
+    
+    # --- 8. Save artifacts ---
+    print("\n[8] Saving artifacts")
+    print("=" * 70)
+    
+    pipeline.save_model("/app/recommender_model.pt")
+    
+    with open("/app/pipeline_summary.json", "w") as f:
+        json.dump(summary, f, indent=2, default=str)
+    
+    print("Model saved to: /app/recommender_model.pt")
+    print("Summary saved to: /app/pipeline_summary.json")
+    
+    print("\n" + "=" * 70)
+    print("PIPELINE EXECUTION COMPLETE")
+    print("=" * 70)
+    
+    return pipeline, results
+
+
+if __name__ == "__main__":
+    pipeline, results = main()