data_layer.py

"""SQL data layer: load the UCI credit card default data into SQLite and run theSQL in ./sql against it. The source copy (Yellowbrick's mirror of UCI dataset 350, checksum verified indata/SOURCE.md) labels the six monthly columns in reverse: its "apr_*" columnholds September 2005, matching UCI attribute X6. Columns are therefore read byposition and named in the UCI order here."""import csvimport gzipimport sqlite3from pathlib import Path import pandas as pd HERE = Path(__file__).resolve().parentDATA = HERE / "data" / "credit.csv.gz"SQL = HERE / "sql" MONTHS = ["sep", "aug", "jul", "jun", "may", "apr"]  # UCI order: X6 = SeptemberCOLUMNS = (    ["credit_limit", "sex", "education", "marital_status", "age"]    + [f"status_{m}" for m in MONTHS]    + [f"bill_{m}" for m in MONTHS]    + [f"paid_{m}" for m in MONTHS]    + ["default_next_month"])  def build_database(path: Path) -> sqlite3.Connection:    """Create the accounts table and load every row, in UCI column order."""    con = sqlite3.connect(path)    con.executescript((SQL / "01_schema.sql").read_text())    with gzip.open(DATA, "rt", newline="") as f:        reader = csv.reader(f)        header = next(reader)        if len(header) != len(COLUMNS):            raise ValueError(f"expected {len(COLUMNS)} columns, found {len(header)}")        rows = [[float(v) for v in row] for row in reader]    placeholders = ", ".join("?" * len(COLUMNS))    con.executemany(f"INSERT INTO accounts ({', '.join(COLUMNS)}) VALUES ({placeholders})", rows)    con.executescript((SQL / "03_features.sql").read_text())    con.commit()    return con  def query(con: sqlite3.Connection, name: str) -> pd.DataFrame:    """Run one of the .sql files and return its result as a DataFrame."""    return pd.read_sql_query((SQL / name).read_text(), con)  def load(path: Path):    """Build the database and return (connection, quality checks, segments, features)."""    con = build_database(path)    quality = query(con, "02_quality.sql")    segments = query(con, "04_segments.sql")    features = pd.read_sql_query("SELECT * FROM account_features ORDER BY account_id", con)    return con, quality, segments, features  if __name__ == "__main__":    db = HERE / "output" / "credit.db"    db.parent.mkdir(exist_ok=True)    db.unlink(missing_ok=True)    _, quality, segments, features = load(db)    print(quality.to_string(index=False))    print(segments.to_string(index=False))    print(features.describe().T[["mean", "min", "max"]].to_string())