monitoring.py

"""CUSUM monitoring for simulated per-parent execution costs."""

from typing import Iterable, Optional, Tuple

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd


def widened_impact_batch(
    parents: pd.DataFrame,
    fills: pd.DataFrame,
    impact_multiplier: float,
    tick_size: float,
) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """Create a controlled second fill batch by widening only temporary impact."""

    required_fill_columns = {
        "data_label",
        "parent_id",
        "side",
        "size",
        "touch_price",
        "impact_ticks",
        "fill_price",
    }
    required_parent_columns = {
        "data_label",
        "parent_id",
        "executed_size",
        "average_fill_price",
    }
    missing_fills = required_fill_columns.difference(fills.columns)
    missing_parents = required_parent_columns.difference(parents.columns)
    if missing_fills or missing_parents:
        raise ValueError(
            f"Missing widened-batch columns: fills={sorted(missing_fills)}, "
            f"parents={sorted(missing_parents)}"
        )
    if set(fills["data_label"].unique()) != {"SIMULATED"} or set(
        parents["data_label"].unique()
    ) != {"SIMULATED"}:
        raise ValueError("Widened-impact inputs must be labelled SIMULATED")
    if impact_multiplier <= 1 or tick_size <= 0:
        raise ValueError("Impact multiplier must exceed one and tick size must be positive")

    stressed_fills = fills.copy()
    stressed_fills["impact_ticks"] = stressed_fills["impact_ticks"] * impact_multiplier
    stressed_fills["fill_price"] = (
        stressed_fills["touch_price"]
        + stressed_fills["side"] * stressed_fills["impact_ticks"] * tick_size
    )
    stressed_parents = parents.copy()
    average_prices = (
        stressed_fills.assign(notional=stressed_fills["fill_price"] * stressed_fills["size"])
        .groupby("parent_id")
        .agg(notional=("notional", "sum"), executed_size=("size", "sum"))
    )
    average_prices["average_fill_price"] = (
        average_prices["notional"] / average_prices["executed_size"]
    )
    stressed_parents = stressed_parents.drop(columns=["average_fill_price"]).merge(
        average_prices[["average_fill_price"]],
        left_on="parent_id",
        right_index=True,
        how="left",
        validate="one_to_one",
    )
    return stressed_parents, stressed_fills


def one_sided_cusum(
    costs_bps: Iterable[float],
    baseline_mean_bps: float,
    slack_bps: float,
    threshold_bps: float,
) -> pd.DataFrame:
    """Calculate a one-sided CUSUM that detects increases in execution cost."""

    costs = np.asarray(tuple(costs_bps), dtype="float64")
    if costs.size == 0 or not np.isfinite(costs).all():
        raise ValueError("CUSUM costs must be non-empty and finite")
    if not np.isfinite(baseline_mean_bps):
        raise ValueError("Baseline mean must be finite")
    if slack_bps < 0 or threshold_bps <= 0:
        raise ValueError("CUSUM slack must be non-negative and threshold must be positive")

    statistic = 0.0
    rows = []
    previously_above = False
    for observation, cost in enumerate(costs, start=1):
        statistic = max(0.0, statistic + cost - baseline_mean_bps - slack_bps)
        above_threshold = statistic > threshold_bps
        rows.append(
            {
                "observation": observation,
                "cost_bps": cost,
                "baseline_mean_bps": baseline_mean_bps,
                "slack_bps": slack_bps,
                "threshold_bps": threshold_bps,
                "cusum_bps": statistic,
                "alarm": above_threshold,
                "new_alarm": above_threshold and not previously_above,
            }
        )
        previously_above = above_threshold
    return pd.DataFrame(rows)


def controlled_deterioration_demo(
    baseline_costs_bps: Iterable[float],
    stressed_costs_bps: Iterable[float],
    slack_bps: float,
    threshold_bps: float,
) -> pd.DataFrame:
    """Run baseline then stressed simulated costs through one continuous CUSUM trace."""

    baseline = np.asarray(tuple(baseline_costs_bps), dtype="float64")
    stressed = np.asarray(tuple(stressed_costs_bps), dtype="float64")
    if baseline.size == 0 or stressed.size == 0:
        raise ValueError("Controlled demo requires non-empty baseline and stressed batches")
    baseline_mean = float(baseline.mean())
    trace = one_sided_cusum(
        np.concatenate((baseline, stressed)),
        baseline_mean,
        slack_bps,
        threshold_bps,
    )
    trace.insert(0, "data_label", "SIMULATED")
    trace.insert(1, "demo_label", "CONTROLLED_DETERIORATION_DEMO")
    trace.insert(
        2,
        "batch",
        ["BASELINE"] * baseline.size + ["WIDENED_IMPACT"] * stressed.size,
    )
    trace["batch_observation"] = list(range(1, baseline.size + 1)) + list(
        range(1, stressed.size + 1)
    )
    return trace


def plot_cusum_demo(
    trace: pd.DataFrame,
    ax: Optional[plt.Axes] = None,
) -> Tuple[plt.Figure, plt.Axes]:
    """Render a clearly labelled controlled CUSUM deterioration demonstration."""

    required = {
        "data_label",
        "demo_label",
        "batch",
        "observation",
        "cusum_bps",
        "threshold_bps",
        "new_alarm",
    }
    missing = required.difference(trace.columns)
    if missing:
        raise ValueError(f"CUSUM trace missing columns: {sorted(missing)}")
    if set(trace["data_label"].unique()) != {"SIMULATED"}:
        raise ValueError("CUSUM demo must be labelled SIMULATED")
    if set(trace["demo_label"].unique()) != {"CONTROLLED_DETERIORATION_DEMO"}:
        raise ValueError("CUSUM plot requires the controlled-demo label")

    if ax is None:
        figure, ax = plt.subplots(figsize=(7, 4))
    else:
        figure = ax.figure
    ax.plot(trace["observation"], trace["cusum_bps"], marker="o", label="One-sided CUSUM")
    ax.axhline(
        trace["threshold_bps"].iloc[0],
        color="red",
        linestyle="--",
        label="Alarm threshold",
    )
    stressed = trace.index[trace["batch"].eq("WIDENED_IMPACT")]
    if len(stressed):
        first_stressed_observation = trace.loc[stressed[0], "observation"]
        ax.axvline(
            first_stressed_observation - 0.5,
            color="grey",
            linestyle=":",
            label="Injected deterioration",
        )
    alarms = trace.loc[trace["new_alarm"]]
    if not alarms.empty:
        ax.scatter(
            alarms["observation"],
            alarms["cusum_bps"],
            color="red",
            zorder=3,
            label="Alarm crossing",
        )
    ax.set(
        title="CONTROLLED DEMO — SIMULATED execution-cost CUSUM",
        xlabel="Parent observation",
        ylabel="CUSUM (bps)",
    )
    ax.legend()
    return figure, ax