Skip to the content.

Complete documentation for predictive analytics and forecasting.

Location: callflow_tracer/predictive_analysis.py (627 lines)


Overview

The Predictive Analysis module predicts future performance issues, capacity limits, and scalability characteristics using historical data and statistical analysis.

Key Components

PerformancePrediction (Dataclass)

Predicts future performance issues.

Attributes:

Risk Levels:

CapacityPrediction (Dataclass)

Predicts capacity limit breaches.

Attributes:

ScalabilityAnalysis (Dataclass)

Analyzes scalability characteristics.

Attributes:

ResourceForecast (Dataclass)

Forecasts resource usage.

Attributes:

Classes

PerformancePredictor

Predicts performance issues using historical data.

Methods:

Confidence Calculation:

CapacityPlanner

Plans for capacity limits.

Methods:

ScalabilityAnalyzer

Analyzes scalability.

Methods:

Complexity Classes:

ResourceForecaster

Forecasts resource usage.

Methods:

Usage Examples

Predict Performance Issues

from callflow_tracer.predictive_analysis import PerformancePredictor

predictor = PerformancePredictor("trace_history.json")
predictions = predictor.predict_performance_issues(current_trace)

for pred in predictions:
    if pred.risk_level == "Critical":
        print(f"CRITICAL: {pred.function_name}")
        print(f"  Current: {pred.current_avg_time:.4f}s")
        print(f"  Predicted: {pred.predicted_time:.4f}s")
        print(f"  Confidence: {pred.confidence:.1%}")

Plan Capacity

from callflow_tracer.predictive_analysis import CapacityPlanner
from datetime import datetime, timedelta

planner = CapacityPlanner()
history = [(datetime.now() - timedelta(days=i), 100 + i*5) for i in range(30)]
prediction = planner.predict_capacity(history, capacity_limit=500)

print(f"Days until limit: {prediction.days_until_limit}")
print(f"Current utilization: {prediction.utilization_percent:.1f}%")
print(f"Recommendations: {prediction.recommendations}")

Analyze Scalability

from callflow_tracer.predictive_analysis import ScalabilityAnalyzer

analyzer = ScalabilityAnalyzer()
load_perf = {100: 0.1, 1000: 1.0, 10000: 100.0}
analysis = analyzer.analyze_scalability("my_func", "my_module", load_perf)

print(f"Complexity: {analysis.complexity_class}")
print(f"Scalability Score: {analysis.scalability_score:.1f}")
print(f"Bottleneck Risk: {analysis.bottleneck_risk}")
print(f"Max Recommended Load: {analysis.max_recommended_load}")

Forecast Resources

from callflow_tracer.predictive_analysis import ResourceForecaster
from datetime import datetime, timedelta

forecaster = ResourceForecaster()
history = [(datetime.now() - timedelta(days=i), 50 + i*2) for i in range(30)]
forecast = forecaster.forecast_resource("Memory", history, days_ahead=30)

print(f"Peak: {forecast.peak_prediction}")
print(f"Trend: {forecast.trend}")
print(f"Days to threshold: {forecast.days_to_threshold}")
print(f"Forecasted usage: {forecast.forecasted_usage}")

CLI Usage

# Generate predictions from trace history
callflow-tracer predict history.json -o predictions.html

# JSON output
callflow-tracer predict history.json --format json

# Custom output file
callflow-tracer predict history.json -o my_predictions.html

Output Format

{
  "performance_predictions": [
    {
      "function_name": "process_data",
      "module": "data_processor",
      "current_avg_time": 0.5,
      "predicted_time": 0.75,
      "confidence": 0.85,
      "risk_level": "High",
      "prediction_basis": "Based on 10 historical measurements",
      "recommendations": [
        "Performance is degrading over time",
        "Consider profiling and optimization"
      ]
    }
  ],
  "summary": {
    "total_predictions": 15,
    "critical_risks": 2,
    "high_risks": 5,
    "average_confidence": 0.82
  },
  "recommendations": [
    "URGENT: 2 functions predicted to have critical performance issues",
    "WARNING: 5 functions showing high performance degradation risk"
  ]
}

Interpretation Guide

Risk Levels

Confidence Levels

Scalability Scores

Complexity Classes