MODULE REFERENCE

results

anomx.detection.results

Shared data contracts flowing between pipeline stages.

` ObservationSet → Model → InferenceOutput → Scorer → AnomalyScores → Calibrator → AnomalyScores → Detector → AnomalyEvents `

AnomalyEvent

AnomalyEvent(start_time: Any = None, end_time: Any = None, peak_score: float = 0.0, severity: float = 0.0, score_level: ScoreLevel = ScoreLevel.SAMPLE, affected_features: list[str] = <factory: list>, affected_entities: list[Any] = <factory: list>, observation_ids: list[Any] = <factory: list>, metadata: dict[str, Any] = <factory: dict>)

One decided anomaly, possibly spanning an interval.

Signature parameters
ParameterTypeDefault
start_timeAnyNone
end_timeAnyNone
peak_scorefloat0.0
severityfloat0.0
score_levelScoreLevelScoreLevel.SAMPLE
affected_featureslist[str]<factory: list>
affected_entitieslist[Any]<factory: list>
observation_idslist[Any]<factory: list>
metadatadict[str, Any]<factory: dict>

Members

start_timeattribute
end_timeattribute
peak_scoreattribute
severityattribute
score_levelattribute
score_level: ScoreLevel = ScoreLevel.SAMPLE
affected_featuresattribute
affected_features: list[str] = field(default_factory=list)
affected_entitiesattribute
affected_entities: list[Any] = field(default_factory=list)
observation_idsattribute
observation_ids: list[Any] = field(default_factory=list)
metadataattribute
metadata: dict[str, Any] = field(default_factory=dict)
to_dictmethod
to_dict() -> dict[str, Any]

Returns dict[str, Any]

AnomalyEvents

AnomalyEvents(events: list[AnomalyEvent] = <factory: list>, frame: pd.DataFrame = <factory: pd.DataFrame>, label_column: str = 'is_anomaly', metadata: dict[str, Any] = <factory: dict>)

Detector decision output: events plus the labeled score table.

Signature parameters
ParameterTypeDefault
eventslist[AnomalyEvent]<factory: list>
framepd.DataFrame<factory: pd.DataFrame>
label_columnstr'is_anomaly'
metadatadict[str, Any]<factory: dict>

Members

eventsattribute
events: list[AnomalyEvent] = field(default_factory=list)
frameattribute
frame: pd.DataFrame = field(default_factory=pd.DataFrame)
label_columnattribute
label_column: str = 'is_anomaly'
metadataattribute
metadata: dict[str, Any] = field(default_factory=dict)
event_countproperty
event_count: int

Returns int

to_dictmethod
to_dict() -> dict[str, Any]

Returns dict[str, Any]

AnomalyScores

AnomalyScores(frame: pd.DataFrame, score_column: str = 'score', score_level: ScoreLevel = ScoreLevel.SAMPLE, is_calibrated: bool = False, metadata: dict[str, Any] = <factory: dict>)

Continuous anomaly scores with explicit alignment information.

Signature parameters
ParameterTypeDefault
framepd.DataFramerequired
score_columnstr'score'
score_levelScoreLevelScoreLevel.SAMPLE
is_calibratedboolFalse
metadatadict[str, Any]<factory: dict>

Members

frameattribute
score_columnattribute
score_column: str = 'score'
score_levelattribute
score_level: ScoreLevel = ScoreLevel.SAMPLE
is_calibratedattribute
is_calibrated: bool = False
metadataattribute
metadata: dict[str, Any] = field(default_factory=dict)
valuesmethod
values() -> pd.Series

Returns pd.Series

to_recordsmethod
to_records() -> list[dict[str, Any]]

Returns list[dict[str, Any]]

ScoreLevel

ScoreLevel

Bases: str, Enum

Alignment level of an anomaly score.

Members

SAMPLEattribute
TIMESTEPattribute
TIMESTEP = 'timestep'
WINDOWattribute
SEQUENCEattribute
SEQUENCE = 'sequence'
FEATUREattribute
ENTITYattribute
NODEattribute
EDGEattribute
GRAPHattribute