From 2df31a1d8c0ab1c31d86047198e2f306b62be5c5 Mon Sep 17 00:00:00 2001 From: Shahar Biron <38566538+shahar-biron@users.noreply.github.com> Date: Sun, 12 Jul 2026 11:50:40 +0300 Subject: [PATCH] feat: add FalkorDB bulk-loader export format Add a new falkordb output mode with schema-enforced headers and Falkor-compatible transaction/fraud rows.\nExpand tests and documentation for loader integration and usage.\n\nCo-Authored-By: Oz --- CHANGELOG.md | 4 +++ README.md | 42 +++++++++++++++++++++++++++++-- src/gen_fraud_graph/cli.py | 8 ++++-- src/gen_fraud_graph/config.py | 7 +++--- src/gen_fraud_graph/exporters.py | 21 +++++++++++++++- src/gen_fraud_graph/generator.py | 23 ++++++++++------- src/gen_fraud_graph/typologies.py | 15 ++++++++--- tests/test_cli.py | 29 +++++++++++++++++++++ tests/test_exporters.py | 12 +++++++++ tests/test_generator.py | 20 +++++++++++++++ tests/test_typologies.py | 20 +++++++++++++++ 11 files changed, 181 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 67dcf1b..6970e70 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- `falkordb` export format for CLI/API output, producing schema-enforced CSV headers compatible with `falkordb-bulk-loader --enforce-schema`. +- FalkorDB transaction exports now include `is_fraud` markers (`false` for normal transactions, `true` for injected fraud edges) and schema-ready endpoint columns (`:START_ID(Account)`, `:END_ID(Account)`). + ## [0.1.0] - 2026-07-06 ### Added diff --git a/README.md b/README.md index 19c0ad2..d3195d4 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,7 @@ The generator creates three types of data: - **Fraud pattern injection** — Cyclic money-laundering rings with configurable depth (4–7 hops) - **Parallel generation** — Multi-process workers for fast generation on high-core machines - **Vector embeddings** — Three providers: `fake` (random, fast), `local` (SentenceTransformers), `openai` (API) -- **Multiple formats** — Generic CSV or AWS Neptune bulk-load format +- **Multiple formats** — Generic CSV, AWS Neptune bulk-load, or FalkorDB bulk-loader schema format - **Resume support** — Interrupted generation can resume from where it left off - **Privacy by design** — All data is 100% synthetic; no real financial data is used @@ -88,6 +88,8 @@ gen-fraud-graph --scale 1.0 --workers 24 --output ./data # Neptune bulk-load format gen-fraud-graph --scale 0.01 --format neptune --output ./neptune_data +# FalkorDB bulk-loader schema format +gen-fraud-graph --scale 0.01 --format falkordb --output ./falkordb_data # Resume interrupted generation (skips completed files) gen-fraud-graph --scale 1.0 --workers 24 --skip-accounts --output ./data @@ -102,7 +104,7 @@ gen-fraud-graph --scale 1.0 --workers 24 --skip-accounts --output ./data | `--output` | `data` | Output directory for generated CSV files. | | `--workers` | `1` | Number of parallel worker processes. | | `--batches` | `1` | Number of file chunks per worker. | -| `--format` | `csv` | Output format: `csv` (generic) or `neptune` (AWS Neptune bulk-load). | +| `--format` | `csv` | Output format: `csv` (generic), `neptune` (AWS Neptune bulk-load), or `falkordb` (FalkorDB bulk-loader schema). | | `--fraud-rings` | auto | Number of fraud rings. Default: auto-scaled from `--scale`. | | `--compress` | off | ZIP-compress output CSV files. | | `--skip-accounts` | off | Skip account generation (useful when resuming). | @@ -184,6 +186,42 @@ data/ | `depth` | int | Number of hops in the ring (4–7) | | `involved_accounts` | string | Pipe-separated list of accounts | +### FalkorDB bulk-loader schema (`--format falkordb`) + +`accounts/accounts_*.csv` header: + +```csv +account_id:ID(Account),customer_name:STRING,balance:DOUBLE,risk_score:DOUBLE,creation_date:STRING +``` + +`transactions/transactions_*.csv` and `fraud/transactions_fraud.csv` header: + +```csv +:START_ID(Account),:END_ID(Account),tx_id:STRING,amount:DOUBLE,timestamp:STRING,description:STRING,embedding:STRING,is_fraud:BOOLEAN +``` + +Row semantics: +- normal transactions are emitted with `is_fraud=false` +- injected fraud transactions are emitted with `is_fraud=true` +- `embedding` remains a pipe-separated `STRING` to keep default comma-delimited loading simple + +Recommended import shape with `falkordb-bulk-loader`: +- pass one `--nodes-with-label Account ` per account shard +- pass one `--relations-with-type TRANSFER ` per transaction shard (including `fraud/transactions_fraud.csv`) +- use `--enforce-schema` + +Example: + +```bash +falkordb-bulk-insert FraudGraph \ + --enforce-schema \ + --nodes-with-label Account ./falkordb_data/accounts/accounts_0_0.csv \ + --relations-with-type TRANSFER ./falkordb_data/transactions/transactions_0_0.csv \ + --relations-with-type TRANSFER ./falkordb_data/fraud/transactions_fraud.csv +``` + +If you generated compressed outputs (`--compress`), unzip them before running the bulk loader. + --- ## Scale Reference diff --git a/src/gen_fraud_graph/cli.py b/src/gen_fraud_graph/cli.py index 5d71a77..fab7bb2 100644 --- a/src/gen_fraud_graph/cli.py +++ b/src/gen_fraud_graph/cli.py @@ -58,8 +58,12 @@ def main(argv: list[str] | None = None) -> None: "--format", type=str, default="csv", - choices=["csv", "neptune"], - help="Output format. 'csv' = generic CSV, 'neptune' = AWS Neptune bulk-load. Default: csv", + choices=["csv", "neptune", "falkordb"], + help=( + "Output format. 'csv' = generic CSV, " + "'neptune' = AWS Neptune bulk-load, " + "'falkordb' = FalkorDB bulk-loader schema CSV. Default: csv" + ), ) parser.add_argument( "--fraud-rings", diff --git a/src/gen_fraud_graph/config.py b/src/gen_fraud_graph/config.py index 5c98fd6..ee6d9bb 100644 --- a/src/gen_fraud_graph/config.py +++ b/src/gen_fraud_graph/config.py @@ -22,8 +22,9 @@ class Config: embedding_dim: Dimensionality of generated embeddings. workers: Parallel processes for account/transaction generation. batches_per_worker: File chunks each worker produces. - output_format: ``"csv"`` (generic) or ``"neptune"`` (AWS Neptune - bulk-load headers). + output_format: ``"csv"`` (generic), ``"neptune"`` (AWS Neptune + bulk-load headers), or ``"falkordb"`` (FalkorDB bulk-loader + schema headers). compress: Whether to ZIP the output CSV files. output_dir: Destination directory for generated files. """ @@ -35,7 +36,7 @@ class Config: embedding_dim: int = 768 workers: int = 1 batches_per_worker: int = 1 - output_format: Literal["csv", "neptune"] = "csv" + output_format: Literal["csv", "neptune", "falkordb"] = "csv" compress: bool = False output_dir: str = "data" diff --git a/src/gen_fraud_graph/exporters.py b/src/gen_fraud_graph/exporters.py index c0cf91c..7d460ce 100644 --- a/src/gen_fraud_graph/exporters.py +++ b/src/gen_fraud_graph/exporters.py @@ -15,7 +15,7 @@ def get_headers( doc_type: Literal["account", "transaction"], - fmt: Literal["csv", "neptune"], + fmt: Literal["csv", "neptune", "falkordb"], ) -> list[str]: """Return CSV column headers for *doc_type* in the given *fmt*.""" if fmt == "neptune": @@ -38,6 +38,25 @@ def get_headers( "timestamp:String", "description:String", ] + if fmt == "falkordb": + if doc_type == "account": + return [ + "account_id:ID(Account)", + "customer_name:STRING", + "balance:DOUBLE", + "risk_score:DOUBLE", + "creation_date:STRING", + ] + return [ + ":START_ID(Account)", + ":END_ID(Account)", + "tx_id:STRING", + "amount:DOUBLE", + "timestamp:STRING", + "description:STRING", + "embedding:STRING", + "is_fraud:BOOLEAN", + ] # Default CSV if doc_type == "account": diff --git a/src/gen_fraud_graph/generator.py b/src/gen_fraud_graph/generator.py index b49096d..f3d21f6 100644 --- a/src/gen_fraud_graph/generator.py +++ b/src/gen_fraud_graph/generator.py @@ -191,16 +191,16 @@ def _generate_transactions_chunk( desc = random.choice(NORMAL_DESCRIPTIONS) batch_texts.append(desc) - row: list = [ - f"tx_{tx_uid}", - src, - dst, - round(random.uniform(10, 500), 2), - "2024-01-01T10:00:00", - desc, - ] + amount = round(random.uniform(10, 500), 2) + timestamp = "2024-01-01T10:00:00" + tx_id = f"tx_{tx_uid}" + if fmt == "neptune": - row.insert(3, "TRANSFER") + row = [tx_id, src, dst, "TRANSFER", amount, timestamp, desc] + elif fmt == "falkordb": + row = [src, dst, tx_id, amount, timestamp, desc] + else: + row = [tx_id, src, dst, amount, timestamp, desc] batch_rows.append(row) embeddings = embedder.generate(batch_texts) @@ -209,6 +209,11 @@ def _generate_transactions_chunk( for idx, r in enumerate(batch_rows): if fmt == "neptune": final_rows.append(r) + elif fmt == "falkordb": + vec = embeddings[idx] + if isinstance(vec, np.ndarray): + vec = vec.tolist() + final_rows.append(r + ["|".join(map(str, vec)), "false"]) else: vec = embeddings[idx] if isinstance(vec, np.ndarray): diff --git a/src/gen_fraud_graph/typologies.py b/src/gen_fraud_graph/typologies.py index c009296..d334d15 100644 --- a/src/gen_fraud_graph/typologies.py +++ b/src/gen_fraud_graph/typologies.py @@ -117,11 +117,15 @@ def generate( dst = accounts[(k + 1) % depth] desc = random.choice(self._descriptions) batch_texts.append(desc) + tx_id = f"tx_{current_tx_id}" + timestamp = "2024-01-01T12:00:00" - row: list = [f"tx_{current_tx_id}", src, dst] if fmt == "neptune": - row.append("TRANSFER") - row.extend([self.amount, "2024-01-01T12:00:00", desc]) + row = [tx_id, src, dst, "TRANSFER", self.amount, timestamp, desc] + elif fmt == "falkordb": + row = [src, dst, tx_id, self.amount, timestamp, desc] + else: + row = [tx_id, src, dst, self.amount, timestamp, desc] batch_rows.append(row) current_tx_id += 1 @@ -130,6 +134,11 @@ def generate( for idx, r in enumerate(batch_rows): if fmt == "neptune": tx_rows.append(r) + elif fmt == "falkordb": + vec = embeddings[idx] + if isinstance(vec, np.ndarray): + vec = vec.tolist() + tx_rows.append(r + ["|".join(map(str, vec)), "true"]) else: vec = embeddings[idx] if isinstance(vec, np.ndarray): diff --git a/tests/test_cli.py b/tests/test_cli.py index 0312761..310cebd 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -5,6 +5,7 @@ from __future__ import annotations +import csv import os @@ -49,3 +50,31 @@ def test_main_skip_accounts_and_compress(self, tmp_dir): ] ) assert os.path.exists(os.path.join(tmp_dir, "fraud", "fraud_cases.csv.zip")) + + def test_main_falkordb_format(self, tmp_dir): + from gen_fraud_graph.cli import main + + main( + [ + "--scale", + "0.0001", + "--provider", + "fake", + "--output", + tmp_dir, + "--workers", + "1", + "--batches", + "1", + "--format", + "falkordb", + "--fraud-rings", + "3", + ] + ) + tx_path = os.path.join(tmp_dir, "transactions", "transactions_0_0.csv") + with open(tx_path) as fh: + header = next(csv.reader(fh)) + assert header[0] == ":START_ID(Account)" + assert header[1] == ":END_ID(Account)" + assert header[-1] == "is_fraud:BOOLEAN" diff --git a/tests/test_exporters.py b/tests/test_exporters.py index 3ea5523..c0ecf6d 100644 --- a/tests/test_exporters.py +++ b/tests/test_exporters.py @@ -33,6 +33,18 @@ def test_neptune_headers_transaction(self): assert "~from" in h assert "~to" in h + def test_falkordb_headers_account(self): + h = get_headers("account", "falkordb") + assert h[0] == "account_id:ID(Account)" + assert "customer_name:STRING" in h + assert "balance:DOUBLE" in h + + def test_falkordb_headers_transaction(self): + h = get_headers("transaction", "falkordb") + assert h[0] == ":START_ID(Account)" + assert h[1] == ":END_ID(Account)" + assert "is_fraud:BOOLEAN" in h + def test_write_output_csv(self, tmp_dir): path = os.path.join(tmp_dir, "test") write_output(path, ["a", "b"], [[1, 2], [3, 4]]) diff --git a/tests/test_generator.py b/tests/test_generator.py index 90bf74b..13dae94 100644 --- a/tests/test_generator.py +++ b/tests/test_generator.py @@ -40,6 +40,15 @@ def test_accounts_chunk_neptune(self, tmp_dir): header = next(csv.reader(fh)) assert "~id" in header + def test_accounts_chunk_falkordb(self, tmp_dir): + _generate_accounts_chunk(0, 0, 0, 10, "fake", 16, tmp_dir, "falkordb") + path = os.path.join(tmp_dir, "accounts", "accounts_0_0.csv") + with open(path) as fh: + rows = list(csv.reader(fh)) + assert rows[0][0] == "account_id:ID(Account)" + assert "risk_score:DOUBLE" in rows[0] + assert len(rows[1]) == len(rows[0]) == 5 + def test_accounts_chunk_resume_complete(self, tmp_dir): _generate_accounts_chunk(0, 0, 0, 5, "fake", 16, tmp_dir, "csv") msg = _generate_accounts_chunk(0, 0, 0, 5, "fake", 16, tmp_dir, "csv") @@ -67,6 +76,17 @@ def test_transactions_chunk_neptune(self, tmp_dir): header = next(csv.reader(fh)) assert "~from" in header + def test_transactions_chunk_falkordb(self, tmp_dir): + _generate_transactions_chunk(0, 0, 0, 20, 100, "fake", 16, tmp_dir, "falkordb") + path = os.path.join(tmp_dir, "transactions", "transactions_0_0.csv") + with open(path) as fh: + rows = list(csv.reader(fh)) + assert rows[0][0] == ":START_ID(Account)" + assert rows[0][1] == ":END_ID(Account)" + assert rows[0][-1] == "is_fraud:BOOLEAN" + assert rows[1][-1] == "false" + assert len(rows[1]) == len(rows[0]) == 8 + def test_transactions_chunk_resume_complete(self, tmp_dir): _generate_transactions_chunk(0, 0, 0, 5, 50, "fake", 16, tmp_dir, "csv") msg = _generate_transactions_chunk(0, 0, 0, 5, 50, "fake", 16, tmp_dir, "csv") diff --git a/tests/test_typologies.py b/tests/test_typologies.py index 5410a39..3175190 100644 --- a/tests/test_typologies.py +++ b/tests/test_typologies.py @@ -87,6 +87,26 @@ def test_neptune_format(self, tmp_dir): header = next(csv.reader(fh)) assert "~from" in header + def test_falkordb_format(self, tmp_dir): + emb = EmbeddingGenerator("fake", dim=16) + gen = FraudRingGenerator(num_rings=2, depth_range=(3, 3)) + n_tx, _ = gen.generate( + max_account_id=50, + start_tx_id=0, + embedder=emb, + output_dir=tmp_dir, + fmt="falkordb", + ) + assert n_tx > 0 + path = os.path.join(tmp_dir, "fraud", "transactions_fraud.csv") + with open(path) as fh: + rows = list(csv.reader(fh)) + assert rows[0][0] == ":START_ID(Account)" + assert rows[0][1] == ":END_ID(Account)" + assert rows[0][-1] == "is_fraud:BOOLEAN" + assert rows[1][-1] == "true" + assert len(rows[1]) == len(rows[0]) == 8 + def test_oversubscribed_rings_raise(self, tmp_dir): """When the rings need more distinct accounts than exist they can't be packed disjointly, so generate() must raise rather than emit rings that