Note
Go to the end to download the full example code.
Static Metrics: Compare sequences and trajectories#
This example illustrates how to define a static metrics, which computes distances between trajectories or sequences with static data (i.e. non-temporal data).
We show that
Setup#
import polars as pl
import matplotlib.pyplot as plt
from tanat import build_events, build_trajectories
from tanat.dataset import simulate_trajectories, simulate_static
from tanat.metric.static import StaticMetric
from tanat.metric.entity import HammingEntityMetric
from tanat.metric.sequence import EditSequenceMetric
from tanat.metric import AggregationTrajectoryMetric
Generate synthetic trajectory data#
Simulate some synthetic data for this example. We generate a sequence pool event_pool which contains events and static data for 10 individuas. Then, we also generate a trajectory pool with # the same data.
N_TRAJ = 10
SEED = 42
data = simulate_trajectories(
sequences={
"events": {"type": "event", "n_ids": N_TRAJ, "features": ["value", "category"]}
},
seed=SEED,
)
static_df = simulate_static(n_ids=N_TRAJ, features=["age", "group"], seed=0)
# Build pools for each sequence type
events_pool = build_events(
temporal_data=data["events"],
id_column="id",
time_column="time",
static_data=static_df,
)
events_pool.cast_features({"category": pl.Categorical})
# Build trajectory pool
traj_pool = build_trajectories(
pools={"events": events_pool}, static_data=static_df, id_column="id"
)
print(traj_pool)
┌─ Event SequenceStore
│
│ Step 1/4: Sorting & preparing data
│
│ Step 2/4: Building sequence index
│
│ Step 3/4: Writing entity, time index & static features
│
│ Step 4/4: Computing & writing metadata
│
└─ Done (10 sequences · 62 entities · 0.01s)
┌─ TrajectoryStore
│
│ Step 1/2: Linking pools: events
/home/runner/work/TanaT/TanaT/src/tanat/store/trajectory/builder.py:204: UserWarning: Pending changes detected in pool 'events' (cast overrides). Materialising a copy into stores/events/. Call pool.save() before build() to avoid data duplication on disk.
links = self._resolve_pool_links(resolved, self._pools, overwrite=exist_ok)
│ ┌─ Event SequenceStore
│ │
│ │ Step 1/4: Sorting & preparing data
│ │
│ │ Step 2/4: Building sequence index
│ │
│ │ Step 3/4: Writing entity, time index & static features
│ │
│ │ Step 4/4: Computing & writing metadata
│ │
│ └─ Done (10 sequences · 62 entities · 0.01s)
│
│ Step 2/2: Building trajectory index & metadata
│
└─ Done (10 trajectories · 1 pool(s) · 0.01s)
┌────────────────────────────────────────────────┐
│ TrajectoryPool Summary │
└────────────────────────────────────────────────┘
Overview
─────────────────────────
Trajectories 10
Store /home/runner/.tanat/_quick_trajectory_8619bc57
id_column id
Time Index
─────────────────────────
Type Datetime(time_unit='us', time_zone=None) [2000-01-16 05:20:09.572188 → 2024-05-29 00:55:21.221110]
t0 position=0, anchor=start
Sequences (1)
─────────────────────────
• events EventSequencePool(n=10, entity_features=2, static_features=2, store='/home/runner/.tanat/_quick_trajectory_8619bc57/stores/events')
Static Features (2)
─────────────────────────
• age Numerical [2 → 86]
• group String [len 1 → 1]
Default static metric#
The default static metric implements a kind of hamming distance stategy: it is 1 when at least one static attributes holds different values, and 0 if they are all similar.
defaut_static_metric = StaticMetric()
# we can now compare two sequences:
seqs_ids = events_pool.unique_ids
seq_a = events_pool[seqs_ids[0]]
seq_b = events_pool[seqs_ids[1]]
print("sequences:")
print(seq_a.static_data())
print(seq_b.static_data())
value = defaut_static_metric(seq_a, seq_b)
print(f"metric value between {seqs_ids[0]} and {seqs_ids[1]}: {value:.1f}.")
sequences:
id age group
0 1 86 D
id age group
0 2 64 E
metric value between 1 and 2: 1.0.
The default metric compares all attributes, only one that differ is necessary to consider the static data are different.
seq_c = events_pool[seqs_ids[3]]
print(seq_c.static_data())
value = defaut_static_metric(seq_a, seq_c)
print(f"metric value between {seqs_ids[0]} and {seqs_ids[3]}: {value:.1f}.")
id age group
0 4 27 D
metric value between 1 and 4: 1.0.
Nonetheless, similar data leads to a 0-valued metric.
value = defaut_static_metric(seq_a, seq_a)
print(f"metric value: {value:.1f}.")
metric value: 0.0.
Define a custom static metric#
Let us now imagine that we would like to compare only the age of the
individuals, but we do not care about the group.
We could drop the group static feature (see
drop_static_features())
but we would loose it for future analysis.
In this case, it is possible to define a custom static metric through the definition of a comparison function. This function must have two parameters that are dictionary with attributes names as keys.
In our example, we simply evalute the absolute difference between the ages.
def age_cmp(s1, s2):
"""Comparison of age static features"""
return abs(s1["age"] - s2["age"])
custom_static_metric = StaticMetric(cmp_fnct=age_cmp)
# Le us now compare the same sequences as before ...
seqs_ids = events_pool.unique_ids
seq_a = events_pool[seqs_ids[0]]
seq_b = events_pool[seqs_ids[1]]
seq_c = events_pool[seqs_ids[3]]
print("sequences:")
print(seq_a.static_data())
print(seq_b.static_data())
print(seq_c.static_data())
value_ab = custom_static_metric(seq_a, seq_b)
value_ac = custom_static_metric(seq_a, seq_c)
print(f"custom metric values: {value_ab:.1f}, {value_ac:.1f}.")
sequences:
id age group
0 1 86 D
id age group
0 2 64 E
id age group
0 4 27 D
custom metric values: 22.0, 59.0.
Compute static distance between pools#
It is also possible to compute the distance matrix based on
a pool of sequences (or trajectories) by simply calling the
compute_matrix() function.
mat = custom_static_metric.compute_matrix(events_pool)
print(mat)
[[ 0. 22. 34. 59. 55. 81. 78. 84. 68. 4.]
[22. 0. 12. 37. 33. 59. 56. 62. 46. 18.]
[34. 12. 0. 25. 21. 47. 44. 50. 34. 30.]
[59. 37. 25. 0. 4. 22. 19. 25. 9. 55.]
[55. 33. 21. 4. 0. 26. 23. 29. 13. 51.]
[81. 59. 47. 22. 26. 0. 3. 3. 13. 77.]
[78. 56. 44. 19. 23. 3. 0. 6. 10. 74.]
[84. 62. 50. 25. 29. 3. 6. 0. 16. 80.]
[68. 46. 34. 9. 13. 13. 10. 16. 0. 64.]
[ 4. 18. 30. 55. 51. 77. 74. 80. 64. 0.]]
Compute static distance between trajectories#
The same StaticMetric objects can
be used also for trajectories. You simply have to take care that
the static feature exists and that the custom comparison function
suits the data.
traj_ids = traj_pool.unique_ids
traj_a = traj_pool[traj_ids[0]]
traj_b = traj_pool[traj_ids[1]]
dist = custom_static_metric(traj_a, traj_b)
print(f"\nDistance between {traj_ids[0]} and {traj_ids[1]}: {dist:.1f}")
# It is also possible to use the default static metric and to compute
# a distance matrix in a similar way as for sequences.
Distance between 1 and 2: 22.0
Use static metric in AggregationMetric#
A StaticMetric can also be used in
an AggregationMetric. This trajectory
metric defines a distance as an aggregate of sequence metrics.
In addition, it is possible to define a static metric as an additional
metric to aggregate (with it own weight).
Let us define a trajectory metric that involves the static data and for that, we first need a sequence metric …
seq_metric = EditSequenceMetric(entity_metric=HammingEntityMetric("category"))
# then the trajectory metric is defined by the sequence metric required
# for the `events` type, and we add a `static_metric` with a very small weight
# (the comparison of ages leads to values between 0 and 120).
traj_metric = AggregationTrajectoryMetric(
sequence_metrics={"events": seq_metric},
static_metric=custom_static_metric,
static_metric_weight=0.01,
)
traj_ids = traj_pool.unique_ids
traj_a = traj_pool[traj_ids[0]]
traj_b = traj_pool[traj_ids[1]]
dist = traj_metric(traj_a, traj_b)
print(f"\nDistance between {traj_ids[0]} and {traj_ids[1]}: {dist:.1f}")
#
# warning ::
# It is possible to use static metric only in a trajectory metric but
# not directly with a `SequenceMetric`.
#
Distance between 1 and 2: 8.1
Total running time of the script: (0 minutes 0.387 seconds)