feat: add FalkorDB bulk-loader export format

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

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 <accounts_csv>` per account shard
+- pass one `--relations-with-type TRANSFER <transactions_csv>` 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

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",

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"
 

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":

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):

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):

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"

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]])

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")

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