tanat.metric.trajectory.type.aggregation package#

Submodules#

tanat.metric.trajectory.type.aggregation.metric module#

AggregationTrajectoryMetric: trajectory distance by per-alias sequence distances with weighted aggregation.

class tanat.metric.trajectory.type.aggregation.metric.AggregationSettings(*, default_metric: SequenceMetric = 'linearpairwise', sequence_metrics: dict[str, SequenceMetric] | None = None, agg_fun: str = 'mean', weights: dict[str, float] | None = None)[source]#

Bases: object

Settings for AggregationTrajectoryMetric.

agg_fun and weights are orthogonal: "mean" + weights computes a weighted mean, "sum" + weights a weighted sum. Aliases absent from weights default to 1.0.

__init__(*args: Any, **kwargs: Any) → None[source]#

agg_fun: str = 'mean'[source]#

default_metric: SequenceMetric = 'linearpairwise'[source]#

model_dump(*, mode='python', **dump_kwargs)[source]#: Dump settings to a dict via Pydantic serialization.

sequence_metrics: dict[str, SequenceMetric] | None = None[source]#

weights: dict[str, float] | None = None[source]#

class tanat.metric.trajectory.type.aggregation.metric.AggregationTrajectoryMetric(default_metric: SequenceMetric | str = 'linearpairwise', sequence_metrics: dict[str, SequenceMetric | str] | None = None, agg_fun: str = 'mean', weights: dict[str, float] | None = None, *, store_path: str | Path | None = None, chunk_size: int = 500, resume: bool = True, dtype: str = 'float32')[source]#

Bases: TrajectoryMetric

Trajectory distance by per-alias sequence distances, then weighted aggregation.

For each store alias visible on both trajectories, computes the sequence-level distance using the configured SequenceMetric. The resulting per-alias distances are aggregated (weighted mean/sum) into a scalar trajectory distance.

Example:

hamming = HammingEntityMetric(entity_feature="status")
lp = LinearPairwiseSequenceMetric(entity_metric=hamming)

agg = AggregationTrajectoryMetric(
    default_metric=lp,
    agg_fun="mean",
    weights={"events": 1.0, "states": 0.5},
)

dist = agg(traj_a, traj_b)
dm   = agg.compute_matrix(traj_pool)

MEMMAP_SUPPORT: bool = True[source]#: Set to True in subclasses that implement disk-backed (memmap) computation.

SETTINGS_CLASS[source]#: alias of AggregationSettings

__init__(default_metric: SequenceMetric | str = 'linearpairwise', sequence_metrics: dict[str, SequenceMetric | str] | None = None, agg_fun: str = 'mean', weights: dict[str, float] | None = None, *, store_path: str | Path | None = None, chunk_size: int = 500, resume: bool = True, dtype: str = 'float32') → None[source]#

Module contents#

Aggregation trajectory metric sub-package.

class tanat.metric.trajectory.type.aggregation.AggregationSettings(*, default_metric: SequenceMetric = 'linearpairwise', sequence_metrics: dict[str, SequenceMetric] | None = None, agg_fun: str = 'mean', weights: dict[str, float] | None = None)[source]#