Skip to content

Core: Models

The live model surface consumes PyG TemporalData event batches.

Current temporal model families:

  • TemporalEventClassifier: stateless MLP baseline over event messages plus ID embeddings.
  • TemporalRNNClassifier: supervised GRU classifier that carries hidden state across adjacent TemporalDataLoader batches and resets at reset_after boundaries.
  • TemporalGAT: causal event-attention classifier.
  • TemporalVGAE: recurrent variational event autoencoder for event-level surprise scoring.
  • TemporalHybridModel: modular temporal hybrid for supervised, anomaly, or joint learning. It composes an event encoder, optional TGN-style arbitration ID memory, a stream backbone (none, gru, ssm_lite, or optional mamba), and independently enabled classifier / self-supervised anomaly heads. Its memory can include per-ID elapsed-time encodings, and optional rhythm and motif contexts add causal CAN schedule features before the stream backbone.
  • id_encoding/: categorical-ID encoders with reserved UNK at index 0.

TemporalHybridModel is configured through the temporal_hybrid primitive. Use objective: supervised for classifier-only training, objective: anomaly for self-supervised attack-free anomaly scoring, and objective: joint when a classifier and anomaly heads should train together. Runtime dimensions (num_ids, in_channels, and num_classes) are still injected from the TemporalDataModule after setup(), so YAML configs do not hard-code dataset vocabulary sizes.

Anomaly heads default to the original regression/error scoring mode. Set anomaly.mode: nll to score IAT and payload-delta heads with Gaussian negative log-likelihood while keeping next-ID categorical NLL.

graphids.core.models

models

Core model families.

base

Shared model infrastructure for temporal modules.

safe_load_checkpoint 

safe_load_checkpoint(model_type: str, ckpt_path, *, map_location='cpu')

Load a checkpoint using the class path stored in the checkpoint.

Source code in graphids/core/models/base.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
def safe_load_checkpoint(model_type: str, ckpt_path, *, map_location="cpu"):
    """Load a checkpoint using the class path stored in the checkpoint."""
    del model_type
    ckpt_path = Path(ckpt_path)
    if not ckpt_path.exists():
        raise FileNotFoundError(f"Checkpoint not found: {ckpt_path}")

    from graphids._fs import atomic_load

    ckpt = atomic_load(ckpt_path, map_location=map_location, weights_only=True)
    dotted = ckpt.get("class_path")
    if not dotted:
        raise KeyError(f"Checkpoint {ckpt_path} missing 'class_path'.")
    module_path, cls_name = dotted.rsplit(".", 1)
    cls = getattr(importlib.import_module(module_path), cls_name)
    hp = dict(ckpt.get("hyper_parameters", {}))
    rebuild = getattr(cls, "_rebuild_excluded_kwargs", None)
    extra_kwargs: dict = rebuild(hp) if rebuild is not None else {}
    module = cls(**{**hp, **extra_kwargs})
    state_dict = strip_orig_mod_prefix(ckpt["state_dict"])
    state_dict = {k: v for k, v in state_dict.items() if not k.startswith("loss_fn.")}
    module.load_state_dict(state_dict)
    module.to(map_location)
    if hasattr(module, "on_load_checkpoint"):
        module.on_load_checkpoint(ckpt)
    return module

strip_orig_mod_prefix 

strip_orig_mod_prefix(state: dict[str, Any]) -> dict[str, Any]

Drop _orig_mod. prefixes injected by torch.compile.

Source code in graphids/core/models/base.py 
15
16
17
def strip_orig_mod_prefix(state: dict[str, Any]) -> dict[str, Any]:
    """Drop ``_orig_mod.`` prefixes injected by ``torch.compile``."""
    return {k.replace("_orig_mod.", ""): v for k, v in state.items()}

id_encoding

Pluggable identity-encoding strategies for event/node IDs.

An IdEncoder maps a node_id LongTensor to per-node embedding vectors. Subclasses implement different strategies (lookup table, k-probe hash, ...) behind a uniform interface so temporal models do not know which strategy is in use.

Research basis: ~/plans/oov-embedding-handling.md.

HashIdEncoder 

HashIdEncoder(num_buckets: int, embedding_dim: int, *, k: int = 2, seed: int = 42)

Bases: IdEncoder

Source code in graphids/core/models/id_encoding/hash_embedding.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    num_buckets: int,
    embedding_dim: int,
    *,
    k: int = 2,
    seed: int = 42,
):
    super().__init__()
    if num_buckets < 2:
        raise ValueError(f"num_buckets must be >= 2, got {num_buckets}")
    if k < 1:
        raise ValueError(f"k must be >= 1, got {k}")
    self.embedding = nn.Embedding(num_buckets, embedding_dim)
    self.out_dim = embedding_dim
    self.num_buckets = num_buckets
    self.k = k
    # ``k`` decorrelated hash offsets, deterministic in ``seed``.
    # Spread across int64 so per-probe bucket distributions are
    # well-separated for small vocabs. Registered as a buffer so
    # checkpoint round-trip is exact.
    offsets = torch.tensor(
        [seed + i * (1 << 30) for i in range(k)],
        dtype=torch.int64,
    )
    self.register_buffer("_hash_offsets", offsets)
from_vocab_size classmethod 
from_vocab_size(num_ids: int, *, embedding_dim: int, k: int = 2, seed: int = 42, num_buckets_factor: int = 4, num_buckets: int | None = None) -> HashIdEncoder

Build from a datamodule-injected num_ids.

Default bucket count: next_pow2(num_buckets_factor · num_ids), minimum 8. Per plan: Yan 2021 / Coleman 2023 use 2–4× vocab size as a sweet spot between collision rate and parameter count. num_buckets can be passed explicitly to override.

Source code in graphids/core/models/id_encoding/hash_embedding.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@classmethod
def from_vocab_size(
    cls,
    num_ids: int,
    *,
    embedding_dim: int,
    k: int = 2,
    seed: int = 42,
    num_buckets_factor: int = 4,
    num_buckets: int | None = None,
) -> HashIdEncoder:
    """Build from a datamodule-injected ``num_ids``.

    Default bucket count: ``next_pow2(num_buckets_factor · num_ids)``,
    minimum 8. Per plan: Yan 2021 / Coleman 2023 use 2–4× vocab size
    as a sweet spot between collision rate and parameter count.
    ``num_buckets`` can be passed explicitly to override.
    """
    if num_buckets is None:
        target = max(8, num_buckets_factor * max(1, num_ids))
        num_buckets = 1 << (target - 1).bit_length()
    return cls(num_buckets=num_buckets, embedding_dim=embedding_dim, k=k, seed=seed)

IdEncoder

Bases: Module

Maps per-node identities to per-node embedding vectors.

Planned subclasses: - LookupIdEncoder — dense nn.Embedding over a shared vocab, with optional stochastic UNK-drop (Stage 3 ablation). - HashIdEncoder (Stage 2 primary, not yet implemented) — k-probe hash embedding per Yan et al. 2021 (CIKM).

build_encoder 

build_encoder(class_path: str, num_ids: int, embedding_dim: int, **kwargs: Any) -> IdEncoder

Resolve a dotted class_path and call from_vocab_size.

num_ids is data-dependent (populated by datamodule.setup), so encoder construction stays at model-build time.

Source code in graphids/core/models/id_encoding/base.py 
21
22
23
24
25
26
27
28
29
def build_encoder(class_path: str, num_ids: int, embedding_dim: int, **kwargs: Any) -> IdEncoder:
    """Resolve a dotted ``class_path`` and call ``from_vocab_size``.

    ``num_ids`` is data-dependent (populated by ``datamodule.setup``), so
    encoder construction stays at model-build time.
    """
    mod, _, cls_name = class_path.rpartition(".")
    cls = getattr(importlib.import_module(mod), cls_name)
    return cls.from_vocab_size(num_ids=num_ids, embedding_dim=embedding_dim, **kwargs)

base

Base class for pluggable identity encoders.

Contract (duck-typed, matching the rest of the codebase):

  • forward(node_id: LongTensor) -> Tensor of shape (N, out_dim).
  • out_dim: int attribute set in __init__.
  • All stateful policy (vocab size, hash seeds, UNK-drop rate) lives on the encoder instance — InputEncoder holds one and does not branch on its type.
IdEncoder

Bases: Module

Maps per-node identities to per-node embedding vectors.

Planned subclasses: - LookupIdEncoder — dense nn.Embedding over a shared vocab, with optional stochastic UNK-drop (Stage 3 ablation). - HashIdEncoder (Stage 2 primary, not yet implemented) — k-probe hash embedding per Yan et al. 2021 (CIKM).

build_encoder 
build_encoder(class_path: str, num_ids: int, embedding_dim: int, **kwargs: Any) -> IdEncoder

Resolve a dotted class_path and call from_vocab_size.

num_ids is data-dependent (populated by datamodule.setup), so encoder construction stays at model-build time.

Source code in graphids/core/models/id_encoding/base.py 
21
22
23
24
25
26
27
28
29
def build_encoder(class_path: str, num_ids: int, embedding_dim: int, **kwargs: Any) -> IdEncoder:
    """Resolve a dotted ``class_path`` and call ``from_vocab_size``.

    ``num_ids`` is data-dependent (populated by ``datamodule.setup``), so
    encoder construction stays at model-build time.
    """
    mod, _, cls_name = class_path.rpartition(".")
    cls = getattr(importlib.import_module(mod), cls_name)
    return cls.from_vocab_size(num_ids=num_ids, embedding_dim=embedding_dim, **kwargs)

config

Explicit ID-encoding configs and factories.

hash_embedding

k-probe hash embedding — primary Stage-2 treatment.

Every id (seen or unseen) deterministically maps to k rows of a bucketed embedding table by k decorrelated hash functions; the per-probe vectors are summed. Because any id hits trained buckets by construction, no special OOV slot is needed.

Shape follows Coleman et al. 2023 Unified Embedding (NeurIPS Spotlight): one shared table, k probes, sum combiner — minimum parameters, clean theoretical analysis. Yan et al. 2021 Binary Code Hash Embedding (CIKM) uses the same k-probe idea with separate tables per hash; at CAN scale (~100 ids, B=512) the shared table has the same expressive power at half the parameters.

Hash: bucket_i(id) = (id * KNUTH + offset_i) mod num_buckets, where KNUTH = 2654435761 (golden-ratio-derived Knuth multiplier) and the k offsets are deterministic functions of the seed constructor arg. The multiplier is coprime to any num_buckets >= 2 that isn't a specific pathological case, and Knuth's value is well-studied for integer-id hashing at tiny scale.

Research basis: ~/plans/oov-embedding-handling.md (Stage 2).

HashIdEncoder 
HashIdEncoder(num_buckets: int, embedding_dim: int, *, k: int = 2, seed: int = 42)

Bases: IdEncoder

Source code in graphids/core/models/id_encoding/hash_embedding.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    num_buckets: int,
    embedding_dim: int,
    *,
    k: int = 2,
    seed: int = 42,
):
    super().__init__()
    if num_buckets < 2:
        raise ValueError(f"num_buckets must be >= 2, got {num_buckets}")
    if k < 1:
        raise ValueError(f"k must be >= 1, got {k}")
    self.embedding = nn.Embedding(num_buckets, embedding_dim)
    self.out_dim = embedding_dim
    self.num_buckets = num_buckets
    self.k = k
    # ``k`` decorrelated hash offsets, deterministic in ``seed``.
    # Spread across int64 so per-probe bucket distributions are
    # well-separated for small vocabs. Registered as a buffer so
    # checkpoint round-trip is exact.
    offsets = torch.tensor(
        [seed + i * (1 << 30) for i in range(k)],
        dtype=torch.int64,
    )
    self.register_buffer("_hash_offsets", offsets)
from_vocab_size classmethod 
from_vocab_size(num_ids: int, *, embedding_dim: int, k: int = 2, seed: int = 42, num_buckets_factor: int = 4, num_buckets: int | None = None) -> HashIdEncoder

Build from a datamodule-injected num_ids.

Default bucket count: next_pow2(num_buckets_factor · num_ids), minimum 8. Per plan: Yan 2021 / Coleman 2023 use 2–4× vocab size as a sweet spot between collision rate and parameter count. num_buckets can be passed explicitly to override.

Source code in graphids/core/models/id_encoding/hash_embedding.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@classmethod
def from_vocab_size(
    cls,
    num_ids: int,
    *,
    embedding_dim: int,
    k: int = 2,
    seed: int = 42,
    num_buckets_factor: int = 4,
    num_buckets: int | None = None,
) -> HashIdEncoder:
    """Build from a datamodule-injected ``num_ids``.

    Default bucket count: ``next_pow2(num_buckets_factor · num_ids)``,
    minimum 8. Per plan: Yan 2021 / Coleman 2023 use 2–4× vocab size
    as a sweet spot between collision rate and parameter count.
    ``num_buckets`` can be passed explicitly to override.
    """
    if num_buckets is None:
        target = max(8, num_buckets_factor * max(1, num_ids))
        num_buckets = 1 << (target - 1).bit_length()
    return cls(num_buckets=num_buckets, embedding_dim=embedding_dim, k=k, seed=seed)

lookup

Dense lookup embedding with optional stochastic UNK-drop.

Default (p_unk_drop=0.0) reproduces the pre-refactor nn.Embedding behavior byte-for-byte so existing single-vocab runs are a no-op change.

p_unk_drop > 0.0 implements the Stage 3 ablation arm from ~/plans/oov-embedding-handling.md: during training, each node_id is remapped to UNK_INDEX with probability p, so the OOV row receives gradient and attack-introduced IDs at inference land in a trained slot instead of init noise.

temporal

Temporal event model family exports.

TemporalEventClassifier 

TemporalEventClassifier(*, loss_fn: Module | None = None, hidden: int | None = None, layers: int | None = None, embedding_dim: int | None = None, dropout: float = 0.2, lr: float = 0.001, weight_decay: float = 0.0001, scale: str = 'small', model_type: str = 'temporal_event_classifier', dataset: str = '', seed: int = 42, num_ids: int = 0, in_channels: int = 0, num_classes: int = 2)

Bases: TemporalModuleBase

MLP baseline over TemporalData event messages plus ID embeddings.

Source code in graphids/core/models/temporal/event_classifier.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    *,
    loss_fn: nn.Module | None = None,
    hidden: int | None = None,
    layers: int | None = None,
    embedding_dim: int | None = None,
    dropout: float = 0.2,
    lr: float = 1e-3,
    weight_decay: float = 1e-4,
    scale: str = "small",
    model_type: str = "temporal_event_classifier",
    dataset: str = "",
    seed: int = 42,
    num_ids: int = 0,
    in_channels: int = 0,
    num_classes: int = 2,
):
    loss_fn = loss_fn if loss_fn is not None else CrossEntropyLoss()
    preset = self._SCALES.get(scale, {})
    hidden = hidden if hidden is not None else preset.get("hidden", 64)
    layers = layers if layers is not None else preset.get("layers", 2)
    embedding_dim = embedding_dim if embedding_dim is not None else preset.get("embedding_dim", 16)
    super().__init__()
    self.test_metrics = classification_test_metrics(num_classes)
    self._val_probs: list[torch.Tensor] = []
    self._val_labels: list[torch.Tensor] = []
    self._init_post(locals())

TemporalGAT 

TemporalGAT(*, loss_fn: Module | None = None, hidden: int | None = None, layers: int | None = None, heads: int | None = None, embedding_dim: int | None = None, dropout: float = 0.2, lr: float = 0.001, weight_decay: float = 0.0001, scale: str = 'small', model_type: str = 'temporal_gat', dataset: str = '', seed: int = 42, num_ids: int = 0, in_channels: int = 0, num_classes: int = 2)

Bases: TemporalModuleBase

Causal event-attention classifier over PyG TemporalData batches.

Source code in graphids/core/models/temporal/gat.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    *,
    loss_fn: nn.Module | None = None,
    hidden: int | None = None,
    layers: int | None = None,
    heads: int | None = None,
    embedding_dim: int | None = None,
    dropout: float = 0.2,
    lr: float = 1e-3,
    weight_decay: float = 1e-4,
    scale: str = "small",
    model_type: str = "temporal_gat",
    dataset: str = "",
    seed: int = 42,
    num_ids: int = 0,
    in_channels: int = 0,
    num_classes: int = 2,
):
    loss_fn = loss_fn if loss_fn is not None else CrossEntropyLoss()
    preset = self._SCALES.get(scale, {})
    hidden = hidden if hidden is not None else preset.get("hidden", 64)
    layers = layers if layers is not None else preset.get("layers", 2)
    heads = heads if heads is not None else preset.get("heads", 4)
    embedding_dim = embedding_dim if embedding_dim is not None else preset.get("embedding_dim", 16)
    super().__init__()
    self.test_metrics = classification_test_metrics(num_classes)
    self._val_probs: list[torch.Tensor] = []
    self._val_labels: list[torch.Tensor] = []
    self._init_post(locals())

TemporalHybridModel 

TemporalHybridModel(*, loss_fn: Module | None = None, scale: str = 'small', objective: Objective = 'supervised', input: dict[str, Any] | None = None, memory: dict[str, Any] | None = None, backbone: dict[str, Any] | None = None, heads: dict[str, bool] | None = None, anomaly: dict[str, Any] | None = None, rhythm: dict[str, Any] | None = None, motif: dict[str, Any] | None = None, loss_weights: dict[str, float] | None = None, anomaly_score_weights: dict[str, float] | None = None, lr: float = 0.001, weight_decay: float = 0.0001, model_type: str = 'temporal_hybrid', dataset: str = '', seed: int = 42, num_ids: int = 0, in_channels: int = 0, num_classes: int = 2)

Bases: TemporalModuleBase

Hybrid classifier/anomaly detector with ID memory and stream backbones.

Source code in graphids/core/models/temporal/_hybrid/model.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def __init__(
    self,
    *,
    loss_fn: nn.Module | None = None,
    scale: str = "small",
    objective: Objective = "supervised",
    input: dict[str, Any] | None = None,
    memory: dict[str, Any] | None = None,
    backbone: dict[str, Any] | None = None,
    heads: dict[str, bool] | None = None,
    anomaly: dict[str, Any] | None = None,
    rhythm: dict[str, Any] | None = None,
    motif: dict[str, Any] | None = None,
    loss_weights: dict[str, float] | None = None,
    anomaly_score_weights: dict[str, float] | None = None,
    lr: float = 1e-3,
    weight_decay: float = 1e-4,
    model_type: str = "temporal_hybrid",
    dataset: str = "",
    seed: int = 42,
    num_ids: int = 0,
    in_channels: int = 0,
    num_classes: int = 2,
):
    if objective not in {"supervised", "anomaly", "joint"}:
        raise ValueError("objective must be one of: supervised, anomaly, joint")
    preset = self._SCALES.get(scale, self._SCALES["small"])
    input = _drop_none(dict(input or {}))
    backbone = _drop_none(dict(backbone or {}))
    memory = _drop_none(dict(memory or {}))
    heads = _drop_none(dict(heads or {}))
    anomaly = _drop_none(dict(anomaly or {}))
    rhythm = _drop_none(dict(rhythm or {}))
    motif = _drop_none(dict(motif or {}))
    loss_weights = dict(loss_weights or {})
    anomaly_score_weights = dict(anomaly_score_weights or {})

    input.setdefault("hidden", preset["hidden"])
    input.setdefault("embedding_dim", preset["embedding_dim"])
    input.setdefault("dropout", 0.1)
    backbone.setdefault("type", "gru")
    backbone.setdefault("layers", preset["layers"])
    backbone.setdefault("dropout", input["dropout"])
    memory.setdefault("type", "tgn")
    memory.setdefault("enabled", True)
    memory.setdefault("reset_on_stream_end", True)
    memory.setdefault("use_source", True)
    memory.setdefault("use_destination", True)
    memory.setdefault("time_encoding_dim", 0)
    anomaly.setdefault("mode", "regression")
    anomaly.setdefault("min_log_scale", -7.0)
    anomaly.setdefault("max_log_scale", 5.0)
    rhythm.setdefault("enabled", False)
    motif.setdefault("enabled", False)
    motif.setdefault("length", 3)
    motif.setdefault("embedding_dim", max(2, int(input["embedding_dim"]) // 2))
    motif.setdefault("time_dim", max(2, int(input["embedding_dim"]) // 2))

    default_ssl = objective in {"anomaly", "joint"}
    heads.setdefault("classification", objective in {"supervised", "joint"})
    heads.setdefault("next_id", default_ssl)
    heads.setdefault("iat", default_ssl)
    heads.setdefault("payload_delta", default_ssl)

    loss_weights.setdefault("classification", 1.0)
    loss_weights.setdefault("next_id", 0.2)
    loss_weights.setdefault("iat", 0.1)
    loss_weights.setdefault("payload_delta", 0.1)
    anomaly_score_weights.setdefault("next_id", 1.0)
    anomaly_score_weights.setdefault("iat", 1.0)
    anomaly_score_weights.setdefault("payload_delta", 1.0)

    self._validate_modular_config(
        objective=objective,
        memory=memory,
        backbone=backbone,
        heads=heads,
        anomaly=anomaly,
        rhythm=rhythm,
        motif=motif,
        loss_fn=loss_fn,
    )
    if heads.get("classification", False) and loss_fn is None:
        loss_fn = CrossEntropyLoss()

    super().__init__()
    self.log_binary_score_metrics = objective in {"anomaly", "joint"} or any(
        heads.get(k, False) for k in ("next_id", "iat", "payload_delta")
    )
    self.test_metrics = classification_test_metrics(num_classes)
    self._train_state: dict[str, torch.Tensor | None] | None = None
    self._val_state: dict[str, torch.Tensor | None] | None = None
    self._test_states: dict[int, dict[str, torch.Tensor | None] | None] = {}
    self._val_cls_probs: list[torch.Tensor] = []
    self._val_cls_labels: list[torch.Tensor] = []
    self._val_anom_scores: list[torch.Tensor] = []
    self._val_anom_labels: list[torch.Tensor] = []
    self._init_post(locals())

TemporalRNNClassifier 

TemporalRNNClassifier(*, loss_fn: Module | None = None, hidden: int | None = None, layers: int | None = None, embedding_dim: int | None = None, dropout: float = 0.2, lr: float = 0.001, weight_decay: float = 0.0001, scale: str = 'small', model_type: str = 'temporal_rnn_classifier', dataset: str = '', seed: int = 42, num_ids: int = 0, in_channels: int = 0, num_classes: int = 2)

Bases: TemporalModuleBase

GRU classifier that carries hidden state across temporal event batches.

Source code in graphids/core/models/temporal/rnn_classifier.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    *,
    loss_fn: nn.Module | None = None,
    hidden: int | None = None,
    layers: int | None = None,
    embedding_dim: int | None = None,
    dropout: float = 0.2,
    lr: float = 1e-3,
    weight_decay: float = 1e-4,
    scale: str = "small",
    model_type: str = "temporal_rnn_classifier",
    dataset: str = "",
    seed: int = 42,
    num_ids: int = 0,
    in_channels: int = 0,
    num_classes: int = 2,
):
    loss_fn = loss_fn if loss_fn is not None else CrossEntropyLoss()
    preset = self._SCALES.get(scale, {})
    hidden = hidden if hidden is not None else preset.get("hidden", 64)
    layers = layers if layers is not None else preset.get("layers", 1)
    embedding_dim = embedding_dim if embedding_dim is not None else preset.get("embedding_dim", 16)
    super().__init__()
    self.test_metrics = classification_test_metrics(num_classes)
    self._train_state: torch.Tensor | None = None
    self._val_state: torch.Tensor | None = None
    self._test_states: dict[int, torch.Tensor | None] = {}
    self._val_probs: list[torch.Tensor] = []
    self._val_labels: list[torch.Tensor] = []
    self._init_post(locals())

TemporalVGAE 

TemporalVGAE(*, hidden: int | None = None, layers: int | None = None, embedding_dim: int | None = None, latent_dim: int | None = None, dropout: float = 0.1, kl_weight: float = 0.01, lr: float = 0.001, weight_decay: float = 0.0001, scale: str = 'small', model_type: str = 'temporal_vgae', dataset: str = '', seed: int = 42, num_ids: int = 0, in_channels: int = 0, num_classes: int = 2)

Bases: TemporalModuleBase

Recurrent variational autoencoder that scores event-level surprise.

Source code in graphids/core/models/temporal/vgae.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    *,
    hidden: int | None = None,
    layers: int | None = None,
    embedding_dim: int | None = None,
    latent_dim: int | None = None,
    dropout: float = 0.1,
    kl_weight: float = 0.01,
    lr: float = 1e-3,
    weight_decay: float = 1e-4,
    scale: str = "small",
    model_type: str = "temporal_vgae",
    dataset: str = "",
    seed: int = 42,
    num_ids: int = 0,
    in_channels: int = 0,
    num_classes: int = 2,
):
    preset = self._SCALES.get(scale, {})
    hidden = hidden if hidden is not None else preset.get("hidden", 64)
    layers = layers if layers is not None else preset.get("layers", 1)
    embedding_dim = embedding_dim if embedding_dim is not None else preset.get("embedding_dim", 16)
    latent_dim = latent_dim if latent_dim is not None else preset.get("latent_dim", 32)
    super().__init__()
    self.log_binary_score_metrics = True
    self._init_post(locals())

base

Shared infrastructure for temporal event models.

TemporalModuleBase

Bases: _ModelBase

Base class for models that consume PyG TemporalData batches.

event_classifier

Stateless supervised classifier for temporal CAN events.

TemporalEventClassifier 
TemporalEventClassifier(*, loss_fn: Module | None = None, hidden: int | None = None, layers: int | None = None, embedding_dim: int | None = None, dropout: float = 0.2, lr: float = 0.001, weight_decay: float = 0.0001, scale: str = 'small', model_type: str = 'temporal_event_classifier', dataset: str = '', seed: int = 42, num_ids: int = 0, in_channels: int = 0, num_classes: int = 2)

Bases: TemporalModuleBase

MLP baseline over TemporalData event messages plus ID embeddings.

Source code in graphids/core/models/temporal/event_classifier.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    *,
    loss_fn: nn.Module | None = None,
    hidden: int | None = None,
    layers: int | None = None,
    embedding_dim: int | None = None,
    dropout: float = 0.2,
    lr: float = 1e-3,
    weight_decay: float = 1e-4,
    scale: str = "small",
    model_type: str = "temporal_event_classifier",
    dataset: str = "",
    seed: int = 42,
    num_ids: int = 0,
    in_channels: int = 0,
    num_classes: int = 2,
):
    loss_fn = loss_fn if loss_fn is not None else CrossEntropyLoss()
    preset = self._SCALES.get(scale, {})
    hidden = hidden if hidden is not None else preset.get("hidden", 64)
    layers = layers if layers is not None else preset.get("layers", 2)
    embedding_dim = embedding_dim if embedding_dim is not None else preset.get("embedding_dim", 16)
    super().__init__()
    self.test_metrics = classification_test_metrics(num_classes)
    self._val_probs: list[torch.Tensor] = []
    self._val_labels: list[torch.Tensor] = []
    self._init_post(locals())

gat

Temporal supervised attention model for CAN event streams.

TemporalGAT 
TemporalGAT(*, loss_fn: Module | None = None, hidden: int | None = None, layers: int | None = None, heads: int | None = None, embedding_dim: int | None = None, dropout: float = 0.2, lr: float = 0.001, weight_decay: float = 0.0001, scale: str = 'small', model_type: str = 'temporal_gat', dataset: str = '', seed: int = 42, num_ids: int = 0, in_channels: int = 0, num_classes: int = 2)

Bases: TemporalModuleBase

Causal event-attention classifier over PyG TemporalData batches.

Source code in graphids/core/models/temporal/gat.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    *,
    loss_fn: nn.Module | None = None,
    hidden: int | None = None,
    layers: int | None = None,
    heads: int | None = None,
    embedding_dim: int | None = None,
    dropout: float = 0.2,
    lr: float = 1e-3,
    weight_decay: float = 1e-4,
    scale: str = "small",
    model_type: str = "temporal_gat",
    dataset: str = "",
    seed: int = 42,
    num_ids: int = 0,
    in_channels: int = 0,
    num_classes: int = 2,
):
    loss_fn = loss_fn if loss_fn is not None else CrossEntropyLoss()
    preset = self._SCALES.get(scale, {})
    hidden = hidden if hidden is not None else preset.get("hidden", 64)
    layers = layers if layers is not None else preset.get("layers", 2)
    heads = heads if heads is not None else preset.get("heads", 4)
    embedding_dim = embedding_dim if embedding_dim is not None else preset.get("embedding_dim", 16)
    super().__init__()
    self.test_metrics = classification_test_metrics(num_classes)
    self._val_probs: list[torch.Tensor] = []
    self._val_labels: list[torch.Tensor] = []
    self._init_post(locals())

hybrid

Compatibility exports for the temporal hybrid model.

Implementation lives under :mod:graphids.core.models.temporal._hybrid; this module remains the public import and checkpoint class path.

TemporalHybridModel 
TemporalHybridModel(*, loss_fn: Module | None = None, scale: str = 'small', objective: Objective = 'supervised', input: dict[str, Any] | None = None, memory: dict[str, Any] | None = None, backbone: dict[str, Any] | None = None, heads: dict[str, bool] | None = None, anomaly: dict[str, Any] | None = None, rhythm: dict[str, Any] | None = None, motif: dict[str, Any] | None = None, loss_weights: dict[str, float] | None = None, anomaly_score_weights: dict[str, float] | None = None, lr: float = 0.001, weight_decay: float = 0.0001, model_type: str = 'temporal_hybrid', dataset: str = '', seed: int = 42, num_ids: int = 0, in_channels: int = 0, num_classes: int = 2)

Bases: TemporalModuleBase

Hybrid classifier/anomaly detector with ID memory and stream backbones.

Source code in graphids/core/models/temporal/_hybrid/model.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def __init__(
    self,
    *,
    loss_fn: nn.Module | None = None,
    scale: str = "small",
    objective: Objective = "supervised",
    input: dict[str, Any] | None = None,
    memory: dict[str, Any] | None = None,
    backbone: dict[str, Any] | None = None,
    heads: dict[str, bool] | None = None,
    anomaly: dict[str, Any] | None = None,
    rhythm: dict[str, Any] | None = None,
    motif: dict[str, Any] | None = None,
    loss_weights: dict[str, float] | None = None,
    anomaly_score_weights: dict[str, float] | None = None,
    lr: float = 1e-3,
    weight_decay: float = 1e-4,
    model_type: str = "temporal_hybrid",
    dataset: str = "",
    seed: int = 42,
    num_ids: int = 0,
    in_channels: int = 0,
    num_classes: int = 2,
):
    if objective not in {"supervised", "anomaly", "joint"}:
        raise ValueError("objective must be one of: supervised, anomaly, joint")
    preset = self._SCALES.get(scale, self._SCALES["small"])
    input = _drop_none(dict(input or {}))
    backbone = _drop_none(dict(backbone or {}))
    memory = _drop_none(dict(memory or {}))
    heads = _drop_none(dict(heads or {}))
    anomaly = _drop_none(dict(anomaly or {}))
    rhythm = _drop_none(dict(rhythm or {}))
    motif = _drop_none(dict(motif or {}))
    loss_weights = dict(loss_weights or {})
    anomaly_score_weights = dict(anomaly_score_weights or {})

    input.setdefault("hidden", preset["hidden"])
    input.setdefault("embedding_dim", preset["embedding_dim"])
    input.setdefault("dropout", 0.1)
    backbone.setdefault("type", "gru")
    backbone.setdefault("layers", preset["layers"])
    backbone.setdefault("dropout", input["dropout"])
    memory.setdefault("type", "tgn")
    memory.setdefault("enabled", True)
    memory.setdefault("reset_on_stream_end", True)
    memory.setdefault("use_source", True)
    memory.setdefault("use_destination", True)
    memory.setdefault("time_encoding_dim", 0)
    anomaly.setdefault("mode", "regression")
    anomaly.setdefault("min_log_scale", -7.0)
    anomaly.setdefault("max_log_scale", 5.0)
    rhythm.setdefault("enabled", False)
    motif.setdefault("enabled", False)
    motif.setdefault("length", 3)
    motif.setdefault("embedding_dim", max(2, int(input["embedding_dim"]) // 2))
    motif.setdefault("time_dim", max(2, int(input["embedding_dim"]) // 2))

    default_ssl = objective in {"anomaly", "joint"}
    heads.setdefault("classification", objective in {"supervised", "joint"})
    heads.setdefault("next_id", default_ssl)
    heads.setdefault("iat", default_ssl)
    heads.setdefault("payload_delta", default_ssl)

    loss_weights.setdefault("classification", 1.0)
    loss_weights.setdefault("next_id", 0.2)
    loss_weights.setdefault("iat", 0.1)
    loss_weights.setdefault("payload_delta", 0.1)
    anomaly_score_weights.setdefault("next_id", 1.0)
    anomaly_score_weights.setdefault("iat", 1.0)
    anomaly_score_weights.setdefault("payload_delta", 1.0)

    self._validate_modular_config(
        objective=objective,
        memory=memory,
        backbone=backbone,
        heads=heads,
        anomaly=anomaly,
        rhythm=rhythm,
        motif=motif,
        loss_fn=loss_fn,
    )
    if heads.get("classification", False) and loss_fn is None:
        loss_fn = CrossEntropyLoss()

    super().__init__()
    self.log_binary_score_metrics = objective in {"anomaly", "joint"} or any(
        heads.get(k, False) for k in ("next_id", "iat", "payload_delta")
    )
    self.test_metrics = classification_test_metrics(num_classes)
    self._train_state: dict[str, torch.Tensor | None] | None = None
    self._val_state: dict[str, torch.Tensor | None] | None = None
    self._test_states: dict[int, dict[str, torch.Tensor | None] | None] = {}
    self._val_cls_probs: list[torch.Tensor] = []
    self._val_cls_labels: list[torch.Tensor] = []
    self._val_anom_scores: list[torch.Tensor] = []
    self._val_anom_labels: list[torch.Tensor] = []
    self._init_post(locals())
TemporalIdMemory 
TemporalIdMemory(*, num_ids: int, hidden: int, use_source: bool = True, use_destination: bool = True, time_encoding_dim: int = 0)

Bases: Module

TGN-style per-ID memory updated causally after each event.

Source code in graphids/core/models/temporal/_hybrid/contexts.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(
    self,
    *,
    num_ids: int,
    hidden: int,
    use_source: bool = True,
    use_destination: bool = True,
    time_encoding_dim: int = 0,
):
    super().__init__()
    self.num_ids = max(1, int(num_ids))
    self.hidden = int(hidden)
    self.use_source = bool(use_source)
    self.use_destination = bool(use_destination)
    self.time_encoding_dim = max(0, int(time_encoding_dim))
    self.update_cell = nn.GRUCell(self.hidden, self.hidden)
    self.time_encoder = (
        TemporalTimeEncoder(self.time_encoding_dim) if self.time_encoding_dim > 0 else None
    )
    context_dim = (self.hidden if self.use_source else 0) + (self.hidden if self.use_destination else 0)
    context_dim += 2 * self.time_encoding_dim
    self._has_context = context_dim > 0
    self.mix = nn.Sequential(
        nn.Linear(max(1, context_dim), self.hidden),
        nn.LayerNorm(self.hidden),
        nn.GELU(),
    )
TemporalInputEncoder 
TemporalInputEncoder(*, num_ids: int, in_channels: int, embedding_dim: int, hidden: int, dropout: float)

Bases: Module

Encode event bytes/features plus source and destination ID embeddings.

Source code in graphids/core/models/temporal/_hybrid/encoders.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def __init__(
    self,
    *,
    num_ids: int,
    in_channels: int,
    embedding_dim: int,
    hidden: int,
    dropout: float,
):
    super().__init__()
    self.src_embedding = nn.Embedding(max(1, int(num_ids)), int(embedding_dim))
    self.dst_embedding = nn.Embedding(max(1, int(num_ids)), int(embedding_dim))
    input_dim = int(in_channels) + (2 * int(embedding_dim))
    self.proj = nn.Sequential(
        nn.Linear(input_dim, int(hidden)),
        nn.LayerNorm(int(hidden)),
        nn.GELU(),
        nn.Dropout(float(dropout)),
    )
TemporalMotifContext 
TemporalMotifContext(*, num_ids: int, hidden: int, length: int, embedding_dim: int, time_dim: int)

Bases: Module

Encode recent destination-ID and IAT motifs from the current stream.

Source code in graphids/core/models/temporal/_hybrid/contexts.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
def __init__(self, *, num_ids: int, hidden: int, length: int, embedding_dim: int, time_dim: int):
    super().__init__()
    self.num_ids = max(1, int(num_ids))
    self.hidden = int(hidden)
    self.length = max(1, int(length))
    self.embedding = nn.Embedding(self.num_ids, int(embedding_dim))
    self.time_encoder = TemporalTimeEncoder(max(1, int(time_dim)))
    input_dim = self.length * (int(embedding_dim) + max(1, int(time_dim)))
    self.proj = nn.Sequential(
        nn.Linear(input_dim, self.hidden),
        nn.LayerNorm(self.hidden),
        nn.GELU(),
    )
TemporalRhythmContext 
TemporalRhythmContext(*, num_ids: int, hidden: int)

Bases: Module

Causal rolling IAT summaries for source and destination IDs.

Source code in graphids/core/models/temporal/_hybrid/contexts.py 
197
198
199
200
201
202
203
204
205
def __init__(self, *, num_ids: int, hidden: int):
    super().__init__()
    self.num_ids = max(1, int(num_ids))
    self.hidden = int(hidden)
    self.proj = nn.Sequential(
        nn.Linear(6, self.hidden),
        nn.LayerNorm(self.hidden),
        nn.GELU(),
    )
TemporalStreamBackbone 
TemporalStreamBackbone(*, backbone_type: str, hidden: int, layers: int, dropout: float)

Bases: Module

Configurable causal stream backbone.

Source code in graphids/core/models/temporal/_hybrid/backbone.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(self, *, backbone_type: str, hidden: int, layers: int, dropout: float):
    super().__init__()
    self.backbone_type = str(backbone_type)
    self.hidden = int(hidden)
    self.layers = max(1, int(layers))
    self.dropout = nn.Dropout(float(dropout))

    if self.backbone_type == "none":
        self.cells = nn.ModuleList()
    elif self.backbone_type == "gru":
        self.cells = nn.ModuleList([nn.GRUCell(self.hidden, self.hidden) for _ in range(self.layers)])
    elif self.backbone_type == "ssm_lite":
        self.cells = nn.ModuleList([_SsmLiteCell(self.hidden, float(dropout)) for _ in range(self.layers)])
    elif self.backbone_type == "mamba":
        try:
            import mamba_ssm  # noqa: F401
        except ImportError as exc:  # pragma: no cover - depends on optional external package.
            raise ImportError(
                "backbone.type='mamba' requires the optional 'mamba_ssm' package. "
                "Install it explicitly or use backbone.type='ssm_lite'."
            ) from exc
        raise NotImplementedError("backbone.type='mamba' is reserved for the optional backend.")
    else:
        raise ValueError("backbone.type must be one of: none, gru, ssm_lite, mamba")
TemporalTimeEncoder 
TemporalTimeEncoder(dim: int)

Bases: Module

Sinusoidal encoding for positive elapsed times.

Source code in graphids/core/models/temporal/_hybrid/encoders.py 
42
43
44
45
46
def __init__(self, dim: int):
    super().__init__()
    self.dim = max(1, int(dim))
    half = max(1, (self.dim + 1) // 2)
    self.register_buffer("freqs", torch.logspace(0, 3, steps=half), persistent=False)

rnn_classifier

Stateful supervised recurrent classifier for temporal CAN events.

TemporalRNNClassifier 
TemporalRNNClassifier(*, loss_fn: Module | None = None, hidden: int | None = None, layers: int | None = None, embedding_dim: int | None = None, dropout: float = 0.2, lr: float = 0.001, weight_decay: float = 0.0001, scale: str = 'small', model_type: str = 'temporal_rnn_classifier', dataset: str = '', seed: int = 42, num_ids: int = 0, in_channels: int = 0, num_classes: int = 2)

Bases: TemporalModuleBase

GRU classifier that carries hidden state across temporal event batches.

Source code in graphids/core/models/temporal/rnn_classifier.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    *,
    loss_fn: nn.Module | None = None,
    hidden: int | None = None,
    layers: int | None = None,
    embedding_dim: int | None = None,
    dropout: float = 0.2,
    lr: float = 1e-3,
    weight_decay: float = 1e-4,
    scale: str = "small",
    model_type: str = "temporal_rnn_classifier",
    dataset: str = "",
    seed: int = 42,
    num_ids: int = 0,
    in_channels: int = 0,
    num_classes: int = 2,
):
    loss_fn = loss_fn if loss_fn is not None else CrossEntropyLoss()
    preset = self._SCALES.get(scale, {})
    hidden = hidden if hidden is not None else preset.get("hidden", 64)
    layers = layers if layers is not None else preset.get("layers", 1)
    embedding_dim = embedding_dim if embedding_dim is not None else preset.get("embedding_dim", 16)
    super().__init__()
    self.test_metrics = classification_test_metrics(num_classes)
    self._train_state: torch.Tensor | None = None
    self._val_state: torch.Tensor | None = None
    self._test_states: dict[int, torch.Tensor | None] = {}
    self._val_probs: list[torch.Tensor] = []
    self._val_labels: list[torch.Tensor] = []
    self._init_post(locals())

vgae

Temporal variational event autoencoder for CAN streams.

TemporalVGAE 
TemporalVGAE(*, hidden: int | None = None, layers: int | None = None, embedding_dim: int | None = None, latent_dim: int | None = None, dropout: float = 0.1, kl_weight: float = 0.01, lr: float = 0.001, weight_decay: float = 0.0001, scale: str = 'small', model_type: str = 'temporal_vgae', dataset: str = '', seed: int = 42, num_ids: int = 0, in_channels: int = 0, num_classes: int = 2)

Bases: TemporalModuleBase

Recurrent variational autoencoder that scores event-level surprise.

Source code in graphids/core/models/temporal/vgae.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    *,
    hidden: int | None = None,
    layers: int | None = None,
    embedding_dim: int | None = None,
    latent_dim: int | None = None,
    dropout: float = 0.1,
    kl_weight: float = 0.01,
    lr: float = 1e-3,
    weight_decay: float = 1e-4,
    scale: str = "small",
    model_type: str = "temporal_vgae",
    dataset: str = "",
    seed: int = 42,
    num_ids: int = 0,
    in_channels: int = 0,
    num_classes: int = 2,
):
    preset = self._SCALES.get(scale, {})
    hidden = hidden if hidden is not None else preset.get("hidden", 64)
    layers = layers if layers is not None else preset.get("layers", 1)
    embedding_dim = embedding_dim if embedding_dim is not None else preset.get("embedding_dim", 16)
    latent_dim = latent_dim if latent_dim is not None else preset.get("latent_dim", 32)
    super().__init__()
    self.log_binary_score_metrics = True
    self._init_post(locals())