Task Scheduling Logic

Documentation of scheduled tasks, background jobs, and automated processes in Hawkeye.

Overview

Hawkeye uses a task scheduling system to handle periodic maintenance, model updates, data processing, and system health checks.

Scheduled Tasks

1. Model Performance Monitoring

Schedule: Daily at 2:00 AM UTC

def scheduled_model_monitoring():
    """
    Daily task to monitor model performance
    """
    # 1. Collect recent predictions
    recent_predictions = get_recent_predictions(days=7)
    
    # 2. Calculate metrics
    metrics = {
        'accuracy': calculate_accuracy(recent_predictions),
        'precision': calculate_precision(recent_predictions),
        'recall': calculate_recall(recent_predictions),
        'f1_score': calculate_f1_score(recent_predictions),
        'confidence_distribution': get_confidence_distribution(recent_predictions)
    }
    
    # 3. Check for degradation
    if metrics['accuracy'] < THRESHOLD_ACCURACY:
        alert_model_degradation(metrics)
    
    # 4. Store metrics
    store_performance_metrics(metrics)
    
    # 5. Generate report
    generate_performance_report(metrics)

2. Data Cleanup Task

Schedule: Weekly on Sunday at 3:00 AM UTC

def scheduled_data_cleanup():
    """
    Weekly cleanup of old data
    """
    # 1. Clean old predictions (older than 90 days)
    deleted_predictions = cleanup_old_predictions(days=90)
    
    # 2. Archive old incidents
    archived_incidents = archive_old_incidents(days=180)
    
    # 3. Clean temporary files
    cleaned_files = cleanup_temporary_files()
    
    # 4. Optimize database
    optimize_database()
    
    # 5. Log cleanup results
    log_cleanup_results({
        'predictions_deleted': deleted_predictions,
        'incidents_archived': archived_incidents,
        'files_cleaned': cleaned_files
    })

3. Model Retraining Check

Schedule: Daily at 1:00 AM UTC

def scheduled_retraining_check():
    """
    Daily check if model needs retraining
    """
    # 1. Check model performance
    performance = get_recent_performance(days=30)
    
    # 2. Check data drift
    drift_score = detect_data_drift()
    
    # 3. Check time since last training
    days_since_training = get_days_since_training()
    
    # 4. Decision logic
    should_retrain = False
    reason = None
    
    if performance['accuracy'] < 0.85:
        should_retrain = True
        reason = "Low accuracy detected"
    elif drift_score > 0.1:
        should_retrain = True
        reason = "Significant data drift"
    elif days_since_training > 90:
        should_retrain = True
        reason = "Model age threshold"
    
    # 5. Trigger retraining if needed
    if should_retrain:
        schedule_model_retraining(reason)
        notify_administrators(f"Model retraining scheduled: {reason}")

4. Statistics Update Task

Schedule: Hourly

def scheduled_statistics_update():
    """
    Hourly update of system statistics
    """
    # 1. Update prediction counts
    update_prediction_counts()
    
    # 2. Update accuracy metrics
    update_accuracy_metrics()
    
    # 3. Update feature statistics
    update_feature_statistics()
    
    # 4. Update system health metrics
    update_health_metrics()
    
    # 5. Cache updated statistics
    cache_statistics()

5. Health Check Task

Schedule: Every 5 minutes

def scheduled_health_check():
    """
    Periodic health check
    """
    health_status = {
        'model_loaded': check_model_loaded(),
        'database_connected': check_database_connection(),
        'api_responding': check_api_health(),
        'disk_space': check_disk_space(),
        'memory_usage': check_memory_usage()
    }
    
    # Update health status
    update_health_status(health_status)
    
    # Alert if unhealthy
    if not all(health_status.values()):
        alert_system_unhealthy(health_status)

Background Jobs

1. Async Prediction Processing

from celery import Celery

app = Celery('hawkeye')

@app.task
def async_prediction_task(incident_data):
    """
    Background task for prediction processing
    """
    try:
        # Process prediction
        result = process_prediction(incident_data)
        
        # Store result
        store_prediction_result(result)
        
        # Notify
        send_notification(result)
        
        return result
    except Exception as e:
        log_error(e)
        raise

2. Batch Processing Job

@app.task
def batch_processing_job(incident_ids):
    """
    Background job for batch processing
    """
    incidents = fetch_incidents(incident_ids)
    results = []
    
    for incident in incidents:
        try:
            result = process_prediction(incident)
            results.append(result)
        except Exception as e:
            log_error(f"Failed to process {incident.id}: {e}")
    
    return results

3. Model Retraining Job

@app.task
def model_retraining_job():
    """
    Background job for model retraining
    """
    # 1. Collect training data
    training_data = collect_training_data()
    
    # 2. Prepare data
    X, y = prepare_training_data(training_data)
    
    # 3. Train model
    new_model = train_model(X, y)
    
    # 4. Validate model
    validation_score = validate_model(new_model)
    
    # 5. If better, deploy
    if validation_score > get_current_model_score():
        deploy_model(new_model)
        notify_model_updated(validation_score)
    else:
        log_info("New model not better, keeping current model")

Cron Schedule Configuration

Schedule Definitions

# Cron schedule definitions
SCHEDULES = {
    'model_monitoring': {
        'schedule': '0 2 * * *',  # Daily at 2 AM
        'task': scheduled_model_monitoring
    },
    'data_cleanup': {
        'schedule': '0 3 * * 0',  # Weekly on Sunday at 3 AM
        'task': scheduled_data_cleanup
    },
    'retraining_check': {
        'schedule': '0 1 * * *',  # Daily at 1 AM
        'task': scheduled_retraining_check
    },
    'statistics_update': {
        'schedule': '0 * * * *',  # Hourly
        'task': scheduled_statistics_update
    },
    'health_check': {
        'schedule': '*/5 * * * *',  # Every 5 minutes
        'task': scheduled_health_check
    }
}

Task Queue Management

Task Priority System

PRIORITY_LEVELS = {
    'CRITICAL': 1,  # Health checks, alerts
    'HIGH': 2,      # Model retraining, performance monitoring
    'MEDIUM': 3,    # Statistics updates, cleanup
    'LOW': 4        # Reporting, archiving
}

def schedule_task(task, priority='MEDIUM'):
    """
    Schedule task with priority
    """
    task_queue.put({
        'task': task,
        'priority': PRIORITY_LEVELS[priority],
        'timestamp': datetime.now()
    })

Task Retry Logic

def retry_task(task, max_retries=3):
    """
    Retry logic for failed tasks
    """
    for attempt in range(max_retries):
        try:
            return task.execute()
        except Exception as e:
            if attempt == max_retries - 1:
                log_error(f"Task failed after {max_retries} attempts: {e}")
                raise
            else:
                wait_time = 2 ** attempt  # Exponential backoff
                time.sleep(wait_time)
                log_warning(f"Task failed, retrying in {wait_time}s: {e}")

Monitoring Scheduled Tasks

Task Execution Monitoring

def monitor_task_execution():
    """
    Monitor scheduled task execution
    """
    tasks = get_recent_tasks(hours=24)
    
    metrics = {
        'total_tasks': len(tasks),
        'successful': sum(1 for t in tasks if t.status == 'success'),
        'failed': sum(1 for t in tasks if t.status == 'failed'),
        'average_duration': calculate_average_duration(tasks),
        'failed_tasks': [t for t in tasks if t.status == 'failed']
    }
    
    # Alert if failure rate is high
    failure_rate = metrics['failed'] / metrics['total_tasks']
    if failure_rate > 0.1:
        alert_high_task_failure_rate(metrics)
    
    return metrics

Task Dependencies

Dependent Task Scheduling

def schedule_dependent_tasks():
    """
    Schedule tasks with dependencies
    """
    # Task 1: Data collection (must complete first)
    data_task = schedule_task(collect_training_data, priority='HIGH')
    
    # Task 2: Data processing (depends on Task 1)
    process_task = schedule_task(
        process_training_data,
        depends_on=data_task,
        priority='HIGH'
    )
    
    # Task 3: Model training (depends on Task 2)
    train_task = schedule_task(
        train_model,
        depends_on=process_task,
        priority='HIGH'
    )
    
    # Task 4: Model validation (depends on Task 3)
    validate_task = schedule_task(
        validate_model,
        depends_on=train_task,
        priority='HIGH'
    )

Next Steps

• Routine Flow - System workflows
• AI Logic - AI processing logic
• DB Schema - Database structure