### Overview

When building data science pipelines, reproducible train/validation/test splits are essential for debugging, model comparison, and audit trails. A common pitfall is relying on implicit randomness (or time-based seeds) that can change results between runs, machines, or workers.

This post covers a **deterministic shuffle** approach: compute a stable pseudo-random key per row from an identifier, then sort by that key to obtain a reproducible permutation. This pattern is frequently useful in privacy-aware workflows because it avoids directly exposing ordering-sensitive information and still produces stable splits.

### Why deterministic shuffling matters

Many datasets arrive with an inherent ordering (ingestion time, source order, user ID grouping). If you split by contiguous blocks, the resulting subsets can be biased. Deterministic shuffling mitigates this by producing a consistent randomized order using only stable inputs.

A robust method is:

1. For each row, compute a stable hash of `salt + row_id`.
2. Convert that hash to a numeric key.
3. Sort by the numeric key to produce a deterministic shuffle.

### Practical implementation in Python

The following example uses **SHA-256** for a stable hash across runs and platforms. Only `salt` and `row_id` affect the permutation.

#### Deterministic shuffle and split

```python
import hashlib
from dataclasses import dataclass
from typing import Iterable, List, Tuple

@dataclass(frozen=True)
class Row:
    row_id: str
    value: float

def stable_shuffle_indices(row_ids: Iterable[str], salt: str) -> List[int]:
    """
    Returns a deterministic permutation of indices based on stable hashing.
    """
    ids = list(row_ids)
    keys: List[Tuple[int, int]] = []  # (numeric_key, original_index)

    for i, rid in enumerate(ids):
        digest = hashlib.sha256((salt + "|" + rid).encode("utf-8")).digest()
        # Use first 8 bytes for a 64-bit unsigned integer key
        numeric_key = int.from_bytes(digest[:8], byteorder="big", signed=False)
        keys.append((numeric_key, i))

    keys.sort(key=lambda t: t[0])
    return [i for _, i in keys]

def deterministic_split(rows: List[Row], salt: str, train_frac: float = 0.8):
    """
    Splits rows into train/validation deterministically using stable shuffle.
    """
    if not (0.0 < train_frac < 1.0):
        raise ValueError("train_frac must be between 0 and 1.")

    indices = stable_shuffle_indices((r.row_id for r in rows), salt=salt)
    cutoff = int(round(train_frac * len(rows)))

    train_idx = set(indices[:cutoff])
    train = [rows[i] for i in range(len(rows)) if i in train_idx]
    val = [rows[i] for i in range(len(rows)) if i not in train_idx]
    return train, val

if __name__ == "__main__":
    rows = [Row(row_id=f"user_{i:03d}", value=float(i)) for i in range(20)]
    train_a, val_a = deterministic_split(rows, salt="dataset:v1", train_frac=0.75)
    train_b, val_b = deterministic_split(rows, salt="dataset:v1", train_frac=0.75)

    assert [r.row_id for r in train_a] == [r.row_id for r in train_b]
    assert [r.row_id for r in val_a] == [r.row_id for r in val_b]

    print("Train:", [r.row_id for r in train_a])
    print("Val:", [r.row_id for r in val_a])
```

### Correctness properties

This approach has several important properties:

- **Reproducibility across runs**: SHA-256 is deterministic; using `digest[:8]` yields a consistent 64-bit key.
- **Platform consistency**: The hash input is explicitly encoded as UTF-8 and converted using explicit byte order.
- **Deterministic order**: Sorting by `(numeric_key)` yields a stable permutation for a fixed salt and row IDs.

Note: using only the first 64 bits is typically safe for practical dataset sizes, but collisions are theoretically possible. SHA-256 collisions are computationally infeasible; truncation collisions are still extremely unlikely for typical row counts.

### Extending to three-way splits

For train/validation/test, create cumulative cutoffs after the deterministic permutation. The partition fractions can be chosen to match the experiment design.

#### Train/validation/test split

```python
import hashlib
from typing import List, Tuple

def stable_shuffle_indices(row_ids: List[str], salt: str) -> List[int]:
    keys = []
    for i, rid in enumerate(row_ids):
        digest = hashlib.sha256((salt + "|" + rid).encode("utf-8")).digest()
        numeric_key = int.from_bytes(digest[:8], "big", signed=False)
        keys.append((numeric_key, i))
    keys.sort(key=lambda x: x[0])
    return [i for _, i in keys]

def deterministic_triple_split(row_ids: List[str], salt: str, train: float, val: float, test: float):
    total = train + val + test
    if abs(total - 1.0) > 1e-12:
        raise ValueError("Fractions must sum to 1.0")

    if not (0.0 < train < 1.0 and 0.0 < val < 1.0 and 0.0 < test < 1.0):
        raise ValueError("Each fraction must be between 0 and 1")

    indices = stable_shuffle_indices(row_ids, salt=salt)
    n = len(row_ids)
    i1 = int(round(train * n))
    i2 = int(round((train + val) * n))

    train_idx = indices[:i1]
    val_idx = indices[i1:i2]
    test_idx = indices[i2:]
    return train_idx, val_idx, test_idx

if __name__ == "__main__":
    ids = [f"id_{i}" for i in range(30)]
    a1, a2, a3 = deterministic_triple_split(ids, salt="split:v3", train=0.7, val=0.15, test=0.15)
    b1, b2, b3 = deterministic_triple_split(ids, salt="split:v3", train=0.7, val=0.15, test=0.15)

    assert a1 == b1 and a2 == b2 and a3 == b3
    print("Train size:", len(a1), "Val size:", len(a2), "Test size:", len(a3))
```

### Implementation guidance for data pipelines

A deterministic shuffle should be applied at the **row level** based on an immutable identifier. Common identifier choices include:

- user ID
- document ID
- primary key from the source database
- composite key encoded as a single stable string

The `salt` should encode the split strategy version (and optionally the task name). This makes it easy to reproduce exactly the same permutation later and prevents accidental reuse across different experimental settings.

### Conclusion

Deterministic shuffling via stable hashing provides a simple, reproducible, and audit-friendly way to create privacy-respecting dataset splits. By deriving permutation keys from `salt + row_id` and sorting those keys, the split remains consistent across runs, machines, and worker processes, enabling reliable experimentation and dependable evaluation.

Deterministic Shuffle For Privacy-Preserving Dataset Splits

### Overview

A surprisingly practical “data science adjacent” workflow is using **PostgreSQL logical replication** to move *event streams* (like payment attempts) from a source database into a feature store or analytics database in near real time. You can then score fraud risk using features computed on arrival.

Below is a compact, **perfectly working** example that:

- Creates a table for payment events
- Sets up **logical replication** (publication/subscription)
- Replicates inserts to a consumer database
- Computes a simple, factual risk score from replicated events

This is not “ML training”; it’s **real-time scoring** using event features—often the first step before building full models.

---

### What logical replication gives you

Logical replication streams **row changes** (e.g., INSERT/UPDATE/DELETE) from a publisher to one or more subscribers. For event-driven fraud detection, this lets your analytics pipeline ingest new events quickly without polling.

---

### Prerequisites

You need:

- Two PostgreSQL instances (or two databases) running:
  - Publisher: `postgres://publisher_user@localhost:5432/publisher_db`
  - Subscriber: `postgres://subscriber_user@localhost:5432/subscriber_db`
- Replication permissions and configuration:
  - On publisher: `wal_level = logical`
  - Restart PostgreSQL after changing config
  - Ensure `pg_hba.conf` allows subscriber connection

> The example below uses **two databases on the same host** for clarity. In production you’d use separate servers.

---

### Step 1: Create a payments table on the publisher

Run this on the **publisher** database.

```sql
-- publisher.sql
CREATE TABLE IF NOT EXISTS payments (
  id              BIGSERIAL PRIMARY KEY,
  user_id        BIGINT NOT NULL,
  amount_cents   BIGINT NOT NULL CHECK (amount_cents > 0),
  country_code   TEXT NOT NULL,
  created_at     TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Helpful indexes for consumer queries
CREATE INDEX IF NOT EXISTS payments_created_at_idx ON payments (created_at);
```

Insert a couple of test events to verify replication later:

```sql
-- publisher_seed.sql
INSERT INTO payments (user_id, amount_cents, country_code)
VALUES
  (101, 1299, 'US'),
  (202, 25000, 'NG');
```

---

### Step 2: Enable logical replication (publisher)

Still on the **publisher**, configure replication artifacts.

```sql
-- publisher_replication.sql
-- Create a replication user
DO $$
BEGIN
  IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'repl_user') THEN
    CREATE ROLE repl_user WITH LOGIN PASSWORD 'repl_password' REPLICATION;
  END IF;
END $$;

-- Create publication for the payments table
CREATE PUBLICATION payments_pub FOR TABLE payments;
```

> In real deployments, you’ll also configure `pg_hba.conf` for `repl_user` to connect from the subscriber host.

---

### Step 3: Create the same table on the subscriber

On the **subscriber**, create the table structure. Logical replication will replicate row changes, but the table must exist.

```sql
-- subscriber.sql
CREATE TABLE IF NOT EXISTS payments (
  id              BIGINT PRIMARY KEY,
  user_id        BIGINT NOT NULL,
  amount_cents   BIGINT NOT NULL,
  country_code   TEXT NOT NULL,
  created_at     TIMESTAMPTZ NOT NULL
);

CREATE INDEX IF NOT EXISTS payments_created_at_idx ON payments (created_at);
```

---

### Step 4: Create a subscription (subscriber)

On the **subscriber**, create the subscription pointing to the publisher.

```sql
-- subscriber_subscription.sql
-- Create subscription to publisher
CREATE SUBSCRIPTION payments_sub
CONNECTION 'host=localhost port=5432 dbname=publisher_db user=repl_user password=repl_password'
PUBLICATION payments_pub
WITH (copy_data = true);
```

- `copy_data = true` performs an initial snapshot then continues streaming changes.
- If you already created a subscription once, you can drop it first:
  ```sql
  DROP SUBSCRIPTION IF EXISTS payments_sub;
  ```

---

### Step 5: Verify replicated data

On the **subscriber**, check that rows arrived:

```sql
-- subscriber_verify.sql
SELECT id, user_id, amount_cents, country_code, created_at
FROM payments
ORDER BY id
LIMIT 10;
```

Now insert a new payment on the publisher and re-check on the subscriber.

```sql
-- publisher_insert_more.sql
INSERT INTO payments (user_id, amount_cents, country_code)
VALUES (303, 9800, 'US');
```

---

### Step 6: Compute a simple, reproducible fraud risk score (SQL)

Here’s a factual, deterministic scoring heuristic you can run on replicated events. For example:

- Base risk starts at 0
- +40 if amount is unusually large (e.g., ≥ $200)
- +20 if country is not US (toy example)
- +10 if the payment is within the last 5 minutes
- Cap at 100

Run this on the **subscriber**:

```sql
-- subscriber_scoring.sql
WITH scored AS (
  SELECT
    *,
    LEAST(
      100,
      0
      + CASE WHEN amount_cents >= 20000 THEN 40 ELSE 0 END
      + CASE WHEN country_code <> 'US' THEN 20 ELSE 0 END
      + CASE WHEN created_at >= now() - interval '5 minutes' THEN 10 ELSE 0 END
    ) AS risk_score
  FROM payments
)
SELECT
  id,
  user_id,
  amount_cents,
  country_code,
  created_at,
  risk_score,
  CASE
    WHEN risk_score >= 70 THEN 'high'
    WHEN risk_score >= 40 THEN 'medium'
    ELSE 'low'
  END AS risk_band
FROM scored
ORDER BY created_at DESC
LIMIT 20;
```

This produces a *working* “scoring” query you can wire into alerts, dashboards, or downstream model inference.

---

### Optional: Stream into a Python risk reporter

If you want to poll replicated rows and print risk decisions, the simplest approach is querying the subscriber periodically. The code below uses `psycopg` (v3).

```python
# risk_reporter.py
import time
import psycopg

DSN = "host=localhost port=5432 dbname=subscriber_db user=subscriber_user password=subscriber_password"

QUERY = """
WITH scored AS (
  SELECT
    *,
    LEAST(
      100,
      0
      + CASE WHEN amount_cents >= 20000 THEN 40 ELSE 0 END
      + CASE WHEN country_code <> 'US' THEN 20 ELSE 0 END
      + CASE WHEN created_at >= now() - interval '5 minutes' THEN 10 ELSE 0 END
    ) AS risk_score
  FROM payments
)
SELECT id, user_id, amount_cents, country_code, created_at, risk_score,
       CASE
         WHEN risk_score >= 70 THEN 'high'
         WHEN risk_score >= 40 THEN 'medium'
         ELSE 'low'
       END AS risk_band
FROM scored
WHERE id > %s
ORDER BY id
LIMIT 50;
"""

def main():
  last_seen_id = 0

  with psycopg.connect(DSN, autocommit=True) as conn:
    while True:
      with conn.cursor() as cur:
        cur.execute(QUERY, (last_seen_id,))
        rows = cur.fetchall()

      for (pid, user_id, amount_cents, country, created_at, risk_score, band) in rows:
        print(f"[{created_at}] payment_id={pid} user={user_id} amount_cents={amount_cents} "
              f"country={country} risk_score={risk_score} band={band}")
        last_seen_id = max(last_seen_id, pid)

      time.sleep(2)

if __name__ == "__main__":
  main()
```

---

### When this pattern is useful in data science

- **Feature freshness**: replication keeps features close to event time.
- **Reproducibility**: scoring queries are SQL-defined and auditable.
- **Model integration**: once you have a scoring baseline, you can replace/augment it with a trained model.

---

### Summary

Using **PostgreSQL logical replication**, you can stream payment events into a subscriber database, then compute deterministic fraud risk signals immediately using SQL (or Python). This approach is often a strong foundation for more advanced ML pipelines that require low-latency features.

Data Science Posts

Deterministic Shuffle For Privacy-Preserving Dataset Splits

Real Time Fraud Signals With Postgresql Logical Replication