Spaces:
Runtime error
Runtime error
File size: 7,599 Bytes
640eb2e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 | """
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.")
|