Add artifact.scatter_plot/src/artifact_scatter_plot/core.py
This commit is contained in:
parent
7de2ff3472
commit
30e75a240d
1 changed files with 72 additions and 0 deletions
72
artifact.scatter_plot/src/artifact_scatter_plot/core.py
Normal file
72
artifact.scatter_plot/src/artifact_scatter_plot/core.py
Normal file
|
|
@ -0,0 +1,72 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from typing import List
|
||||||
|
|
||||||
|
import matplotlib.pyplot as plt
|
||||||
|
import pandas as pd
|
||||||
|
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class ScatterData:
|
||||||
|
"""Repräsentiert einen einzelnen Messdatensatz für den Scatter-Plot."""
|
||||||
|
|
||||||
|
band_width: float
|
||||||
|
near_expiry_unpinned: float
|
||||||
|
|
||||||
|
def __post_init__(self) -> None:
|
||||||
|
if not isinstance(self.band_width, (float, int)):
|
||||||
|
raise TypeError("band_width muss vom Typ float oder int sein.")
|
||||||
|
if not isinstance(self.near_expiry_unpinned, (float, int)):
|
||||||
|
raise TypeError("near_expiry_unpinned muss vom Typ float oder int sein.")
|
||||||
|
|
||||||
|
self.band_width = float(self.band_width)
|
||||||
|
self.near_expiry_unpinned = float(self.near_expiry_unpinned)
|
||||||
|
|
||||||
|
|
||||||
|
def create_scatter_plot(data: List[ScatterData]) -> None:
|
||||||
|
"""Erstellt und zeigt einen Scatter-Plot aus einer Liste von ScatterData-Objekten.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
data: Liste der Messpunkte mit band_width und near_expiry_unpinned-Werten.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: Falls die Eingabedaten leer oder ungültig sind.
|
||||||
|
"""
|
||||||
|
if not isinstance(data, list):
|
||||||
|
raise TypeError("data muss eine Liste von ScatterData-Objekten sein.")
|
||||||
|
if not data:
|
||||||
|
raise ValueError("Die Eingabedatenliste ist leer.")
|
||||||
|
|
||||||
|
# Validierung und Konvertierung in DataFrame
|
||||||
|
records = []
|
||||||
|
for idx, item in enumerate(data):
|
||||||
|
if not isinstance(item, ScatterData):
|
||||||
|
raise TypeError(f"Eintrag {idx} ist kein ScatterData-Objekt.")
|
||||||
|
records.append({
|
||||||
|
"band_width": item.band_width,
|
||||||
|
"near_expiry_unpinned": item.near_expiry_unpinned,
|
||||||
|
})
|
||||||
|
|
||||||
|
df = pd.DataFrame.from_records(records)
|
||||||
|
|
||||||
|
if df.empty:
|
||||||
|
raise ValueError("Keine gültigen Daten zum Plotten gefunden.")
|
||||||
|
|
||||||
|
logger.debug("Erstelle Scatter-Plot mit %d Punkten", len(df))
|
||||||
|
|
||||||
|
plt.figure(figsize=(8, 6))
|
||||||
|
plt.scatter(df["band_width"], df["near_expiry_unpinned"], alpha=0.7, edgecolors='k')
|
||||||
|
plt.title("Mix Freeze Scatter Analysis")
|
||||||
|
plt.xlabel("Band Width")
|
||||||
|
plt.ylabel("Near-Expiry Unpinned")
|
||||||
|
plt.grid(True)
|
||||||
|
plt.tight_layout()
|
||||||
|
|
||||||
|
plt.show()
|
||||||
|
|
||||||
|
logger.info("Scatter-Plot erfolgreich angezeigt.")
|
||||||
Loading…
Reference in a new issue