diff --git a/sfm/database.py b/sfm/database.py
index 90f54ea..c1f596c 100644
--- a/sfm/database.py
+++ b/sfm/database.py
@@ -603,6 +603,34 @@ class SeismoDb:
             ).fetchall()
         return [dict(r) for r in rows]
 
+    def find_twins(self, event_id: str, *, window_seconds: int = 300) -> list[dict]:
+        """
+        Find this event's histogram/waveform twin(s): rows sharing the same
+        serial and an identical peak_vector_sum, whose timestamp falls
+        within ``window_seconds`` of this event's timestamp.  Excludes the
+        event itself.  Returns [] if the event or any required field
+        (serial / peak_vector_sum / timestamp) is missing.
+        """
+        row = self.get_event(event_id)
+        if not row:
+            return []
+        serial = row.get("serial"); pvs = row.get("peak_vector_sum"); ts = row.get("timestamp")
+        if serial is None or pvs is None or not ts:
+            return []
+        try:
+            t = datetime.datetime.fromisoformat(ts.replace(" ", "T"))
+        except ValueError:
+            return []
+        lo = (t - datetime.timedelta(seconds=window_seconds)).isoformat()
+        hi = (t + datetime.timedelta(seconds=window_seconds)).isoformat()
+        with self._connect() as conn:
+            rows = conn.execute(
+                "SELECT * FROM events WHERE serial=? AND id!=? AND peak_vector_sum=? "
+                "AND timestamp BETWEEN ? AND ?",
+                (serial, event_id, pvs, lo, hi),
+            ).fetchall()
+        return [dict(r) for r in rows]
+
     def set_false_trigger(self, event_id: str, value: bool) -> bool:
         """Set or clear the false_trigger flag on an event. Returns True if found."""
         with self._connect() as conn:
diff --git a/tests/test_find_twins.py b/tests/test_find_twins.py
new file mode 100644
index 0000000..d0b356c
--- /dev/null
+++ b/tests/test_find_twins.py
@@ -0,0 +1,33 @@
+import datetime
+from sfm.database import SeismoDb
+from minimateplus.models import Event, Timestamp
+
+
+def _ins(db, key, serial, pvs, ts):
+    ev = Event(index=0)
+    ev._waveform_key = bytes.fromhex(key)
+    ev.timestamp = ts
+    # peak_vector_sum comes from peak_values; simplest: insert then UPDATE pvs directly
+    db.insert_events([ev], serial=serial)
+    row = [r for r in db.query_events(serial=serial) if r["waveform_key"] == key][0]
+    import sqlite3
+    with sqlite3.connect(db.db_path) as c:
+        c.execute("UPDATE events SET peak_vector_sum=? WHERE id=?", (pvs, row["id"]))
+    return row["id"]
+
+
+def test_find_twins_matches_same_serial_pvs_near_time(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    base = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=5)
+    twin = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=45)
+    far = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=21, minute=0, second=0)
+    # d needs a timestamp distinct from `twin` (UNIQUE(serial, timestamp) would
+    # otherwise collide with b and UPSERT onto its row instead of inserting a
+    # new one) while staying near `base` in time.
+    near = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=44)
+    a = _ins(db, "01110001", "BE1", 0.4763, base)
+    b = _ins(db, "01110002", "BE1", 0.4763, twin)   # twin: same serial+pvs, 40s apart
+    c = _ins(db, "01110003", "BE1", 0.4763, far)    # same pvs but >window away
+    d = _ins(db, "01110004", "BE1", 0.9999, near)   # near time but different pvs
+    ids = {r["id"] for r in db.find_twins(a, window_seconds=300)}
+    assert ids == {b}