GoldenFlow

Moved. This repo has moved into the benzsevern/goldenmatch monorepo at packages/python/goldenflow (and packages/typescript/goldenflow)/. This repo is archived; new development happens in the monorepo.

GoldenFlow

Data transformation toolkit — standardize, reshape, and normalize messy data before it hits your pipeline. Built by Ben Severn.

Works on files (CSV, Excel, Parquet), cloud storage (S3, GCS), or live databases. Zero-config mode auto-detects what needs fixing. One command to clean what GoldenCheck found and prep what GoldenMatch needs. Available in Python and TypeScript with full feature parity.

[PyPI] [npm] [CI] [codecov] [Downloads] [Python 3.11+] [Node 20+] [License: MIT] [Docs] [DQBench] [Open In Colab]

# Pythonpip install goldenflowgoldenflow transform data.csv
# TypeScript / Node.jsnpm install goldenflownpx goldenflow-js transform data.csv

The Problem

Your data arrives broken in predictable ways:

• Phone numbers come in 15 different formats
• Dates are mixed between MM/DD/YYYY and YYYY-MM-DD
• Addresses have inconsistent abbreviations
• Column names don't match between systems ("fname" vs "first_name" vs "given_name")
• Values have leading whitespace, unicode garbage, smart quotes
• Categoricals are inconsistent ("USA", "US", "United States")

Every data engineer writes throwaway scripts to fix these. Every script is slightly different. None of them are reusable.

GoldenFlow makes the transforms reusable, composable, and automatic.

Quick Start

Python

pip install goldenflow
# Auto-transform (zero-config)goldenflow transform messy_data.csv
# Try the demo firstgoldenflow demogoldenflow transform demo_data.csv -c demo_config.yaml
# With configgoldenflow learn messy_data.csv -o config.yamlgoldenflow transform messy_data.csv -c config.yaml
# Schema mappinggoldenflow map --source system_a.csv --target system_b.csv
# Full pipelinegoldencheck scan data.csvgoldenflow transform data.csvgoldenmatch dedupe data_transformed.csv

TypeScript / Node.js

npm install goldenflow
# Auto-transform (zero-config)npx goldenflow-js transform messy_data.csv
# Try the demo firstnpx goldenflow-js demonpx goldenflow-js transform demo_data.csv -c demo_config.yaml
# With confignpx goldenflow-js learn messy_data.csv -o config.yamlnpx goldenflow-js transform messy_data.csv -c config.yaml
# Schema mappingnpx goldenflow-js map -s system_a.csv -t system_b.csv

Programmatic (TypeScript)

import { TransformEngine } from "goldenflow";
const result = new TransformEngine().transformDf([  { name: "  JOHN  ", phone: "(555) 123-4567", email: "[email protected]" },]);console.log(result.rows[0]);// { name: "JOHN", phone: "+15551234567", email: "[email protected]" }

Zero-Config Mode

goldenflow transform customers.csv# or just:goldenflow customers.csv

GoldenFlow profiles every column and applies safe transforms automatically:

• Strips whitespace and normalizes unicode
• Standardizes phone numbers to E.164 format
• Normalizes email casing
• Parses and standardizes date formats to ISO 8601
• Normalizes zip codes (zero-padding, strip +4)
• Replaces smart/curly quotes with straight quotes
• Auto-corrects categorical misspellings via fuzzy matching

Output: a clean CSV with a sidecar manifest showing every transform applied.

customers.csv           -> customers_transformed.csv                        -> customers_manifest.json

The manifest is an audit trail — what changed, why, and which rows were affected.

CLI Commands

GoldenFlow has 14 commands. The most common ones:

# Core transformsgoldenflow transform data.csv                    # Zero-config: auto-detect and fixgoldenflow transform data.csv -c config.yaml     # Apply saved configgoldenflow transform data.csv --domain healthcare # Use a domain packgoldenflow transform data.csv --strict           # Fail on any transform errorgoldenflow transform data.csv --llm              # Enable LLM-enhanced correctionsgoldenflow data.csv                              # Shorthand: auto-routes to transform
# Schema & profilinggoldenflow map -s a.csv -t b.csv                 # Auto-map schemas between filesgoldenflow profile data.csv                      # Show column profilesgoldenflow learn data.csv -o config.yaml         # Generate config from data patternsgoldenflow validate data.csv                     # Dry-run: show what would changegoldenflow diff before.csv after.csv             # Compare pre/post transform
# Continuous & scheduledgoldenflow watch ./data/                         # Auto-transform new/changed filesgoldenflow schedule data.csv --every 1h          # Run on a schedule (5m, 1h, 30s...)goldenflow stream large_file.csv --chunk-size 50000  # Stream-process in batches
# Discovery & historygoldenflow init data.csv                         # Interactive setup wizardgoldenflow demo                                  # Generate sample data to trygoldenflow history                               # Show recent transform runsgoldenflow history -n 50                         # Last 50 runs
# Integrationsgoldenflow interactive data.csv                  # Launch TUIgoldenflow serve                                 # REST API for real-time transformsgoldenflow mcp-serve                             # MCP server for Claude Desktop

Default Routing

Running goldenflow <file> without a subcommand auto-routes to transform:

goldenflow customers.csv       # equivalent to: goldenflow transform customers.csvgoldenflow -                   # read from stdin, write to stdout

Streaming

For files too large to load into memory, use StreamProcessor or the stream command:

goldenflow stream large_file.csv --chunk-size 50000

from goldenflow.streaming import StreamProcessor
processor = StreamProcessor(config=config)
# Process a single recordresult = processor.transform_one({"name": "  John  ", "phone": "(555) 123-4567"})
# Process a batchresult = processor.transform_batch(df_batch)
# Stream a large file in chunksfor result in processor.stream_file("large_data.csv", chunk_size=10_000):    write_to_output(result.df)
print(f"Processed {processor.batches_processed} batches")

Cloud Connectors

GoldenFlow reads from and writes to S3 and Google Cloud Storage transparently:

# S3goldenflow transform s3://my-bucket/raw/customers.csv -o s3://my-bucket/clean/
# GCSgoldenflow transform gs://my-bucket/data/records.csv

from goldenflow.connectors.s3 import read_s3, write_s3from goldenflow.connectors.gcs import read_gcs, write_gcs
df = read_s3("s3://my-bucket/raw/customers.csv")df = read_gcs("gs://my-bucket/data/records.csv")

Cloud paths are detected automatically — no extra flags needed.

Watch Mode

Auto-transform files as they arrive in a directory:

goldenflow watch ./data/goldenflow watch ./incoming/ -c config.yaml -o ./processed/

GoldenFlow polls the directory and applies transforms to any new or changed files.

Scheduling

Run transforms on a repeating schedule:

goldenflow schedule data.csv --every 1hgoldenflow schedule data.csv --every 30m -c config.yaml -o ./output/

Supported intervals: 30s, 5m, 1h, 2h, etc.

Setup Wizard

Generate a YAML config interactively:

goldenflow init data.csv

The wizard profiles your data, suggests transforms, and saves a goldenflow.yaml ready to use.

History

GoldenFlow tracks every transform run in ~/.goldenflow/history/:

goldenflow history         # Last 20 runsgoldenflow history -n 50   # Last 50 runs

Each run record captures: source file, row count, transforms applied, errors, and duration.

Schema Mapping

When you need to merge data from different systems:

goldenflow map --source crm_export.csv --target warehouse_schema.csv

GoldenFlow auto-maps columns between schemas using name similarity and data profiling:

crm_export.csv              warehouse schema-----                       -----email_address      ->       email (rename)phone_number       ->       phone (rename)fname              ->       first_name (alias match)st                 ->       state (alias match)

Ambiguous mappings get flagged for human review. Confident mappings apply automatically.

Domain Packs

Pre-configured transform sets for common industries. All 5 are now implemented:

goldenflow transform patients.csv --domain healthcaregoldenflow transform employees.csv --domain people_hrgoldenflow transform transactions.csv --domain financegoldenflow transform orders.csv --domain ecommercegoldenflow transform listings.csv --domain real_estate

Transform Library (76 transforms)

Text Transforms (18)

Phone Transforms (5)

Name Transforms (8)

Address Transforms (8)

Date Transforms (13)

Categorical Transforms (6)

Numeric Transforms (9)

Email Transforms (4)

Identifier Transforms (3)

URL Transforms (2)

Special Modes

Strict Mode

Fail immediately if any transform error occurs — useful in CI or production pipelines:

goldenflow transform data.csv --strict

Exits with code 1 and prints the first 5 errors if any transform fails.

LLM Mode

Use an LLM to enhance categorical corrections and handle edge cases that fuzzy matching misses:

goldenflow transform data.csv --llm

Requires OPENAI_API_KEY or ANTHROPIC_API_KEY in your environment. Falls back to standard transforms gracefully.

Auto-Correct

category_auto_correct uses fuzzy matching to fix misspelled categorical values automatically. It is suppressed on high-cardinality columns (>10% unique values) to avoid false positives.

"actve" -> "active""Pennsylvnia" -> "Pennsylvania""Unted States" -> "United States"

YAML Config

For repeatable pipelines:

# goldenflow.yamlsource: customers.csvoutput: customers_clean.csv
transforms:  - column: name    ops: [strip, title_case]  - column: email    ops: [lowercase, strip]  - column: phone    ops: [phone_e164]  - column: state    ops: [state_abbreviate]  - column: signup_date    ops: [date_iso8601]
renames:  email_address: email  phone_number: phone
drop: [internal_id, temp_notes]
dedup:  columns: [email]  keep: first

goldenflow transform customers.csv -c goldenflow.yaml

Generate a config from your data automatically:

goldenflow learn data.csv -o config.yaml

Python API

import goldenflow
# Zero-configresult = goldenflow.transform_file("messy_data.csv")print(result.df)          # Clean Polars DataFrameprint(result.manifest)    # Audit trail
# With configfrom goldenflow import GoldenFlowConfig, TransformSpec, TransformEngine
config = GoldenFlowConfig(    transforms=[        TransformSpec(column="phone", ops=["phone_e164"]),        TransformSpec(column="date", ops=["date_iso8601"]),    ])engine = TransformEngine(config=config)result = engine.transform_df(df)

Jupyter Notebook Support

TransformResult, Manifest, and DatasetProfile all have _repr_html_() — they render as rich HTML tables automatically in Jupyter:

import goldenflow
result = goldenflow.transform_file("messy_data.csv")result           # renders as HTML table in Jupyterresult.manifest  # renders transform audit trail

TypeScript / JavaScript

GoldenFlow has a full TypeScript port with feature parity — same 83 transforms, same engine, same config format. The core is edge-safe (runs in browsers, Cloudflare Workers, Vercel Edge) with a Node layer for file I/O and CLI.

Install

npm install goldenflow

CLI

npx goldenflow-js transform data.csv              # Zero-confignpx goldenflow-js transform data.csv -c config.yaml  # With confignpx goldenflow-js profile data.csv                 # Column profilesnpx goldenflow-js learn data.csv -o config.yaml    # Generate confignpx goldenflow-js diff before.csv after.csv        # Compare filesnpx goldenflow-js map -s source.csv -t target.csv  # Schema mappingnpx goldenflow-js stream large.csv --chunk-size 50000  # Streamingnpx goldenflow-js demo                             # Generate sample datanpx goldenflow-js history                          # Show recent runs

TypeScript API

import { TransformEngine, makeConfig } from "goldenflow";
// Zero-config — auto-detect and fixconst engine = new TransformEngine();const result = engine.transformDf([  { name: "  John Smith  ", email: "[email protected]", phone: "(555) 123-4567" },  { name: "DR. JANE DOE", email: "  [email protected]  ", phone: "555.987.6543" },]);
console.log(result.rows);// [//   { name: "John Smith", email: "[email protected]", phone: "+15551234567" },//   { name: "Jane Doe", email: "[email protected]", phone: "+15559876543" },// ]console.log(result.manifest.records.length); // transforms applied

// Configured — explicit transforms per columnconst engine = new TransformEngine({  transforms: [    { column: "phone", ops: ["phone_e164"] },    { column: "email", ops: ["strip", "email_normalize"] },    { column: "name", ops: ["strip", "title_case"] },    { column: "state", ops: ["state_abbreviate"] },    { column: "price", ops: ["currency_strip"] },    { column: "signup_date", ops: ["date_iso8601"] },  ],  renames: { email_address: "email" },  drop: ["internal_id"],  dedup: { columns: ["email"], keep: "first" },});const result = engine.transformDf(rows);

// Schema mappingimport { SchemaMapper } from "goldenflow";
const mapper = new SchemaMapper();const mappings = mapper.map(  [{ fname: "John", lname: "Smith", email_address: "[email protected]" }],  [{ first_name: "", last_name: "", email: "" }],);// [{ source: "fname", target: "first_name", confidence: 0.95 }, ...]

// Streaming large datasetsimport { StreamProcessor } from "goldenflow";
const processor = new StreamProcessor({ transforms: [{ column: "name", ops: ["strip"] }] });for (const result of processor.streamRows(largeDataset, 10_000)) {  await writeChunk(result.rows);}

// Profilingimport { profileDataframe } from "goldenflow";
const profile = profileDataframe(rows, "customers.csv");for (const col of profile.columns) {  console.log(`${col.name}: ${col.inferredType}, ${col.nullCount} nulls, ${col.uniqueCount} unique`);}

// Edge-safe import (browsers, Workers, Edge Runtime)import { TransformEngine } from "goldenflow/core";
// Node-only import (includes file I/O, MCP, CLI)import { readFile, TransformEngine } from "goldenflow/node";

MCP Server (TypeScript)

import { TOOL_DEFINITIONS, handleTool } from "goldenflow/node";
// TOOL_DEFINITIONS: 10 MCP tools for Claude Desktop// handleTool("transform", { path: "data.csv" }) → JSON string

REST API (TypeScript)

import { runServer } from "goldenflow/node";runServer(8000); // Starts HTTP server with /health, /transforms, /transform

Public API (34 exports)

from goldenflow import (    # Core engine    TransformEngine, TransformResult,    # Config    GoldenFlowConfig, TransformSpec, SplitSpec, FilterSpec, DedupSpec, MappingSpec,    # Convenience    transform_file, transform_df,    # Manifest    Manifest, TransformRecord, TransformError,    # Profiler    DatasetProfile, ColumnProfile,    # Selector & differ    select_transforms, diff_dataframes, DiffResult,    # Transform registry    TransformInfo, register_transform, get_transform, list_transforms, parse_transform_name,    # Mapping    SchemaMapper, ColumnMapping,    # Config helpers    load_config, save_config, merge_configs, learn_config,    # Domains    DomainPack, load_domain,    # Connectors    read_file, write_file,)

Integrations

REST API

goldenflow serve --host 0.0.0.0 --port 8000

POST CSV data, get transformed CSV back. Built with FastAPI.

MCP Server

goldenflow mcp-serve

Exposes GoldenFlow as an MCP tool for Claude Desktop. Configure in your Claude Desktop settings.

TUI

goldenflow interactive data.csv

Full-featured terminal UI built with Textual. Browse profiles, apply transforms, preview results.

Remote MCP Server

GoldenFlow is available as a hosted MCP server on Smithery — connect from any MCP client without installing anything.

Claude Desktop / Claude Code:

{  "mcpServers": {    "goldenflow": {      "url": "https://goldenflow-mcp-production.up.railway.app/mcp/"    }  }}

Local server:

pip install goldenflow[mcp]goldenflow mcp-serve

10 tools available: transform files, auto-map schemas, profile columns, generate configs, diff before/after, apply domain packs.

Part of the Golden Suite

Raw Data   |   v+--------------+|  GoldenCheck |  <- Discover quality issues|  goldencheck ||  scan data   |+------+-------+       | findings       v+--------------+|  GoldenFlow  |  <- Fix issues, standardize, reshape|  goldenflow  ||  transform   |+------+-------+       | clean data       v+--------------+|  GoldenMatch |  <- Deduplicate, match, create golden records|  goldenmatch ||  dedupe      |+------+-------+       | golden records       v   Clean, deduplicated,   production-ready data

Chain them:

goldencheck scan data.csv | goldenflow transform --from-findings | goldenmatch dedupe

Performance

Built on Polars (Rust-backed DataFrames). Transforms use a hybrid approach: native Polars expressions stay in the Rust engine for simple transforms (strip, lowercase), while complex transforms (phone parsing, date parsing) use optimized Python via map_batches.

Benchmarks

GoldenFlow scores 100/100 on the DQBench transform benchmark across all three tiers (customer database, e-commerce, healthcare claims).

pip install dqbenchdqbench run goldenflow

Why GoldenFlow?

GoldenFlow is purpose-built for the transform step between validation and matching — not a general ETL tool. It turns messy data into clean, standardized data automatically.

Error Handling

GoldenFlow catches errors at the CLI boundary and shows friendly, actionable messages — no raw stack traces. Individual transform errors are captured in the manifest rather than crashing the run. Use --strict to change this behavior.

Progress Bars

Long-running operations (streaming, watch mode, scheduling) display a Rich progress spinner showing batch count, rows processed, and estimated completion.

GitHub: github.com/benzsevern/goldenflow Author: Ben Severn License: MIT Python: 3.11+ | Node.js: 20+ | npm: goldenflow | PyPI: goldenflow