predictive_engine module

Predictive Health Engine (Observability v7)

  • Maintains sliding window for metrics: error_rate_percent, latency_seconds, memory_usage_percent

  • Exponential smoothing (EW-Regression) for trend estimation, fallback to linear regression

  • Adaptive Feedback Loop: compare predictions vs actuals and adjust smoothing reactiveness

  • If forecast crosses adaptive threshold within horizon (default 15 minutes), logs a predictive incident

  • Predictive incidents are appended to data/predictions_log.json (JSONL)

  • Optionally triggers preemptive actions (cache clear, GC, controlled restart) and logs PREDICTIVE_ACTION_TRIGGERED

  • Auto-cleanup for predictions older than 24h

This module is intentionally best-effort and fail-open. It should never raise.

predictive_engine.emit_event(event, severity='info', **fields)[מקור]
פרמטרים:
class predictive_engine.Trend(metric, slope_per_minute, intercept, current_value, threshold, predicted_cross_ts)[מקור]

Bases: object

פרמטרים:
metric: str
slope_per_minute: float
intercept: float
current_value: float
threshold: float
predicted_cross_ts: Optional[float]
__init__(metric, slope_per_minute, intercept, current_value, threshold, predicted_cross_ts)
פרמטרים:
Return type:

None

predictive_engine.note_observation(*, error_rate_percent=None, latency_seconds=None, memory_usage_percent=None, ts=None)[מקור]

Record a single observation into sliding windows.

Values default to current snapshot from alert_manager/psutil.

Return type:

None

פרמטרים:

  • error_rate_percent (float | None)

  • latency_seconds (float | None)

  • memory_usage_percent (float | None)

  • ts (float | None)

predictive_engine.evaluate_predictions(now_ts=None, horizon_seconds=None)[מקור]

Compute predictions for all metrics. Returns list of Trend objects.

Return type:

List[Trend]

פרמטרים:

  • now_ts (float | None)

  • horizon_seconds (int | None)

predictive_engine.reset_state_for_tests()[מקור]

Clear in-memory state for unit tests (not for production use).

Return type:

None

predictive_engine.maybe_recompute_and_preempt(now_ts=None)[מקור]

Throttled evaluation. If any metric predicted to breach within horizon, trigger preemptive actions.

Returns the trends (for observability/tests).

Return type:

List[Trend]

פרמטרים:

now_ts (float | None)

predictive_engine.get_recent_predictions(limit=5)[מקור]
Return type:

List[Dict[str, Any]]

פרמטרים:

limit (int)

predictive_engine.get_trend_snapshot()[מקור]

Return a snapshot with direction and slope for UI (/predict).

Keys are metrics; values include direction emoji and slope.

Return type:

Dict[str, Dict[str, Any]]