Convert numeric interaction timestamps

src/lenskit/data/_adapt.py

@@ -111,6 +111,7 @@ def from_interactions_df(
     users: IDSequence | pd.Index | Iterable[ID] | Vocabulary | None = None,
     items: IDSequence | pd.Index | Iterable[ID] | Vocabulary | None = None,
     class_name: str = "rating",
+    timestamp_unit: Literal["s", "ms", "us", "ns"] | None = None,
 ) -> Dataset:
     """
     Create a dataset from a data frame of ratings or other user-item
@@ -132,6 +133,9 @@ def from_interactions_df(
             The name of the rating column.
         timestamp_col:
             The name of the timestamp column.
+        timestamp_unit:
+            The unit of numeric timestamp values. If omitted, integer units are
+            inferred and floating-point values are assumed to be seconds.
         user_ids:
             A vocabulary of user IDs.  The data frame is subset to this set of IDs.
         item_ids:
@@ -176,6 +180,7 @@ def from_interactions_df(
         missing=missing,
         allow_repeats=False,
         default=True,
+        timestamp_unit=timestamp_unit,
     )
 
     return dsb.build()

src/lenskit/data/_builder.py

@@ -391,6 +391,7 @@ def add_relationships(
         interaction: bool | Literal["default"] = False,
         _warning_parent: int = 0,
         remove_repeats: bool | Literal["exact"] = False,
+        timestamp_unit: Literal["s", "ms", "us", "ns"] | None = None,
     ) -> None:
         """
         Add relationship records to the data set.
@@ -423,6 +424,10 @@ def add_relationships(
             remove_repeats:
                 If ``True``, repeated interactions will be removed. If ``"exact"``,
                 duplicated interactions will be removed.
+            timestamp_unit:
+                The unit of numeric values in the ``timestamp`` column. If omitted,
+                integer units are inferred and floating-point values are assumed to
+                be seconds. Numeric timestamps are converted to Arrow timestamps.
         """
         if isinstance(data, pd.DataFrame):
             table = pa.Table.from_pandas(data, preserve_index=False)
@@ -431,6 +436,8 @@ def add_relationships(
         else:
             table = data
 
+        table = _convert_relationship_timestamps(table, timestamp_unit)
+
         log = self._log.bind(class_name=cls, count=table.num_rows)
 
         rc_def = self.schema.relationships.get(cls, None)
@@ -570,6 +577,7 @@ def add_interactions(
         allow_repeats: bool = True,
         default: bool = False,
         remove_repeats: bool | Literal["exact"] = False,
+        timestamp_unit: Literal["s", "ms", "us", "ns"] | None = None,
     ) -> None:
         """
         Add a interaction records to the data set.
@@ -607,6 +615,10 @@ def add_interactions(
             remove_repeats:
                 If ``True``, repeated interactions will be removed. If ``"exact"``,
                 duplicated interactions will be removed.
+            timestamp_unit:
+                The unit of numeric values in the ``timestamp`` column. If omitted,
+                integer units are inferred and floating-point values are assumed to
+                be seconds. Numeric timestamps are converted to Arrow timestamps.
         """
         self.add_relationships(
             cls,
@@ -617,6 +629,7 @@ def add_interactions(
             interaction="default" if default else True,
             _warning_parent=1,
             remove_repeats=remove_repeats,
+            timestamp_unit=timestamp_unit,
         )
 
     def filter_interactions(
@@ -1228,6 +1241,85 @@ def _empty_rel_table(types: list[str]) -> pa.Table:
     return pa.table({num_col_name(t): pa.array([], pa.int32()) for t in types})
 
 
+_TIMESTAMP_UNITS: tuple[Literal["s", "ms", "us", "ns"], ...] = (
+    "s",
+    "ms",
+    "us",
+    "ns",
+)
+_TIMESTAMP_UNIT_FACTORS: dict[Literal["s", "ms", "us", "ns"], int] = {
+    "s": 1,
+    "ms": 1_000,
+    "us": 1_000_000,
+    "ns": 1_000_000_000,
+}
+_TIMESTAMP_MIN_SECONDS = int(dt.datetime.fromisoformat("1900-01-01T00:00:00+00:00").timestamp())
+_TIMESTAMP_MAX_SECONDS = int(dt.datetime.fromisoformat("2100-01-01T00:00:00+00:00").timestamp())
+
+
+def _convert_relationship_timestamps(
+    table: pa.Table,
+    timestamp_unit: Literal["s", "ms", "us", "ns"] | None,
+) -> pa.Table:
+    if timestamp_unit is not None and timestamp_unit not in _TIMESTAMP_UNITS:
+        raise ValueError(f"invalid timestamp unit {timestamp_unit!r}")
+
+    column_idx = table.schema.get_field_index("timestamp")
+    if column_idx < 0:
+        return table
+
+    timestamps = table.column(column_idx)
+    column_type = timestamps.type
+    if pa.types.is_timestamp(column_type):
+        return table
+
+    if pa.types.is_null(column_type):
+        converted = pa.nulls(len(timestamps), pa.timestamp(timestamp_unit or "s"))
+    elif pa.types.is_floating(column_type):
+        values = pc.if_else(
+            pc.is_nan(timestamps),
+            pa.scalar(None, column_type),
+            timestamps,
+        )
+        if timestamp_unit is None:
+            values = pc.multiply(values, 1_000)
+            timestamp_unit = "ms"
+        values = pc.cast(pc.round(values), pa.int64())
+        converted = pc.cast(values, pa.timestamp(timestamp_unit))
+    elif pa.types.is_integer(column_type):
+        timestamp_unit = timestamp_unit or _infer_timestamp_unit(timestamps)
+        converted = pc.cast(timestamps, pa.timestamp(timestamp_unit))
+    else:
+        return table
+
+    return table.set_column(column_idx, "timestamp", converted)
+
+
+def _infer_timestamp_unit(
+    timestamps: pa.ChunkedArray,
+) -> Literal["s", "ms", "us", "ns"]:
+    valid_count = len(timestamps) - timestamps.null_count
+    if valid_count == 0:
+        return "s"
+
+    # Loss of integer precision is harmless for this coarse range check, and
+    # permitting it lets us inspect microsecond and nanosecond epoch values.
+    values = pc.cast(timestamps, pa.float64(), safe=False)
+    for unit in _TIMESTAMP_UNITS:
+        factor = _TIMESTAMP_UNIT_FACTORS[unit]
+        in_range = pc.and_(
+            pc.greater_equal(values, float(_TIMESTAMP_MIN_SECONDS * factor)),
+            pc.less(values, float(_TIMESTAMP_MAX_SECONDS * factor)),
+        )
+        in_range_count = pc.sum(pc.fill_null(in_range, False)).as_py() or 0
+        if in_range_count / valid_count >= 0.95:
+            return unit
+
+    raise ValueError(
+        "could not infer timestamp unit because fewer than 95% of values fall between 1900 and 2099"
+    )
+
+
 def _conform_time(time: float | str | dt.datetime, col_type: pa.DataType):
     if isinstance(time, str):
         time = dt.datetime.fromisoformat(time)

src/lenskit/data/_relationships.py

@@ -630,6 +630,8 @@ def scipy(
             values = np.ones(nnz, dtype=np.float32)
         else:
             value_col = self._table.column(attribute)
+            if pa.types.is_timestamp(self._table.field(attribute).type):
+                value_col = value_col.cast(pa.timestamp("s")).cast(pa.int64())
             if value_col.null_count:
                 mask = value_col.is_valid()
                 values = value_col.filter(mask).to_numpy()

tests/data/test_builder_filter.py

@@ -72,14 +72,17 @@ def test_filter_ratings_min_time(
         ml_ratings = ml_ratings.assign(
             timestamp=(ml_ratings["timestamp"] - pd.Timestamp("1970-01-01")) // pd.Timedelta("1s")
         )
+        expected_timestamps = pd.to_datetime(ml_ratings["timestamp"], unit="s")
+    else:
+        expected_timestamps = ml_ratings["timestamp"]
     dsb.add_interactions(
         "rating", ml_ratings, entities=["user", "item"], missing="insert", default=True
     )
-    q = QueryDT.create("2001-01-01", q_fmt, ts_fmt)
+    q = QueryDT.create("2001-01-01", q_fmt, "timestamp")
     dsb.filter_interactions("rating", min_time=q.thresh)
     ds = dsb.build()
     assert ds.interactions().pandas()["timestamp"].min() >= q.compare
-    assert ds.interactions().pandas()["timestamp"].max() == ml_ratings["timestamp"].max()
+    assert ds.interactions().pandas()["timestamp"].max() == expected_timestamps.max()
 
 
 @mark.parametrize(
@@ -95,13 +98,16 @@ def test_filter_ratings_max_time(
         ml_ratings = ml_ratings.assign(
             timestamp=(ml_ratings["timestamp"] - pd.Timestamp("1970-01-01")) // pd.Timedelta("1s")
         )
+        expected_timestamps = pd.to_datetime(ml_ratings["timestamp"], unit="s")
+    else:
+        expected_timestamps = ml_ratings["timestamp"]
     dsb.add_interactions(
         "rating", ml_ratings, entities=["user", "item"], missing="insert", default=True
     )
-    q = QueryDT.create("2001-01-01", q_fmt, ts_fmt)
+    q = QueryDT.create("2001-01-01", q_fmt, "timestamp")
     dsb.filter_interactions("rating", max_time=q.thresh)
     ds = dsb.build()
-    assert ds.interactions().pandas()["timestamp"].min() == ml_ratings["timestamp"].min()
+    assert ds.interactions().pandas()["timestamp"].min() == expected_timestamps.min()
     assert ds.interactions().pandas()["timestamp"].max() < q.compare
 
 
@@ -118,11 +124,12 @@ def test_filter_ratings_min_max_time(
         ml_ratings = ml_ratings.assign(
             timestamp=(ml_ratings["timestamp"] - pd.Timestamp("1970-01-01")) // pd.Timedelta("1s")
         )
+
     dsb.add_interactions(
         "rating", ml_ratings, entities=["user", "item"], missing="insert", default=True
     )
-    q1 = QueryDT.create("2001-01-01", q_fmt, ts_fmt)
-    q2 = QueryDT.create("2004-01-01", q_fmt, ts_fmt)
+    q1 = QueryDT.create("2001-01-01", q_fmt, "timestamp")
+    q2 = QueryDT.create("2004-01-01", q_fmt, "timestamp")
     dsb.filter_interactions("rating", min_time=q1.thresh, max_time=q2.thresh)
     ds = dsb.build()
     assert ds.interactions().pandas()["timestamp"].min() >= q1.compare