Spaces:
Runtime error
Runtime error
| """ | |
| 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) | |
| # =========================================================================== | |
| 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 | |
| # =========================================================================== | |
| 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.") | |