Data Governance e Data Quality per AI Affidabile

# data_quality_pipeline.py
# Framework completo di data quality con Great Expectations

import great_expectations as gx
from great_expectations.core.batch import RuntimeBatchRequest
import pandas as pd
import numpy as np
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

class DataQualityFramework:
    """
    Framework di data quality per pipeline ML enterprise.
    Implementa le 6 dimensioni DAMA con Great Expectations.
    """

    def __init__(self, datasource_name: str = "ml_training_data"):
        self.context = gx.get_context()
        self.datasource_name = datasource_name
        self.validation_results = {}

    def build_expectation_suite_for_ml(
        self,
        suite_name: str,
        target_column: str,
        feature_columns: list[str]
    ):
        """
        Crea una suite di expectations per dataset ML.
        Copre tutte le 6 dimensioni DAMA.
        """
        suite = self.context.add_expectation_suite(
            expectation_suite_name=suite_name
        )

        # === DIMENSIONE: COMPLETENESS ===
        # Feature critiche devono avere meno del 5% di valori nulli
        for col in feature_columns:
            suite.add_expectation(
                gx.core.ExpectationConfiguration(
                    expectation_type="expect_column_values_to_not_be_null",
                    kwargs={
                        "column": col,
                        "mostly": 0.95  # 95% non-null
                    }
                )
            )

        # Target column deve essere 100% non-null
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_not_be_null",
                kwargs={"column": target_column}
            )
        )

        # === DIMENSIONE: VALIDITY ===
        # Valori numerici in range atteso (es. eta 0-120)
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_be_between",
                kwargs={
                    "column": "customer_age",
                    "min_value": 18,
                    "max_value": 120,
                    "mostly": 0.999
                }
            )
        )

        # === DIMENSIONE: UNIQUENESS ===
        # Chiavi primarie devono essere uniche
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_be_unique",
                kwargs={"column": "transaction_id"}
            )
        )

        # === DIMENSIONE: CONSISTENCY ===
        # Il target deve avere solo valori attesi (no label noise)
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_be_in_set",
                kwargs={
                    "column": target_column,
                    "value_set": [0, 1],  # Classificazione binaria
                }
            )
        )

        # === DIMENSIONE: ACCURACY - Distribuzione statistica ===
        # La distribuzione non deve essere degenere
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_proportion_of_unique_values_to_be_between",
                kwargs={
                    "column": "customer_id",
                    "min_value": 0.3,   # Almeno 30% utenti unici
                    "max_value": 1.0
                }
            )
        )

        # Bilanciamento classi per training set (max 10:1 ratio)
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_most_common_value_to_be_in_set",
                kwargs={
                    "column": target_column,
                    "value_set": [0, 1],
                    "ties_okay": True
                }
            )
        )

        # === DIMENSIONE: TIMELINESS ===
        # Dati non più vecchi di 90 giorni
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_be_between",
                kwargs={
                    "column": "event_timestamp",
                    "min_value": "2024-11-01",  # Rolling window
                    "max_value": datetime.now().strftime("%Y-%m-%d"),
                    "parse_strings_as_datetimes": True,
                    "mostly": 0.99
                }
            )
        )

        self.context.save_expectation_suite(suite)
        logger.info(f"Suite {suite_name} creata con {len(suite.expectations)} expectations")
        return suite

    def validate_dataset(
        self,
        df: pd.DataFrame,
        suite_name: str,
        run_name: str = None
    ) -> dict:
        """
        Valida un DataFrame contro la suite definita.
        Restituisce risultati strutturati per monitoring.
        """
        run_name = run_name or f"run_{datetime.now().isoformat()}"

        batch_request = RuntimeBatchRequest(
            datasource_name=self.datasource_name,
            data_connector_name="runtime_connector",
            data_asset_name="training_data",
            runtime_parameters={"batch_data": df},
            batch_identifiers={"run_id": run_name}
        )

        checkpoint = self.context.add_or_update_checkpoint(
            name="ml_data_checkpoint",
            validations=[
                {
                    "batch_request": batch_request,
                    "expectation_suite_name": suite_name
                }
            ]
        )

        results = checkpoint.run(run_name=run_name)

        # Estrai metriche aggregate
        validation_result = results.list_validation_results()[0]
        stats = validation_result.statistics

        quality_report = {
            "run_name": run_name,
            "timestamp": datetime.now().isoformat(),
            "success": results.success,
            "evaluated_expectations": stats["evaluated_expectations"],
            "successful_expectations": stats["successful_expectations"],
            "unsuccessful_expectations": stats["unsuccessful_expectations"],
            "success_rate": stats["success_percent"] / 100,
            "failed_checks": [
                {
                    "expectation": r.expectation_config.expectation_type,
                    "column": r.expectation_config.kwargs.get("column"),
                    "details": r.result
                }
                for r in validation_result.results
                if not r.success
            ]
        }

        self.validation_results[run_name] = quality_report
        return quality_report


# Utilizzo nella pipeline
def run_data_quality_checks(raw_data_path: str):
    framework = DataQualityFramework()

    # Definisci le expectations per il dataset
    framework.build_expectation_suite_for_ml(
        suite_name="ml_training_v1",
        target_column="churn_label",
        feature_columns=["customer_age", "tenure_months", "monthly_charges",
                        "total_charges", "contract_type", "payment_method"]
    )

    # Carica e valida i dati
    df = pd.read_parquet(raw_data_path)
    report = framework.validate_dataset(df, "ml_training_v1")

    if not report["success"]:
        failed = report["unsuccessful_expectations"]
        logger.error(f"Data quality FALLITA: {failed} check falliti su {report['evaluated_expectations']}")
        for check in report["failed_checks"]:
            logger.error(f"  - {check['expectation']} su colonna '{check['column']}'")
        raise DataQualityError(f"Dataset non conforme agli standard di qualità")

    logger.info(f"Data quality OK: {report['success_rate']:.1%} success rate")
    return report

# models/schema.yml
# Test di data quality con dbt-expectations

version: 2

models:
  - name: ml_features_customer
    description: "Feature store per modello churn prediction"
    columns:
      - name: customer_id
        description: "Identificativo univoco cliente"
        tests:
          - unique
          - not_null
          - dbt_expectations.expect_column_values_to_be_of_type:
              column_type: VARCHAR

      - name: customer_age
        description: "Eta del cliente in anni"
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 18
              max_value: 120
              mostly: 0.999

      - name: monthly_charges
        description: "Addebiti mensili in EUR"
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 10000
          - dbt_expectations.expect_column_mean_to_be_between:
              min_value: 50
              max_value: 200

      - name: churn_label
        description: "Label target: 1=churned, 0=retained"
        tests:
          - not_null
          - accepted_values:
              values: [0, 1]
          # Verifica che il dataset non sia troppo sbilanciato
          - dbt_expectations.expect_column_proportion_of_unique_values_to_be_between:
              min_value: 0.01  # Almeno 1% di churn (altrimenti dataset anomalo)
              max_value: 0.5   # Max 50% di churn (dataset equilibrato)

      - name: event_date
        description: "Data dell'evento"
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: "2023-01-01"
              max_value: "{{ var('max_date', modules.datetime.date.today().isoformat()) }}"
              parse_strings_as_datetimes: true

  - name: ml_features_customer
    tests:
      # Test a livello di tabella: volumetria attesa
      - dbt_expectations.expect_table_row_count_to_be_between:
          min_value: 10000
          max_value: 10000000

      # Nessuna riga duplicata su chiave composta
      - dbt_expectations.expect_compound_columns_to_be_unique:
          column_list: ["customer_id", "snapshot_date"]

# openmetadata_lineage.py
# Registrazione automatica della data lineage per pipeline ML

from metadata.ingestion.api.workflow import Workflow
from metadata.generated.schema.entity.data.pipeline import Pipeline
from metadata.generated.schema.entity.data.table import Table
from metadata.generated.schema.type.entityLineage import (
    ColumnLineage,
    EntitiesEdge,
    LineageDetails
)
from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest
from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.generated.schema.security.client.openMetadataJWTClientConfig import (
    OpenMetadataJWTClientConfig
)
import json
from typing import Optional


class MLPipelineLineageTracker:
    """
    Tracker automatico di data lineage per pipeline ML.
    Registra ogni trasformazione in OpenMetadata.
    """

    def __init__(self, server_url: str, jwt_token: str):
        from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
            OpenMetadataConnection
        )
        server_config = OpenMetadataConnection(
            hostPort=server_url,
            authProvider="openmetadata",
            securityConfig=OpenMetadataJWTClientConfig(jwtToken=jwt_token)
        )
        self.metadata = OpenMetadata(server_config)

    def register_training_data_lineage(
        self,
        source_tables: list[str],
        feature_store_table: str,
        ml_model_name: str,
        transformation_description: str
    ):
        """
        Registra il percorso: raw_data -> feature_store -> ml_model
        Essenziale per compliance AI Act (Art. 10 - Data Governance)
        """

        # 1. Recupera entità sorgente
        source_entities = []
        for table_fqn in source_tables:
            table = self.metadata.get_by_name(
                entity=Table,
                fqn=table_fqn
            )
            if table:
                source_entities.append(table)

        # 2. Recupera/crea feature store table
        feature_table = self.metadata.get_by_name(
            entity=Table,
            fqn=feature_store_table
        )

        # 3. Costruisci lineage con dettagli di trasformazione
        for source in source_entities:
            lineage_request = AddLineageRequest(
                edge=EntitiesEdge(
                    fromEntity={
                        "id": str(source.id.__root__),
                        "type": "table"
                    },
                    toEntity={
                        "id": str(feature_table.id.__root__),
                        "type": "table"
                    },
                    lineageDetails=LineageDetails(
                        description=transformation_description,
                        columnsLineage=[
                            ColumnLineage(
                                fromColumns=[f"{source.fullyQualifiedName.__root__}.customer_id"],
                                toColumn=f"{feature_store_table}.customer_id"
                            ),
                            ColumnLineage(
                                fromColumns=[f"{source.fullyQualifiedName.__root__}.event_date"],
                                toColumn=f"{feature_store_table}.snapshot_date"
                            )
                        ]
                    )
                )
            )
            self.metadata.add_lineage(lineage_request)

        # 4. Aggiungi tag di compliance (AI Act)
        self._tag_for_ai_act_compliance(feature_store_table, ml_model_name)

    def _tag_for_ai_act_compliance(
        self,
        table_fqn: str,
        model_name: str
    ):
        """
        Aggiunge tag AI Act per tracciabilita compliance.
        Essenziale per sistemi high-risk (AI Act Art. 10).
        """
        table = self.metadata.get_by_name(entity=Table, fqn=table_fqn)

        # Tag strutturati per AI Act
        compliance_tags = {
            "ai_act_article_10": "training_data",
            "ai_system": model_name,
            "data_governance_reviewed": "true",
            "bias_assessment_date": "2025-01-15",
            "data_quality_score": "98.5"
        }

        # Aggiorna custom properties
        self.metadata.patch_entity(
            entity=Table,
            source=table,
            destination=table.copy(
                update={"customMetrics": compliance_tags}
            )
        )

        print(f"Compliance tags aggiunti per {table_fqn} -> modello {model_name}")


# Utilizzo pratico
def setup_ml_lineage():
    tracker = MLPipelineLineageTracker(
        server_url="http://openmetadata.internal:8585/api",
        jwt_token="your-jwt-token"
    )

    tracker.register_training_data_lineage(
        source_tables=[
            "default.raw_crm.customers",
            "default.raw_billing.transactions",
            "default.raw_support.tickets"
        ],
        feature_store_table="default.feature_store.ml_churn_features_v3",
        ml_model_name="churn_prediction_xgboost_v2",
        transformation_description="Feature engineering per churn prediction: "
                                   "aggregazioni mensili, calcolo tenure, "
                                   "encoding categoriche. Approvato data steward 2025-01-15."
    )

# ai_act_compliance_checker.py
# Verifica compliance AI Act Art. 10 per dataset di training

from dataclasses import dataclass, field
from typing import Optional
import pandas as pd
import numpy as np
from scipy import stats
import json
from datetime import datetime


@dataclass
class DatasetComplianceReport:
    """Report di compliance AI Act Articolo 10."""
    dataset_name: str
    assessment_date: str
    assessor: str
    compliant: bool = False
    checks: dict = field(default_factory=dict)
    recommendations: list = field(default_factory=list)


class AIActArticle10Checker:
    """
    Verifica conformità AI Act Art. 10 per dataset di training high-risk.

    Art. 10 richiede che i dataset siano:
    1. Rilevanti e sufficientemente rappresentativi
    2. Liberi da errori per quanto possibile
    3. Completi rispetto allo scopo
    4. Con proprietà statistiche appropriate
    5. Privi di bias che potrebbero discriminare gruppi protetti
    """

    def __init__(self, dataset_name: str, assessor: str):
        self.report = DatasetComplianceReport(
            dataset_name=dataset_name,
            assessment_date=datetime.now().isoformat(),
            assessor=assessor
        )

    def check_representativeness(
        self,
        df: pd.DataFrame,
        demographic_columns: list[str],
        reference_distributions: dict
    ) -> bool:
        """
        Art. 10(3): Verifica rappresentativita demografica.
        Confronta distribuzione del dataset con popolazione di riferimento.
        """
        is_representative = True
        self.report.checks["representativeness"] = {}

        for col in demographic_columns:
            if col not in df.columns:
                continue

            # Distribuzione nel dataset
            observed = df[col].value_counts(normalize=True).to_dict()

            # Distribuzione attesa (es. ISTAT per demografia italiana)
            expected = reference_distributions.get(col, {})

            if not expected:
                continue

            # Test chi-quadro per significativita statistica
            categories = list(set(observed.keys()) | set(expected.keys()))
            obs_counts = [df[col].value_counts().get(c, 0) for c in categories]
            exp_props = [expected.get(c, 0.001) for c in categories]

            # Normalizza expected
            total_exp = sum(exp_props)
            exp_normalized = [p / total_exp * len(df) for p in exp_props]

            chi2, p_value = stats.chisquare(obs_counts, exp_normalized)

            deviation = {
                "chi2_statistic": chi2,
                "p_value": p_value,
                "is_representative": p_value > 0.05,  # p > 0.05 = no significant deviation
                "observed_distribution": observed,
                "expected_distribution": expected
            }

            self.report.checks["representativeness"][col] = deviation

            if p_value <= 0.05:
                is_representative = False
                self.report.recommendations.append(
                    f"CRITICO: Colonna '{col}' non e rappresentativa della popolazione target "
                    f"(chi2={chi2:.2f}, p={p_value:.4f}). "
                    f"Richiesta oversampling o raccolta dati aggiuntiva."
                )

        return is_representative

    def check_bias_protected_attributes(
        self,
        df: pd.DataFrame,
        target_column: str,
        protected_attributes: list[str]
    ) -> bool:
        """
        Art. 10(5): Verifica assenza bias su attributi protetti.
        Usa disparate impact ratio (soglia 0.8 = 80% rule).
        """
        is_unbiased = True
        self.report.checks["bias_assessment"] = {}

        for attr in protected_attributes:
            if attr not in df.columns:
                continue

            groups = df[attr].unique()
            positive_rates = {}

            for group in groups:
                mask = df[attr] == group
                group_df = df[mask]
                if len(group_df) == 0:
                    continue
                positive_rate = (group_df[target_column] == 1).mean()
                positive_rates[str(group)] = positive_rate

            if len(positive_rates) < 2:
                continue

            max_rate = max(positive_rates.values())
            min_rate = min(positive_rates.values())

            # Disparate Impact = min_rate / max_rate
            # AI Act richiede > 0.8 (80% rule)
            disparate_impact = min_rate / max_rate if max_rate > 0 else 1.0

            bias_result = {
                "disparate_impact": disparate_impact,
                "compliant": disparate_impact >= 0.8,
                "group_rates": positive_rates,
                "max_group": max(positive_rates, key=positive_rates.get),
                "min_group": min(positive_rates, key=positive_rates.get)
            }

            self.report.checks["bias_assessment"][attr] = bias_result

            if disparate_impact < 0.8:
                is_unbiased = False
                self.report.recommendations.append(
                    f"CRITICO: Bias rilevato su attributo protetto '{attr}'. "
                    f"Disparate impact = {disparate_impact:.3f} (soglia AI Act: 0.80). "
                    f"Applicare re-weighting, resampling o fairness constraints."
                )

        return is_unbiased

    def check_completeness_and_errors(
        self,
        df: pd.DataFrame,
        critical_features: list[str]
    ) -> bool:
        """
        Art. 10(2): Verifica completezza e assenza di errori.
        """
        is_complete = True
        self.report.checks["completeness"] = {}

        for col in critical_features:
            null_rate = df[col].isna().mean()
            is_col_complete = null_rate < 0.05  # Soglia 5%

            self.report.checks["completeness"][col] = {
                "null_rate": null_rate,
                "compliant": is_col_complete
            }

            if not is_col_complete:
                is_complete = False
                self.report.recommendations.append(
                    f"AVVISO: Feature '{col}' ha {null_rate:.1%} valori mancanti. "
                    f"Applicare imputation strategy o raccogliere dati aggiuntivi."
                )

        return is_complete

    def generate_compliance_report(self) -> str:
        """Genera report JSON per audit AI Act."""
        self.report.compliant = all([
            check.get("is_representative", True)
            for check in self.report.checks.get("representativeness", {}).values()
        ] + [
            check.get("compliant", True)
            for check in self.report.checks.get("bias_assessment", {}).values()
        ] + [
            check.get("compliant", True)
            for check in self.report.checks.get("completeness", {}).values()
        ])

        return json.dumps({
            "dataset": self.report.dataset_name,
            "assessment_date": self.report.assessment_date,
            "assessor": self.report.assessor,
            "ai_act_article_10_compliant": self.report.compliant,
            "checks": self.report.checks,
            "recommendations": self.report.recommendations,
            "next_review_date": "2025-04-15"
        }, indent=2, default=str)


# Esempio di utilizzo per un sistema HR high-risk
def assess_hr_model_dataset(df: pd.DataFrame):
    checker = AIActArticle10Checker(
        dataset_name="hr_promotion_training_v3",
        assessor="Maria Rossi, Data Steward HR"
    )

    # Distribuzione di riferimento: composizione forza lavoro italiana
    reference_distributions = {
        "gender": {"M": 0.54, "F": 0.46},
        "age_group": {"18-30": 0.20, "31-45": 0.40, "46-60": 0.35, "60+": 0.05}
    }

    # Esegui tutti i check
    checker.check_representativeness(df, ["gender", "age_group"], reference_distributions)
    checker.check_bias_protected_attributes(df, "promoted", ["gender", "age_group", "nationality"])
    checker.check_completeness_and_errors(df, ["performance_score", "tenure", "education_level"])

    report = checker.generate_compliance_report()
    print(report)
    return report

# soda_checks_ml_pipeline.yml
# Check Soda Core per data observability pipeline ML

checks for ml_features_customer:

  # === FRESHNESS: timeliness dei dati ===
  - freshness(event_date) < 24h:
      name: "Dati non più vecchi di 24 ore"
      fail:
        when freshness(event_date) > 24h
      warn:
        when freshness(event_date) > 12h

  # === VOLUME: anomalie di volumetria ===
  - row_count > 50000:
      name: "Minimo 50K record per inference affidabile"

  - row_count between 50000 and 5000000:
      name: "Volume nel range atteso (no spike anomali)"

  # === COMPLETENESS ===
  - missing_count(customer_age) = 0:
      name: "Customer age: zero null tollerati"

  - missing_percent(monthly_charges) < 2%:
      name: "Monthly charges: max 2% null"

  # === VALIDITY ===
  - invalid_percent(customer_age) < 0.1%:
      name: "Customer age nel range 18-120"
      valid min: 18
      valid max: 120

  - duplicate_count(customer_id) = 0:
      name: "Nessun customer_id duplicato"

  # === DISTRIBUTION DRIFT ===
  # Confronto con baseline storica (riferimento: settimana precedente)
  - distribution_difference_index(monthly_charges) < 0.1:
      name: "Monthly charges: drift < 10% vs baseline"
      method: psi  # Population Stability Index
      baseline: soda://ml_features_customer/monthly_charges_baseline

  - distribution_difference_index(customer_age) < 0.1:
      name: "Customer age: drift < 10% vs baseline"
      method: ks   # Kolmogorov-Smirnov test

  # === BUSINESS RULES ===
  - failed_rows(monthly_charges_anomaly):
      name: "Nessun addebito negativo"
      fail condition: monthly_charges < 0

  - failed_rows(contract_consistency):
      name: "Contratti annuali: addebito >= 100 EUR"
      fail condition: contract_type = 'annual' AND monthly_charges < 100

# Configurazione alerting
alert config:
  slack:
    webhook: "https://hooks.slack.com/services/..."
    channel: "#data-quality-alerts"
  email:
    to:
      - "data-team@azienda.it"
      - "ml-team@azienda.it"
  severity:
    fail: immediate
    warn: daily_digest