""" 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) @classmethod 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(), } @classmethod 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