kumar6591/data-quality-env

1from tasks.base import BaseTask2from env.models import AuditReport3 4 5class Task2(BaseTask):6    def get_description(self) -> str:7        return (8            "Audit the 'orders' table. Detect: (1) type violations (amounts stored as strings "9            "like '$12.50', dates in human-readable format), (2) range violations (negative "10            "quantity), (3) unparseable values in amount field. Report each violation type, "11            "an example value, and your confidence."12        )13 14    def get_table_names(self) -> list[str]:15        return ["orders"]16 17    def grade(self, report: AuditReport, gold: dict) -> tuple[float, dict]:18        scores: dict[str, float] = {}19 20        amt_detected = any(21            "amount" in str(v.get("column", "")).lower() and "type" in str(v.get("issue_type", "")).lower()22            for v in report.schema_violations23        )24        conf = next((float(v.get("confidence", 0.5)) for v in report.schema_violations if "amount" in str(v.get("column", "")).lower()), 0.5)25        scores["amount_type"] = self.brier_adjust(1.0 if amt_detected else 0.0, conf, amt_detected)26 27        date_detected = any("date" in str(v.get("column", "")).lower() for v in report.schema_violations)28        conf = next((float(v.get("confidence", 0.5)) for v in report.schema_violations if "date" in str(v.get("column", "")).lower()), 0.5)29        scores["date_format"] = self.brier_adjust(1.0 if date_detected else 0.0, conf, date_detected)30 31        neg_qty_violations = [32            v33            for v in report.schema_violations34            if "quantity" in str(v.get("column", "")).lower() and "negative" in str(v.get("issue_type", "")).lower()35        ]36        if neg_qty_violations:37            reported_count = int(neg_qty_violations[0].get("count", 0))38            acc = self.count_accuracy(reported_count, int(gold["negative_quantity_rows"]))39            conf = float(neg_qty_violations[0].get("confidence", 0.5))40            scores["neg_qty"] = self.brier_adjust(acc, conf, acc > 0.5)41        else:42            scores["neg_qty"] = 0.043 44        bad_detected = any(45            "unparseable" in str(v.get("issue_type", "")).lower()46            or ("amount" in str(v.get("column", "")).lower() and "invalid" in str(v.get("issue_type", "")).lower())47            for v in report.schema_violations48        )49        scores["bad_amount"] = self.brier_adjust(0.8 if bad_detected else 0.0, 0.5, bad_detected)50 51        scores = {k: self.strict_score(v) for k, v in scores.items()}52 53        weights = {"amount_type": 0.25, "date_format": 0.25, "neg_qty": 0.25, "bad_amount": 0.25}54        total = sum(scores[k] * weights[k] for k in weights)55        return self.strict_score(round(total, 4)), scores56