Examples

1import json
2import featuresmith as fs
3
4def run_featuresmith_pipeline(data_path: str, target: str):
5    # 1. Load data safely into Dataset layer
6    print(f"Loading data from {data_path}...")
7    dataset = fs.load(data_path)
8
9    # 2. Run data audit checks (load → profile → rules evaluation)
10    print("Running rule audit...")
11    result = fs.analyze(
12        dataset,
13        target_column=target,
14        rule_config={
15            "quality.missing_value_threshold": {"threshold": 10.0},
16            "statistical.high_correlation": {"threshold": 0.85}
17        }
18    )
19
20    # 3. Handle rules results
21    print(f"Audit completed. Findings: {len(result.findings)}")
22    for finding in result.findings:
23        print(f"[{finding.severity.upper()}] Column: {finding.column_name}")
24        print(f"  Issue  : {finding.title}")
25        print(f"  Detail : {finding.description}")
26
27    # 4. Serialize result to dictionary/JSON
28    report_dict = result.to_dict()
29    with open("report.json", "w") as f:
30        json.dump(report_dict, f, indent=2, default=str)
31
32if __name__ == "__main__":
33    run_featuresmith_pipeline("customers.csv", target="churn")

1# .github/workflows/data-quality-gate.yml
2name: Data Quality Gate
3
4on:
5  push:
6    branches: [ main ]
7  schedule:
8    - cron: '0 0 * * *' # Run daily audits
9
10jobs:
11  audit:
12    runs-on: ubuntu-latest
13    steps:
14      - name: Checkout code
15        uses: actions/checkout@v4
16
17      - name: Set up Python
18        uses: actions/setup-python@v5
19        with:
20          python-version: '3.11'
21
22      - name: Install dependencies
23        run: |
24          pip install featuresmith-core featuresmith-cli
25
26      - name: Audit dataset quality
27        run: |
28          # Gate CI build: if critical rule violations exist, exit code 1 fails step
29          featuresmith analyze data/incoming_leads.csv --target converted --severity critical --output audit_report.txt
30
31      - name: Upload audit report
32        if: always()
33        uses: actions/upload-artifact@v4
34        with:
35          name: data-audit-report
36          path: audit_report.txt

1from featuresmith.core.profile_result import ProfileResult
2from featuresmith.rules.base import BaseRule
3from featuresmith.core.rule_finding import RuleFinding
4
5class ZeroVarianceRule(BaseRule):
6    """Detect numeric columns with zero standard deviation."""
7
8    @property
9    def id(self) -> str:
10        return "statistical.zero_variance"
11
12    @property
13    def name(self) -> str:
14        return "Zero Variance Columns"
15
16    @property
17    def description(self) -> str:
18        return "Flags numeric columns with no observed variance."
19
20    @property
21    def category(self) -> str:
22        return "statistical"
23
24    @property
25    def severity(self) -> str:
26        return "warning"
27
28    @property
29    def enabled_by_default(self) -> bool:
30        return True
31
32    def evaluate(self, profile: ProfileResult) -> list[RuleFinding]:
33        findings: list[RuleFinding] = []
34        for col_name, numeric_profile in profile.numeric_profiles.items():
35            if numeric_profile.std_dev == 0.0:
36                    findings.append(
37                        RuleFinding(
38                            rule_id=self.id,
39                            rule_name=self.name,
40                            category=self.category,
41                            severity=self.severity,
42                            column_name=col_name,
43                            title="Zero Variance Detected",
44                            description=f"Column '{col_name}' has standard deviation of 0.0 (no variance).",
45                            evidence={"std_dev": numeric_profile.std_dev}
46                        )
47                    )
48        return findings