Spaces:
Runtime error
Runtime error
| """ | |
| Taxy ML — Shared Preprocessing Module | |
| ===================================== | |
| Single source of truth untuk transformasi fitur. Dipakai oleh: | |
| - data_prep.py (membangun dataset training) | |
| - inference.py (inferensi lokal) | |
| - serve.py (REST API FastAPI) | |
| Tujuan: memastikan fitur yang dilihat model saat training IDENTIK dengan | |
| fitur saat inferensi (menghindari training/serving skew). | |
| Business problem: Tax-Compliance Risk Profiling. | |
| Diberikan profil finansial & perilaku seorang wajib pajak, model memprediksi: | |
| - risk_index : skor risiko ketidakpatuhan pajak kontinu [0,1] (regresi) | |
| - risk_category: {Rendah, Sedang, Tinggi} (klasifikasi) | |
| """ | |
| import json | |
| import math | |
| import numpy as np | |
| # --------------------------------------------------------------------------- | |
| # Definisi fitur & kategori (urutan WAJIB stabil) | |
| # --------------------------------------------------------------------------- | |
| PEKERJAAN_CATS = ["Fresh Graduate", "Junior Staff", "Senior Staff", "Manager", "Director"] | |
| # Bobot senioritas dipakai untuk salah satu interaksi non-linear pada target | |
| PEKERJAAN_RANK = {c: i for i, c in enumerate(PEKERJAAN_CATS)} | |
| # Urutan fitur numerik final yang masuk ke model | |
| NUMERIC_FEATURES = [ | |
| "usia_norm", # usia dinormalisasi | |
| "log_gaji", # log(gaji_bulanan) | |
| "literasi_norm", # skor literasi / 100 | |
| "punya_npwp", # 0/1 | |
| "aktif_taxy", # 0/1 | |
| "is_kawin", # 0/1 (status PTKP diawali 'K') | |
| "tanggungan_norm", # jumlah tanggungan / 3 | |
| "income_bracket", # ordinal pendapatan [0..4] / 4 | |
| ] | |
| # Ditambah one-hot pekerjaan (5 kolom) -> total fitur = 8 + 5 = 13 | |
| FEATURE_NAMES = NUMERIC_FEATURES + [f"pekerjaan_{c}" for c in PEKERJAAN_CATS] | |
| N_FEATURES = len(FEATURE_NAMES) | |
| CLASS_NAMES = ["Rendah", "Sedang", "Tinggi"] | |
| N_CLASSES = len(CLASS_NAMES) | |
| # Batas kategori risiko (pada risk_index yang sudah dinormalisasi [0,1]) | |
| RISK_BINS = [0.40, 0.66] # < .40 Rendah, < .66 Sedang, else Tinggi | |
| INCOME_BRACKET_ORDER = ["<3Jt", "3-6Jt", "6-12Jt", "12-25Jt", ">25Jt"] | |
| INCOME_BRACKET_MAP = {c: i for i, c in enumerate(INCOME_BRACKET_ORDER)} | |
| def _to_float(x, default=0.0): | |
| try: | |
| if x is None or x == "" or (isinstance(x, float) and math.isnan(x)): | |
| return default | |
| return float(x) | |
| except (TypeError, ValueError): | |
| return default | |
| def parse_dependents(status_ptkp): | |
| """Ambil jumlah tanggungan (0-3) dari status PTKP seperti 'K/2' atau 'TK/0'.""" | |
| if not status_ptkp or "/" not in str(status_ptkp): | |
| return 0 | |
| try: | |
| return max(0, min(3, int(str(status_ptkp).split("/")[-1]))) | |
| except ValueError: | |
| return 0 | |
| def income_bracket_from(record): | |
| """Tentukan bracket pendapatan ordinal [0..4]. Pakai kolom jika ada, | |
| selain itu turunkan dari gaji_bulanan agar tetap konsisten saat serving.""" | |
| kat = record.get("kategori_pendapatan") | |
| if kat in INCOME_BRACKET_MAP: | |
| return INCOME_BRACKET_MAP[kat] | |
| gaji = _to_float(record.get("gaji_bulanan")) | |
| if gaji < 3e6: | |
| return 0 | |
| if gaji < 6e6: | |
| return 1 | |
| if gaji < 12e6: | |
| return 2 | |
| if gaji < 25e6: | |
| return 3 | |
| return 4 | |
| def build_raw_features(record): | |
| """Bangun vektor fitur MENTAH (sebelum standardisasi) dari satu record dict. | |
| Record dapat berasal dari CSV (kolom: usia, gaji_bulanan, status_ptkp, | |
| skor_literasi_pajak, punya_npwp, aktif_taxy, pekerjaan, kategori_pendapatan) | |
| atau dari payload API (kunci sama / serupa). | |
| """ | |
| usia = _to_float(record.get("usia"), 30.0) | |
| gaji = _to_float(record.get("gaji_bulanan"), 0.0) | |
| literasi = _to_float(record.get("skor_literasi_pajak"), 50.0) | |
| npwp = 1.0 if str(record.get("punya_npwp", 0)) in ("1", "1.0", "True", "true") else 0.0 | |
| aktif = 1.0 if str(record.get("aktif_taxy", 0)) in ("1", "1.0", "True", "true") else 0.0 | |
| status = str(record.get("status_ptkp", "TK/0")) | |
| is_kawin = 1.0 if status.upper().startswith("K") else 0.0 | |
| tanggungan = parse_dependents(status) | |
| bracket = income_bracket_from(record) | |
| feats = { | |
| "usia_norm": (usia - 18.0) / (60.0 - 18.0), # ~[0,1] | |
| "log_gaji": math.log1p(max(0.0, gaji)), # ~[14..17] | |
| "literasi_norm": literasi / 100.0, | |
| "punya_npwp": npwp, | |
| "aktif_taxy": aktif, | |
| "is_kawin": is_kawin, | |
| "tanggungan_norm": tanggungan / 3.0, | |
| "income_bracket": bracket / 4.0, | |
| } | |
| pekerjaan = str(record.get("pekerjaan", "")).strip() | |
| vec = [feats[k] for k in NUMERIC_FEATURES] | |
| for c in PEKERJAAN_CATS: | |
| vec.append(1.0 if pekerjaan == c else 0.0) | |
| return np.asarray(vec, dtype=np.float32) | |
| def compute_risk_index(record): | |
| """Hitung target risk_index [0,1] secara deterministik (smooth, non-linear). | |
| Semakin tinggi = semakin berisiko TIDAK patuh pajak. Komponen: | |
| - tidak punya NPWP (penalti besar) | |
| - literasi pajak rendah (penalti proporsional) | |
| - tidak aktif memakai Taxy (penalti perilaku) | |
| - pendapatan tinggi (eksposur kewajiban lebih besar) | |
| - banyak tanggungan (beban administratif, penalti kecil) | |
| - interaksi non-linear: tanpa-NPWP DAN literasi-rendah saling memperkuat | |
| -> membuat permukaan target tidak bisa ditebak model linear sederhana. | |
| Output dinormalisasi ke [0,1] dengan pembagi konstan W_MAX. | |
| """ | |
| npwp = 1.0 if str(record.get("punya_npwp", 0)) in ("1", "1.0", "True", "true") else 0.0 | |
| aktif = 1.0 if str(record.get("aktif_taxy", 0)) in ("1", "1.0", "True", "true") else 0.0 | |
| literasi = _to_float(record.get("skor_literasi_pajak"), 50.0) / 100.0 | |
| bracket = income_bracket_from(record) / 4.0 | |
| tanggungan = parse_dependents(record.get("status_ptkp")) / 3.0 | |
| no_npwp = 1.0 - npwp | |
| low_lit = 1.0 - literasi | |
| no_aktif = 1.0 - aktif | |
| raw = ( | |
| 0.32 * no_npwp | |
| + 0.22 * low_lit | |
| + 0.15 * no_aktif | |
| + 0.10 * bracket | |
| + 0.05 * tanggungan | |
| + 0.18 * (no_npwp * low_lit) # interaksi non-linear | |
| + 0.06 * (no_aktif * low_lit) # interaksi non-linear | |
| ) | |
| W_MAX = 0.32 + 0.22 + 0.15 + 0.10 + 0.05 + 0.18 + 0.06 # = 1.08 | |
| return float(np.clip(raw / W_MAX, 0.0, 1.0)) | |
| def risk_category(risk_index): | |
| """Petakan risk_index -> indeks kelas (0=Rendah,1=Sedang,2=Tinggi).""" | |
| if risk_index < RISK_BINS[0]: | |
| return 0 | |
| if risk_index < RISK_BINS[1]: | |
| return 1 | |
| return 2 | |
| # --------------------------------------------------------------------------- | |
| # Standardisasi (StandardScaler manual agar tak bergantung sklearn saat serving) | |
| # --------------------------------------------------------------------------- | |
| class Scaler: | |
| """StandardScaler ringan; hanya menstandardisasi kolom NON-biner/one-hot.""" | |
| def __init__(self, mean, std, scale_mask): | |
| self.mean = np.asarray(mean, dtype=np.float32) | |
| self.std = np.asarray(std, dtype=np.float32) | |
| self.scale_mask = np.asarray(scale_mask, dtype=bool) | |
| def fit(cls, X): | |
| mean = X.mean(axis=0) | |
| std = X.std(axis=0) | |
| std[std == 0] = 1.0 | |
| # Hanya standardisasi 3 kolom kontinu pertama (usia, log_gaji, literasi); | |
| # sisanya sudah dalam skala [0,1] / biner. | |
| scale_mask = np.zeros(X.shape[1], dtype=bool) | |
| for name in ("usia_norm", "log_gaji", "literasi_norm", "income_bracket", "tanggungan_norm"): | |
| scale_mask[FEATURE_NAMES.index(name)] = True | |
| return cls(mean, std, scale_mask) | |
| def transform(self, X): | |
| X = np.asarray(X, dtype=np.float32).copy() | |
| m = self.scale_mask | |
| X[:, m] = (X[:, m] - self.mean[m]) / self.std[m] | |
| return X | |
| def to_dict(self): | |
| return { | |
| "mean": self.mean.tolist(), | |
| "std": self.std.tolist(), | |
| "scale_mask": self.scale_mask.tolist(), | |
| } | |
| def from_dict(cls, d): | |
| return cls(d["mean"], d["std"], d["scale_mask"]) | |
| def save_preprocessing(path, scaler): | |
| meta = { | |
| "feature_names": FEATURE_NAMES, | |
| "n_features": N_FEATURES, | |
| "class_names": CLASS_NAMES, | |
| "risk_bins": RISK_BINS, | |
| "pekerjaan_cats": PEKERJAAN_CATS, | |
| "income_bracket_order": INCOME_BRACKET_ORDER, | |
| "scaler": scaler.to_dict(), | |
| } | |
| with open(path, "w", encoding="utf-8") as f: | |
| json.dump(meta, f, indent=2, ensure_ascii=False) | |
| def load_preprocessing(path): | |
| with open(path, "r", encoding="utf-8") as f: | |
| meta = json.load(f) | |
| meta["scaler"] = Scaler.from_dict(meta["scaler"]) | |
| return meta | |