transactions

name	transactions
description	Use for tasks over transactional / event data: aggregating raw event logs into entity-level features (RFM, time-window stats), per-entity sequence representations (CoLES embeddings or hand-crafted lag features), double-entry ledger reconciliation and balance audits, point-of-sale category-share / basket analysis, and per-transaction anomaly / fraud scoring.
metadata	{"dependencies":["pandas","numpy","scikit-learn"],"optional_dependencies":["pytorch-lifestream"]}

Transactions Dispatch

Transactions = sequential, financial, or behavioural event data. Five distinct shapes show up; the right Python is determined by which shape the data is in, not by what the downstream model is. Identify the shape first, then pick a path.

Identify the shape

Input you see	Path
Long event log: entity_id, ts, amount, [type] — many rows per entity, no fixed schema per entity	§A Aggregate to entity features
Per-entity sequences of events; need a fixed-length vector per entity	§B Sequence representation
Two-sided postings: (account, debit, credit, ts) or (from, to, amount) — must balance	§C Ledger reconciliation
POS / basket rows with category or sku per line	§D Category / basket analysis
Stream of transactions, label is_fraud (or unlabeled), score per row	§E Anomaly / fraud scoring

Cross-cutting: parse ts to datetime64[ns], sort by (entity, ts) before any windowing, never shuffle in time-aware paths, and persist artifacts at the exact paths instruction.md names.

§A — Aggregate to entity features (RFM + time windows)

Take this when the task asks for one row per entity (user, account, card) with engineered features. Default for tabular models downstream.

import pandas as pd
import numpy as np

events = pd.read_csv("events.csv", parse_dates=["ts"]).sort_values(["entity_id", "ts"])
ref = events["ts"].max()  # snapshot date

agg = events.groupby("entity_id").agg(
    recency_days=("ts", lambda s: (ref - s.max()).days),
    frequency=("amount", "size"),
    monetary_sum=("amount", "sum"),
    monetary_mean=("amount", "mean"),
    amount_std=("amount", "std"),
    active_days=("ts", lambda s: s.dt.normalize().nunique()),
).fillna(0.0)

# Rolling windows: counts and sums in last 7 / 30 / 90 days
for w in (7, 30, 90):
    cutoff = ref - pd.Timedelta(days=w)
    win = events[events["ts"] >= cutoff].groupby("entity_id")["amount"]
    agg[f"cnt_{w}d"] = win.size()
    agg[f"sum_{w}d"] = win.sum()

agg = agg.fillna(0.0).reset_index()
agg.to_parquet("entity_features.parquet", index=False)

Notes: keep entity_id as a column on output (downstream joins break without it); recency measured in days from a fixed reference, not from now(); std is NaN for one-event entities — fill with 0.

§B — Sequence representation

Take this when each entity has an ordered list of events and the task wants a fixed-size embedding or explicit lag features.

Hand-crafted lag features (always available, no extra deps):

import pandas as pd

events = pd.read_csv("events.csv", parse_dates=["ts"]).sort_values(["entity_id", "ts"])
g = events.groupby("entity_id")
events["amount_lag1"] = g["amount"].shift(1)
events["amount_lag3"] = g["amount"].shift(3)
events["delta_secs"] = g["ts"].diff().dt.total_seconds()
events["amount_roll7"] = g["amount"].transform(lambda s: s.rolling(7, min_periods=1).mean())
events.fillna(0.0).to_parquet("event_features.parquet", index=False)

CoLES embeddings (use only if instruction.md asks for embeddings and pytorch-lifestream is available; CPU-only is fine for inference):

# Optional path; requires pip install pytorch-lifestream
from functools import partial
import torch.optim as optim
from ptls.nn import TrxEncoder, RnnSeqEncoder
from ptls.frames.coles import CoLESModule

embeddings = {"category": {"in": 64, "out": 8}}      # categorical fields
numeric = ["amount"]                                  # continuous fields
trx_encoder = TrxEncoder(embeddings=embeddings, numeric_values={k: "identity" for k in numeric})
seq_encoder = RnnSeqEncoder(trx_encoder=trx_encoder, hidden_size=32, type="gru")
model = CoLESModule(
    seq_encoder=seq_encoder,
    optimizer_partial=partial(optim.Adam, lr=1e-3),
    lr_scheduler_partial=partial(optim.lr_scheduler.StepLR, step_size=10, gamma=0.9),
)
# Inference produces a (n_entities, hidden_size) matrix; persist as parquet/npy.

Notes: CoLES needs lr_scheduler_partial to construct; pad variable lengths via ptls.data_load.PaddedBatch. For CPU-only environments, prefer the lag-feature variant unless the task specifically grades contrastive embeddings.

§C — Ledger reconciliation (double-entry)

Take this when the schema implies postings — every transaction has matched debit + credit, and the task is auditing balance, finding unmatched entries, or computing per-account net.

import pandas as pd

post = pd.read_csv("postings.csv", parse_dates=["ts"])
# Convention: positive = debit, negative = credit; or split columns.
if {"debit", "credit"}.issubset(post.columns):
    post["signed"] = post["debit"].fillna(0) - post["credit"].fillna(0)
else:
    post["signed"] = post["amount"]

# Invariant: total debits must equal total credits per txn_id.
per_txn = post.groupby("txn_id")["signed"].sum()
unbalanced = per_txn[per_txn.abs() > 1e-6]

balances = post.groupby("account")["signed"].sum().rename("balance").reset_index()
balances.to_csv("balances.csv", index=False)
unbalanced.to_csv("unbalanced_txns.csv", header=["delta"])

Notes: tolerance 1e-6 handles float drift; for currency, prefer Decimal or integer cents and require exact zero. Always emit both the per-account balance and the list of unbalanced txn_ids — tests typically check both.

§D — Category / basket analysis

Take this for POS-style data where each row is a line item with a category or SKU, and the task asks for share-of-wallet, top categories, or basket-level features.

import pandas as pd

lines = pd.read_csv("pos_lines.csv", parse_dates=["ts"])

# Category share per entity
share = (lines.groupby(["entity_id", "category"])["amount"].sum()
              .groupby(level=0).apply(lambda s: s / s.sum())
              .rename("share").reset_index())

# Basket-level features (one row per basket / receipt)
basket = lines.groupby("basket_id").agg(
    n_items=("sku", "size"),
    n_unique_sku=("sku", "nunique"),
    basket_total=("amount", "sum"),
    top_category=("category", lambda s: s.value_counts().idxmax()),
).reset_index()

share.to_csv("category_share.csv", index=False)
basket.to_parquet("basket_features.parquet", index=False)

Notes: idxmax on value_counts() is deterministic only when the top category is unique — if ties matter, sort by (count desc, category asc) explicitly. Shares should sum to ~1.0 per entity; assert before saving.

§E — Anomaly / fraud scoring

Take this for per-transaction scoring with weak or no labels. Default baseline: IsolationForest on numeric + one-hot categoricals.

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import OneHotEncoder

txns = pd.read_csv("txns.csv", parse_dates=["ts"])
num_cols = ["amount"]
cat_cols = [c for c in ("merchant_type", "country") if c in txns.columns]

X_num = txns[num_cols].to_numpy(dtype=float)
if cat_cols:
    enc = OneHotEncoder(handle_unknown="ignore", sparse_output=False)
    X_cat = enc.fit_transform(txns[cat_cols].astype(str))
    X = np.hstack([X_num, X_cat])
else:
    X = X_num

clf = IsolationForest(n_estimators=200, contamination="auto", random_state=42)
clf.fit(X)
# Higher = more anomalous; flip sign of decision_function.
txns["anomaly_score"] = -clf.decision_function(X)
txns[["txn_id", "anomaly_score"]].to_csv("scores.csv", index=False)

Notes: set random_state=42 for reproducibility; if the task supplies labels, evaluate with roc_auc_score(y, anomaly_score) rather than hard threshold accuracy. Log scale amount if it spans many decades — IsolationForest tolerates skew but tree splits become coarse.

Validation self-checks

• §A: row count of output equals nunique(entity_id); entity_id is a column, not just an index.
• §B: no leakage across entity boundaries — features for entity X must use only events of entity X (groupby before shift / rolling).
• §C: unbalanced_txns.csv is empty only if the source data is truly balanced — do not silently filter.
• §D: per-entity category shares sum to 1.0 within 1e-6.
• §E: scores are finite, length matches input row count, higher means more anomalous (document the convention).

Pitfalls

• Mixing time zones in ts parsing — normalize with utc=True if any offset appears.
• Using now() for recency — makes outputs non-reproducible across runs; use events["ts"].max() or a task-specified reference date.
• Calling shift / rolling on the full frame instead of per-entity — silently leaks across users.
• Treating ledger amounts as floats and asserting exact zero — drift bites; either tolerate 1e-6 or switch to integer cents.
• Picking CoLES when the task only needs lag features — adds heavy deps and a GPU expectation that the benchmark does not require.

When in doubt, start with §A; the others are upgrade paths.