""" Taxy ML — Komponen Kustom Lanjutan (TensorFlow / Keras) ======================================================= Memenuhi syarat "minimal satu komponen kustom lanjutan". Di sini diimplementasikan KETIGA-tiganya agar pipeline robust: 1. Custom Layer -> FeatureGatingLayer (Model Subclassing dari keras.layers.Layer) 2. Custom Loss -> CombinedTaxLoss (gabungan focal-CE + Huber, multi-task) 3. Custom Callback -> TargetMetricCallback (pantau target akurasi/MAE + early stop) Semua kelas memakai @register_keras_serializable + get_config agar bisa diserialisasi ke format .keras / SavedModel. """ import tensorflow as tf import keras # =========================================================================== # 1. CUSTOM LAYER — Feature Gating (self-attention sederhana antar fitur) # =========================================================================== @keras.saving.register_keras_serializable(package="taxy") class FeatureGatingLayer(keras.layers.Layer): """Learnable feature-wise gating. Setiap unit input dikalikan gerbang (gate) sigmoid yang dipelajari dari seluruh vektor input -> model belajar menonjolkan fitur paling relevan untuk risiko pajak (mis. NPWP, literasi) dan meredam yang kurang penting. gate = sigmoid(W_g x + b_g) out = (x * gate) @ W_t + b_t # proyeksi linear setelah gating Mengimplementasikan build() (membuat bobot) dan call() secara manual. """ def __init__(self, units, **kwargs): super().__init__(**kwargs) self.units = int(units) def build(self, input_shape): in_dim = int(input_shape[-1]) # Bobot gerbang (gating) — menghasilkan satu gate per fitur input self.w_gate = self.add_weight( name="w_gate", shape=(in_dim, in_dim), initializer="glorot_uniform", trainable=True) self.b_gate = self.add_weight( name="b_gate", shape=(in_dim,), initializer="zeros", trainable=True) # Bobot transformasi (proyeksi ke `units`) self.w_transform = self.add_weight( name="w_transform", shape=(in_dim, self.units), initializer="glorot_uniform", trainable=True) self.b_transform = self.add_weight( name="b_transform", shape=(self.units,), initializer="zeros", trainable=True) super().build(input_shape) def call(self, inputs): gate = tf.sigmoid(tf.matmul(inputs, self.w_gate) + self.b_gate) gated = inputs * gate # penekanan fitur out = tf.matmul(gated, self.w_transform) + self.b_transform return tf.nn.gelu(out) def get_config(self): cfg = super().get_config() cfg.update({"units": self.units}) return cfg # =========================================================================== # 2. CUSTOM LOSS — Combined multi-task loss # =========================================================================== @keras.saving.register_keras_serializable(package="taxy") class CombinedTaxLoss(keras.losses.Loss): """Loss multi-task yang menggabungkan dua kepala output: - Klasifikasi (risk_category): Focal Categorical Cross-Entropy (focal -> menangani ketidakseimbangan kelas Tinggi/Sedang) - Regresi (risk_index): Huber loss (robust terhadap outlier) Karena Keras memanggil loss per-output, kelas ini dipakai dalam DUA mode: mode="classification" atau mode="regression". Bobot gabungan diatur di training loop (lihat train.py). """ def __init__(self, mode="classification", gamma=2.0, alpha=0.25, huber_delta=0.1, name="combined_tax_loss", **kwargs): super().__init__(name=name, **kwargs) self.mode = mode self.gamma = float(gamma) self.alpha = float(alpha) self.huber_delta = float(huber_delta) def call(self, y_true, y_pred): if self.mode == "classification": # y_true: one-hot [B,C], y_pred: probabilitas softmax [B,C] eps = keras.backend.epsilon() y_pred = tf.clip_by_value(y_pred, eps, 1.0 - eps) ce = -y_true * tf.math.log(y_pred) focal_weight = self.alpha * tf.pow(1.0 - y_pred, self.gamma) loss = tf.reduce_sum(focal_weight * ce, axis=-1) return loss else: # Huber loss untuk regresi err = y_true - tf.squeeze(y_pred, axis=-1) if len(y_pred.shape) > 1 else y_true - y_pred abs_err = tf.abs(err) quad = tf.minimum(abs_err, self.huber_delta) lin = abs_err - quad return 0.5 * quad * quad + self.huber_delta * lin def get_config(self): cfg = super().get_config() cfg.update({ "mode": self.mode, "gamma": self.gamma, "alpha": self.alpha, "huber_delta": self.huber_delta, }) return cfg # =========================================================================== # 3. CUSTOM CALLBACK — Target metric monitor + early stop # =========================================================================== class TargetMetricCallback(keras.callbacks.Callback): """Memantau metrik validasi terhadap target bisnis dan: - mencatat ke TensorBoard (scalar 'targets/...') - menyimpan bobot terbaik (berdasar skor gabungan) - early-stop saat KEDUA target tercapai (akurasi>=85% & MAE<=0.02) dan bertahan `patience_after_hit` epoch. Dipakai oleh custom training loop (train.py) yang memanggil on_epoch_end(epoch, logs={'val_acc':..., 'val_mae':...}). """ def __init__(self, target_acc=0.85, target_mae=0.02, summary_writer=None, patience_after_hit=3): super().__init__() self.target_acc = target_acc self.target_mae = target_mae self.writer = summary_writer self.patience_after_hit = patience_after_hit self.best_score = -1.0 self.best_epoch = -1 self.hit_streak = 0 self.should_stop = False self.targets_met_epoch = None def on_epoch_end(self, epoch, logs=None): logs = logs or {} acc = logs.get("val_acc", 0.0) mae = logs.get("val_mae", 1.0) # Skor gabungan: akurasi tinggi & MAE rendah lebih baik score = acc - mae if score > self.best_score: self.best_score = score self.best_epoch = epoch met = (acc >= self.target_acc) and (mae <= self.target_mae) if met: if self.targets_met_epoch is None: self.targets_met_epoch = epoch self.hit_streak += 1 else: self.hit_streak = 0 if self.writer is not None: with self.writer.as_default(): tf.summary.scalar("targets/val_accuracy", acc, step=epoch) tf.summary.scalar("targets/val_mae", mae, step=epoch) tf.summary.scalar("targets/acc_target_met", 1.0 if acc >= self.target_acc else 0.0, step=epoch) tf.summary.scalar("targets/mae_target_met", 1.0 if mae <= self.target_mae else 0.0, step=epoch) flag = "[OK]" if met else " " print(f" {flag} val_acc={acc:.4f} (>= {self.target_acc}) | " f"val_mae={mae:.5f} (<= {self.target_mae}) | streak={self.hit_streak}") if met and self.hit_streak >= self.patience_after_hit: self.should_stop = True print(f" >> Kedua target tercapai & stabil {self.hit_streak} epoch. Early stop.")