Spaces:
Paused
Paused
File size: 3,812 Bytes
5a3b9db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | """
Implements: 02_Data/06_SYNTHETIC_DATA_GENERATION.md
"""
import pandas as pd
import logging
from typing import Dict, Any
logger = logging.getLogger(__name__)
class DataValidator:
"""
Validates synthetic datasets against interface contracts.
Implements Stage 9 of the Data Generation Pipeline.
"""
def validate_auth_logs(self, df: pd.DataFrame) -> bool:
required_columns = [
"event_id", "timestamp", "user_id", "device_id",
"session_id", "resource_id", "ip_address", "country",
"city", "latitude", "longitude", "authentication_method",
"authentication_result", "is_attack"
]
missing_cols = [c for c in required_columns if c not in df.columns]
if missing_cols:
logger.error(f"Missing columns: {missing_cols}")
return False
required_no_nulls = [c for c in required_columns if c != "failure_reason"]
if df[required_no_nulls].isnull().any().any():
logger.error("Missing values in required columns.")
return False
if not df["ip_address"].str.match(r"^\d{1,3}(\.\d{1,3}){3}$").all():
logger.error("Invalid IP addresses found.")
return False
if not df["timestamp"].is_monotonic_increasing:
logger.error("Timestamps are not strictly chronological.")
return False
logger.info("Validation passed successfully.")
return True
def validate_all(self, datasets: Dict[str, pd.DataFrame]) -> bool:
if "authentication_events" not in datasets:
return False
return self.validate_auth_logs(datasets["authentication_events"])
def generate_validation_report(self, datasets: Dict[str, pd.DataFrame], features_df: pd.DataFrame = None) -> Dict[str, Any]:
"""
Generates a comprehensive pipeline validation report containing:
- Generated artifacts
- Schema validation results
- Record counts
- Feature counts
- Data quality summary
- Validation errors
- Assumptions made
"""
report = {
"generated_artifacts": list(datasets.keys()),
"schema_validation": {},
"record_counts": {name: len(df) for name, df in datasets.items()},
"feature_counts": len(features_df.columns) if features_df is not None else 0,
"data_quality_summary": {},
"validation_errors": [],
"assumptions_made": [
"Synthetic data simulates enterprise behavior and represents chronological order.",
"Nulls in 'failure_reason' are structurally valid for successful logins.",
"Feature generation correctly maps historical aggregates for profiling."
]
}
for name, df in datasets.items():
if name == "authentication_events":
is_valid = self.validate_auth_logs(df)
report["schema_validation"][name] = "Passed" if is_valid else "Failed"
if not is_valid:
report["validation_errors"].append("authentication_events failed schema validation.")
else:
report["schema_validation"][name] = "Not Evaluated (No strict schema enforced)"
report["data_quality_summary"][name] = {
"missing_values": int(df.isnull().sum().sum()),
"duplicate_rows": int(df.duplicated().sum())
}
if features_df is not None:
report["generated_artifacts"].append("engineered_features")
report["record_counts"]["engineered_features"] = len(features_df)
return report
|