Debby
1330 lines
63 KiB
Python
1330 lines
63 KiB
Python
"""
|
|
BIGQUERY ANALYSIS LAYER - INDICATOR NORM AGGREGATION
|
|
|
|
PERUBAHAN ARSITEKTUR:
|
|
- ASEAN aggregate DIGABUNG ke dalam tabel yang sama (country_id=0, country_name="ASEAN")
|
|
sehingga Looker Studio dapat memfilter: per negara, atau ASEAN saja.
|
|
- agg_narrative_indicator: granularity tetap per indicator_id (all years, all countries),
|
|
ASEAN summary ditambahkan sebagai kolom terpisah (asean_avg_value_first/last).
|
|
|
|
Output 2 tabel:
|
|
1. agg_indicator_norm -> per baris (year x country x indicator), termasuk ASEAN rows
|
|
2. agg_narrative_indicator -> per indicator_id, ada kolom asean_avg_* tambahan
|
|
|
|
BUGFIX (diteruskan dari versi sebelumnya):
|
|
- INDICATOR_NAME_ID_MAP: semua key lowercase agar match dengan .lower().strip() lookup.
|
|
"""
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime
|
|
import logging
|
|
import json
|
|
|
|
from scripts.bigquery_config import get_bigquery_client
|
|
from scripts.bigquery_helpers import (
|
|
log_update,
|
|
load_to_bigquery,
|
|
read_from_bigquery,
|
|
setup_logging,
|
|
save_etl_metadata,
|
|
)
|
|
from google.cloud import bigquery
|
|
|
|
|
|
# =============================================================================
|
|
# KONSTANTA
|
|
# =============================================================================
|
|
|
|
ASEAN_COUNTRY_ID = 0
|
|
ASEAN_COUNTRY_NAME = "ASEAN"
|
|
ASEAN_COUNTRY_NAME_ID = "ASEAN"
|
|
|
|
|
|
# =============================================================================
|
|
# MAPPING BAHASA INDONESIA
|
|
# CHANGED: Other / Lainnya
|
|
# =============================================================================
|
|
|
|
COUNTRY_NAME_ID_MAP: dict = {
|
|
"Brunei Darussalam" : "Brunei Darussalam",
|
|
"Cambodia" : "Kamboja",
|
|
"Indonesia" : "Indonesia",
|
|
"Lao People's Democratic Republic" : "Laos",
|
|
"Lao PDR" : "Laos",
|
|
"Malaysia" : "Malaysia",
|
|
"Myanmar" : "Myanmar",
|
|
"Philippines" : "Filipina",
|
|
"Singapore" : "Singapura",
|
|
"Thailand" : "Thailand",
|
|
"Timor-Leste" : "Timor-Leste",
|
|
"Viet Nam" : "Vietnam",
|
|
"Vietnam" : "Vietnam",
|
|
"ASEAN" : "ASEAN",
|
|
}
|
|
|
|
PILLAR_NAME_ID_MAP: dict = {
|
|
# Mapping nama pilar (Inggris dengan prefix Food) -> Bahasa Indonesia
|
|
"Food Availability" : "Ketersediaan Pangan",
|
|
"Food Access" : "Akses Pangan",
|
|
"Food Utilization" : "Pemanfaatan Pangan",
|
|
"Food Stability" : "Stabilitas Pangan",
|
|
"Food Other" : "Indikator Tambahan",
|
|
# Variasi tanpa prefix Food
|
|
"Availability" : "Ketersediaan Pangan",
|
|
"Access" : "Akses Pangan",
|
|
"Utilization" : "Pemanfaatan Pangan",
|
|
"Stability" : "Stabilitas Pangan",
|
|
"Other" : "Indikator Tambahan",
|
|
# lowercase
|
|
"food availability" : "Ketersediaan Pangan",
|
|
"food access" : "Akses Pangan",
|
|
"food utilization" : "Pemanfaatan Pangan",
|
|
"food stability" : "Stabilitas Pangan",
|
|
"food other" : "Indikator Tambahan",
|
|
"availability" : "Ketersediaan Pangan",
|
|
"access" : "Akses Pangan",
|
|
"utilization" : "Pemanfaatan Pangan",
|
|
"stability" : "Stabilitas Pangan",
|
|
"other" : "Indikator Tambahan",
|
|
}
|
|
|
|
# BUGFIX: semua key lowercase
|
|
INDICATOR_NAME_ID_MAP: dict = {
|
|
"dietary energy supply used in the estimation of the prevalence of undernourishment (kcal/cap/day)":
|
|
"Pasokan energi makanan yang digunakan dalam estimasi prevalensi kekurangan gizi (kkal/kapita/hari)",
|
|
"dietary energy supply used in the estimation of the prevalence of undernourishment (kcal/cap/day) (3-year average)":
|
|
"Pasokan energi makanan yang digunakan dalam estimasi prevalensi kekurangan gizi (kkal/kapita/hari) (rata-rata 3 tahun)",
|
|
"percentage of population using at least basic drinking water services (percent)":
|
|
"Persentase penduduk yang menggunakan layanan air minum dasar (persen)",
|
|
"percentage of population using at least basic sanitation services (percent)":
|
|
"Persentase penduduk yang menggunakan layanan sanitasi dasar (persen)",
|
|
"percentage of population using safely managed drinking water services (percent)":
|
|
"Persentase penduduk yang menggunakan layanan air minum yang dikelola dengan aman (persen)",
|
|
"percentage of population using safely managed sanitation services (percent)":
|
|
"Persentase penduduk yang menggunakan layanan sanitasi yang dikelola dengan aman (persen)",
|
|
"rail lines density (total route in km per 100 square km of land area)":
|
|
"Kepadatan jalur kereta api (total rute dalam km per 100 km² lahan)",
|
|
"average dietary energy requirement (kcal/cap/day)":
|
|
"Rata-rata kebutuhan energi makanan (kkal/kapita/hari)",
|
|
"average dietary energy supply adequacy (percent) (3-year average)":
|
|
"Kecukupan rata-rata pasokan energi makanan (persen) (rata-rata 3 tahun)",
|
|
"average fat supply (g/cap/day) (3-year average)":
|
|
"Rata-rata pasokan lemak (g/kapita/hari) (rata-rata 3 tahun)",
|
|
"average protein supply (g/cap/day) (3-year average)":
|
|
"Rata-rata pasokan protein (g/kapita/hari) (rata-rata 3 tahun)",
|
|
"average supply of protein of animal origin (g/cap/day) (3-year average)":
|
|
"Rata-rata pasokan protein hewani (g/kapita/hari) (rata-rata 3 tahun)",
|
|
"percent of arable land equipped for irrigation (percent) (3-year average)":
|
|
"Persentase lahan pertanian yang dilengkapi irigasi (persen) (rata-rata 3 tahun)",
|
|
"cereal import dependency ratio (percent) (3-year average)":
|
|
"Rasio ketergantungan impor sereal (persen) (rata-rata 3 tahun)",
|
|
"share of dietary energy supply derived from cereals, roots and tubers (percent) (3-year average)":
|
|
"Proporsi pasokan energi makanan dari serealia, akar, dan umbi-umbian (persen) (rata-rata 3 tahun)",
|
|
"per capita food supply variability (kcal/cap/day)":
|
|
"Variabilitas pasokan pangan per kapita (kkal/kapita/hari)",
|
|
"value of food imports in total merchandise exports (percent) (3-year average)":
|
|
"Nilai impor pangan terhadap total ekspor barang (persen) (rata-rata 3 tahun)",
|
|
"gross domestic product per capita, ppp, (constant 2021 international $)":
|
|
"Produk domestik bruto per kapita, PPP (internasional konstan 2021 US$)",
|
|
"political stability and absence of violence/terrorism (index)":
|
|
"Stabilitas politik dan tidak adanya kekerasan/terorisme (indeks)",
|
|
"prevalence of undernourishment (percent) (3-year average)":
|
|
"Prevalensi kekurangan gizi (persen) (rata-rata 3 tahun)",
|
|
"number of people undernourished (million) (3-year average)":
|
|
"Jumlah penduduk kekurangan gizi (juta jiwa) (rata-rata 3 tahun)",
|
|
"minimum dietary energy requirement (kcal/cap/day)":
|
|
"Kebutuhan energi makanan minimum (kkal/kapita/hari)",
|
|
"prevalence of exclusive breastfeeding among infants 0-5 months of age (percent)":
|
|
"Prevalensi pemberian ASI eksklusif pada bayi usia 0-5 bulan (persen)",
|
|
"number of children under 5 years affected by wasting (million)":
|
|
"Jumlah anak di bawah 5 tahun yang mengalami wasting (juta jiwa)",
|
|
"number of moderately or severely food insecure female adults (million) (3-year average)":
|
|
"Jumlah perempuan dewasa yang mengalami kerawanan pangan sedang atau berat (juta jiwa) (rata-rata 3 tahun)",
|
|
"number of moderately or severely food insecure male adults (million) (3-year average)":
|
|
"Jumlah laki-laki dewasa yang mengalami kerawanan pangan sedang atau berat (juta jiwa) (rata-rata 3 tahun)",
|
|
"number of moderately or severely food insecure people (million) (3-year average)":
|
|
"Jumlah penduduk yang mengalami kerawanan pangan sedang atau berat (juta jiwa) (rata-rata 3 tahun)",
|
|
"number of severely food insecure female adults (million) (3-year average)":
|
|
"Jumlah perempuan dewasa yang mengalami kerawanan pangan berat (juta jiwa) (rata-rata 3 tahun)",
|
|
"number of severely food insecure male adults (million) (3-year average)":
|
|
"Jumlah laki-laki dewasa yang mengalami kerawanan pangan berat (juta jiwa) (rata-rata 3 tahun)",
|
|
"number of severely food insecure people (million) (3-year average)":
|
|
"Jumlah penduduk yang mengalami kerawanan pangan berat (juta jiwa) (rata-rata 3 tahun)",
|
|
"number of women of reproductive age (15-49 years) affected by anemia (million)":
|
|
"Jumlah perempuan usia reproduksi (15-49 tahun) yang menderita anemia (juta jiwa)",
|
|
"percentage of children under 5 years affected by wasting (percent)":
|
|
"Persentase anak di bawah 5 tahun yang mengalami wasting (persen)",
|
|
"prevalence of anemia among women of reproductive age (15-49 years) (percent)":
|
|
"Prevalensi anemia pada perempuan usia reproduksi (15-49 tahun) (persen)",
|
|
"coefficient of variation of habitual caloric consumption distribution (real number)":
|
|
"Koefisien variasi distribusi konsumsi kalori kebiasaan (bilangan riil)",
|
|
"incidence of caloric losses at retail distribution level (percent)":
|
|
"Insidensi kehilangan kalori pada tingkat distribusi ritel (persen)",
|
|
"number of children under 5 years of age who are overweight (modeled estimates) (million)":
|
|
"Jumlah anak di bawah 5 tahun yang mengalami kelebihan berat badan (estimasi model) (juta jiwa)",
|
|
"number of children under 5 years of age who are stunted (modeled estimates) (million)":
|
|
"Jumlah anak di bawah 5 tahun yang mengalami stunting (estimasi model) (juta jiwa)",
|
|
"number of newborns with low birthweight (million)":
|
|
"Jumlah bayi baru lahir dengan berat badan lahir rendah (juta jiwa)",
|
|
"number of obese adults (18 years and older) (million)":
|
|
"Jumlah orang dewasa yang mengalami obesitas (18 tahun ke atas) (juta jiwa)",
|
|
"percentage of children under 5 years of age who are overweight (modelled estimates) (percent)":
|
|
"Persentase anak di bawah 5 tahun yang mengalami kelebihan berat badan (estimasi model) (persen)",
|
|
"percentage of children under 5 years of age who are stunted (modelled estimates) (percent)":
|
|
"Persentase anak di bawah 5 tahun yang mengalami stunting (estimasi model) (persen)",
|
|
"prevalence of low birthweight (percent)":
|
|
"Prevalensi berat badan lahir rendah (persen)",
|
|
"prevalence of moderate or severe food insecurity in the female adult population (percent) (3-year average)":
|
|
"Prevalensi kerawanan pangan sedang atau berat pada penduduk perempuan dewasa (persen) (rata-rata 3 tahun)",
|
|
"prevalence of moderate or severe food insecurity in the male adult population (percent) (3-year average)":
|
|
"Prevalensi kerawanan pangan sedang atau berat pada penduduk laki-laki dewasa (persen) (rata-rata 3 tahun)",
|
|
"prevalence of moderate or severe food insecurity in the total population (percent) (3-year average)":
|
|
"Prevalensi kerawanan pangan sedang atau berat pada total penduduk (persen) (rata-rata 3 tahun)",
|
|
"prevalence of obesity in the adult population (18 years and older) (percent)":
|
|
"Prevalensi obesitas pada penduduk dewasa (18 tahun ke atas) (persen)",
|
|
"prevalence of severe food insecurity in the female adult population (percent) (3-year average)":
|
|
"Prevalensi kerawanan pangan berat pada penduduk perempuan dewasa (persen) (rata-rata 3 tahun)",
|
|
"prevalence of severe food insecurity in the male adult population (percent) (3-year average)":
|
|
"Prevalensi kerawanan pangan berat pada penduduk laki-laki dewasa (persen) (rata-rata 3 tahun)",
|
|
"prevalence of severe food insecurity in the total population (percent) (3-year average)":
|
|
"Prevalensi kerawanan pangan berat pada total penduduk (persen) (rata-rata 3 tahun)",
|
|
}
|
|
|
|
|
|
def get_country_name_id(country_name: str) -> str:
|
|
return COUNTRY_NAME_ID_MAP.get(str(country_name).strip(), str(country_name))
|
|
|
|
|
|
def get_indicator_name_id(indicator_name: str) -> str:
|
|
return INDICATOR_NAME_ID_MAP.get(str(indicator_name).lower().strip(), str(indicator_name))
|
|
|
|
|
|
def get_pillar_name_id(pillar_name: str) -> str:
|
|
return PILLAR_NAME_ID_MAP.get(str(pillar_name).strip(), str(pillar_name))
|
|
|
|
|
|
# =============================================================================
|
|
# SDG-ONLY KEYWORD SET
|
|
# =============================================================================
|
|
|
|
SDG_ONLY_KEYWORDS: frozenset = frozenset([
|
|
"prevalence of undernourishment (percent) (3-year average)",
|
|
"number of people undernourished (million) (3-year average)",
|
|
"prevalence of severe food insecurity in the total population (percent) (3-year average)",
|
|
"prevalence of severe food insecurity in the male adult population (percent) (3-year average)",
|
|
"prevalence of severe food insecurity in the female adult population (percent) (3-year average)",
|
|
"prevalence of moderate or severe food insecurity in the total population (percent) (3-year average)",
|
|
"prevalence of moderate or severe food insecurity in the male adult population (percent) (3-year average)",
|
|
"prevalence of moderate or severe food insecurity in the female adult population (percent) (3-year average)",
|
|
"number of severely food insecure people (million) (3-year average)",
|
|
"number of severely food insecure male adults (million) (3-year average)",
|
|
"number of severely food insecure female adults (million) (3-year average)",
|
|
"number of moderately or severely food insecure people (million) (3-year average)",
|
|
"number of moderately or severely food insecure male adults (million) (3-year average)",
|
|
"number of moderately or severely food insecure female adults (million) (3-year average)",
|
|
"percentage of children under 5 years of age who are stunted (modelled estimates) (percent)",
|
|
"number of children under 5 years of age who are stunted (modeled estimates) (million)",
|
|
"percentage of children under 5 years affected by wasting (percent)",
|
|
"number of children under 5 years affected by wasting (million)",
|
|
"percentage of children under 5 years of age who are overweight (modelled estimates) (percent)",
|
|
"number of children under 5 years of age who are overweight (modeled estimates) (million)",
|
|
"prevalence of anemia among women of reproductive age (15-49 years) (percent)",
|
|
"number of women of reproductive age (15-49 years) affected by anemia (million)",
|
|
])
|
|
|
|
_SDG_ONLY_LOWER: frozenset = frozenset(k.lower() for k in SDG_ONLY_KEYWORDS)
|
|
|
|
_FIES_DETECTION_KEYWORDS: frozenset = frozenset([
|
|
"prevalence of severe food insecurity in the total population (percent) (3-year average)",
|
|
"prevalence of moderate or severe food insecurity in the total population (percent) (3-year average)",
|
|
"number of severely food insecure people (million) (3-year average)",
|
|
"number of moderately or severely food insecure people (million) (3-year average)",
|
|
])
|
|
_FIES_DETECTION_LOWER: frozenset = frozenset(k.lower() for k in _FIES_DETECTION_KEYWORDS)
|
|
|
|
DIRECTION_INVERT_KEYWORDS = frozenset({
|
|
"negative", "lower_better", "lower_is_better", "inverse", "neg",
|
|
})
|
|
DIRECTION_POSITIVE_KEYWORDS = frozenset({
|
|
"positive", "higher_better", "higher_is_better",
|
|
})
|
|
|
|
_PERFORMANCE_THRESHOLD: float = 60.0
|
|
|
|
|
|
# =============================================================================
|
|
# PURE HELPERS
|
|
# =============================================================================
|
|
|
|
def _should_invert(direction: str, logger=None, context: str = "") -> bool:
|
|
d = str(direction).lower().strip()
|
|
if d in DIRECTION_INVERT_KEYWORDS:
|
|
return True
|
|
if d in DIRECTION_POSITIVE_KEYWORDS:
|
|
return False
|
|
if logger:
|
|
logger.warning(
|
|
f" [DIRECTION WARNING] Unknown direction '{direction}' "
|
|
f"{'(' + context + ')' if context else ''}. Defaulting to positive (no invert)."
|
|
)
|
|
return False
|
|
|
|
|
|
def global_minmax(series: pd.Series, lo: float = 1.0, hi: float = 100.0) -> pd.Series:
|
|
values = series.dropna().values
|
|
if len(values) == 0:
|
|
return pd.Series(np.nan, index=series.index)
|
|
v_min, v_max = values.min(), values.max()
|
|
if v_min == v_max:
|
|
return pd.Series((lo + hi) / 2.0, index=series.index)
|
|
result = np.full(len(series), np.nan)
|
|
not_nan = series.notna()
|
|
result[not_nan.values] = lo + (series[not_nan].values - v_min) / (v_max - v_min) * (hi - lo)
|
|
return pd.Series(result, index=series.index)
|
|
|
|
|
|
def _compute_yoy(df: pd.DataFrame) -> pd.DataFrame:
|
|
df = df.sort_values("year").copy()
|
|
df["value_prev"] = df["value"].shift(1)
|
|
df["norm_value_prev"] = df["norm_value"].shift(1)
|
|
df["yoy_value"] = np.where(
|
|
df["value"].notna() & df["value_prev"].notna(),
|
|
df["value"] - df["value_prev"],
|
|
np.nan,
|
|
)
|
|
df["yoy_norm_value"] = np.where(
|
|
df["norm_value"].notna() & df["norm_value_prev"].notna(),
|
|
df["norm_value"] - df["norm_value_prev"],
|
|
np.nan,
|
|
)
|
|
df = df.drop(columns=["value_prev", "norm_value_prev"])
|
|
return df
|
|
|
|
|
|
def _is_lower_better(direction: str) -> bool:
|
|
return str(direction).lower().strip() in DIRECTION_INVERT_KEYWORDS
|
|
|
|
|
|
# =============================================================================
|
|
# NARRATIVE CONDITION DETECTORS
|
|
# =============================================================================
|
|
|
|
def _detect_trend(scores_by_year: pd.Series, lower_better: bool) -> str:
|
|
if len(scores_by_year) < 3:
|
|
return "insufficient_data"
|
|
|
|
years = sorted(scores_by_year.index)
|
|
vals = [scores_by_year[y] for y in years if not pd.isna(scores_by_year.get(y, np.nan))]
|
|
|
|
if len(vals) < 3:
|
|
return "insufficient_data"
|
|
|
|
x = np.arange(len(vals))
|
|
slope = np.polyfit(x, vals, 1)[0]
|
|
improving = (slope > 0 and not lower_better) or (slope < 0 and lower_better)
|
|
|
|
mid = len(vals) // 2
|
|
first_half = vals[:mid]
|
|
second_half = vals[mid:]
|
|
slope1 = np.polyfit(np.arange(len(first_half)), first_half, 1)[0] if len(first_half) > 1 else 0
|
|
slope2 = np.polyfit(np.arange(len(second_half)), second_half, 1)[0] if len(second_half) > 1 else 0
|
|
|
|
cv = np.std(vals) / (np.mean(vals) + 1e-9)
|
|
if cv > 0.25:
|
|
return "fluctuating"
|
|
|
|
if improving:
|
|
if lower_better:
|
|
slowing = slope2 > slope1
|
|
else:
|
|
slowing = slope2 < slope1
|
|
return "improving_slowing" if slowing else "improving_consistent"
|
|
else:
|
|
return "deteriorating"
|
|
|
|
|
|
def _detect_gap_trend(df_ind: pd.DataFrame, lower_better: bool) -> str:
|
|
# Hanya hitung gap di antara negara asli (bukan ASEAN)
|
|
df_real = df_ind[df_ind["country_id"] != ASEAN_COUNTRY_ID]
|
|
std_by_year = (
|
|
df_real.groupby("year")["value"]
|
|
.std()
|
|
.dropna()
|
|
)
|
|
if len(std_by_year) < 3:
|
|
return "unknown"
|
|
|
|
years = sorted(std_by_year.index)
|
|
stds = [std_by_year[y] for y in years]
|
|
slope = np.polyfit(np.arange(len(stds)), stds, 1)[0]
|
|
|
|
if abs(slope) < 0.01 * np.mean(stds):
|
|
return "stable"
|
|
return "widening" if slope > 0 else "narrowing"
|
|
|
|
|
|
def _detect_anomaly_year(scores_by_year: pd.Series) -> tuple:
|
|
if len(scores_by_year) < 3:
|
|
return None, None
|
|
|
|
years = sorted(scores_by_year.index)
|
|
deltas = {}
|
|
for i in range(1, len(years)):
|
|
y_prev = years[i - 1]
|
|
y_curr = years[i]
|
|
v_prev = scores_by_year.get(y_prev, np.nan)
|
|
v_curr = scores_by_year.get(y_curr, np.nan)
|
|
if not pd.isna(v_prev) and not pd.isna(v_curr):
|
|
deltas[y_curr] = v_curr - v_prev
|
|
|
|
if not deltas:
|
|
return None, None
|
|
|
|
max_drop_year = min(deltas, key=deltas.get)
|
|
max_rise_year = max(deltas, key=deltas.get)
|
|
|
|
threshold = 1.5 * np.std(list(deltas.values()))
|
|
if abs(deltas[max_drop_year]) > threshold and deltas[max_drop_year] < 0:
|
|
return max_drop_year, "drop"
|
|
if abs(deltas[max_rise_year]) > threshold and deltas[max_rise_year] > 0:
|
|
return max_rise_year, "rise"
|
|
|
|
return None, None
|
|
|
|
|
|
def _detect_consistency(df_ind: pd.DataFrame, lower_better: bool) -> tuple:
|
|
"""Hanya negara asli (bukan ASEAN) yang di-analisa konsistensinya."""
|
|
df_real = df_ind[df_ind["country_id"] != ASEAN_COUNTRY_ID]
|
|
country_avg = (
|
|
df_real.groupby("country_name")["value"]
|
|
.mean()
|
|
.dropna()
|
|
)
|
|
if country_avg.empty:
|
|
return None, None, False
|
|
|
|
if lower_better:
|
|
best = country_avg.idxmin()
|
|
worst = country_avg.idxmax()
|
|
else:
|
|
best = country_avg.idxmax()
|
|
worst = country_avg.idxmin()
|
|
|
|
asean_avg_by_year = df_real.groupby("year")["value"].mean()
|
|
country_by_year = df_real[df_real["country_name"] == best].set_index("year")["value"]
|
|
|
|
years_both = set(asean_avg_by_year.index) & set(country_by_year.index)
|
|
if not years_both:
|
|
return best, worst, False
|
|
|
|
if lower_better:
|
|
consistent = all(
|
|
country_by_year[y] <= asean_avg_by_year[y]
|
|
for y in years_both
|
|
if not pd.isna(country_by_year.get(y, np.nan))
|
|
)
|
|
else:
|
|
consistent = all(
|
|
country_by_year[y] >= asean_avg_by_year[y]
|
|
for y in years_both
|
|
if not pd.isna(country_by_year.get(y, np.nan))
|
|
)
|
|
|
|
return best, worst, consistent
|
|
|
|
|
|
# =============================================================================
|
|
# NARRATIVE BUILDER
|
|
# =============================================================================
|
|
|
|
def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tuple:
|
|
ind_id = int(row["indicator_id"])
|
|
ind_name_en = str(row["indicator_name"]).strip()
|
|
ind_name_id = str(row.get("indicator_name_id", ind_name_en)).strip()
|
|
unit = str(row["unit"]).strip() if row["unit"] else ""
|
|
direction = str(row["direction"]).strip()
|
|
pillar_en = str(row["pillar_name"]).strip()
|
|
pillar_id_ = get_pillar_name_id(pillar_en)
|
|
framework = str(row["framework"]).strip()
|
|
year_min = int(row["year_min"])
|
|
year_max = int(row["year_max"])
|
|
lower_better = _is_lower_better(direction)
|
|
|
|
# Gunakan hanya negara asli (bukan ASEAN) untuk analisa tren/gap/konsistensi
|
|
df_ind = df_full[
|
|
(df_full["indicator_id"] == ind_id) &
|
|
(df_full["country_id"] != ASEAN_COUNTRY_ID)
|
|
].copy()
|
|
|
|
if df_ind.empty:
|
|
na_en = f"{ind_name_en} ({framework}, {pillar_en}): Insufficient data for analysis."
|
|
na_id = f"{ind_name_id} ({framework}, {pillar_id_}): Data tidak cukup untuk dianalisis."
|
|
return na_en, na_id
|
|
|
|
asean_avg_by_year = (
|
|
df_ind.groupby("year")["value"].mean().dropna()
|
|
)
|
|
|
|
trend_label = _detect_trend(asean_avg_by_year, lower_better)
|
|
gap_label = _detect_gap_trend(df_ind, lower_better)
|
|
anomaly_year, anomaly_dir = _detect_anomaly_year(asean_avg_by_year)
|
|
best_country_en, worst_country_en, is_consistent = _detect_consistency(df_ind, lower_better)
|
|
|
|
best_country_id = get_country_name_id(best_country_en) if best_country_en else None
|
|
worst_country_id = get_country_name_id(worst_country_en) if worst_country_en else None
|
|
|
|
avg_first = row.get("avg_value_first", np.nan)
|
|
avg_last = row.get("avg_value_last", np.nan)
|
|
|
|
def fmt(v):
|
|
if pd.isna(v):
|
|
return "N/A"
|
|
abs_v = abs(v)
|
|
s = f"{v:,.1f}" if abs_v >= 1000 else (f"{v:.2f}" if abs_v >= 10 else f"{v:.3f}")
|
|
return f"{s} {unit}".strip() if unit else s
|
|
|
|
sentences_en = []
|
|
sentences_id = []
|
|
|
|
s1_en = f"{ind_name_en} ({framework}, {pillar_en}, {year_min}-{year_max}):"
|
|
s1_id = f"{ind_name_id} ({framework}, {pillar_id_}, {year_min}-{year_max}):"
|
|
sentences_en.append(s1_en)
|
|
sentences_id.append(s1_id)
|
|
|
|
trend_map_en = {
|
|
"improving_consistent": f"Regional average improved consistently from {fmt(avg_first)} to {fmt(avg_last)}.",
|
|
"improving_slowing": f"Regional average improved from {fmt(avg_first)} to {fmt(avg_last)}, though the pace slowed in recent years.",
|
|
"deteriorating": f"Regional average worsened from {fmt(avg_first)} to {fmt(avg_last)} over the period.",
|
|
"fluctuating": f"Regional average fluctuated between {fmt(avg_first)} and {fmt(avg_last)} with no clear trend.",
|
|
"insufficient_data": f"Trend analysis is limited due to sparse data.",
|
|
}
|
|
trend_map_id = {
|
|
"improving_consistent": f"Rata-rata regional membaik secara konsisten dari {fmt(avg_first)} menjadi {fmt(avg_last)}.",
|
|
"improving_slowing": f"Rata-rata regional membaik dari {fmt(avg_first)} menjadi {fmt(avg_last)}, namun lajunya melambat dalam beberapa tahun terakhir.",
|
|
"deteriorating": f"Rata-rata regional memburuk dari {fmt(avg_first)} menjadi {fmt(avg_last)} sepanjang periode.",
|
|
"fluctuating": f"Rata-rata regional berfluktuasi antara {fmt(avg_first)} dan {fmt(avg_last)} tanpa tren yang jelas.",
|
|
"insufficient_data": f"Analisis tren terbatas karena data yang tersedia tidak cukup.",
|
|
}
|
|
sentences_en.append(trend_map_en.get(trend_label, ""))
|
|
sentences_id.append(trend_map_id.get(trend_label, ""))
|
|
|
|
if gap_label == "widening":
|
|
sentences_en.append("Disparity among ASEAN countries has widened over time, indicating unequal progress.")
|
|
sentences_id.append("Kesenjangan antar negara ASEAN melebar seiring waktu, menunjukkan kemajuan yang tidak merata.")
|
|
elif gap_label == "narrowing":
|
|
sentences_en.append("Disparity among ASEAN countries has narrowed, suggesting more balanced regional progress.")
|
|
sentences_id.append("Kesenjangan antar negara ASEAN menyempit, mengindikasikan kemajuan regional yang lebih merata.")
|
|
elif gap_label == "stable":
|
|
sentences_en.append("The gap among ASEAN countries remained relatively stable throughout the period.")
|
|
sentences_id.append("Kesenjangan antar negara ASEAN relatif stabil sepanjang periode.")
|
|
|
|
if anomaly_year is not None:
|
|
if anomaly_dir == "drop":
|
|
sentences_en.append(f"A notable decline was recorded in {anomaly_year}, which stood out from the overall pattern.")
|
|
sentences_id.append(f"Penurunan signifikan tercatat pada tahun {anomaly_year}, yang menyimpang dari pola keseluruhan.")
|
|
elif anomaly_dir == "rise":
|
|
sentences_en.append(f"A sharp improvement was observed in {anomaly_year}, standing out from the overall pattern.")
|
|
sentences_id.append(f"Peningkatan tajam tercatat pada tahun {anomaly_year}, yang menyimpang dari pola keseluruhan.")
|
|
|
|
if best_country_en and worst_country_en:
|
|
if is_consistent:
|
|
sentences_en.append(
|
|
f"{best_country_en} consistently performed above the regional average, "
|
|
f"while {worst_country_en} consistently lagged behind."
|
|
)
|
|
sentences_id.append(
|
|
f"{best_country_id} secara konsisten berada di atas rata-rata regional, "
|
|
f"sementara {worst_country_id} secara konsisten tertinggal."
|
|
)
|
|
else:
|
|
sentences_en.append(
|
|
f"Overall, {best_country_en} showed the best performance, "
|
|
f"while {worst_country_en} had the weakest results across the period."
|
|
)
|
|
sentences_id.append(
|
|
f"Secara keseluruhan, {best_country_id} menunjukkan performa terbaik, "
|
|
f"sementara {worst_country_id} memiliki hasil terlemah sepanjang periode."
|
|
)
|
|
|
|
narrative_en = " ".join(s for s in sentences_en if s)
|
|
narrative_id = " ".join(s for s in sentences_id if s)
|
|
return narrative_en, narrative_id
|
|
|
|
|
|
# =============================================================================
|
|
# MAIN CLASS
|
|
# =============================================================================
|
|
|
|
class IndicatorNormAggregator:
|
|
|
|
def __init__(self, client: bigquery.Client):
|
|
self.client = client
|
|
self.logger = logging.getLogger(self.__class__.__name__)
|
|
self.logger.propagate = False
|
|
|
|
self.df = None
|
|
self.df_unit = None
|
|
self.sdgs_start_year = None
|
|
|
|
self.pipeline_start = None
|
|
self.pipeline_metadata = {
|
|
"rows_fetched" : 0,
|
|
"rows_loaded" : 0,
|
|
"rows_loaded_narrative" : 0,
|
|
"start_time" : None,
|
|
"end_time" : None,
|
|
}
|
|
|
|
# =========================================================================
|
|
# STEP 1: Load data
|
|
# =========================================================================
|
|
|
|
def load_data(self):
|
|
self.logger.info("\n" + "=" * 80)
|
|
self.logger.info("STEP 1: LOAD DATA — fact_asean_food_security_selected")
|
|
self.logger.info("=" * 80)
|
|
|
|
self.df = read_from_bigquery(
|
|
self.client, "fact_asean_food_security_selected", layer="gold"
|
|
)
|
|
|
|
required = {
|
|
"country_id", "country_name",
|
|
"indicator_id", "indicator_name", "direction",
|
|
"pillar_id", "pillar_name",
|
|
"year", "value",
|
|
}
|
|
missing = required - set(self.df.columns)
|
|
if missing:
|
|
raise ValueError(f"Kolom tidak ditemukan: {missing}")
|
|
|
|
n_null = self.df["direction"].isna().sum()
|
|
if n_null > 0:
|
|
self.logger.warning(f" {n_null} rows direction NULL -> diisi 'positive'")
|
|
self.df["direction"] = self.df["direction"].fillna("positive")
|
|
|
|
# Rename pillar_name: add 'Food ' prefix, remove
|
|
PILLAR_RENAME_MAP = {
|
|
'Availability' : 'Food Availability',
|
|
'Access' : 'Food Access',
|
|
'Utilization' : 'Food Utilization',
|
|
'Stability' : 'Food Stability',
|
|
'Other' : 'Food Other',
|
|
}
|
|
self.df["pillar_name"] = self.df["pillar_name"].replace(PILLAR_RENAME_MAP)
|
|
|
|
self.pipeline_metadata["rows_fetched"] = len(self.df)
|
|
self.logger.info(f" Rows : {len(self.df):,}")
|
|
self.logger.info(f" Countries : {self.df['country_id'].nunique()}")
|
|
self.logger.info(f" Indicators: {self.df['indicator_id'].nunique()}")
|
|
self.logger.info(
|
|
f" Years : {int(self.df['year'].min())} - {int(self.df['year'].max())}"
|
|
)
|
|
|
|
# =========================================================================
|
|
# STEP 2: Load unit
|
|
# =========================================================================
|
|
|
|
def load_units(self):
|
|
self.logger.info("\n" + "=" * 80)
|
|
self.logger.info("STEP 2: LOAD UNIT — dim_indicator")
|
|
self.logger.info("=" * 80)
|
|
|
|
dim = read_from_bigquery(self.client, "dim_indicator", layer="gold")
|
|
|
|
if "indicator_id" not in dim.columns or "unit" not in dim.columns:
|
|
raise ValueError(
|
|
f"dim_indicator harus punya kolom 'indicator_id' dan 'unit'. "
|
|
f"Kolom tersedia: {list(dim.columns)}"
|
|
)
|
|
|
|
self.df_unit = (
|
|
dim[["indicator_id", "unit"]]
|
|
.drop_duplicates(subset=["indicator_id"])
|
|
.copy()
|
|
)
|
|
self.df_unit["indicator_id"] = self.df_unit["indicator_id"].astype(int)
|
|
self.df_unit["unit"] = self.df_unit["unit"].fillna("").astype(str)
|
|
|
|
self.logger.info(f" dim_indicator rows (unique indicator_id): {len(self.df_unit):,}")
|
|
|
|
# =========================================================================
|
|
# STEP 3: Merge unit
|
|
# =========================================================================
|
|
|
|
def _merge_unit(self):
|
|
before = len(self.df)
|
|
self.df = self.df.merge(self.df_unit, on="indicator_id", how="left")
|
|
self.df["unit"] = self.df["unit"].fillna("").astype(str)
|
|
after = len(self.df)
|
|
assert before == after, f"Row count berubah: {before} -> {after}"
|
|
self.logger.info(f" Merge unit OK. Rows: {after:,}")
|
|
|
|
# =========================================================================
|
|
# STEP 3b: Tambah kolom nama Bahasa Indonesia
|
|
# =========================================================================
|
|
|
|
def _add_indonesia_name_columns(self):
|
|
self.df["country_name_id"] = self.df["country_name"].apply(get_country_name_id).astype(str)
|
|
self.df["indicator_name_id"] = self.df["indicator_name"].apply(get_indicator_name_id).astype(str)
|
|
self.df["pillar_name_id"] = self.df["pillar_name"].apply(get_pillar_name_id).astype(str)
|
|
|
|
self.logger.info(" Kolom terjemahan Indonesia ditambahkan.")
|
|
sample_pil = self.df[["pillar_name", "pillar_name_id"]].drop_duplicates()
|
|
self.logger.info(" Pillar mapping:")
|
|
for _, r in sample_pil.iterrows():
|
|
self.logger.info(f" {r['pillar_name']:<20} -> {r['pillar_name_id']}")
|
|
|
|
# =========================================================================
|
|
# STEP 4: Deteksi sdgs_start_year
|
|
# =========================================================================
|
|
|
|
def _detect_sdgs_start_year(self) -> int:
|
|
fies_rows = self.df[
|
|
self.df["indicator_name"].str.lower().str.strip().isin(_FIES_DETECTION_LOWER)
|
|
]
|
|
if not fies_rows.empty:
|
|
sdgs_start = int(fies_rows["year"].min())
|
|
self.logger.info(f" [Metode 1 - FIES explicit] sdgs_start_year = {sdgs_start}")
|
|
return sdgs_start
|
|
|
|
ind_min_year = (
|
|
self.df.groupby("indicator_id")["year"]
|
|
.min().reset_index()
|
|
.rename(columns={"year": "min_year"})
|
|
)
|
|
unique_years = sorted(ind_min_year["min_year"].unique())
|
|
|
|
if len(unique_years) == 1:
|
|
sdgs_start = int(unique_years[0]) + 9999
|
|
else:
|
|
gaps = [
|
|
(unique_years[i+1] - unique_years[i], unique_years[i], unique_years[i+1])
|
|
for i in range(len(unique_years) - 1)
|
|
]
|
|
gaps.sort(reverse=True)
|
|
_, y_before, y_after = gaps[0]
|
|
sdgs_start = int(y_after)
|
|
self.logger.info(f" Gap terbesar: {y_before} -> {y_after} -> sdgs_start_year = {sdgs_start}")
|
|
|
|
return sdgs_start
|
|
|
|
# =========================================================================
|
|
# STEP 5: Assign framework
|
|
# =========================================================================
|
|
|
|
def _assign_framework(self):
|
|
df = self.df.copy()
|
|
df["_is_sdg_kw"] = df["indicator_name"].str.lower().str.strip().isin(_SDG_ONLY_LOWER)
|
|
df["framework"] = "MDGs"
|
|
|
|
mask_sdgs = df["_is_sdg_kw"] & (df["year"] >= self.sdgs_start_year)
|
|
df.loc[mask_sdgs, "framework"] = "SDGs"
|
|
df = df.drop(columns=["_is_sdg_kw"])
|
|
self.df = df
|
|
|
|
fw_dist = self.df["framework"].value_counts()
|
|
self.logger.info("\n Framework distribution (rows):")
|
|
for fw, cnt in fw_dist.items():
|
|
self.logger.info(f" {fw:<6}: {cnt:,} rows")
|
|
|
|
# =========================================================================
|
|
# STEP 6: Hitung norm_value per indikator
|
|
# =========================================================================
|
|
|
|
def _compute_norm_values(self) -> pd.DataFrame:
|
|
df = self.df.copy()
|
|
norm_parts = []
|
|
|
|
for ind_id, grp in df.groupby("indicator_id"):
|
|
grp = grp.copy()
|
|
direction = str(grp["direction"].iloc[0])
|
|
do_invert = _should_invert(
|
|
direction, self.logger, context=f"indicator_id={ind_id}"
|
|
)
|
|
|
|
valid_mask = grp["value"].notna()
|
|
n_valid = valid_mask.sum()
|
|
|
|
if n_valid < 2:
|
|
grp["norm_value"] = np.nan
|
|
norm_parts.append(grp)
|
|
continue
|
|
|
|
raw = grp.loc[valid_mask, "value"].values
|
|
v_min = raw.min()
|
|
v_max = raw.max()
|
|
|
|
normed = np.full(len(grp), np.nan)
|
|
if v_min == v_max:
|
|
normed[valid_mask.values] = 0.5
|
|
else:
|
|
normed[valid_mask.values] = (raw - v_min) / (v_max - v_min)
|
|
|
|
if do_invert:
|
|
normed = np.where(np.isnan(normed), np.nan, 1.0 - normed)
|
|
|
|
grp["norm_value"] = normed
|
|
norm_parts.append(grp)
|
|
|
|
return pd.concat(norm_parts, ignore_index=True)
|
|
|
|
# =========================================================================
|
|
# STEP 6b: Tambah baris ASEAN (rata-rata dari semua negara per ind per year)
|
|
# =========================================================================
|
|
|
|
def _add_asean_rows(self, df_normed: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
Buat baris ASEAN aggregate: nilai = rata-rata nilai semua negara asli.
|
|
norm_value = rata-rata norm_value semua negara asli.
|
|
"""
|
|
real_rows = df_normed[df_normed["country_id"] != ASEAN_COUNTRY_ID]
|
|
|
|
dim_cols = [
|
|
"indicator_id", "indicator_name", "indicator_name_id",
|
|
"unit", "direction",
|
|
"pillar_id", "pillar_name", "pillar_name_id",
|
|
"framework",
|
|
]
|
|
|
|
asean_agg = (
|
|
real_rows.groupby(["indicator_id", "year"])
|
|
.agg(
|
|
value =("value", "mean"),
|
|
norm_value=("norm_value", "mean"),
|
|
)
|
|
.reset_index()
|
|
)
|
|
|
|
# Gabung kolom dimensi dari baris pertama per indicator_id
|
|
dim_ref = (
|
|
real_rows[dim_cols]
|
|
.drop_duplicates(subset=["indicator_id"])
|
|
.copy()
|
|
)
|
|
asean_agg = asean_agg.merge(dim_ref, on="indicator_id", how="left")
|
|
|
|
asean_agg["country_id"] = ASEAN_COUNTRY_ID
|
|
asean_agg["country_name"] = ASEAN_COUNTRY_NAME
|
|
asean_agg["country_name_id"] = ASEAN_COUNTRY_NAME_ID
|
|
|
|
# Hanya kolom yang ada di df_normed
|
|
cols_needed = [c for c in df_normed.columns if c in asean_agg.columns]
|
|
for c in df_normed.columns:
|
|
if c not in asean_agg.columns:
|
|
asean_agg[c] = np.nan
|
|
|
|
return pd.concat([df_normed, asean_agg[df_normed.columns]], ignore_index=True)
|
|
|
|
# =========================================================================
|
|
# STEP 7: Hitung YoY
|
|
# =========================================================================
|
|
|
|
def _compute_yoy_columns(self, df: pd.DataFrame) -> pd.DataFrame:
|
|
parts = []
|
|
groups = df.groupby(["indicator_id", "country_id"], sort=False)
|
|
|
|
for (ind_id, country_id), grp in groups:
|
|
parts.append(_compute_yoy(grp))
|
|
|
|
return pd.concat(parts, ignore_index=True)
|
|
|
|
# =========================================================================
|
|
# STEP 8: Scale ke 1-100
|
|
# =========================================================================
|
|
|
|
def _compute_scores(self, df: pd.DataFrame) -> pd.DataFrame:
|
|
score_parts = []
|
|
for ind_id, grp in df.groupby("indicator_id"):
|
|
grp = grp.copy()
|
|
grp["norm_score_1_100"] = global_minmax(grp["norm_value"])
|
|
score_parts.append(grp)
|
|
return pd.concat(score_parts, ignore_index=True)
|
|
|
|
# =========================================================================
|
|
# STEP 9: Assign performance label
|
|
# =========================================================================
|
|
|
|
def _assign_performance(self, df: pd.DataFrame) -> pd.DataFrame:
|
|
df = df.copy()
|
|
df["performance"] = pd.NA
|
|
|
|
has_score = df["norm_score_1_100"].notna()
|
|
df.loc[has_score & (df["norm_score_1_100"] >= _PERFORMANCE_THRESHOLD), "performance"] = "Good"
|
|
df.loc[has_score & (df["norm_score_1_100"] < _PERFORMANCE_THRESHOLD), "performance"] = "Bad"
|
|
return df
|
|
|
|
# =========================================================================
|
|
# STEP 10: Save agg_indicator_norm (termasuk ASEAN rows)
|
|
# =========================================================================
|
|
|
|
def _save(self, df: pd.DataFrame) -> int:
|
|
table_name = "agg_indicator_norm"
|
|
|
|
out = df[[
|
|
"year", "country_id", "country_name", "country_name_id",
|
|
"indicator_id", "indicator_name", "indicator_name_id",
|
|
"unit", "direction",
|
|
"pillar_id", "pillar_name", "pillar_name_id",
|
|
"framework",
|
|
"value", "norm_value", "norm_score_1_100",
|
|
"yoy_value", "yoy_norm_value", "performance",
|
|
]].copy()
|
|
|
|
out = out.sort_values(
|
|
["year", "country_name", "pillar_name", "indicator_name"]
|
|
).reset_index(drop=True)
|
|
|
|
out["year"] = out["year"].astype(int)
|
|
out["country_id"] = out["country_id"].astype(int)
|
|
out["country_name"] = out["country_name"].astype(str)
|
|
out["country_name_id"] = out["country_name_id"].astype(str)
|
|
out["indicator_id"] = out["indicator_id"].astype(int)
|
|
out["indicator_name"] = out["indicator_name"].astype(str)
|
|
out["indicator_name_id"] = out["indicator_name_id"].astype(str)
|
|
out["unit"] = out["unit"].astype(str)
|
|
out["direction"] = out["direction"].astype(str)
|
|
out["pillar_id"] = out["pillar_id"].astype(int)
|
|
out["pillar_name"] = out["pillar_name"].astype(str)
|
|
out["pillar_name_id"] = out["pillar_name_id"].astype(str)
|
|
out["framework"] = out["framework"].astype(str)
|
|
out["value"] = out["value"].astype(float)
|
|
out["norm_value"] = out["norm_value"].astype(float)
|
|
out["norm_score_1_100"] = out["norm_score_1_100"].astype(float)
|
|
out["yoy_value"] = pd.to_numeric(out["yoy_value"], errors="coerce").astype(float)
|
|
out["yoy_norm_value"] = pd.to_numeric(out["yoy_norm_value"], errors="coerce").astype(float)
|
|
out["performance"] = out["performance"].astype(str).replace("nan", pd.NA).astype("string")
|
|
|
|
n_asean = (out["country_id"] == ASEAN_COUNTRY_ID).sum()
|
|
n_country = (out["country_id"] != ASEAN_COUNTRY_ID).sum()
|
|
self.logger.info(f" Total rows : {len(out):,} ({n_country:,} country + {n_asean:,} ASEAN)")
|
|
|
|
schema = [
|
|
bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("country_id", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("country_name", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("country_name_id", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("indicator_id", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("indicator_name", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("indicator_name_id", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("unit", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("direction", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_id", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_name_id", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("framework", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("value", "FLOAT", mode="REQUIRED"),
|
|
bigquery.SchemaField("norm_value", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("norm_score_1_100", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("yoy_value", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("yoy_norm_value", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("performance", "STRING", mode="NULLABLE"),
|
|
]
|
|
|
|
rows_loaded = load_to_bigquery(
|
|
self.client, out, table_name,
|
|
layer="gold", write_disposition="WRITE_TRUNCATE", schema=schema,
|
|
)
|
|
|
|
log_update(self.client, "DW", table_name, "full_load", rows_loaded)
|
|
self.logger.info(f" [OK] {table_name}: {rows_loaded:,} rows -> [Gold] fs_asean_gold")
|
|
|
|
metadata = {
|
|
"source_class" : self.__class__.__name__,
|
|
"table_name" : table_name,
|
|
"execution_timestamp": self.pipeline_start,
|
|
"duration_seconds" : (datetime.now() - self.pipeline_start).total_seconds(),
|
|
"rows_fetched" : self.pipeline_metadata["rows_fetched"],
|
|
"rows_transformed" : rows_loaded,
|
|
"rows_loaded" : rows_loaded,
|
|
"completeness_pct" : 100.0,
|
|
"config_snapshot" : json.dumps({
|
|
"sdgs_start_year" : self.sdgs_start_year,
|
|
"layer" : "gold",
|
|
"normalization" : "per_indicator_global_minmax",
|
|
"performance_threshold": _PERFORMANCE_THRESHOLD,
|
|
"asean_country_id" : ASEAN_COUNTRY_ID,
|
|
"architecture" : "ASEAN merged into country table (country_id=0)",
|
|
"pillar_change" : "Renamed to Food Other; all pillars use 'Food ' prefix",
|
|
}),
|
|
"validation_metrics" : json.dumps({
|
|
"total_rows" : rows_loaded,
|
|
"n_indicators" : int(out["indicator_id"].nunique()),
|
|
"n_countries" : int(out[out["country_id"] != ASEAN_COUNTRY_ID]["country_id"].nunique()),
|
|
"asean_rows" : int(n_asean),
|
|
}),
|
|
}
|
|
save_etl_metadata(self.client, metadata)
|
|
return rows_loaded
|
|
|
|
# =========================================================================
|
|
# STEP 11: agg_narrative_indicator (per indicator, ASEAN summary sebagai kolom)
|
|
# =========================================================================
|
|
|
|
def _build_narrative_table(self, df_final: pd.DataFrame):
|
|
self.logger.info("\n" + "=" * 80)
|
|
self.logger.info("STEP 11: agg_narrative_indicator")
|
|
self.logger.info(" Granularity: per indicator_id")
|
|
self.logger.info(" ASEAN data: digunakan untuk asean_avg_value_first/last")
|
|
self.logger.info("=" * 80)
|
|
|
|
# Negara asli saja untuk analisa statistik
|
|
df_real = df_final[df_final["country_id"] != ASEAN_COUNTRY_ID]
|
|
df_asean = df_final[df_final["country_id"] == ASEAN_COUNTRY_ID]
|
|
|
|
# ---- Statistik per indikator (negara asli) ----
|
|
df_yr = (
|
|
df_real.groupby(["indicator_id", "year"])
|
|
.agg(
|
|
avg_value =("value", "mean"),
|
|
avg_norm_score =("norm_score_1_100", "mean"),
|
|
n_countries_yr =("country_id", "nunique"),
|
|
)
|
|
.reset_index()
|
|
)
|
|
|
|
df_first = (
|
|
df_yr.sort_values("year").groupby("indicator_id").first().reset_index()
|
|
[["indicator_id", "year", "avg_value"]]
|
|
.rename(columns={"year": "year_min", "avg_value": "avg_value_first"})
|
|
)
|
|
df_last = (
|
|
df_yr.sort_values("year").groupby("indicator_id").last().reset_index()
|
|
[["indicator_id", "year", "avg_value"]]
|
|
.rename(columns={"year": "year_max", "avg_value": "avg_value_last"})
|
|
)
|
|
df_score_avg = (
|
|
df_yr.groupby("indicator_id")
|
|
.agg(avg_norm_score_1_100=("avg_norm_score", "mean"))
|
|
.reset_index()
|
|
)
|
|
df_nc = (
|
|
df_real.groupby("indicator_id")["country_id"]
|
|
.nunique().reset_index()
|
|
.rename(columns={"country_id": "n_countries"})
|
|
)
|
|
|
|
# ASEAN avg per indikator
|
|
df_asean_yr = (
|
|
df_asean.groupby(["indicator_id", "year"])
|
|
.agg(asean_avg_value=("value", "mean"))
|
|
.reset_index()
|
|
)
|
|
df_asean_first = (
|
|
df_asean_yr.sort_values("year").groupby("indicator_id").first().reset_index()
|
|
[["indicator_id", "asean_avg_value"]]
|
|
.rename(columns={"asean_avg_value": "asean_avg_value_first"})
|
|
)
|
|
df_asean_last = (
|
|
df_asean_yr.sort_values("year").groupby("indicator_id").last().reset_index()
|
|
[["indicator_id", "asean_avg_value"]]
|
|
.rename(columns={"asean_avg_value": "asean_avg_value_last"})
|
|
)
|
|
|
|
# YoY stats (negara asli)
|
|
dir_map = (
|
|
df_real[["indicator_id", "direction"]]
|
|
.drop_duplicates(subset=["indicator_id"])
|
|
.set_index("indicator_id")["direction"]
|
|
.to_dict()
|
|
)
|
|
|
|
yoy_parts = []
|
|
for ind_id, grp in df_yr.groupby("indicator_id"):
|
|
grp = grp.sort_values("year").copy()
|
|
grp["prev_avg"] = grp["avg_value"].shift(1)
|
|
grp["yoy"] = np.where(
|
|
grp["avg_value"].notna() & grp["prev_avg"].notna(),
|
|
grp["avg_value"] - grp["prev_avg"],
|
|
np.nan,
|
|
)
|
|
grp = grp.drop(columns=["prev_avg"])
|
|
yoy_parts.append(grp)
|
|
df_yr = pd.concat(yoy_parts, ignore_index=True)
|
|
|
|
def _is_positive_yoy(ind_id, yoy_val):
|
|
if pd.isna(yoy_val):
|
|
return False
|
|
lb = _is_lower_better(dir_map.get(ind_id, "positive"))
|
|
return (yoy_val < 0) if lb else (yoy_val > 0)
|
|
|
|
yoy_stats = []
|
|
for ind_id, grp in df_yr.groupby("indicator_id"):
|
|
grp_yoy = grp[grp["yoy"].notna()].copy()
|
|
lb = _is_lower_better(dir_map.get(ind_id, "positive"))
|
|
n_total = len(grp_yoy)
|
|
n_positive = int(sum(_is_positive_yoy(ind_id, v) for v in grp_yoy["yoy"]))
|
|
|
|
if n_total > 0:
|
|
idx_best = grp_yoy["yoy"].idxmin() if lb else grp_yoy["yoy"].idxmax()
|
|
best_row = grp_yoy.loc[idx_best]
|
|
best_yoy_from = best_row["year"] - 1
|
|
best_yoy_to = best_row["year"]
|
|
else:
|
|
best_yoy_from = np.nan
|
|
best_yoy_to = np.nan
|
|
|
|
yoy_stats.append({
|
|
"indicator_id" : ind_id,
|
|
"n_yoy_total" : n_total,
|
|
"n_yoy_positive": n_positive,
|
|
"best_yoy_from" : best_yoy_from,
|
|
"best_yoy_to" : best_yoy_to,
|
|
})
|
|
df_yoy_stats = pd.DataFrame(yoy_stats)
|
|
|
|
# Country best/worst
|
|
df_country_avg = (
|
|
df_real.groupby(["indicator_id", "country_id", "country_name"])
|
|
.agg(country_avg_value=("value", "mean"))
|
|
.reset_index()
|
|
)
|
|
country_stats = []
|
|
for ind_id, grp in df_country_avg.groupby("indicator_id"):
|
|
lb = _is_lower_better(dir_map.get(ind_id, "positive"))
|
|
if lb:
|
|
worst_row = grp.loc[grp["country_avg_value"].idxmax()]
|
|
best_row = grp.loc[grp["country_avg_value"].idxmin()]
|
|
else:
|
|
worst_row = grp.loc[grp["country_avg_value"].idxmin()]
|
|
best_row = grp.loc[grp["country_avg_value"].idxmax()]
|
|
country_stats.append({
|
|
"indicator_id" : ind_id,
|
|
"country_worst" : worst_row["country_name"],
|
|
"country_best" : best_row["country_name"],
|
|
"country_worst_id": get_country_name_id(worst_row["country_name"]),
|
|
"country_best_id" : get_country_name_id(best_row["country_name"]),
|
|
})
|
|
df_country_stats = pd.DataFrame(country_stats)
|
|
|
|
# Dim cols
|
|
dim_cols = [
|
|
"indicator_name", "indicator_name_id",
|
|
"unit", "direction",
|
|
"pillar_name", "pillar_name_id",
|
|
"framework",
|
|
]
|
|
df_dim = df_real[["indicator_id"] + dim_cols].drop_duplicates(subset=["indicator_id"])
|
|
|
|
# Merge semua
|
|
df_agg = (
|
|
df_dim
|
|
.merge(df_first, on="indicator_id", how="left")
|
|
.merge(df_last, on="indicator_id", how="left")
|
|
.merge(df_score_avg, on="indicator_id", how="left")
|
|
.merge(df_nc, on="indicator_id", how="left")
|
|
.merge(df_yoy_stats, on="indicator_id", how="left")
|
|
.merge(df_country_stats, on="indicator_id", how="left")
|
|
.merge(df_asean_first, on="indicator_id", how="left")
|
|
.merge(df_asean_last, on="indicator_id", how="left")
|
|
)
|
|
|
|
# Performance
|
|
df_agg["performance"] = pd.NA
|
|
has_score = df_agg["avg_norm_score_1_100"].notna()
|
|
df_agg.loc[has_score & (df_agg["avg_norm_score_1_100"] >= _PERFORMANCE_THRESHOLD), "performance"] = "Good"
|
|
df_agg.loc[has_score & (df_agg["avg_norm_score_1_100"] < _PERFORMANCE_THRESHOLD), "performance"] = "Bad"
|
|
|
|
# Build narrative
|
|
narratives_en = []
|
|
narratives_id = []
|
|
for _, row in df_agg.iterrows():
|
|
n_en, n_id = _build_narrative_per_indicator(row, df_final)
|
|
narratives_en.append(n_en)
|
|
narratives_id.append(n_id)
|
|
|
|
df_agg["narrative_en"] = narratives_en
|
|
df_agg["narrative_id"] = narratives_id
|
|
|
|
# Output
|
|
out = df_agg[[
|
|
"indicator_id", "indicator_name", "indicator_name_id",
|
|
"unit", "direction",
|
|
"pillar_name", "pillar_name_id",
|
|
"framework",
|
|
"year_min", "year_max", "n_countries",
|
|
"avg_value_first", "avg_value_last",
|
|
"asean_avg_value_first", "asean_avg_value_last",
|
|
"avg_norm_score_1_100", "performance",
|
|
"n_yoy_total", "n_yoy_positive",
|
|
"best_yoy_from", "best_yoy_to",
|
|
"country_worst", "country_best",
|
|
"country_worst_id", "country_best_id",
|
|
"narrative_en", "narrative_id",
|
|
]].copy()
|
|
|
|
out = out.sort_values(["pillar_name", "indicator_name"]).reset_index(drop=True)
|
|
|
|
out["indicator_id"] = out["indicator_id"].astype(int)
|
|
out["indicator_name"] = out["indicator_name"].astype(str)
|
|
out["indicator_name_id"] = out["indicator_name_id"].astype(str)
|
|
out["unit"] = out["unit"].fillna("").astype(str)
|
|
out["direction"] = out["direction"].astype(str)
|
|
out["pillar_name"] = out["pillar_name"].astype(str)
|
|
out["pillar_name_id"] = out["pillar_name_id"].astype(str)
|
|
out["framework"] = out["framework"].astype(str)
|
|
out["year_min"] = out["year_min"].astype(int)
|
|
out["year_max"] = out["year_max"].astype(int)
|
|
out["n_countries"] = out["n_countries"].astype(int)
|
|
out["avg_value_first"] = pd.to_numeric(out["avg_value_first"], errors="coerce").astype(float)
|
|
out["avg_value_last"] = pd.to_numeric(out["avg_value_last"], errors="coerce").astype(float)
|
|
out["asean_avg_value_first"]= pd.to_numeric(out["asean_avg_value_first"], errors="coerce").astype(float)
|
|
out["asean_avg_value_last"] = pd.to_numeric(out["asean_avg_value_last"], errors="coerce").astype(float)
|
|
out["avg_norm_score_1_100"] = pd.to_numeric(out["avg_norm_score_1_100"], errors="coerce").astype(float)
|
|
out["performance"] = out["performance"].astype(str).replace("nan", pd.NA).astype("string")
|
|
out["n_yoy_total"] = pd.to_numeric(out["n_yoy_total"], errors="coerce").astype("Int64")
|
|
out["n_yoy_positive"] = pd.to_numeric(out["n_yoy_positive"], errors="coerce").astype("Int64")
|
|
out["best_yoy_from"] = pd.to_numeric(out["best_yoy_from"], errors="coerce").astype("Int64")
|
|
out["best_yoy_to"] = pd.to_numeric(out["best_yoy_to"], errors="coerce").astype("Int64")
|
|
out["country_worst"] = out["country_worst"].astype(str).replace("nan", pd.NA).astype("string")
|
|
out["country_best"] = out["country_best"].astype(str).replace("nan", pd.NA).astype("string")
|
|
out["country_worst_id"] = out["country_worst_id"].astype(str).replace("nan", pd.NA).astype("string")
|
|
out["country_best_id"] = out["country_best_id"].astype(str).replace("nan", pd.NA).astype("string")
|
|
out["narrative_en"] = out["narrative_en"].astype(str)
|
|
out["narrative_id"] = out["narrative_id"].astype(str)
|
|
|
|
schema = [
|
|
bigquery.SchemaField("indicator_id", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("indicator_name", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("indicator_name_id", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("unit", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("direction", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_name_id", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("framework", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("year_min", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("year_max", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("n_countries", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("avg_value_first", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("avg_value_last", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("asean_avg_value_first", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("asean_avg_value_last", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("avg_norm_score_1_100", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("performance", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("n_yoy_total", "INTEGER", mode="NULLABLE"),
|
|
bigquery.SchemaField("n_yoy_positive", "INTEGER", mode="NULLABLE"),
|
|
bigquery.SchemaField("best_yoy_from", "INTEGER", mode="NULLABLE"),
|
|
bigquery.SchemaField("best_yoy_to", "INTEGER", mode="NULLABLE"),
|
|
bigquery.SchemaField("country_worst", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("country_best", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("country_worst_id", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("country_best_id", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("narrative_en", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("narrative_id", "STRING", mode="NULLABLE"),
|
|
]
|
|
|
|
rows_loaded = load_to_bigquery(
|
|
self.client, out, "agg_narrative_indicator",
|
|
layer="gold", write_disposition="WRITE_TRUNCATE", schema=schema,
|
|
)
|
|
|
|
log_update(self.client, "DW", "agg_narrative_indicator", "full_load", rows_loaded)
|
|
self.logger.info(
|
|
f" [OK] agg_narrative_indicator: {rows_loaded:,} rows -> [Gold] fs_asean_gold"
|
|
)
|
|
|
|
metadata = {
|
|
"source_class" : self.__class__.__name__,
|
|
"table_name" : "agg_narrative_indicator",
|
|
"execution_timestamp": self.pipeline_start,
|
|
"duration_seconds" : (datetime.now() - self.pipeline_start).total_seconds(),
|
|
"rows_fetched" : self.pipeline_metadata["rows_fetched"],
|
|
"rows_transformed" : rows_loaded,
|
|
"rows_loaded" : rows_loaded,
|
|
"completeness_pct" : 100.0,
|
|
"config_snapshot" : json.dumps({
|
|
"granularity" : "indicator_id only",
|
|
"narrative_style" : "interpretive, plain text, bilingual EN/ID",
|
|
"asean_columns" : ["asean_avg_value_first", "asean_avg_value_last"],
|
|
"architecture" : "ASEAN rows included in agg_indicator_norm",
|
|
"pillar_change" : "Renamed to Food Other; all pillars use 'Food ' prefix",
|
|
}),
|
|
"validation_metrics" : json.dumps({
|
|
"total_rows" : rows_loaded,
|
|
"n_indicators": int(out["indicator_id"].nunique()),
|
|
}),
|
|
}
|
|
save_etl_metadata(self.client, metadata)
|
|
self.pipeline_metadata["rows_loaded_narrative"] = rows_loaded
|
|
|
|
# =========================================================================
|
|
# RUN
|
|
# =========================================================================
|
|
|
|
def run(self):
|
|
self.pipeline_start = datetime.now()
|
|
self.pipeline_metadata["start_time"] = self.pipeline_start
|
|
|
|
self.logger.info("\n" + "=" * 80)
|
|
self.logger.info("INDICATOR NORM AGGREGATION")
|
|
self.logger.info(" ASEAN rows ditambahkan ke agg_indicator_norm (country_id=0)")
|
|
self.logger.info(" Rename Food Other; all pillars use Food prefix")
|
|
self.logger.info("=" * 80)
|
|
|
|
self.load_data()
|
|
self.load_units()
|
|
self._merge_unit()
|
|
self._add_indonesia_name_columns()
|
|
self.sdgs_start_year = self._detect_sdgs_start_year()
|
|
self._assign_framework()
|
|
df_normed = self._compute_norm_values()
|
|
df_with_asean = self._add_asean_rows(df_normed) # <-- ASEAN ditambahkan di sini
|
|
df_yoy = self._compute_yoy_columns(df_with_asean)
|
|
df_scored = self._compute_scores(df_yoy)
|
|
df_final = self._assign_performance(df_scored)
|
|
rows_loaded = self._save(df_final)
|
|
self.pipeline_metadata["rows_loaded"] = rows_loaded
|
|
self._build_narrative_table(df_final)
|
|
|
|
self.pipeline_metadata["end_time"] = datetime.now()
|
|
duration = (
|
|
self.pipeline_metadata["end_time"] - self.pipeline_start
|
|
).total_seconds()
|
|
|
|
self.logger.info("\n" + "=" * 80)
|
|
self.logger.info("COMPLETED")
|
|
self.logger.info("=" * 80)
|
|
self.logger.info(f" Duration : {duration:.2f}s")
|
|
self.logger.info(f" Rows Fetched : {self.pipeline_metadata['rows_fetched']:,}")
|
|
self.logger.info(f" Rows Loaded (norm) : {rows_loaded:,}")
|
|
self.logger.info(f" Rows Loaded (narrative) : {self.pipeline_metadata['rows_loaded_narrative']:,}")
|
|
self.logger.info(f" sdgs_start_year : {self.sdgs_start_year}")
|
|
|
|
|
|
# =============================================================================
|
|
# AIRFLOW TASK
|
|
# =============================================================================
|
|
|
|
def run_indicator_norm_aggregation():
|
|
client = get_bigquery_client()
|
|
agg = IndicatorNormAggregator(client)
|
|
agg.run()
|
|
print(f"agg_indicator_norm loaded : {agg.pipeline_metadata['rows_loaded']:,} rows")
|
|
print(f"agg_narrative_indicator loaded: {agg.pipeline_metadata['rows_loaded_narrative']:,} rows")
|
|
|
|
|
|
# =============================================================================
|
|
# MAIN
|
|
# =============================================================================
|
|
|
|
if __name__ == "__main__":
|
|
import sys, io
|
|
if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"):
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
|
|
if sys.stderr.encoding and sys.stderr.encoding.lower() not in ("utf-8", "utf8"):
|
|
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
|
|
|
|
print("=" * 80)
|
|
print("INDICATOR NORM AGGREGATION -> fs_asean_gold")
|
|
print(f" ASEAN merged into country tables (country_id={ASEAN_COUNTRY_ID})")
|
|
print("=" * 80)
|
|
|
|
logger = setup_logging()
|
|
client = get_bigquery_client()
|
|
agg = IndicatorNormAggregator(client)
|
|
agg.run()
|
|
|
|
print("\n" + "=" * 80)
|
|
print("[OK] COMPLETED")
|
|
print("=" * 80) |