{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.12.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[],"dockerImageVersionId":28755,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"\"\"\"建立 train/test 建模資料集 — WOE/IV 特徵組(依 build_model_dataset_WOEIV.md)。\n\n本檔是 build_model_dataset.py 的獨立變體(不覆蓋原檔/原輸出),供 WOE/IV 特徵工程使用。\n欄位清單改以 credit_bureau_a_2 為重心(56 個 L1×L2 組合特徵)+ credit_bureau_a_1(21)+\nstatic_0 直取(12)+ applprev_1(3)+ reject_rate + 3 個財務比率;7 個財務基礎欄僅作為比率\n輸入,算完即從輸出移除。\n\n本檔以【資料工程 DATA ENGINEERING】與【特徵工程 FEATURE ENGINEERING】兩大區塊組織:\n\n【資料工程】把原始分區檔整理成乾淨、正確型別的表:\n  1. union 各表分區檔(train_.../test_...)。\n  2. 依欄位大寫標籤標準化型別(transform_by_label):A/P→Float64、M→字串、T/L→維持原生型別。\n\n【特徵工程】從乾淨表建構模型特徵:\n  3. 合理值清洗:連續型 99%PR 截尾(p99==0 不補)、A 欄負值→null。\n  4. 依後贅詞聚合(depth-0 直接 join、depth-1 一次聚合、depth-2 兩次聚合)。\n  5. 7 個財務基礎欄聚合 → 3 個財務比率(dti_ratio/overdue_debt_ratio/deposit_to_debt_ratio)\n     → 財務基礎欄本身從輸出移除。\n  6. reject_rate 衍生欄。\n  7. join 回 base,輸出 df_train / df_test(Polars DataFrame + parquet)。\n\ncredit_bureau_a_2(depth-2)資料量達 1.88 億列,使用 DuckDB 聚合(regr_slope 等視窗/聚合函式\n對超大表更穩健,並以分階段 parquet checkpoint 控制記憶體);其餘表用 Polars。\ncredit_bureau_b(b1/b2)依使用者指示忽略,不納入。\n\"\"\"\n\nimport glob\nimport math\nimport os\n\nimport duckdb\nimport polars as pl\nimport lightgbm as lgb\nfrom pathlib import Path\nfrom collections import defaultdict\nfrom lightgbm import LGBMClassifier\n\nimport numpy as np\nimport pandas as pd\nfrom IPython.display import display\n\nfrom sklearn.metrics import (\n    roc_auc_score,\n    roc_curve,\n    log_loss,\n    brier_score_loss\n)\n\nROOT            = Path(\"/kaggle/input/competitions/home-credit-credit-risk-model-stability\")\nTRAIN_DIR       = ROOT / \"parquet_files\" / \"train\"\nTEST_DIR        = ROOT / \"parquet_files\" / \"test\"\n\nREFERENCE_YEAR = 2024\nINVALID_YEAR_LO, INVALID_YEAR_HI = 2025, 2028  # bureau_a2_merge_GroupRule.txt 規則\n\n\n# #############################################################################\n# 【資料工程 DATA ENGINEERING】union 分區 → 標籤型別標準化\n# #############################################################################\n\ndef read_table(data_dir: str, split: str, table: str) -> pl.LazyFrame:\n    \"\"\"[資料工程] 讀取一張表,若有分區檔(depth 後贅詞_N)則 union,否則讀單檔。\"\"\"\n    exact = f\"{data_dir}/{split}_{table}.parquet\"\n    if os.path.exists(exact):\n        return pl.scan_parquet(exact)\n    pattern = f\"{data_dir}/{split}_{table}_*.parquet\"\n    files = sorted(glob.glob(pattern))\n    if not files:\n        raise FileNotFoundError(f\"no files matched {exact} or {pattern}\")\n    return pl.concat([pl.scan_parquet(f) for f in files], how=\"vertical_relaxed\")\n\n\ndef raw_glob(data_dir: str, split: str, table: str) -> str:\n    \"\"\"[資料工程] 回傳供 DuckDB read_parquet 用的 glob pattern(單檔或多分區)。\"\"\"\n    exact = f\"{data_dir}/{split}_{table}.parquet\"\n    if os.path.exists(exact):\n        return exact\n    return f\"{data_dir}/{split}_{table}_*.parquet\"\n\n\ndef transform_by_label(df: pl.DataFrame, cols: list[str]) -> pl.DataFrame:\n    \"\"\"[資料工程] 依欄位大寫尾碼做格式(型別)標準化:\n    A→Float64(Transform amount)、P→Float64(Transform DPD)、\n    M→Utf8(Masking categories,維持遮罩雜湊字串)、T/L→維持原生型別(Unspecified)。\n    \"\"\"\n    exprs = []\n    for c in cols:\n        if c not in df.columns:\n            continue\n        dtype = df.schema[c]\n        if c.endswith(\"A\") or c.endswith(\"P\"):\n            if dtype != pl.Float64:\n                exprs.append(pl.col(c).cast(pl.Float64, strict=False).alias(c))\n        elif c.endswith(\"M\"):\n            if dtype != pl.Utf8:\n                exprs.append(pl.col(c).cast(pl.Utf8).alias(c))\n        # T / L:未指定 Transform,維持原生型別\n    return df.with_columns(exprs) if exprs else df\n\n\n# #############################################################################\n# 【特徵工程 FEATURE ENGINEERING】清洗 → 聚合 → 財務指標 → 衍生欄 → join\n# #############################################################################\n\ndef winsorize_p99(df: pl.DataFrame, cols: list[str]) -> pl.DataFrame:\n    \"\"\"[特徵工程] 連續型 99%PR 截尾(不刪列);若 p99 為 0 或 null 則不截尾。\"\"\"\n    cols = [c for c in cols if c in df.columns]\n    if not cols:\n        return df\n    p99 = df.select([pl.col(c).quantile(0.99).alias(c) for c in cols]).row(0)\n    exprs = []\n    for c, cap in zip(cols, p99):\n        if cap is None or abs(cap) < 1e-9:  # 視為 0(容忍近似分位數的浮點雜訊)\n            continue\n        exprs.append(pl.when(pl.col(c) > cap).then(cap).otherwise(pl.col(c)).alias(c))\n    return df.with_columns(exprs) if exprs else df\n\n\ndef clean_by_label(df: pl.DataFrame, cols: list[str]) -> pl.DataFrame:\n    \"\"\"[特徵工程] A 尾碼(金額)負值 -> null。\"\"\"\n    exprs = [\n        pl.when(pl.col(c) < 0).then(None).otherwise(pl.col(c)).alias(c)\n        for c in cols\n        if c in df.columns and c.endswith(\"A\")\n    ]\n    return df.with_columns(exprs) if exprs else df\n\n\ndef clean_future_d_cols(df: pl.DataFrame, cols: list[str], ref_col: str) -> pl.DataFrame:\n    \"\"\"[特徵工程] 大寫 D 結尾(日期)欄位晚於 ref_col 者 -> null(內部衍生用)。\"\"\"\n    d_cols = [c for c in cols if c.endswith(\"D\") and c in df.columns]\n    if not d_cols:\n        return df\n    return df.with_columns(\n        [\n            pl.when(pl.col(c) > pl.col(ref_col)).then(None).otherwise(pl.col(c)).alias(c)\n            for c in d_cols\n        ]\n    )\n\n\ndef cat_group_stats(\n    df: pl.DataFrame, group_keys: list[str], col: str, stats: tuple[str, ...]\n) -> pl.DataFrame:\n    \"\"\"類別欄分組統計:non_placeholder_ratio(沿用既有 build_cat_agg 慣例)。\"\"\"\n    M_PLACEHOLDER = \"a55475b1\"\n    vc = df.drop_nulls(col).group_by(group_keys + [col]).agg(pl.len().alias(\"cnt\"))\n    exprs = []\n    if \"non_placeholder_ratio\" in stats:\n        exprs.append(\n            (pl.col(\"cnt\").filter(pl.col(col) != M_PLACEHOLDER).sum() / pl.col(\"cnt\").sum())\n            .fill_null(0.0)\n            .alias(f\"{col}_non_placeholder_ratio\")\n        )\n    return vc.group_by(group_keys).agg(exprs)\n\n\nNUMERIC_STAT_FUNCS = {\n    \"mean\": lambda c: c.mean(),\n    \"max\": lambda c: c.max(),\n    \"min\": lambda c: c.min(),\n    \"std\": lambda c: c.std(),\n    \"sum\": lambda c: c.sum(),\n}\n\n\ndef stat_expr(raw: str, stat: str) -> pl.Expr:\n    return NUMERIC_STAT_FUNCS[stat](pl.col(raw)).alias(f\"{raw}__{stat}\")\n\n\ndef financial_stat_expr(raw: str, stat: str) -> pl.Expr:\n    \"\"\"財務基礎欄聚合:輸出沿用原始欄名(代表 case 層級值),不加 __suffix。\"\"\"\n    return NUMERIC_STAT_FUNCS[stat](pl.col(raw)).alias(raw)\n\n\n# =============================================================================\n# depth-0:static_0(直取 12 欄 + 2 財務基礎欄)\n# =============================================================================\n\nSTATIC_0_COLS = [\n    \"avgdpdtolclosure24_3658938P\", \"pctinstlsallpaidlate1d_3546856L\", \"maxdbddpdtollast12m_3658940P\",\n    \"pctinstlsallpaidlate4d_3546849L\", \"avgdbddpdlast24m_3658932P\", \"pctinstlsallpaidlate6d_3546844L\",\n    \"maxdpdlast24m_143P\", \"pctinstlsallpaidearl3d_427L\", \"daysoverduetolerancedd_3976961L\",\n    \"numinstlswithdpd10_728L\", \"pctinstlsallpaidlat10d_839L\", \"maxdpdlast12m_727P\",\n]\nSTATIC_0_FINANCIAL = [\"totaldebt_9A\", \"maininc_215A\"]  # 財務基礎欄(直取,1:1)\n\n\ndef build_static(data_dir: str, split: str) -> pl.DataFrame:\n    cols = STATIC_0_COLS + STATIC_0_FINANCIAL\n    s0 = read_table(data_dir, split, \"static_0\").select([\"case_id\"] + cols).collect()\n    # [資料工程] 標籤型別標準化(A/P→Float64)\n    s0 = transform_by_label(s0, cols)\n    # [特徵工程] 清洗(A 負值→null)+ 連續型 99%PR 截尾\n    s0 = clean_by_label(s0, cols)\n    numeric_cols = [c for c in cols if s0.schema[c].is_numeric()]\n    s0 = winsorize_p99(s0, numeric_cols)\n    return s0\n\n\n# =============================================================================\n# depth-1:credit_bureau_a_1(21 個後贅詞統計 + 2 財務基礎欄)\n# =============================================================================\n\nA1_SPEC = [\n    (\"dpdmax_139P\", \"max\"), (\"dpdmax_139P\", \"mean\"), (\"dpdmax_139P\", \"sum\"),\n    (\"dpdmax_757P\", \"max\"), (\"dpdmax_757P\", \"mean\"), (\"dpdmax_757P\", \"std\"), (\"dpdmax_757P\", \"sum\"),\n    (\"numberofoverdueinstlmax_1039L\", \"max\"), (\"numberofoverdueinstlmax_1039L\", \"mean\"),\n    (\"numberofoverdueinstlmax_1039L\", \"sum\"),\n    (\"numberofoverdueinstlmax_1151L\", \"max\"), (\"numberofoverdueinstlmax_1151L\", \"mean\"),\n    (\"numberofoverdueinstlmax_1151L\", \"std\"), (\"numberofoverdueinstlmax_1151L\", \"sum\"),\n    (\"overdueamountmax2_14A\", \"mean\"), (\"overdueamountmax2_398A\", \"mean\"),\n    (\"overdueamountmax_155A\", \"max\"), (\"overdueamountmax_155A\", \"mean\"), (\"overdueamountmax_155A\", \"sum\"),\n    (\"overdueamountmax_35A\", \"mean\"), (\"overdueamountmax_35A\", \"std\"),\n]\n# 財務基礎欄(F3):存續合約逾期/未償債務,跨合約 sum,輸出用原始欄名(算完比率即移除)\nA1_FINANCIAL = [\n    (\"totaldebtoverduevalue_178A\", \"sum\"),\n    (\"totaloutstanddebtvalue_39A\", \"sum\"),\n]\n\n\ndef build_bureau_a1(data_dir: str, split: str) -> pl.DataFrame:\n    raw_cols = sorted({raw for raw, _ in A1_SPEC} | {raw for raw, _ in A1_FINANCIAL})\n    df = read_table(data_dir, split, \"credit_bureau_a_1\").select([\"case_id\"] + raw_cols).collect()\n    # [資料工程] 標籤型別標準化\n    df = transform_by_label(df, raw_cols)\n    # [特徵工程] 清洗 + 截尾\n    df = clean_by_label(df, raw_cols)\n    numeric_cols = [c for c in raw_cols if df.schema[c].is_numeric()]\n    df = winsorize_p99(df, numeric_cols)\n    # [特徵工程] depth-1 一次聚合 by case_id(既有後贅詞統計 + 財務欄 sum)\n    exprs = [stat_expr(raw, stat) for raw, stat in A1_SPEC]\n    exprs += [financial_stat_expr(raw, stat) for raw, stat in A1_FINANCIAL]\n    return df.group_by(\"case_id\").agg(exprs)\n\n\n# =============================================================================\n# depth-1:applprev_1(maxdpdtolerance_577P mean+max、rejectreasonclient non_placeholder_ratio、reject_rate)\n# =============================================================================\n\ndef build_applprev(data_dir: str, split: str) -> pl.DataFrame:\n    cols = [\n        \"case_id\", \"num_group1\", \"status_219L\", \"creationdate_885D\",\n        \"maxdpdtolerance_577P\", \"rejectreasonclient_4145042M\",\n    ]\n    df = read_table(data_dir, split, \"applprev_1\").select(cols).collect()\n    df = df.with_columns(pl.col(\"creationdate_885D\").str.strptime(pl.Date, strict=False))\n    df = transform_by_label(df, [\"maxdpdtolerance_577P\", \"rejectreasonclient_4145042M\"])\n    df = clean_by_label(df, [\"maxdpdtolerance_577P\"])\n    df = winsorize_p99(df, [\"maxdpdtolerance_577P\"])\n\n    numeric_agg = df.group_by(\"case_id\").agg(\n        pl.col(\"maxdpdtolerance_577P\").mean().alias(\"maxdpdtolerance_577P_mean\"),\n        pl.col(\"maxdpdtolerance_577P\").max().alias(\"maxdpdtolerance_577P_max\"),\n        (pl.col(\"status_219L\") == \"D\").sum().alias(\"_n_rejected\"),\n        pl.col(\"status_219L\").is_not_null().sum().alias(\"_n_status_valid\"),\n    ).with_columns(\n        (pl.col(\"_n_rejected\") / pl.col(\"_n_status_valid\")).alias(\"reject_rate\")\n    ).drop([\"_n_rejected\", \"_n_status_valid\"])\n\n    reject_stats = cat_group_stats(\n        df, [\"case_id\"], \"rejectreasonclient_4145042M\", (\"non_placeholder_ratio\",)\n    )\n    return numeric_agg.join(reject_stats, on=\"case_id\", how=\"left\")\n\n\n# =============================================================================\n# depth-1:person_1(僅財務基礎欄 mainoccupationinc_384A)\n# =============================================================================\n\ndef build_person(data_dir: str, split: str) -> pl.DataFrame:\n    cols = [\"case_id\", \"num_group1\", \"mainoccupationinc_384A\"]\n    df = read_table(data_dir, split, \"person_1\").select(cols).collect()\n    df = transform_by_label(df, [\"mainoccupationinc_384A\"])\n    df = clean_by_label(df, [\"mainoccupationinc_384A\"])\n    df = winsorize_p99(df, [\"mainoccupationinc_384A\"])\n    # [特徵工程] case 層級 max(僅本人列有值,max 即取回本人收入)\n    return df.group_by(\"case_id\").agg(financial_stat_expr(\"mainoccupationinc_384A\", \"max\"))\n\n\n# =============================================================================\n# depth-1:other_1(財務基礎欄:存款流入 / 支出流出,case 層級 sum)\n# =============================================================================\n\nOTHER_FINANCIAL = [\n    (\"amtdepositincoming_4809444A\", \"sum\"),\n    (\"amtdebitoutgoing_4809440A\", \"sum\"),\n]\n\n\ndef build_other(data_dir: str, split: str) -> pl.DataFrame:\n    raw_cols = [raw for raw, _ in OTHER_FINANCIAL]\n    df = read_table(data_dir, split, \"other_1\").select([\"case_id\"] + raw_cols).collect()\n    df = transform_by_label(df, raw_cols)\n    df = clean_by_label(df, raw_cols)\n    df = winsorize_p99(df, raw_cols)\n    return df.group_by(\"case_id\").agg(\n        [financial_stat_expr(raw, stat) for raw, stat in OTHER_FINANCIAL]\n    )\n\n\n# =============================================================================\n# 財務比率特徵:以聚合後 case 層級欄位計算,算完即從輸出移除 7 個財務基礎欄\n# =============================================================================\n\nFINANCIAL_BASE_COLS = [\n    \"totaldebt_9A\", \"maininc_215A\", \"mainoccupationinc_384A\",\n    \"totaldebtoverduevalue_178A\", \"totaloutstanddebtvalue_39A\",\n    \"amtdepositincoming_4809444A\", \"amtdebitoutgoing_4809440A\",\n]\n\n\ndef _safe_ratio(numerator: pl.Expr, denominator: pl.Expr) -> pl.Expr:\n    \"\"\"除零/除 null 保護:分母為 null 或 0 時回傳 null。\"\"\"\n    return pl.when(denominator.is_null() | (denominator == 0)).then(None).otherwise(\n        numerator / denominator\n    )\n\n\ndef build_financial_indicators(df: pl.DataFrame) -> pl.DataFrame:\n    \"\"\"[特徵工程] 依 Financial indicators.csv 建構 3 個財務比率(需 df 已含 7 財務基礎欄)。\"\"\"\n    income = pl.coalesce([\"maininc_215A\", \"mainoccupationinc_384A\"])  # 主要收入,職業收入備援\n    return df.with_columns(\n        _safe_ratio(pl.col(\"totaldebt_9A\"), income).alias(\"dti_ratio\"),\n        _safe_ratio(\n            pl.col(\"totaldebtoverduevalue_178A\"), pl.col(\"totaloutstanddebtvalue_39A\")\n        ).alias(\"overdue_debt_ratio\"),\n        _safe_ratio(\n            pl.col(\"amtdepositincoming_4809444A\"), pl.col(\"amtdebitoutgoing_4809440A\")\n        ).alias(\"deposit_to_debt_ratio\"),\n    )\n\n\n# =============================================================================\n# depth-2:credit_bureau_a_2(DuckDB;規模達 1.88 億列)— spec-driven,由 93-特徵清單解析\n# =============================================================================\n\nA2_RAW_COLS = [\n    \"case_id\", \"num_group1\", \"num_group2\",\n    \"pmts_dpd_1073P\", \"pmts_dpd_303P\", \"pmts_overdue_1140A\", \"pmts_overdue_1152A\",\n    \"pmts_year_1139T\", \"pmts_month_158T\", \"pmts_year_507T\", \"pmts_month_706T\",\n]\nA2_WINSORIZE_COLS = [\"pmts_dpd_1073P\", \"pmts_dpd_303P\", \"pmts_overdue_1140A\", \"pmts_overdue_1152A\"]\nA2_RAW_PREFIXES = [\"pmts_dpd_1073P\", \"pmts_dpd_303P\", \"pmts_overdue_1140A\", \"pmts_overdue_1152A\"]\nA2_SHORT_NAME = {\n    \"pmts_dpd_1073P\": \"dpd1073\", \"pmts_dpd_303P\": \"dpd303\",\n    \"pmts_overdue_1140A\": \"ov1140\", \"pmts_overdue_1152A\": \"ov1152\",\n}\nA2_TIME_KEY = {\"pmts_dpd_1073P\": \"active_time_key\", \"pmts_dpd_303P\": \"closed_time_key\"}\n\n# 93-特徵清單第 7 點:credit_bureau_a_2 相關的 56 個特徵名(逐字照抄使用者清單)\nA2_FEATURES = [\n    \"pmts_overdue_1152A_overdue_rate__weighted_avg\", \"pmts_dpd_303P_overdue_rate__weighted_avg\",\n    \"pmts_dpd_303P_std__mean\", \"pmts_dpd_303P_mean__mean\", \"pmts_overdue_1152A_mean__mean\",\n    \"pmts_dpd_303P_recent12_mean__mean\", \"pmts_overdue_1140A_overdue_rate__weighted_avg\",\n    \"pmts_dpd_303P_recent6_mean__mean\", \"pmts_dpd_1073P_overdue_rate__weighted_avg\",\n    \"pmts_dpd_1073P_std__mean\", \"pmts_overdue_1152A_std__mean\", \"pmts_overdue_1140A_mean__mean\",\n    \"pmts_dpd_303P_recent3_mean__mean\", \"pmts_dpd_1073P_mean__mean\", \"pmts_dpd_1073P_trend__mean\",\n    \"pmts_dpd_1073P_std__max\", \"pmts_dpd_303P_std__max\", \"pmts_overdue_1140A_std__mean\",\n    \"pmts_overdue_1140A_mean__max\", \"pmts_dpd_1073P_mean__max\", \"pmts_dpd_1073P_recent12_mean__mean\",\n    \"pmts_dpd_303P_mean__max\", \"pmts_dpd_303P_recent12_mean__max\", \"pmts_dpd_1073P_max__max\",\n    \"pmts_dpd_1073P_recent12_mean__max\", \"pmts_dpd_303P_recent6_mean__max\",\n    \"pmts_overdue_1152A_median__mean\", \"pmts_dpd_303P_max__max\", \"pmts_dpd_303P_median__mean\",\n    \"pmts_overdue_1140A_std__max\", \"pmts_dpd_1073P_trend__max\", \"pmts_dpd_303P_recent3_mean__max\",\n    \"pmts_overdue_1152A_mean__max\", \"pmts_dpd_1073P_consecutive_max__max\",\n    \"pmts_dpd_1073P_n_unique__max\", \"pmts_dpd_303P_trend__mean\", \"pmts_dpd_1073P_recent6_mean__mean\",\n    \"pmts_overdue_1140A_n_unique__max\", \"pmts_overdue_1140A_sum_positive__sum\",\n    \"pmts_overdue_1140A_max__max\", \"pmts_dpd_1073P_longest_good_streak__max\",\n    \"pmts_dpd_303P_median__max\", \"pmts_dpd_1073P_sum_positive__sum\",\n    \"pmts_overdue_1140A_sum_positive__max\", \"pmts_overdue_1140A_positive_count__sum\",\n    \"pmts_dpd_1073P_sum_positive__max\", \"pmts_dpd_303P_n_unique__max\",\n    \"pmts_dpd_1073P_recent6_mean__max\", \"pmts_overdue_1152A_median__max\",\n    \"pmts_dpd_303P_longest_good_streak__max\", \"pmts_overdue_1140A_positive_count__max\",\n    \"pmts_dpd_303P_last__mean_fallback\", \"pmts_dpd_1073P_positive_count__sum\",\n    \"pmts_dpd_303P_trend__max\", \"pmts_dpd_1073P_positive_count__max\",\n    \"pmts_dpd_1073P_mean_positive__max\",\n]\n\n\ndef parse_a2_feature(name: str) -> tuple[str, str, str]:\n    for raw in A2_RAW_PREFIXES:\n        if name.startswith(raw + \"_\"):\n            rest = name[len(raw) + 1:]\n            l1, l2 = rest.split(\"__\", 1)\n            return raw, l1, l2\n    raise ValueError(f\"無法解析 a2 特徵名: {name}\")\n\n\nA2_SPEC = [parse_a2_feature(f) for f in A2_FEATURES]\nassert len(A2_SPEC) == len(A2_FEATURES) == 56, len(A2_SPEC)\n\nBASIC_L1_SQL = {\n    \"mean\": \"AVG({raw})\",\n    \"max\": \"MAX({raw})\",\n    \"std\": \"STDDEV_SAMP({raw})\",\n    \"median\": \"MEDIAN({raw})\",\n    \"n_unique\": \"approx_count_distinct({raw})::BIGINT\",\n    \"overdue_rate\": \"SUM(CASE WHEN {raw}>0 THEN 1 ELSE 0 END)::DOUBLE / NULLIF(COUNT({raw}),0)\",\n    \"positive_count\": \"SUM(CASE WHEN {raw}>0 THEN 1 ELSE 0 END)::BIGINT\",\n    \"sum_positive\": \"SUM(CASE WHEN {raw}>0 THEN {raw} ELSE 0 END)\",\n    \"mean_positive\": \"AVG({raw}) FILTER (WHERE {raw}>0)\",\n}\nTEMPORAL_L1_STATS = {\"consecutive_max\", \"longest_good_streak\", \"recent3_mean\", \"recent6_mean\", \"recent12_mean\", \"trend\", \"last\"}\n\n\ndef build_bureau_a2(data_dir: str, split: str) -> pl.DataFrame:\n    os.makedirs(\".tmp\", exist_ok=True)\n    keyed_path = f\".tmp/a2woe_keyed_{split}.parquet\"\n    l1_basic_path = f\".tmp/a2woe_l1basic_{split}.parquet\"\n    l1_1073_path = f\".tmp/a2woe_l1_1073_{split}.parquet\"\n    l1_303_path = f\".tmp/a2woe_l1_303_{split}.parquet\"\n    l1_path = f\".tmp/a2woe_l1_{split}.parquet\"\n\n    con = duckdb.connect()\n    con.sql(\"SET memory_limit='8GB'\")\n    con.sql(\"SET preserve_insertion_order=false\")\n    con.sql(f\"SET threads TO {max(1, os.cpu_count() // 2)}\")\n    src = raw_glob(data_dir, split, \"credit_bureau_a_2\")\n    cols_sql = \", \".join(A2_RAW_COLS)\n\n    # --- p99(approx)+ 若為 0 則不截尾 ---\n    p99_row = con.sql(\n        f\"\"\"\n        SELECT {\", \".join(f\"approx_quantile({c}, 0.99) AS {c}\" for c in A2_WINSORIZE_COLS)}\n        FROM (SELECT {cols_sql} FROM read_parquet('{src}'))\n        WHERE {\" OR \".join(f\"{c} IS NOT NULL\" for c in A2_WINSORIZE_COLS)}\n        \"\"\"\n    ).fetchone()\n    caps = dict(zip(A2_WINSORIZE_COLS, p99_row))\n    print(f\"  a2 p99 caps: {caps}\")\n\n    def cap_expr(col: str) -> str:\n        cap = caps.get(col)\n        if cap is None or abs(cap) < 1e-9:\n            return col\n        return f\"LEAST({col}, {cap})\"\n\n    # --- 清洗 + 時間鍵,checkpoint(避免重複掃 1.88 億列原始表) ---\n    con.sql(\n        f\"\"\"\n        COPY (\n            SELECT\n                case_id, num_group1, num_group2,\n                {cap_expr(\"CASE WHEN pmts_dpd_1073P < 0 THEN NULL ELSE pmts_dpd_1073P END\")} AS pmts_dpd_1073P,\n                {cap_expr(\"CASE WHEN pmts_dpd_303P < 0 THEN NULL ELSE pmts_dpd_303P END\")} AS pmts_dpd_303P,\n                {cap_expr(\"pmts_overdue_1140A\")} AS pmts_overdue_1140A,\n                {cap_expr(\"pmts_overdue_1152A\")} AS pmts_overdue_1152A,\n                CASE WHEN pmts_month_158T BETWEEN 1 AND 12\n                          AND pmts_year_1139T NOT BETWEEN {INVALID_YEAR_LO} AND {INVALID_YEAR_HI}\n                     THEN pmts_year_1139T * 12 + pmts_month_158T END AS active_time_key,\n                CASE WHEN pmts_month_706T BETWEEN 1 AND 12\n                          AND pmts_year_507T NOT BETWEEN {INVALID_YEAR_LO} AND {INVALID_YEAR_HI}\n                     THEN pmts_year_507T * 12 + pmts_month_706T END AS closed_time_key\n            FROM (SELECT {cols_sql} FROM read_parquet('{src}'))\n        ) TO '{keyed_path}' (FORMAT PARQUET)\n        \"\"\"\n    )\n    a2_keyed = f\"read_parquet('{keyed_path}')\"\n\n    # --- 依 spec 決定各原始欄需要的 L1 統計 ---\n    specs_by_raw: dict[str, list[tuple[str, str]]] = defaultdict(list)\n    for raw, l1, l2 in A2_SPEC:\n        specs_by_raw[raw].append((l1, l2))\n    l1_stats_by_raw = {raw: sorted({l1 for l1, _ in pairs}) for raw, pairs in specs_by_raw.items()}\n\n    # --- L1 基本統計(單趟,四原始欄一起算;非時序統計 + non_null_count(供 weighted_avg 權重)+ trend) ---\n    basic_exprs = []\n    for raw in A2_RAW_PREFIXES:\n        short = A2_SHORT_NAME[raw]\n        needed = set(l1_stats_by_raw.get(raw, []))\n        for l1_stat in sorted(needed - TEMPORAL_L1_STATS):\n            sql = BASIC_L1_SQL[l1_stat].format(raw=raw)\n            basic_exprs.append(f\"{sql} AS {short}_{l1_stat}\")\n        # non_null_count:凡有 weighted_avg 需求即需要(此清單四欄皆需要)\n        basic_exprs.append(f\"COUNT({raw}) AS {short}_non_null_count\")\n        if \"trend\" in needed:\n            time_key = A2_TIME_KEY[raw]\n            basic_exprs.append(f\"regr_slope({raw}, {time_key}) AS {short}_trend\")\n\n    con.sql(\n        f\"\"\"\n        COPY (\n            SELECT case_id, num_group1, {\", \".join(basic_exprs)}\n            FROM {a2_keyed}\n            GROUP BY case_id, num_group1\n        ) TO '{l1_basic_path}' (FORMAT PARQUET)\n        \"\"\"\n    )\n    l1_basic_src = f\"read_parquet('{l1_basic_path}')\"\n\n    # --- 時序型 L1:pmts_dpd_1073P(active)—— consecutive_max/longest_good_streak/recent6·12_mean ---\n    con.sql(\n        f\"\"\"\n        COPY (\n            WITH ranked AS (\n                SELECT case_id, num_group1, pmts_dpd_1073P AS x,\n                    ROW_NUMBER() OVER (PARTITION BY case_id, num_group1\n                        ORDER BY active_time_key DESC, num_group2 DESC) AS rn_desc,\n                    ROW_NUMBER() OVER (PARTITION BY case_id, num_group1\n                        ORDER BY active_time_key ASC, num_group2 ASC) AS rn_asc\n                FROM {a2_keyed}\n                WHERE pmts_dpd_1073P IS NOT NULL AND active_time_key IS NOT NULL\n            ),\n            recent_means AS (\n                SELECT case_id, num_group1,\n                    AVG(x) FILTER (WHERE rn_desc <= 6) AS dpd1073_recent6_mean,\n                    AVG(x) FILTER (WHERE rn_desc <= 12) AS dpd1073_recent12_mean\n                FROM ranked GROUP BY case_id, num_group1\n            ),\n            islands AS (\n                SELECT case_id, num_group1, x,\n                    rn_asc - ROW_NUMBER() OVER (PARTITION BY case_id, num_group1, (x > 0)\n                        ORDER BY rn_asc) AS grp\n                FROM ranked\n            ),\n            runs AS (\n                SELECT case_id, num_group1, (x > 0) AS is_pos, COUNT(*) AS run_len\n                FROM islands GROUP BY case_id, num_group1, grp, (x > 0)\n            ),\n            streaks AS (\n                SELECT case_id, num_group1,\n                    MAX(run_len) FILTER (WHERE is_pos) AS dpd1073_consecutive_max,\n                    MAX(run_len) FILTER (WHERE NOT is_pos) AS dpd1073_longest_good_streak\n                FROM runs GROUP BY case_id, num_group1\n            )\n            SELECT rm.case_id, rm.num_group1, rm.dpd1073_recent6_mean, rm.dpd1073_recent12_mean,\n                   st.dpd1073_consecutive_max, st.dpd1073_longest_good_streak\n            FROM recent_means rm LEFT JOIN streaks st USING (case_id, num_group1)\n        ) TO '{l1_1073_path}' (FORMAT PARQUET)\n        \"\"\"\n    )\n    l1_1073_src = f\"read_parquet('{l1_1073_path}')\"\n\n    # --- 時序型 L1:pmts_dpd_303P(closed)—— last/longest_good_streak/recent3·6·12_mean ---\n    con.sql(\n        f\"\"\"\n        COPY (\n            WITH ranked AS (\n                SELECT case_id, num_group1, pmts_dpd_303P AS x,\n                    ROW_NUMBER() OVER (PARTITION BY case_id, num_group1\n                        ORDER BY closed_time_key DESC, num_group2 DESC) AS rn_desc,\n                    ROW_NUMBER() OVER (PARTITION BY case_id, num_group1\n                        ORDER BY closed_time_key ASC, num_group2 ASC) AS rn_asc\n                FROM {a2_keyed}\n                WHERE pmts_dpd_303P IS NOT NULL AND closed_time_key IS NOT NULL\n            ),\n            recent_means AS (\n                SELECT case_id, num_group1,\n                    AVG(x) FILTER (WHERE rn_desc <= 3) AS dpd303_recent3_mean,\n                    AVG(x) FILTER (WHERE rn_desc <= 6) AS dpd303_recent6_mean,\n                    AVG(x) FILTER (WHERE rn_desc <= 12) AS dpd303_recent12_mean,\n                    MAX(x) FILTER (WHERE rn_desc = 1) AS dpd303_last\n                FROM ranked GROUP BY case_id, num_group1\n            ),\n            islands AS (\n                SELECT case_id, num_group1, x,\n                    rn_asc - ROW_NUMBER() OVER (PARTITION BY case_id, num_group1, (x = 0)\n                        ORDER BY rn_asc) AS grp\n                FROM ranked\n            ),\n            runs AS (\n                SELECT case_id, num_group1, (x = 0) AS is_zero, COUNT(*) AS run_len\n                FROM islands GROUP BY case_id, num_group1, grp, (x = 0)\n            ),\n            streaks AS (\n                SELECT case_id, num_group1,\n                    MAX(run_len) FILTER (WHERE is_zero) AS dpd303_longest_good_streak\n                FROM runs GROUP BY case_id, num_group1\n            )\n            SELECT rm.*, st.dpd303_longest_good_streak\n            FROM recent_means rm LEFT JOIN streaks st USING (case_id, num_group1)\n        ) TO '{l1_303_path}' (FORMAT PARQUET)\n        \"\"\"\n    )\n    l1_303_src = f\"read_parquet('{l1_303_path}')\"\n\n    # --- 合併三個 L1 checkpoint ---\n    con.sql(\n        f\"\"\"\n        COPY (\n            SELECT b.*,\n                r1.dpd1073_recent6_mean, r1.dpd1073_recent12_mean,\n                r1.dpd1073_consecutive_max, r1.dpd1073_longest_good_streak,\n                r3.dpd303_recent3_mean, r3.dpd303_recent6_mean, r3.dpd303_recent12_mean,\n                r3.dpd303_last, r3.dpd303_longest_good_streak\n            FROM {l1_basic_src} b\n            LEFT JOIN {l1_1073_src} r1 USING (case_id, num_group1)\n            LEFT JOIN {l1_303_src} r3 USING (case_id, num_group1)\n        ) TO '{l1_path}' (FORMAT PARQUET)\n        \"\"\"\n    )\n    l1_src = f\"read_parquet('{l1_path}')\"\n\n    # --- L2:二次聚合 by case_id,依 A2_SPEC 逐一產生 ---\n    l2_exprs = []\n    for raw, l1, l2 in A2_SPEC:\n        short = A2_SHORT_NAME[raw]\n        col = f\"{short}_{l1}\"\n        final_name = f\"{raw}_{l1}__{l2}\"\n        if l2 == \"weighted_avg\":\n            weight = f\"{short}_non_null_count\"\n            l2_exprs.append(\n                f\"SUM({col} * {weight}) / NULLIF(SUM({weight}), 0) AS {final_name}\"\n            )\n        elif l2 == \"mean_fallback\":\n            l2_exprs.append(f\"AVG(COALESCE({col}, 0)) AS {final_name}\")\n        elif l2 == \"sum\":\n            l2_exprs.append(f\"SUM({col})::DOUBLE AS {final_name}\")\n        elif l2 in (\"mean\", \"max\"):\n            l2_exprs.append(f\"{l2.upper()}({col}) AS {final_name}\")\n        else:\n            raise ValueError(f\"未知 L2 統計: {l2}\")\n\n    l2 = con.sql(\n        f\"\"\"\n        SELECT case_id, {\", \".join(l2_exprs)}\n        FROM {l1_src}\n        GROUP BY case_id\n        \"\"\"\n    ).pl()\n\n    con.close()\n    for p in (keyed_path, l1_basic_path, l1_1073_path, l1_303_path, l1_path):\n        if os.path.exists(p):\n            os.remove(p)\n\n    # regr_slope 在樣本不足/x 無變異時回傳 NaN(而非 SQL NULL);統一轉為 null 避免下游誤判\n    float_cols = [c for c, dt in l2.schema.items() if dt in (pl.Float32, pl.Float64)]\n    l2 = l2.with_columns([pl.col(c).fill_nan(None) for c in float_cols])\n    return l2\n\n\n# =============================================================================\n# base + 組裝\n# =============================================================================\n\nFINAL_FEATURE_COLS = (\n    [f\"{raw}_{l1}__{l2}\" for raw, l1, l2 in A2_SPEC]  # 56 個 credit_bureau_a_2 特徵\n    + [f\"{raw}__{stat}\" for raw, stat in A1_SPEC]  # 21 個 credit_bureau_a_1 特徵\n    + STATIC_0_COLS  # 12 個 static_0 直取特徵\n    + [\n        \"maxdpdtolerance_577P_mean\", \"maxdpdtolerance_577P_max\",\n        \"rejectreasonclient_4145042M_non_placeholder_ratio\",  # applprev_1(3)\n        \"reject_rate\",  # 衍生欄\n        \"dti_ratio\", \"overdue_debt_ratio\", \"deposit_to_debt_ratio\",  # 3 個財務比率\n    ]\n)\nassert len(FINAL_FEATURE_COLS) == 96, len(FINAL_FEATURE_COLS)\n\n\ndef load_base(data_dir: str, split: str) -> pl.DataFrame:\n    df = read_table(data_dir, split, \"base\").collect()\n    return df.with_columns(pl.col(\"date_decision\").str.strptime(pl.Date, strict=False))\n\n\ndef build_dataset(data_dir: str, split: str) -> pl.DataFrame:\n    print(f\"=== building {split} (WOEIV) ===\")\n    base = load_base(data_dir, split)\n\n    print(\"  static...\")\n    static = build_static(data_dir, split)\n    print(\"  applprev_1...\")\n    applprev = build_applprev(data_dir, split)\n    print(\"  credit_bureau_a_1...\")\n    a1 = build_bureau_a1(data_dir, split)\n    print(\"  credit_bureau_a_2 (duckdb, may take a while)...\")\n    a2 = build_bureau_a2(data_dir, split)\n    print(\"  person_1...\")\n    person = build_person(data_dir, split)\n    print(\"  other_1...\")\n    other = build_other(data_dir, split)\n\n    # [特徵工程] join 回 base\n    has_target = \"target\" in base.columns\n    key_cols = [\"case_id\", \"WEEK_NUM\", \"date_decision\"] + ([\"target\"] if has_target else [])\n    out = base.select(key_cols)\n    for part in [static, applprev, a1, a2, person, other]:\n        out = out.join(part, on=\"case_id\", how=\"left\")\n\n    # [特徵工程] 財務比率(需 7 財務基礎欄皆已 join 齊全),算完即移除財務基礎欄\n    out = build_financial_indicators(out)\n    out = out.drop(FINANCIAL_BASE_COLS)\n\n    final_cols = key_cols + FINAL_FEATURE_COLS\n    out = out.select(final_cols)\n    print(f\"  {split} shape: {out.shape}\")\n    return out\n\n\ndef main():\n    df_train = build_dataset(TRAIN_DIR, \"train\")\n    df_test = build_dataset(TEST_DIR, \"test\")\n\n    # df_train.write_parquet(TRAIN_OUT)\n    # df_test.write_parquet(TEST_OUT)\n    # print(f\"written: {TRAIN_OUT} {df_train.shape}\")\n    # print(f\"written: {TEST_OUT} {df_test.shape}\")\n    return df_train, df_test\n\n\nif __name__ == \"__main__\":\n    df_train, df_test = main()\n\n\n#訓練模型\ntrain = df_train\ntest = df_test\n\nx_train = train.drop([\"case_id\",\"target\",\"WEEK_NUM\",\"date_decision\"])\ny_train = train.select(\"target\")\nx_test = test.drop([\"WEEK_NUM\",\"date_decision\"])\n\n# week_train_pd = (\n#     train\n#     .select(\"WEEK_NUM\")\n#     .to_numpy()\n#     .ravel()\n# )\n\n# week_test_pd = (\n#     test\n#     .select(\"WEEK_NUM\")\n#     .to_numpy()\n#     .ravel()\n# )\n\n\nx_train_pd = x_train.to_pandas()\ny_train_pd = y_train.to_numpy().ravel()\nx_test_pd = x_test.to_pandas()\n\n\n\ncategorical_cols = list(\n    x_train_pd.select_dtypes(include=[\"object\"]).columns\n)\n\nfor col in categorical_cols:\n    x_train_pd[col] = x_train_pd[col].astype(\"category\")\n    x_test_pd[col] = x_test_pd[col].astype(\"category\")\n\nmodel = LGBMClassifier(\n    objective=\"binary\",\n    boosting_type=\"gbdt\",\n\n    n_estimators=500,\n\n    learning_rate=0.05,\n\n    num_leaves=31,\n\n    random_state=42,\n\n    n_jobs=-1\n)\n\nmodel.fit(\n    x_train_pd,\n    y_train_pd\n)\n\n## 這個就是輸出機率\n\n\nx_test_pd = x_test_pd.set_index(\"case_id\")\n\ntest_probability = pd.Series(model.predict_proba(x_test_pd)[:, 1], index=x_test_pd.index)\n\ndf_subm = pd.read_csv(ROOT / \"sample_submission.csv\")\ndf_subm = df_subm.set_index(\"case_id\")\n\ndf_subm[\"score\"] = test_probability\nprint(\"Check null: \", df_subm[\"score\"].isnull().any())\n\ndf_subm.head()\nprint(df_subm)\n\ndf_subm.to_csv(\"submission.csv\")","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2026-07-24T08:29:53.466451Z","iopub.execute_input":"2026-07-24T08:29:53.466876Z","iopub.status.idle":"2026-07-24T08:34:22.651945Z","shell.execute_reply.started":"2026-07-24T08:29:53.466843Z","shell.execute_reply":"2026-07-24T08:34:22.650353Z"}},"outputs":[],"execution_count":null}]}