from __future__ import annotations import time from dataclasses import dataclass from pathlib import Path from typing import Generator, Optional import gradio as gr import joblib import matplotlib.pyplot as plt import numpy as np import pandas as pd import tensorflow as tf from sklearn.preprocessing import StandardScaler @dataclass(frozen=True) class DemoConfig: model_path: Path data_path: Path scaler_path: Path lookback: int = 30 horizon: int = 60 threshold: float = 0.5 refresh_delay: float = 0.08 class DataRepository: def __init__(self, data_path: Path) -> None: self.data_path = data_path self._data: Optional[pd.DataFrame] = None def get_data(self) -> pd.DataFrame: if self._data is None: df = pd.read_csv(self.data_path, parse_dates=["datetime"]) df = df.sort_values(["user_id", "datetime"]).reset_index(drop=True) self._data = df return self._data.copy() class SequenceBuilder: def __init__(self, lookback: int) -> None: self.lookback = lookback def build_window(self, user_frame: pd.DataFrame, feature_cols: list[str], end_index: int) -> np.ndarray: start_index = end_index - self.lookback + 1 window = user_frame.iloc[start_index:end_index + 1][feature_cols].to_numpy(dtype=np.float32) if len(window) != self.lookback: raise ValueError("Недостаточно данных для построения окна наблюдений.") return window class PredictionService: def __init__(self, config: DemoConfig) -> None: self.config = config self._model: Optional[tf.keras.Model] = None self._scaler: Optional[StandardScaler] = None @staticmethod def _make_weighted_bce(pos_weight: float = 1.0): pos_weight_tensor = tf.constant(pos_weight, dtype=tf.float32) def weighted_bce(y_true, y_pred): y_true_cast = tf.cast(y_true, tf.float32) y_pred_clip = tf.clip_by_value(y_pred, 1e-7, 1 - 1e-7) loss_pos = -y_true_cast * tf.math.log(y_pred_clip) * pos_weight_tensor loss_neg = -(1.0 - y_true_cast) * tf.math.log(1.0 - y_pred_clip) return tf.reduce_mean(loss_pos + loss_neg) return weighted_bce def _load_model(self) -> tf.keras.Model: if self._model is None: self._model = tf.keras.models.load_model( self.config.model_path, custom_objects={"weighted_bce": self._make_weighted_bce(1.0)}, compile=False, ) return self._model def _load_scaler(self) -> StandardScaler: if self._scaler is None: if self.config.scaler_path.exists(): self._scaler = joblib.load(self.config.scaler_path) else: self._scaler = StandardScaler() return self._scaler def predict(self, window: np.ndarray) -> np.ndarray: model = self._load_model() scaler = self._load_scaler() if hasattr(scaler, "mean_"): window_2d = window.reshape(-1, window.shape[-1]) scaled_window = scaler.transform(window_2d).reshape(1, window.shape[0], window.shape[1]) else: scaled_window = window.reshape(1, window.shape[0], window.shape[1]) prediction = model.predict(scaled_window, verbose=0)[0] return np.asarray(prediction, dtype=np.float32) class DemoFormatter: @staticmethod def risk_band(score: float) -> str: if score >= 0.8: return "Критический" if score >= 0.6: return "Высокий" if score >= 0.4: return "Средний" return "Низкий" @staticmethod def summary_html(max_risk_so_far: float, mean_risk_so_far: float, current_probability: float, horizon_minutes: int) -> str: level = DemoFormatter.risk_band(max_risk_so_far) color = "#d7263d" if current_probability >= 0.5 else "#16a34a" glow = "rgba(215,38,61,0.22)" if current_probability >= 0.5 else "rgba(22,163,74,0.18)" return f"""
{level} риск отказа
Текущий шаг
{current_probability:.1%}
Пиковая вероятность
{max_risk_so_far:.1%}
Средняя вероятность
{mean_risk_so_far:.1%}
Окно прогноза
{horizon_minutes} мин
""".strip() class FailureForecastDemoApp: def __init__(self, config: DemoConfig) -> None: self.config = config self.repository = DataRepository(config.data_path) self.sequence_builder = SequenceBuilder(config.lookback) self.prediction_service = PredictionService(config) self.df = self.repository.get_data() self.feature_cols = [col for col in self.df.columns if col not in ["user_id", "datetime"]] self.valid_users = self._collect_valid_users() self.theme = gr.themes.Soft(primary_hue="blue", secondary_hue="slate", neutral_hue="slate") self.custom_css = """ .gradio-container { max-width: 1880px !important; margin: 0 auto !important; } .hero-card { background: linear-gradient(135deg, #0f172a 0%, #1e3a8a 45%, #2563eb 100%); border-radius: 24px; padding: 34px; color: white; box-shadow: 0 18px 60px rgba(37,99,235,.22); margin: 0 auto 22px auto; text-align: center; max-width: 1320px; } .hero-card h1 { margin: 0 0 10px 0; font-size: 42px; } .hero-card p { margin: 0; font-size: 20px; opacity: .92; } #controls-wrap, #status-wrap, #plots-wrap, #table-wrap { max-width: 1700px; margin: 0 auto; } #plots-wrap .gr-plot { min-height: 720px !important; } .forecast-table { margin: 0 auto; max-width: 1400px; } """ def _collect_valid_users(self) -> list[int]: counts = self.df.groupby("user_id").size() valid = counts[counts >= self.config.lookback + self.config.horizon].index.tolist() return [int(user_id) for user_id in valid] def get_user_choices(self) -> list[int]: return self.valid_users def get_user_index_bounds(self, user_id: int) -> tuple[int, int, int]: frame = self.df[self.df["user_id"] == user_id].reset_index(drop=True) min_index = self.config.lookback - 1 max_index = len(frame) - self.config.horizon - 1 default_index = max_index return min_index, max_index, default_index def _build_history_plot( self, user_frame: pd.DataFrame, end_index: int, metric_name: str, prediction_df: pd.DataFrame, current_step: int, ): fig, ax1 = plt.subplots(figsize=(18, 8.5)) history = user_frame.iloc[max(0, end_index - self.config.lookback + 1): end_index + 1] future = prediction_df.iloc[:current_step].copy() ax1.plot( history["datetime"], history[metric_name], linewidth=3.0, label=f"История: {metric_name}", ) ax1.set_xlabel("Время", fontsize=13) ax1.set_ylabel(metric_name, fontsize=13) ax1.grid(alpha=0.25) ax2 = ax1.twinx() ax2.plot( future["datetime"], future["failure_probability"], linestyle="--", linewidth=2.8, label="Вероятность отказа", ) if not future.empty: current_point = future.iloc[-1] ax2.scatter( [current_point["datetime"]], [current_point["failure_probability"]], s=120, label="Текущий шаг", ) ax2.set_ylabel("P(failure)", fontsize=13) ax2.set_ylim(0, 1.05) lines1, labels1 = ax1.get_legend_handles_labels() lines2, labels2 = ax2.get_legend_handles_labels() ax1.legend(lines1 + lines2, labels1 + labels2, loc="upper left", fontsize=12) title = f"История и текущий прогноз по метрике '{metric_name}'" if current_step > 0: current_point = future.iloc[-1] title += f" • +{int(current_point['step'])} мин • риск {current_point['failure_probability']:.1%}" ax1.set_title(title, fontsize=16) ax1.tick_params(axis="both", labelsize=11) ax2.tick_params(axis="y", labelsize=11) fig.autofmt_xdate() plt.tight_layout() return fig def _build_probability_plot(self, prediction_df: pd.DataFrame, current_step: int): fig, ax = plt.subplots(figsize=(18, 7.8)) shown = prediction_df.iloc[:current_step].copy() remaining = prediction_df.iloc[current_step:].copy() if not shown.empty: ax.plot( shown["step"], shown["failure_probability"], linewidth=3.0, label="Раскрытый участок", ) peak_idx = shown["failure_probability"].idxmax() peak_row = shown.loc[peak_idx] ax.scatter( [peak_row["step"]], [peak_row["failure_probability"]], s=120, label="Пик на текущий момент", ) current_row = shown.iloc[-1] ax.scatter( [current_row["step"]], [current_row["failure_probability"]], s=120, label="Текущий шаг", ) if not remaining.empty: ax.plot( remaining["step"], remaining["failure_probability"], linestyle=":", linewidth=2.3, alpha=0.45, label="Оставшийся горизонт", ) ax.set_ylim(0, 1.05) ax.set_xlabel("Шаг прогноза, мин", fontsize=13) ax.set_ylabel("Вероятность отказа", fontsize=13) ax.set_title("Профиль риска на горизонте 60 минут", fontsize=16) ax.grid(alpha=0.25) ax.legend(loc="upper left", fontsize=12) ax.tick_params(axis="both", labelsize=11) plt.tight_layout() return fig def _build_prediction_table(self, forecast_start: pd.Timestamp, prediction: np.ndarray, current_step: int) -> pd.DataFrame: current_index = current_step - 1 current_probability = float(prediction[current_index]) current_datetime = (forecast_start + pd.Timedelta(minutes=current_index)).strftime("%Y-%m-%d %H:%M:%S") current_state = "ALERT" if current_probability >= self.config.threshold else "OK" return pd.DataFrame( [ { "step": current_step, "datetime": current_datetime, "failure_probability": f"{current_probability:.2%}", "state": current_state, "progress": "← текущий", } ] ) def simulate(self, user_id: int, metric_name: str, end_index: int) -> Generator: user_frame = self.df[self.df["user_id"] == user_id].reset_index(drop=True) min_index, max_index, _ = self.get_user_index_bounds(user_id) end_index = int(np.clip(end_index, min_index, max_index)) history_end_dt = user_frame.loc[end_index, "datetime"] forecast_start_dt = history_end_dt + pd.Timedelta(minutes=1) window = self.sequence_builder.build_window(user_frame, self.feature_cols, end_index) prediction = self.prediction_service.predict(window) prediction_df = pd.DataFrame( { "step": np.arange(1, len(prediction) + 1), "datetime": pd.date_range(forecast_start_dt, periods=len(prediction), freq="min"), "failure_probability": prediction, } ) for step in range(1, self.config.horizon + 1): shown_prediction = prediction[:step] current_probability = float(shown_prediction[-1]) max_risk_so_far = float(shown_prediction.max()) mean_risk_so_far = float(shown_prediction.mean()) summary_html = DemoFormatter.summary_html( max_risk_so_far=max_risk_so_far, mean_risk_so_far=mean_risk_so_far, current_probability=current_probability, horizon_minutes=self.config.horizon, ) history_plot = self._build_history_plot(user_frame, end_index, metric_name, prediction_df, step) risk_plot = self._build_probability_plot(prediction_df, step) table = self._build_prediction_table(forecast_start_dt, prediction, step) yield summary_html, history_plot, risk_plot, table time.sleep(self.config.refresh_delay) @staticmethod def refresh_slider(user_id: int, app: "FailureForecastDemoApp"): minimum, maximum, value = app.get_user_index_bounds(user_id) return gr.Slider(minimum=minimum, maximum=maximum, value=value, step=1) def build(self) -> gr.Blocks: metric_choices = [ metric for metric in ["cpus", "memory", "assigned_memory", "page_cache_memory", "failed"] if metric in self.df.columns ] default_user = self.valid_users[0] slider_min, slider_max, slider_value = self.get_user_index_bounds(default_user) with gr.Blocks(title="Демонстрация прогнозирования отказов") as demo: gr.HTML( """

Предвестник отказов

Интерактивная демонстрация многократного прогноза отказов на горизонте 60 минут. Нажми кнопку запуска — и интерфейс поэтапно покажет, как меняется риск во времени.

""" ) with gr.Column(elem_id="controls-wrap"): with gr.Row(equal_height=True): user_input = gr.Dropdown( choices=self.get_user_choices(), value=default_user, label="Объект мониторинга (user_id)", info="Доступны только ряды, где хватает истории и горизонта прогноза.", scale=1, ) metric_input = gr.Dropdown( choices=metric_choices, value=metric_choices[0], label="Ключевая метрика для визуализации", scale=1, ) end_index_input = gr.Slider( minimum=slider_min, maximum=slider_max, value=slider_value, step=1, label="Позиция конца окна истории", info="Чем правее, тем ближе к последним данным выбранного объекта.", ) launch_button = gr.Button("Запустить симуляцию", variant="primary", size="lg") with gr.Column(elem_id="status-wrap"): summary_output = gr.HTML(label="Сводка") with gr.Column(elem_id="plots-wrap"): with gr.Row(equal_height=True): history_plot_output = gr.Plot(label="История + текущий прогноз", scale=1) risk_plot_output = gr.Plot(label="Кривая риска", scale=1) with gr.Column(elem_id="table-wrap"): table_output = gr.Dataframe( headers=["step", "datetime", "failure_probability", "state", "progress"], datatype=["number", "str", "str", "str", "str"], interactive=False, label="Поточечный прогноз", wrap=True, row_count=1, column_count=5, max_height=120, elem_classes="forecast-table", ) user_input.change( fn=lambda selected_user: self.refresh_slider(selected_user, self), inputs=user_input, outputs=end_index_input, ) launch_button.click( fn=self.simulate, inputs=[user_input, metric_input, end_index_input], outputs=[summary_output, history_plot_output, risk_plot_output, table_output], ) return demo def resolve_project_paths() -> DemoConfig: base_dir = Path(__file__).resolve().parent model_path = (base_dir / "models/gru_direct_multistep/final_model.keras").resolve() data_path = (base_dir / "data/processed/data_processed.csv").resolve() scaler_path = (base_dir / "models/gru_direct_multistep/seq_scaler.pkl").resolve() return DemoConfig( model_path=model_path, data_path=data_path, scaler_path=scaler_path, ) def validate_paths(config: DemoConfig) -> None: missing_paths = [str(path) for path in [config.model_path, config.data_path] if not path.exists()] if missing_paths: joined = "\n - ".join(missing_paths) raise FileNotFoundError(f"Не найдены обязательные файлы:\n - {joined}") if __name__ == "__main__": tf.keras.utils.set_random_seed(42) config = resolve_project_paths() validate_paths(config) app = FailureForecastDemoApp(config) demo = app.build() demo.queue().launch( server_name="0.0.0.0", server_port=7860, theme=app.theme, css=app.custom_css, )