Blame for preprocessing.py

"""
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