Python API

import goldenmatch as gm

gm.dedupe(
    *files: str,
    config: str | GoldenMatchConfig | None = None,
    exact: list[str] | None = None,
    fuzzy: dict[str, float] | None = None,
    blocking: list[str] | None = None,
    threshold: float | None = None,
    llm_scorer: bool = False,
    backend: str | None = None,
) -> DedupeResult

result = gm.dedupe("customers.csv")
result = gm.dedupe("a.csv", "b.csv", exact=["email"], fuzzy={"name": 0.85})
result = gm.dedupe("products.csv", config="config.yaml", llm_scorer=True)
result = gm.dedupe("huge.parquet", exact=["email"], backend="ray")

gm.dedupe_df(
    df: pl.DataFrame,
    *,
    config: GoldenMatchConfig | None = None,
    exact: list[str] | None = None,
    fuzzy: dict[str, float] | None = None,
    blocking: list[str] | None = None,
    threshold: float | None = None,
    llm_scorer: bool = False,
    llm_auto: bool = False,
    backend: str | None = None,
    source_name: str = "dataframe",
) -> DedupeResult

df = pl.read_csv("customers.csv")
result = gm.dedupe_df(df, exact=["email"], fuzzy={"name": 0.85})

gm.match(
    target: str,
    reference: str,
    *,
    config: str | GoldenMatchConfig | None = None,
    exact: list[str] | None = None,
    fuzzy: dict[str, float] | None = None,
    blocking: list[str] | None = None,
    threshold: float | None = None,
    backend: str | None = None,
) -> MatchResult

result = gm.match("new.csv", "master.csv", fuzzy={"name": 0.85})
result.matched.write_csv("matches.csv")

gm.match_df(
    target: pl.DataFrame,
    reference: pl.DataFrame,
    *,
    config: GoldenMatchConfig | None = None,
    exact: list[str] | None = None,
    fuzzy: dict[str, float] | None = None,
    blocking: list[str] | None = None,
    threshold: float | None = None,
    backend: str | None = None,
) -> MatchResult

result = gm.match_df(target_df, reference_df, fuzzy={"name": 0.85})

gm.score_strings(value_a: str, value_b: str, scorer: str = "jaro_winkler") -> float

gm.score_strings("John Smith", "Jon Smyth", "jaro_winkler")  # 0.884
gm.score_strings("hello", "helo", "levenshtein")              # 0.8
gm.score_strings("Smith", "Smyth", "soundex_match")           # 1.0

gm.score_pair_df(
    record_a: dict, record_b: dict,
    *, fuzzy: dict[str, float] | None = None,
    exact: list[str] | None = None,
    scorer: str = "jaro_winkler",
) -> float

score = gm.score_pair_df(
    {"name": "John Smith", "zip": "10001"},
    {"name": "Jon Smyth", "zip": "10001"},
    fuzzy={"name": 0.7, "zip": 0.3},
)

gm.explain_pair_df(
    record_a: dict, record_b: dict,
    *, fuzzy: dict[str, float] | None = None,
    exact: list[str] | None = None,
    scorer: str = "jaro_winkler",
) -> str

explanation = gm.explain_pair_df(
    {"name": "John Smith", "email": "john@example.com"},
    {"name": "Jon Smyth", "email": "john@example.com"},
    fuzzy={"name": 0.5}, exact=["email"],
)
print(explanation)

gm.pprl_link(
    file_a: str, file_b: str,
    *, fields: list[str] | None = None,
    threshold: float | None = None,
    security_level: str = "high",
    protocol: str = "trusted_third_party",
    auto_config: bool = True,
) -> dict

result = gm.pprl_link("hospital_a.csv", "hospital_b.csv")
result = gm.pprl_link("a.csv", "b.csv", fields=["name", "dob", "zip"], security_level="paranoid")

gm.pprl_auto_config(df: pl.DataFrame, security_level: str = "high") -> AutoConfigResult

config = gm.pprl_auto_config(df)
print(config.recommended_fields)  # ['first_name', 'last_name', 'zip_code', 'birth_year']

gm.run_pprl(df_a: pl.DataFrame, df_b: pl.DataFrame, config: PPRLConfig) -> LinkageResult

gm.PPRLConfig(
    fields: list[str],
    threshold: float = 0.85,
    security_level: str = "high",
    ngram_size: int = 2,
    hash_functions: int = 30,
    bloom_filter_size: int = 1024,
    protocol: str = "trusted_third_party",
)

gm.evaluate(
    *files: str,
    config: str | GoldenMatchConfig,
    ground_truth: str,
    col_a: str = "id_a",
    col_b: str = "id_b",
) -> dict

metrics = gm.evaluate("data.csv", config="config.yaml", ground_truth="gt.csv")

gm.evaluate_pairs(predicted: set[tuple], ground_truth: set[tuple]) -> EvalResult

gm.evaluate_clusters(clusters: dict, ground_truth: set[tuple]) -> EvalResult

@dataclass
class EvalResult:
    precision: float
    recall: float
    f1: float
    tp: int
    fp: int
    fn: int

    def summary(self) -> dict  # Returns dict representation

gm.load_ground_truth_csv(path: str, col_a: str = "id_a", col_b: str = "id_b") -> set[tuple]

gm.compare_clusters(
    clusters_a: dict[int, dict],
    clusters_b: dict[int, dict],
) -> CompareResult

result = gm.compare_clusters(clusters_run1, clusters_run2)
print(result.summary())  # {"unchanged": 42, "merged": 3, "twi": 0.92, ...}

@dataclass
class CompareResult:
    unchanged: int          # Clusters identical in both runs
    merged: int             # Clusters absorbed into larger clusters
    partitioned: int        # Clusters split into smaller clusters
    overlapping: int        # Complex reorganization
    rc: int                 # Total reference count
    cc1: int                # Cluster count in run A
    cc2: int                # Cluster count in run B
    sc1: int                # Singleton count in run A
    sc2: int                # Singleton count in run B
    twi: float              # Talburt-Wang Index (1.0 = identical)
    cases: list[ClusterCase]  # Per-cluster details

    def summary(self) -> dict

gm.run_sensitivity(
    file_specs: list,
    config: GoldenMatchConfig,
    sweep_params: list[SweepParam],
    sample_size: int | None = None,
) -> list[SensitivityResult]

results = gm.run_sensitivity(
    file_specs=[("data.csv", "src")],
    config=config,
    sweep_params=[gm.SweepParam("threshold", 0.70, 0.95, 0.05)],
    sample_size=5000,
)
print(results[0].stability_report())

gm.SweepParam(field: str, start: float, stop: float, step: float)

gm.load_config(path: str) -> GoldenMatchConfig

config = gm.load_config("config.yaml")
matchkeys = config.get_matchkeys()

gm.MatchkeyConfig(
    name: str,
    type: Literal["exact", "weighted", "probabilistic"],
    fields: list[MatchkeyField],
    threshold: float | None = None,      # required for weighted
    rerank: bool = False,                 # cross-encoder reranking
    rerank_band: float = 0.1,
    em_iterations: int = 20,             # for probabilistic
)

gm.MatchkeyField(
    field: str,
    scorer: str | None = None,           # jaro_winkler, levenshtein, exact, etc.
    weight: float | None = None,         # required for weighted matchkeys
    transforms: list[str] = [],          # lowercase, strip, soundex, etc.
    columns: list[str] | None = None,    # for record_embedding scorer
    column_weights: dict[str, float] | None = None,
    levels: int = 2,                     # for probabilistic: 2 or 3
)

@dataclass
class DedupeResult:
    golden: pl.DataFrame | None       # Canonical merged records
    clusters: dict[int, dict]         # Cluster info (members, pair_scores, confidence)
    dupes: pl.DataFrame | None        # Duplicate records
    unique: pl.DataFrame | None       # Non-duplicate records
    stats: dict                       # total_records, total_clusters, match_rate
    scored_pairs: list[tuple]         # (id_a, id_b, score) tuples
    config: GoldenMatchConfig
    postflight_report: PostflightReport | None  # v1.5.0: signals + advisories + adjustments

    def to_csv(path, which="golden")  # Write results to CSV
    match_rate: float                 # Property: percentage of dupes
    total_records: int                # Property
    total_clusters: int               # Property

@dataclass
class MatchResult:
    matched: pl.DataFrame | None      # Matched target records with scores
    unmatched: pl.DataFrame | None    # Unmatched target records
    stats: dict
    postflight_report: PostflightReport | None  # v1.5.0: signals + advisories + adjustments

    def to_csv(path)

gm.match_one(record: dict, df: pl.DataFrame, matchkey: MatchkeyConfig) -> list[tuple]

matches = gm.match_one({"name": "John Smith", "zip": "10001"}, existing_df, matchkey)

gm.StreamProcessor(df: pl.DataFrame, config: GoldenMatchConfig)

processor = gm.StreamProcessor(df, config)
matches = processor.process_record(new_record)

gm.run_stream(df, config, records) -> dict

gm.run_dedupe(file_specs: list[tuple], config: GoldenMatchConfig) -> dict
gm.run_match(target_spec: tuple, ref_specs: list[tuple], config: GoldenMatchConfig) -> dict

gm.llm_score_pairs(pairs, df, config: LLMScorerConfig) -> list[tuple]

gm.llm_cluster_pairs(pairs, df, config: LLMScorerConfig) -> list[tuple]

gm.BudgetTracker(max_cost_usd=0.05, max_calls=100)

gm.train_em(df, matchkey, n_sample_pairs=10000, blocking_fields=None) -> EMResult

gm.score_probabilistic(block_df, matchkey, em_result) -> list[tuple]

gm.learn_blocking_rules(df, matchkey, ...) -> LearnedBlockingResult
gm.apply_learned_blocks(df, rules) -> list[pl.DataFrame]

gm.explain_pair(record_a, record_b, field_scores, overall) -> str
gm.explain_cluster(cluster, df) -> str

gm.discover_rulebooks() -> dict[str, DomainRulebook]  # 7 built-in packs
gm.load_rulebook(name) -> DomainRulebook
gm.save_rulebook(name, rulebook)
gm.match_domain(df, column) -> str | None
gm.extract_with_rulebook(df, column, rulebook) -> tuple[pl.DataFrame, pl.DataFrame]

gm.auto_fix_dataframe(df) -> tuple[pl.DataFrame, list]
gm.validate_dataframe(df, config) -> list[dict]
gm.detect_anomalies(df) -> list[dict]
gm.auto_map_columns(df_a, df_b) -> dict[str, str]  # Schema matching

gm.build_lineage(clusters, df, matchkeys) -> dict
gm.save_lineage(lineage, path)
gm.profile_dataframe(df) -> dict

gm.auto_configure(file_specs) -> GoldenMatchConfig
gm.auto_configure_df(
    df,
    llm_provider=None,
    domain_config=None,
    llm_auto=False,
    strict=False,              # v1.5.0
    allow_remote_assets=False, # v1.5.0
) -> GoldenMatchConfig
gm.suggest_threshold(df, matchkey) -> float

from goldenmatch.core.autoconfig import _LAST_CONTROLLER_RUN, auto_configure_df

cfg = auto_configure_df(df)
profile, history = _LAST_CONTROLLER_RUN.get()  # ComplexityProfile, RunHistory
print(history.stop_reason)            # StopReason enum
print(profile.health())                # HealthVerdict enum
for entry in history.entries:
    if entry.decision:
        print(f"iter {entry.iteration}: {entry.decision.rule_name}")

from goldenmatch.web.controller_telemetry import serialize_telemetry

telemetry = serialize_telemetry(
    profile=profile,
    history=history,
    committed_config=cfg,
    source="my_app",
    run_name=None,
    recorded_at=None,
)
# telemetry["stop_reason"], telemetry["health"], telemetry["decisions"], ...

from goldenmatch import AgentSession

session = AgentSession()
result = session.autoconfigure("customers.csv")
# {"config": {...}, "telemetry": {"stop_reason": "green", "health": "green", ...}}

# `deduplicate` / `match_sources` also cache telemetry for follow-up reads.
session.deduplicate("customers.csv")
print(session.last_telemetry)  # same shape as result["telemetry"]

gm.preflight(
    df: pl.DataFrame,
    config: GoldenMatchConfig,
    *,
    profiles: list[ColumnProfile] | None = None,
    allow_remote_assets: bool = False,
) -> PreflightReport

report = gm.preflight(df, config)
for f in report.findings:
    print(f"[{f.severity}] {f.check}: {f.message} (repaired={f.repaired})")

gm.postflight(
    df: pl.DataFrame,
    config: GoldenMatchConfig,
    *,
    pair_scores: list[tuple[int, int, float]],
    current_threshold: float | None = None,
) -> PostflightReport

report = gm.postflight(df, config, pair_scores=scored, current_threshold=0.85)
print(report.signals["threshold_overlap_pct"])
for adj in report.adjustments:
    print(f"{adj.field}: {adj.from_value} -> {adj.to_value} ({adj.reason})")

@dataclass
class PreflightFinding:
    check: str                    # "missing_column" | "cardinality" | "block_size" |
                                  # "remote_asset" | "weight_confidence"
    severity: str                 # "error" | "warning" | "info"
    subject: str                  # column / matchkey name
    message: str
    repaired: bool
    repair_note: str | None

@dataclass
class PreflightReport:
    findings: list[PreflightFinding]
    config_was_modified: bool
    has_errors: bool              # property: True if any unrepaired error

class ConfigValidationError(Exception):
    report: PreflightReport       # full report attached for programmatic inspection

@dataclass
class PostflightAdjustment:
    field: str                    # e.g. "threshold"
    from_value: Any
    to_value: Any
    reason: str
    signal: str                   # which signal motivated the change

@dataclass
class PostflightReport:
    signals: PostflightSignals    # TypedDict, schema below
    adjustments: list[PostflightAdjustment]
    advisories: list[str]

class PostflightSignals(TypedDict):
    score_histogram: ScoreHistogram           # {"bins": list[float], "counts": list[int]}
    blocking_recall: float | Literal["deferred"]  # "deferred" when <10K rows
    block_size_percentiles: BlockSizePercentiles  # {"p50", "p95", "p99", "max"}
    threshold_overlap_pct: float              # fraction of pairs within 0.02 of threshold
    total_pairs_scored: int
    current_threshold: float
    preliminary_cluster_sizes: ClusterSizePercentiles
        # {"p50", "p95", "p99", "max", "count"}
    oversized_clusters: list[OversizedCluster]
        # each: {"cluster_id": int, "size": int, "bottleneck_pair": [int, int]}

gm.boost_accuracy(pairs, df, config) -> dict

gm.generate_diff(before, after) -> dict
gm.rollback_run(run_id) -> dict

gm.run_graph_er(entities, relationships) -> dict

gm.write_output(result, config) -> dict
gm.generate_dedupe_report(result) -> str  # HTML report

import goldenmatch

# Programmatically register a correction (same effect as the review TUI)
goldenmatch.add_correction(
    id_a=42, id_b=87, decision="reject", source="steward",
    reason="Different EIN despite name match", dataset="customers",
)

# Force a learning pass (otherwise auto-runs at next pipeline call)
adjustments = goldenmatch.learn()
print(f"Adjusted {len(adjustments)} matchkey thresholds")

# Inspect what's stored
print(goldenmatch.memory_stats())

# Direct store access (for custom queries / migrations)
store = goldenmatch.get_memory()
for c in store.get_corrections(dataset="customers"):
    print(c.id_a, c.id_b, c.decision, c.trust, c.reason)

result = goldenmatch.dedupe_df(df, config=config)
print(result.memory_stats)
# {'applied': 12, 'stale': 0, 'stale_ambiguous': 0, 'unanchorable': 0}

client = gm.Client("http://localhost:8080")
client.match(record)
client.list_clusters()
client.explain(id_a, id_b)
client.reviews()