MODULE REFERENCE
results
anomx.detection.resultsShared data contracts flowing between pipeline stages.
` ObservationSet → Model → InferenceOutput → Scorer → AnomalyScores → Calibrator → AnomalyScores → Detector → AnomalyEvents `
classView source ↗
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.
| Parameter | Type | Default |
|---|---|---|
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> |
Members
start_timeattribute
start_time: Any = Noneend_timeattribute
end_time: Any = Nonepeak_scoreattribute
peak_score: float = 0.0severityattribute
severity: float = 0.0score_levelattribute
score_level: ScoreLevel = ScoreLevel.SAMPLEaffected_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
classView source ↗
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.
| Parameter | Type | Default |
|---|---|---|
events | list[AnomalyEvent] | <factory: list> |
frame | pd.DataFrame | <factory: pd.DataFrame> |
label_column | str | 'is_anomaly' |
metadata | dict[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
to_dictmethod
classView source ↗
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.
| Parameter | Type | Default |
|---|---|---|
frame | pd.DataFrame | required |
score_column | str | 'score' |
score_level | ScoreLevel | ScoreLevel.SAMPLE |
is_calibrated | bool | False |
metadata | dict[str, Any] | <factory: dict> |
Members
frameattribute
frame: pd.DataFramescore_columnattribute
score_column: str = 'score'score_levelattribute
score_level: ScoreLevel = ScoreLevel.SAMPLEis_calibratedattribute
is_calibrated: bool = Falsemetadataattribute
metadata: dict[str, Any] = field(default_factory=dict)valuesmethod
to_recordsmethod
classView source ↗
ScoreLevel
ScoreLevelBases: str, Enum
Alignment level of an anomaly score.
Members
SAMPLEattribute
SAMPLE = 'sample'TIMESTEPattribute
TIMESTEP = 'timestep'WINDOWattribute
WINDOW = 'window'SEQUENCEattribute
SEQUENCE = 'sequence'FEATUREattribute
FEATURE = 'feature'ENTITYattribute
ENTITY = 'entity'NODEattribute
NODE = 'node'EDGEattribute
EDGE = 'edge'GRAPHattribute
GRAPH = 'graph'