Signal Alert System

pip install pandas numpy matplotlib seaborn scipy

import logging
import time
import random
from collections import deque
from typing import Dict, Any, List, Union

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from scipy.stats import norm

def configure_logging(level: int = logging.INFO) -> None:
    """
    Configures the basic logging setup for the system.

    Parameters
    ----------
    level : int, optional
        The logging level (e.g., logging.INFO, logging.DEBUG), defaults to logging.INFO.

    Returns
    -------
    None
    """
    logging.basicConfig(level=level,
                        format='%(asctime)s - %(levelname)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')
    logging.info(f"Logging configured at level: {logging.getLevelName(level)}")

def create_system_state(window_size: int, threshold: float, initial_value: float = 100.0) -> Dict[str, Any]:
    """
    Initializes the system's state dictionary.

    Parameters
    ----------
    window_size : int
        The number of data points to keep in the rolling window for calculations.
    threshold : float
        The value that, if exceeded, triggers an alert.
    initial_value : float, optional
        The starting value for the simulated data, defaults to 100.0.

    Returns
    -------
    dict
        An initialized dictionary representing the system's state.
    """
    state = {
        "data_history": deque([initial_value] * window_size, maxlen=window_size),
        "alerts": [],
        "window_size": window_size,
        "threshold": threshold,
        "current_timestamp": pd.Timestamp.now(),
        "last_alert_time": None
    }
    logging.info(f"System state initialized with window_size={window_size}, threshold={threshold}")
    return state

def simulate_data_point(state: Dict[str, Any], random_walk_factor: float = 0.01, noise_std: float = 0.5) -> float:
    """
    Simulates a new data point based on the last observed value.

    Parameters
    ----------
    state : dict
        Current system state, containing 'data_history' deque.
    random_walk_factor : float, optional
        Factor influencing the random walk, defaults to 0.01.
    noise_std : float, optional
        Standard deviation of the random noise, defaults to 0.5.

    Returns
    -------
    float
        The newly simulated data point.

    Examples
    --------
    >>> state = create_system_state(window_size=3, threshold=105.0, initial_value=100.0)
    >>> new_point = simulate_data_point(state)
    >>> isinstance(new_point, float)
    True
    """
    last_value = state["data_history"][-1]
    change = last_value * random_walk_factor * np.random.randn() + np.random.normal(0, noise_std)
    new_value = last_value + change
    logging.debug(f"Simulated new data point: {new_value:.2f}")
    return new_value

def calculate_moving_average(state: Dict[str, Any]) -> float:
    """
    Calculates the simple moving average (SMA) of the data in the history.

    Parameters
    ----------
    state : dict
        Current system state, containing 'data_history' deque.

    Returns
    -------
    float
        The calculated simple moving average.

    Examples
    --------
    >>> state = create_system_state(window_size=3, threshold=105.0)
    >>> state["data_history"] = deque([90, 100, 110], maxlen=3)
    >>> calculate_moving_average(state)
    100.0
    """
    if not state["data_history"]:
        logging.warning("Cannot calculate moving average: data_history is empty.")
        return 0.0
    sma = sum(state["data_history"]) / len(state["data_history"])
    logging.debug(f"Calculated moving average: {sma:.2f}")
    return sma

def detect_signal(state: Dict[str, Any], current_moving_average: float) -> bool:
    """
    Detects if a signal (threshold breach) has occurred.

    Parameters
    ----------
    state : dict
        Current system state, containing 'threshold'.
    current_moving_average : float
        The most recently calculated moving average.

    Returns
    -------
    bool
        True if a signal is detected, False otherwise.

    Examples
    --------
    >>> state = create_system_state(window_size=3, threshold=105.0)
    >>> detect_signal(state, 106.0)
    True
    >>> detect_signal(state, 104.0)
    False
    """
    signal_fired = current_moving_average > state["threshold"]
    if signal_fired:
        logging.info(f"Signal detected! Moving average ({current_moving_average:.2f}) > Threshold ({state['threshold']:.2f})")
    else:
        logging.debug(f"No signal. Moving average ({current_moving_average:.2f}) <= Threshold ({state['threshold']:.2f})")
    return signal_fired

def generate_alert(state: Dict[str, Any], timestamp: pd.Timestamp, value: float) -> Dict[str, Any]:
    """
    Generates and records an alert in the system state.

    Parameters
    ----------
    state : dict
        Current system state, to which the alert will be added.
    timestamp : pd.Timestamp
        The timestamp when the alert was generated.
    value : float
        The value that triggered the alert.

    Returns
    -------
    dict
        The updated system state with the new alert.

    Examples
    --------
    >>> state = create_system_state(window_size=3, threshold=105.0)
    >>> current_time = pd.Timestamp('2023-01-01 10:00:00')
    >>> updated_state = generate_alert(state, current_time, 106.5)
    >>> len(updated_state['alerts']) == 1
    True
    >>> updated_state['alerts'][0]['value'] == 106.5
    True
    """
    alert_message = f"ALERT! Threshold ({state['threshold']:.2f}) breached at {timestamp} by value {value:.2f}"
    alert_entry = {
        "timestamp": timestamp,
        "value": value,
        "message": alert_message
    }
    state["alerts"].append(alert_entry)
    state["last_alert_time"] = timestamp
    logging.critical(alert_message) # Use critical for actual alerts
    return state

def apply_exponential_backoff(retries: int, base_delay: float = 0.1, max_delay: float = 10.0) -> float:
    """
    Calculates an exponential backoff delay with random jitter.

    Parameters
    ----------
    retries : int
        The current number of retry attempts.
    base_delay : float, optional
        The base delay in seconds, defaults to 0.1.
    max_delay : float, optional
        The maximum allowed delay in seconds, defaults to 10.0.

    Returns
    -------
    float
        The calculated delay time in seconds.

    Raises
    ------
    ValueError
        If retries is negative.

    Examples
    --------
    >>> delay = apply_exponential_backoff(1)
    >>> 0.1 <= delay <= 0.2 # Roughly, due to jitter
    True
    """
    if retries < 0:
        raise ValueError("Retries cannot be negative.")

    delay = min(max_delay, base_delay * (2 ** retries))
    jitter = random.uniform(0, delay * 0.1) # Add 0-10% random jitter
    final_delay = delay + jitter
    logging.debug(f"Calculated backoff delay for {retries} retries: {final_delay:.2f}s")
    return final_delay

def process_data_point(state: Dict[str, Any], new_data_point: float) -> Dict[str, Any]:
    """
    Processes a single new data point, updates state, calculates SMA, detects signal, and generates alerts.

    Parameters
    ----------
    state : dict
        Current system state, which will be updated.
    new_data_point : float
        The latest data point to be processed.

    Returns
    -------
    dict
        The updated system state after processing the new data point.

    Examples
    --------
    >>> state = create_system_state(window_size=3, threshold=105.0, initial_value=100.0)
    >>> state['data_history'] = deque([100, 101, 102], maxlen=3)
    >>> updated_state = process_data_point(state, 103.0) # MA = 102, no alert
    >>> updated_state['data_history'][-1] == 103.0
    True
    >>> updated_state = process_data_point(state, 110.0) # MA = (101+102+110)/3 = 107.67, alert
    >>> len(updated_state['alerts']) > 0
    True
    """
    state["data_history"].append(new_data_point)
    state["current_timestamp"] += pd.Timedelta(minutes=1) # Advance time

    current_ma = calculate_moving_average(state)

    if detect_signal(state, current_ma):
        # Optional: Implement a cooldown period for alerts
        if state["last_alert_time"] is None or \
           (state["current_timestamp"] - state["last_alert_time"]) > pd.Timedelta(minutes=5):
            state = generate_alert(state, state["current_timestamp"], current_ma)
        else:
            logging.info("Alert cooldown period active, skipping alert for now.")

    logging.debug(f"Processed data point: {new_data_point:.2f}, MA: {current_ma:.2f}")
    return state

# 1. Configure logging
configure_logging(level=logging.INFO)

# 2. Initialize system state
WINDOW_SIZE = 10
THRESHOLD = 50.0
INITIAL_VALUE = 100.0

system_state = create_system_state(WINDOW_SIZE, THRESHOLD, INITIAL_VALUE)

# Simulation parameters
NUM_DATA_POINTS = 200
SIMULATION_HISTORY = [] # To store all data points and MA for plotting
ALERTS_DF = pd.DataFrame(columns=['timestamp', 'value', 'message'])

logging.info("Starting data simulation and alert system processing...")

for i in range(NUM_DATA_POINTS):
    # Simulate a new data point
    # Introduce a 'spike' around the middle of the simulation to trigger alerts
    if 80 < i < 120:
        new_point = simulate_data_point(system_state, random_walk_factor=0.05, noise_std=1.5)
    else:
        new_point = simulate_data_point(system_state, random_walk_factor=0.01, noise_std=0.5)

    # Process the data point
    system_state = process_data_point(system_state, new_point)

    # Store data for visualization
    current_ma = calculate_moving_average(system_state) # Recalculate for current point
    SIMULATION_HISTORY.append({
        'timestamp': system_state["current_timestamp"],
        'value': new_point,
        'moving_average': current_ma if len(system_state["data_history"]) == WINDOW_SIZE else np.nan,
        'is_alert': True if system_state["last_alert_time"] == system_state["current_timestamp"] else False
    })

    # Capture alerts into a DataFrame
    if system_state["alerts"] and system_state["alerts"][-1]["timestamp"] == system_state["current_timestamp"]:
        latest_alert = system_state["alerts"][-1]
        ALERTS_DF = pd.concat([ALERTS_DF, pd.DataFrame([latest_alert])], ignore_index=True)

logging.info("Simulation complete.")

# Convert simulation history to DataFrame for easier plotting
df_sim = pd.DataFrame(SIMULATION_HISTORY)
df_sim['timestamp'] = pd.to_datetime(df_sim['timestamp']) # Ensure datetime type

# Display summary of alerts
print("\n--- Alerts Generated ---")
display(ALERTS_DF)

### Visualization: Data Trend with Moving Average and Alerts

plt.figure(figsize=(15, 7))
sns.lineplot(x='timestamp', y='value', data=df_sim, label='Raw Data', color='gray', alpha=0.7)
sns.lineplot(x='timestamp', y='moving_average', data=df_sim, label=f'SMA ({WINDOW_SIZE} periods)', color='blue', linewidth=2)

# Plot the threshold line
plt.axhline(y=THRESHOLD, color='red', linestyle='--', label='Alert Threshold')

# Plot alert points
alert_points = df_sim[df_sim['is_alert'] == True]
plt.scatter(alert_points['timestamp'], alert_points['moving_average'], color='red', s=100, marker='X', zorder=5, label='Alert Fired')

plt.title('Simulated Time Series Data, Moving Average, and Alerts')
plt.xlabel('Time')
plt.ylabel('Value')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()