Merge pull request #41 from FreshCode-Org/lean-core

Lean core

Author: JohnnyWilson16

CONTRIBUTING_DOMAINS.md

@@ -0,0 +1,102 @@
+# Contributing a domain validator pack
+
+A *domain pack* teaches `fd.clean(df, domain="…")` how to validate and repair a
+specific kind of tabular data against versioned, reviewable rules. Built-in packs
+live in `src/freshdata/domains/<name>/`; third-party packs ship as separate
+PyPI distributions and register through an entry point. Both implement the same
+interface.
+
+## The interface
+
+Every pack implements `freshdata.domains.DomainValidator`:
+
+```python
+class DomainValidator(ABC):
+    domain_name: str        # "finance"
+    version: str            # "0.1.0"
+    schema_version: str     # "2024-01"
+
+    def detect_columns(self, df) -> ColumnMapping: ...
+    def validate(self, df) -> ValidationReport: ...
+    def repair(self, df, report) -> tuple[DataFrame, RepairLog]: ...
+    def describe(self) -> dict: ...
+```
+
+Most packs subclass **`ConfigDrivenValidator`**, which implements the layered
+engine for you — you supply a `rules.yaml`, declare canonical fields, and add any
+custom checks. The finance pack (`src/freshdata/domains/finance/`) is the
+reference example.
+
+## Anatomy of a pack
+
+```
+mypack/
+  __init__.py        # exports your Validator class
+  validator.py       # subclass ConfigDrivenValidator; register custom checks
+  rules.yaml         # the rules (see below)
+  reference/*.json   # bundled code sets, each with a _meta block
+```
+
+### `validator.py`
+
+```python
+from freshdata.domains import ConfigDrivenValidator
+
+class MyValidator(ConfigDrivenValidator):
+    domain_name = "mydomain"
+    version = "0.1.0"
+    schema_version = "2024-01"
+
+    canonical_fields = ("id", "amount", "code")
+    required_fields = ("id", "amount")
+    id_fields = ("id",)                      # never repaired
+    aliases = {"id": (r"my_?id", r"identifier")}   # regex, case-insensitive
+    rules_path = str(Path(__file__).parent / "rules.yaml")
+
+    def register_extensions(self):
+        self.register_check("my_check", self._my_check)        # params.func
+        self.register_repair("my_fix", self._my_fix)           # repair_params.func
+
+    def load_reference_values(self, name):                     # for reference checks
+        ...
+```
+
+### `rules.yaml`
+
+Each rule carries `id`, `name`, `layer` (`schema|format|reference|business|
+semantic`), `severity` (`error|warning|info`), `field`/`fields`, `check`
+(`not_null|required|regex|enum|reference|range|custom`), optional `params`, and
+an optional `repair` (`fill_default|coerce|flag_only|reject|none`) with
+`repair_params`. Layers always run in order; a rule whose target column is
+absent is skipped (the schema layer reports it as `MISSING_REQUIRED_FIELD`).
+
+### Reference data
+
+Bundle code sets as static JSON with a `_meta` block so they stay auditable:
+
+```json
+{ "_meta": {"source": "ISO 4217", "retrieved_date": "2024-01-15", "version": "2024-01"},
+  "codes": ["USD", "EUR", "..."] }
+```
+
+## Rules of the road
+
+- **Validation never mutates data.** All changes happen in `repair`, and every
+  attempt is logged to the `RepairLog` with from/to values, the rule id, and a
+  status (`applied`/`flagged`/`unresolvable`).
+- **Identifier columns are never repaired** — list them in `id_fields`.
+- **Never guess column mappings silently** — detection logs how each match was made.
+- **No network calls or LLM calls** in a pack.
+
+## Registering a third-party pack
+
+Expose your validator through the `freshdata.domains` entry-point group in your
+package's `pyproject.toml`:
+
+```toml
+[project.entry-points."freshdata.domains"]
+mydomain = "mypack.validator:MyValidator"
+```
+
+Once installed, `fd.clean(df, domain="mydomain")` finds it automatically.
+Built-in names take precedence, so you cannot shadow `finance` (etc.).

pyproject.toml

@@ -70,6 +70,10 @@ semantic = [
 cli = [
     "pyyaml>=5.1",
 ]
+# Domain validator packs (finance/retail/transport) read their rules from YAML.
+domains = [
+    "pyyaml>=5.1",
+]
 cleanlab = [
     "cleanlab>=2.0",
     "scikit-learn>=1.0",
@@ -113,6 +117,12 @@ docs = [
 [project.scripts]
 freshdata = "freshdata.enterprise.cli:main"
 
+# Built-in domain packs are also exposed here so third-party packs can register
+# the same way (the registry resolves built-ins directly and does not require
+# this entry to be installed).
+[project.entry-points."freshdata.domains"]
+finance = "freshdata.domains.finance:FinanceValidator"
+
 [project.urls]
 Homepage = "https://freshcode-org.github.io/freshdata/"
 Documentation = "https://freshcode-org.github.io/freshdata/"
@@ -193,6 +203,8 @@ ignore = [
 "src/freshdata/enterprise/interface.py" = ["PLC0415"]
 "src/freshdata/enterprise/cli.py" = ["PLC0415", "PLR0915"]
 "src/freshdata/__init__.py" = ["PLC0415"]
+# Domain packs defer the PyYAML import so the base infra stays dependency-free.
+"src/freshdata/domains/base.py" = ["PLC0415"]
 
 [tool.mypy]
 python_version = "3.10"

src/freshdata/api.py

@@ -5,8 +5,9 @@
 import pandas as pd
 
 from .adapters.polars import from_pandas, to_pandas
-from .cleaner import Cleaner
+from .cleaner import Cleaner, run_pipeline
 from .config import CleanConfig, merge_options
+from .domains import SEVERITY_TO_RISK, DomainOutcome, run_domain
 from .engine.context import build_contexts
 from .engine.model_select import EngineMode, rank_missing_models
 from .plan import suggest_plan
@@ -19,6 +20,8 @@ def clean(
     *,
     config: CleanConfig | None = None,
     return_report: bool = False,
+    domain: str | None = None,
+    column_map: dict[str, str] | None = None,
     **options: object,
 ) -> pd.DataFrame | tuple[pd.DataFrame, CleanReport]:
     """Clean a DataFrame and return a new, repaired one.
@@ -54,6 +57,16 @@ def clean(
         If True, return ``(cleaned_df, CleanReport)``. The report carries
         per-action rationale/risk/confidence, missing counts before/after,
         warnings, and recommendations for manual review.
+    domain:
+        Optional domain validator pack (e.g. ``"finance"``). When set, generic
+        cleaning runs first (defaulting to ``strategy="conservative"`` so the
+        statistical engine never silently alters ledgers/IDs unless you pass an
+        explicit ``strategy``), then the pack validates in layers and repairs
+        separately; findings and a ``domain_trust_score`` are folded into the
+        report. Unknown names raise :class:`~freshdata.domains.UnknownDomainError`.
+    column_map:
+        Optional ``{actual_column: canonical_field}`` overrides for the domain
+        pack's column detection. Requires ``domain`` to be set.
     **options:
         Any :class:`~freshdata.CleanConfig` field as a keyword override — e.g.
         ``strategy`` (``"balanced"`` default / ``"aggressive"`` / ``"conservative"``),
@@ -71,7 +84,15 @@ def clean(
 
     >>> fd.clean(df, outlier_action="flag", target_column="churn",
     ...          preserve_columns=("notes",), verbose=False)
+
+    >>> ledger = fd.clean(df, domain="finance")          # validate + repair
+    >>> ledger, rep = fd.clean(df, domain="finance", return_report=True)
+    >>> rep.domain_trust_score                            # 0–1
     """
+    if domain is not None:
+        return _clean_with_domain(df, domain, column_map, config, return_report, options)
+    if column_map is not None:
+        raise TypeError("column_map requires a domain= to be set")
     cleaner = Cleaner(config=config, **options)
     result = cleaner.clean(df, report=return_report)
     if return_report:
@@ -80,6 +101,65 @@ def clean(
     return from_pandas(result, df)
 
 
+def _clean_with_domain(
+    df: pd.DataFrame,
+    domain: str,
+    column_map: dict[str, str] | None,
+    config: CleanConfig | None,
+    return_report: bool,
+    options: dict[str, object],
+) -> pd.DataFrame | tuple[pd.DataFrame, CleanReport]:
+    """Generic clean (conservative by default) then domain validate + repair."""
+    # With an explicit config the caller owns every setting. Otherwise default to
+    # a conservative base that does *not* infer dtypes: the domain pack owns
+    # format validation/coercion (per its audited rules), and generic dtype
+    # inference would otherwise silently retype dates/amounts before validation.
+    if config is None:
+        options = {
+            "strategy": "conservative",
+            "fix_dtypes": False,
+            **options,  # explicit caller options win
+        }
+    cfg = merge_options(config, **options)
+    cleaned, rep = run_pipeline(df, cfg)
+    repaired, outcome = run_domain(cleaned, domain, column_map=column_map)
+    _fold_domain_outcome(rep, outcome)
+    if cfg.verbose:
+        print(rep.brief())
+    out = from_pandas(repaired, df)
+    return (out, rep) if return_report else out
+
+
+def _fold_domain_outcome(rep: CleanReport, outcome: DomainOutcome) -> None:
+    """Merge a domain run's findings/repairs into the existing CleanReport."""
+    report = outcome.report
+    rep.domain = outcome.domain
+    rep.domain_trust_score = outcome.trust_score
+    rep.domain_findings = [r.to_dict() for r in report.results]
+    rep.domain_repairs = [a.to_dict() for a in outcome.repairs.actions]
+    for result in report.results:
+        if not result.violated:
+            continue
+        col = report.mapping.actual(result.fields[0]) if result.fields else None
+        rep.add(
+            step=f"domain:{outcome.domain}:{result.rule_id}",
+            description=result.message or result.name,
+            column=col,
+            count=result.n_violations,
+            risk=SEVERITY_TO_RISK.get(result.severity, "low"),
+            rationale=result.name,
+        )
+        if result.severity == "error":
+            rep.add_warning(
+                f"[{outcome.domain}] {result.rule_id}: {result.message or result.name}"
+            )
+    applied = sum(1 for a in outcome.repairs.actions if a.status == "applied")
+    if applied:
+        rep.add_recommendation(
+            f"{outcome.domain}: {applied} domain repair(s) applied — see domain_repairs"
+        )
+
+
 def _engine_mode(cfg: CleanConfig) -> EngineMode:
     mode = cfg.engine_mode or "balanced"
     return "balanced" if mode == "balanced" else "aggressive"

src/freshdata/domains/__init__.py

@@ -0,0 +1,100 @@
+"""Domain-specific validator packs for :func:`freshdata.clean`.
+
+A domain pack validates and (separately) repairs a specific kind of tabular data
+against versioned, config-driven rules, extending the existing clean audit trail
+with domain findings and a trust score. Use it through the normal entry point::
+
+    import freshdata as fd
+    df_out = fd.clean(df, domain="finance")
+    df_out, report = fd.clean(df, domain="finance", return_report=True)
+
+Third-party packs register via the ``freshdata.domains`` entry-point group; see
+``CONTRIBUTING_DOMAINS.md``.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+import pandas as pd
+
+from .base import (
+    LAYERS,
+    MISSING_REQUIRED_FIELD,
+    SEVERITIES,
+    SEVERITY_TO_RISK,
+    ColumnMapping,
+    ConfigDrivenValidator,
+    DomainError,
+    DomainValidator,
+    RepairAction,
+    RepairLog,
+    Rule,
+    RuleResult,
+    ValidationReport,
+)
+from .registry import (
+    UnknownDomainError,
+    available,
+    get_validator,
+    register,
+)
+
+__all__ = [
+    "LAYERS",
+    "MISSING_REQUIRED_FIELD",
+    "SEVERITIES",
+    "SEVERITY_TO_RISK",
+    "ColumnMapping",
+    "ConfigDrivenValidator",
+    "DomainError",
+    "DomainOutcome",
+    "DomainValidator",
+    "RepairAction",
+    "RepairLog",
+    "Rule",
+    "RuleResult",
+    "UnknownDomainError",
+    "ValidationReport",
+    "available",
+    "get_validator",
+    "register",
+    "run_domain",
+]
+
+
+@dataclass
+class DomainOutcome:
+    """Everything a domain run produced, beyond the repaired frame."""
+
+    description: dict[str, Any]
+    report: ValidationReport
+    repairs: RepairLog
+
+    @property
+    def domain(self) -> str:
+        return self.report.domain
+
+    @property
+    def trust_score(self) -> float:
+        return self.report.domain_trust_score
+
+
+def run_domain(
+    df: pd.DataFrame,
+    domain: str,
+    *,
+    column_map: Mapping[str, str] | None = None,
+) -> tuple[pd.DataFrame, DomainOutcome]:
+    """Validate then (separately) repair *df* with the named domain pack.
+
+    Returns ``(repaired_df, outcome)``. Validation never mutates *df*; repair
+    runs afterward and never touches identifier columns. Raises
+    :class:`UnknownDomainError` if *domain* is not registered.
+    """
+    validator = get_validator(domain, column_map=column_map)
+    report = validator.validate(df)
+    repaired, repairs = validator.repair(df, report)
+    return repaired, DomainOutcome(validator.describe(), report, repairs)