Skip to content

features/dynamics

Temporal dynamics features: rolling means over 5/15-minute windows and inter-bucket deltas for interaction metrics. DynamicsTracker is implemented as a slotted dataclass with the same constructor arguments (rolling_5, rolling_15).

taskclf.features.dynamics

Temporal dynamics: rolling means and inter-bucket deltas.

These features capture how interaction patterns change over time, which is a strong signal for distinguishing tasks that look similar in a single bucket but differ in trajectory (e.g. sustained coding vs. switching between reading and chatting).

All functions are pure — they take a history buffer and return a single value (or None when insufficient history is available).

DynamicsTracker dataclass

Stateful tracker that accumulates per-bucket metrics and emits dynamics features.

Maintains rolling buffers for keys_per_min, clicks_per_min, and mouse_distance so that each call to :meth:update returns the computed rolling means and deltas for the latest bucket.

For batch mode, call :meth:compute_batch instead.

Source code in src/taskclf/features/dynamics.py 
 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
142
143
144
145
146
147
148
149
150
151
152
@dataclass(eq=False)
class DynamicsTracker:
    """Stateful tracker that accumulates per-bucket metrics and emits dynamics features.

    Maintains rolling buffers for ``keys_per_min``, ``clicks_per_min``,
    and ``mouse_distance`` so that each call to :meth:`update` returns
    the computed rolling means and deltas for the latest bucket.

    For batch mode, call :meth:`compute_batch` instead.
    """

    rolling_5: int = 5
    rolling_15: int = 15
    _keys_buf: deque[float | None] = field(init=False)
    _clicks_buf: deque[float | None] = field(init=False)
    _mouse_buf: deque[float | None] = field(init=False)
    _prev_keys: float | None = field(init=False, default=None)
    _prev_clicks: float | None = field(init=False, default=None)
    _prev_mouse: float | None = field(init=False, default=None)

    def __post_init__(self) -> None:
        max_len = max(self.rolling_5, self.rolling_15)
        self._keys_buf = deque(maxlen=max_len)
        self._clicks_buf = deque(maxlen=max_len)
        self._mouse_buf = deque(maxlen=max_len)

    def update(
        self,
        keys_per_min: float | None,
        clicks_per_min: float | None,
        mouse_distance: float | None,
    ) -> dict[str, float | None]:
        """Record one bucket's metrics and return dynamics features.

        Args:
            keys_per_min: Current bucket's keys_per_min (may be None).
            clicks_per_min: Current bucket's clicks_per_min (may be None).
            mouse_distance: Current bucket's mouse_distance (may be None).

        Returns:
            Dict with keys matching the FeatureRow field names.
        """
        self._keys_buf.append(keys_per_min)
        self._clicks_buf.append(clicks_per_min)
        self._mouse_buf.append(mouse_distance)

        result = {
            "keys_per_min_rolling_5": rolling_mean(self._keys_buf, self.rolling_5),
            "keys_per_min_rolling_15": rolling_mean(self._keys_buf, self.rolling_15),
            "mouse_distance_rolling_5": rolling_mean(self._mouse_buf, self.rolling_5),
            "mouse_distance_rolling_15": rolling_mean(self._mouse_buf, self.rolling_15),
            "keys_per_min_delta": delta_from_previous(keys_per_min, self._prev_keys),
            "clicks_per_min_delta": delta_from_previous(
                clicks_per_min, self._prev_clicks
            ),
            "mouse_distance_delta": delta_from_previous(
                mouse_distance, self._prev_mouse
            ),
        }

        self._prev_keys = keys_per_min
        self._prev_clicks = clicks_per_min
        self._prev_mouse = mouse_distance

        return result

    @staticmethod
    def compute_batch(
        keys_series: Sequence[float | None],
        clicks_series: Sequence[float | None],
        mouse_series: Sequence[float | None],
        *,
        rolling_5: int = 5,
        rolling_15: int = 15,
    ) -> list[dict[str, float | None]]:
        """Compute dynamics features for an entire ordered sequence of buckets.

        Args:
            keys_series: keys_per_min values, one per bucket.
            clicks_series: clicks_per_min values, one per bucket.

            mouse_series: mouse_distance values, one per bucket.
            rolling_5: Short rolling window size.
            rolling_15: Long rolling window size.

        Returns:
            List of dicts (one per bucket) with dynamics feature values.
        """
        tracker = DynamicsTracker(rolling_5=rolling_5, rolling_15=rolling_15)
        return [
            tracker.update(k, c, m)
            for k, c, m in zip(keys_series, clicks_series, mouse_series)
        ]

update(keys_per_min, clicks_per_min, mouse_distance)

Record one bucket's metrics and return dynamics features.

Parameters:

Name Type Description Default
keys_per_min float | None

Current bucket's keys_per_min (may be None).

 required
clicks_per_min float | None

Current bucket's clicks_per_min (may be None).

 required
mouse_distance float | None

Current bucket's mouse_distance (may be None).

 required

Returns:

Type Description
dict[str, float | None]

Dict with keys matching the FeatureRow field names.
Source code in src/taskclf/features/dynamics.py 
 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
def update(
    self,
    keys_per_min: float | None,
    clicks_per_min: float | None,
    mouse_distance: float | None,
) -> dict[str, float | None]:
    """Record one bucket's metrics and return dynamics features.

    Args:
        keys_per_min: Current bucket's keys_per_min (may be None).
        clicks_per_min: Current bucket's clicks_per_min (may be None).
        mouse_distance: Current bucket's mouse_distance (may be None).

    Returns:
        Dict with keys matching the FeatureRow field names.
    """
    self._keys_buf.append(keys_per_min)
    self._clicks_buf.append(clicks_per_min)
    self._mouse_buf.append(mouse_distance)

    result = {
        "keys_per_min_rolling_5": rolling_mean(self._keys_buf, self.rolling_5),
        "keys_per_min_rolling_15": rolling_mean(self._keys_buf, self.rolling_15),
        "mouse_distance_rolling_5": rolling_mean(self._mouse_buf, self.rolling_5),
        "mouse_distance_rolling_15": rolling_mean(self._mouse_buf, self.rolling_15),
        "keys_per_min_delta": delta_from_previous(keys_per_min, self._prev_keys),
        "clicks_per_min_delta": delta_from_previous(
            clicks_per_min, self._prev_clicks
        ),
        "mouse_distance_delta": delta_from_previous(
            mouse_distance, self._prev_mouse
        ),
    }

    self._prev_keys = keys_per_min
    self._prev_clicks = clicks_per_min
    self._prev_mouse = mouse_distance

    return result

compute_batch(keys_series, clicks_series, mouse_series, *, rolling_5=5, rolling_15=15) staticmethod

Compute dynamics features for an entire ordered sequence of buckets.

Parameters:

Name Type Description Default
keys_series Sequence[float | None]

keys_per_min values, one per bucket.

 required
clicks_series Sequence[float | None]

clicks_per_min values, one per bucket.

 required
mouse_series Sequence[float | None]

mouse_distance values, one per bucket.

 required
rolling_5 int

Short rolling window size.

 5
rolling_15 int

Long rolling window size.

 15

Returns:

Type Description
list[dict[str, float | None]]

List of dicts (one per bucket) with dynamics feature values.
Source code in src/taskclf/features/dynamics.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@staticmethod
def compute_batch(
    keys_series: Sequence[float | None],
    clicks_series: Sequence[float | None],
    mouse_series: Sequence[float | None],
    *,
    rolling_5: int = 5,
    rolling_15: int = 15,
) -> list[dict[str, float | None]]:
    """Compute dynamics features for an entire ordered sequence of buckets.

    Args:
        keys_series: keys_per_min values, one per bucket.
        clicks_series: clicks_per_min values, one per bucket.

        mouse_series: mouse_distance values, one per bucket.
        rolling_5: Short rolling window size.
        rolling_15: Long rolling window size.

    Returns:
        List of dicts (one per bucket) with dynamics feature values.
    """
    tracker = DynamicsTracker(rolling_5=rolling_5, rolling_15=rolling_15)
    return [
        tracker.update(k, c, m)
        for k, c, m in zip(keys_series, clicks_series, mouse_series)
    ]

rolling_mean(history, window)

Compute the mean of the last window non-None values in history.

Returns None when there are zero non-None values in the window.

Parameters:

Name Type Description Default
history deque[float | None]

Ordered buffer of recent metric values (newest last).

 required
window int

Number of recent entries to consider.

 required

Returns:

Type Description
float | None

Mean of available values, or None.
Source code in src/taskclf/features/dynamics.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def rolling_mean(
    history: deque[float | None],
    window: int,
) -> float | None:
    """Compute the mean of the last *window* non-None values in *history*.

    Returns ``None`` when there are zero non-None values in the window.

    Args:
        history: Ordered buffer of recent metric values (newest last).
        window: Number of recent entries to consider.

    Returns:
        Mean of available values, or ``None``.
    """
    values = [v for v in list(history)[-window:] if v is not None]
    if not values:
        return None
    return round(sum(values) / len(values), 4)

delta_from_previous(current, previous)

Compute the change from previous to current.

Returns None when either value is None.

Parameters:

Name Type Description Default
current float | None

Current bucket's metric value.

 required
previous float | None

Previous bucket's metric value.

 required

Returns:

Type Description
float | None

current - previous, or None.
Source code in src/taskclf/features/dynamics.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def delta_from_previous(
    current: float | None,
    previous: float | None,
) -> float | None:
    """Compute the change from *previous* to *current*.

    Returns ``None`` when either value is ``None``.

    Args:
        current: Current bucket's metric value.
        previous: Previous bucket's metric value.

    Returns:
        ``current - previous``, or ``None``.
    """
    if current is None or previous is None:
        return None
    return round(current - previous, 4)