Demo 1: Anomaly Detection using Passersby
This notebook shows the use of one of Tracktable’s built-in applications: simple anomaly detection.
Here, we define an anomaly as a trajectory that does not have any other trajectories nearby. We sample the trajectories at 4 evenly spaced points each (the num_control_points argument to anomaly_detection()) and search for passers-by.
In [1]:
from tracktable.render.render_trajectories import render_trajectories
from tracktable.applications.anomaly_detection import anomaly_detection
import tracktable.examples.tutorials.tutorial_helper as tutorial
Import Historical and Test Datasets
For this demo, we will use one week of maritime traffic in NY Harbor in December 2020\(^1\) as our historical dataset, and the following day of maritime traffic in NY Harbor \(^2\) for our test trajectory dataset.
In [2]:
historical_trajectories = tutorial.get_trajectory_list('anomaly-historical')
test_trajectories = tutorial.get_trajectory_list('anomaly-test')
Loading Trajectories: 513 trajectory [00:00, 1781.08 trajectory/s]
[2025-06-11 12:41:20.512333] [0x00000001f1aadf00] [info] Read a total of 513 trajectories.
Loading Trajectories: 38 trajectory [00:00, 2694.84 trajectory/s]
[2025-06-11 12:41:20.538751] [0x00000001f1aadf00] [info] Read a total of 38 trajectories.
Detect Anomalies
We send the anomaly detection function our historical and test trajectories, and for each test trajectory it detects how many historical trajectories passed by all num_control_points points equally-spaced along the trajectory (defaulted to 4), within a radius of nearness_radius kilometers (defaulted to 5km) of each point.
In [3]:
%%time
nearby_trajectory_indices = anomaly_detection(test_trajectories,
historical_trajectories=historical_trajectories,
nearness_radius=1,
num_control_points=4,
filename="test.log")
100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 513/513 [00:00<00:00, 1676.21it/s]
100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 38/38 [00:00<00:00, 497.16it/s]
CPU times: user 665 ms, sys: 48.5 ms, total: 713 ms
Wall time: 710 ms
Store our anomaly score as a trajectory property.
We can create new properties to be stored with a given trajectory. Let’s store the number of passersby from the historical trajectories as an “anomaly score”, where 0 is the most anomalous, as this means no historical trajectories passed along the same route.
In [4]:
for i, traj in enumerate(test_trajectories):
traj.set_property('anomalous_score', len(nearby_trajectory_indices[i]))
Sort the trajectories by their anomaly score.
In [5]:
sorted_trajectories = sorted(enumerate(test_trajectories), key=lambda x: x[1].properties['anomalous_score'])
Visualize Anomalies
Visualize all anomalous trajectories
You can scroll and zoom to examine each anomalous trajectory more closely.
In [6]:
render_trajectories([trajectory for trajectory in test_trajectories if trajectory.properties['anomalous_score'] == 0])
Out[6]:
Visualize the i-th most anomalous trajectory, where i = 0 is the most and i = -1 is the least.
In [7]:
# Change i to change the trajectory being visualized in the following cells.
i = -1
Render the trajectory alone.
In [8]:
canvas = render_trajectories(sorted_trajectories[i][1], show=True)
Render again with historical passersby.
Create a list of the historical trajectories that passed by the same route as this trajectory.
In [9]:
passersby = []
for index in nearby_trajectory_indices[sorted_trajectories[i][0]]:
passersby.append(historical_trajectories[index])
How many historical passersby does this trajectory have?
In [10]:
len(passersby)
Out[10]:
131
Let’s render the first twenty alongside the i-th most anomalous test trajectory.
In [11]:
render_trajectories(passersby[:20], canvas=canvas)
Out[11]: