izu/airflow-coolify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bismillah capekk

Browse Source
This commit is contained in:
Debby
2026-04-03 08:40:30 +07:00
parent f652f2f730
commit 5313039b50
3 changed files with 361 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

538
scripts/bigquery_aggregate_layer.py
View File

@@ -5,11 +5,19 @@ UPDATED: Simpan 6 tabel ke fs_asean_gold (layer='gold'):
- agg_pillar_composite
- agg_pillar_by_country
- agg_framework_by_country
- agg_framework_asean
- agg_framework_asean (+ kolom performance_status: 'Good'/'Bad', threshold=60)
- agg_narrative_overview
- agg_narrative_pillar
SOURCE TABLE: fact_asean_food_security_selected (sudah include nama + ID)
n_indicators logic (sesuai agg_indicator_norm):
- Setiap tahun dihitung dari indikator yang benar-benar hadir di tahun tsb.
- Framework MDGs/SDGs per tahun mengikuti SDG_ONLY_KEYWORDS:
* Indikator tidak di SDG_ONLY -> selalu MDGs
* Indikator di SDG_ONLY + year >= sdgs_start_year -> SDGs
* Indikator di SDG_ONLY + year < sdgs_start_year -> MDGs
- Sehingga n_indicators MDGs dan SDGs bisa berbeda antar tahun.
"""
import pandas as pd
@@ -37,13 +45,49 @@ from google.cloud import bigquery
DIRECTION_INVERT_KEYWORDS = frozenset({
"negative", "lower_better", "lower_is_better", "inverse", "neg",
})
DIRECTION_POSITIVE_KEYWORDS = frozenset({
"positive", "higher_better", "higher_is_better",
})
NORMALIZE_FRAMEWORKS_JOINTLY = False
# Threshold performance_status di agg_framework_asean
PERFORMANCE_THRESHOLD = 60.0 # score >= 60 -> "Good", < 60 -> "Bad"
# SDG_ONLY_KEYWORDS (sama persis dengan bigquery_aggraget_fact_selected_layer.py)
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_LOWER: 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)",
])
# =============================================================================
# Windows CP1252 safe logging
@@ -133,19 +177,24 @@ def check_and_dedup(df: pd.DataFrame, key_cols: list, context: str = "", logger=
return df
def _performance_status(score) -> str:
"""Classify score into 'Good' or 'Bad' based on PERFORMANCE_THRESHOLD."""
if score is None or (isinstance(score, float) and np.isnan(score)):
return "N/A"
return "Good" if score >= PERFORMANCE_THRESHOLD else "Bad"
# =============================================================================
# NARRATIVE BUILDER FUNCTIONS (pure functions, easy to unit-test)
# NARRATIVE HELPERS
# =============================================================================
def _fmt_score(score) -> str:
"""Format score to 2 decimal places."""
if score is None or (isinstance(score, float) and np.isnan(score)):
return "N/A"
return f"{score:.2f}"
def _fmt_delta(delta) -> str:
"""Format YoY delta with sign and 2 decimal places."""
if delta is None or (isinstance(delta, float) and np.isnan(delta)):
return "N/A"
sign = "+" if delta >= 0 else ""
@@ -158,16 +207,19 @@ def _build_overview_narrative(
n_sdg: int,
n_total_ind: int,
score: float,
performance_status: str,
yoy_val,
yoy_pct,
prev_year: int,
prev_score,
prev_performance_status: str,
ranking_list: list,
most_improved_country,
most_improved_delta,
most_declined_country,
most_declined_delta,
) -> str:
# Sentence 1: indicator breakdown
parts_ind = []
if n_mdg > 0:
parts_ind.append(f"{n_mdg} MDG indicator{'s' if n_mdg > 1 else ''}")
@@ -187,24 +239,39 @@ def _build_overview_narrative(
f"{n_total_ind} indicator{'s' if n_total_ind != 1 else ''}."
)
# Sentence 2: score + performance status + YoY
status_phrase = (
f"classified as \"{performance_status}\" performance "
f"(threshold: {PERFORMANCE_THRESHOLD:.0f})"
)
if yoy_val is not None and prev_score is not None:
direction_word = "increasing" if yoy_val >= 0 else "decreasing"
pct_clause = ""
pct_clause = ""
if yoy_pct is not None:
abs_pct = abs(yoy_pct)
trend_word = "improvement" if yoy_val >= 0 else "decline"
pct_clause = f", which represents a {abs_pct:.2f}% {trend_word} year-over-year"
pct_clause = f", representing a {abs_pct:.2f}% {trend_word} year-over-year"
# Note if performance status changed
status_change = ""
if prev_performance_status not in ("N/A", None) and prev_performance_status != performance_status:
status_change = (
f" This marks a shift from \"{prev_performance_status}\" in {prev_year} "
f"to \"{performance_status}\" in {year}."
)
sent2 = (
f"The ASEAN overall score (Total framework) reached {_fmt_score(score)}, "
f"{direction_word} by {abs(yoy_val):.2f} points compared to the previous year "
f"({_fmt_score(prev_score)} in {prev_year}){pct_clause}."
f"{status_phrase}, {direction_word} by {abs(yoy_val):.2f} points compared to "
f"{prev_year} ({_fmt_score(prev_score)}, \"{prev_performance_status}\"){pct_clause}.{status_change}"
)
else:
sent2 = (
f"The ASEAN overall score (Total framework) reached {_fmt_score(score)} in {year}; "
f"no prior-year data is available for year-over-year comparison."
f"The ASEAN overall score (Total framework) reached {_fmt_score(score)} in {year}, "
f"{status_phrase}. No prior-year data is available for year-over-year comparison."
)
# Sentence 3: country ranking
sent3 = ""
if ranking_list:
first = ranking_list[0]
@@ -225,14 +292,12 @@ def _build_overview_narrative(
)
else:
middle_parts = [
f"{c['country_name']} ({_fmt_score(c['score'])})"
for c in middle
f"{c['country_name']} ({_fmt_score(c['score'])})" for c in middle
]
if len(middle_parts) == 1:
middle_str = middle_parts[0]
else:
middle_str = ", ".join(middle_parts[:-1]) + f", and {middle_parts[-1]}"
sent3 = (
f"In terms of country performance, {first['country_name']} led the region "
f"with a score of {_fmt_score(first['score'])}, followed by {middle_str}. "
@@ -240,6 +305,7 @@ def _build_overview_narrative(
f"of {_fmt_score(last['score'])} in {year}."
)
# Sentence 4: most improved / declined country
sent4_parts = []
if most_improved_country and most_improved_delta is not None:
sent4_parts.append(
@@ -339,9 +405,9 @@ def _build_pillar_narrative(
f"for the {pillar_name} pillar in {year}"
)
if most_improved_pillar and most_improved_delta is not None \
and most_declined_pillar and most_declined_delta is not None \
and most_improved_pillar != most_declined_pillar:
if (most_improved_pillar and most_improved_delta is not None
and most_declined_pillar and most_declined_delta is not None
and most_improved_pillar != most_declined_pillar):
sent4 += (
f". Across all pillars, {most_improved_pillar} showed the greatest improvement "
f"({_fmt_delta(most_improved_delta)} pts), while {most_declined_pillar} "
@@ -374,15 +440,15 @@ class FoodSecurityAggregator:
"agg_narrative_pillar": {"rows_loaded": 0, "status": "pending", "start_time": None, "end_time": None},
}
self.df = None
self.dims = {}
self.df = None
self.sdgs_start_year = None
self.sdgs_start_year = None
self.mdgs_indicator_ids = set()
self.sdgs_indicator_ids = set()
# Lookup: (indicator_id, year) -> framework label
# Dibangun di _assign_framework_labels(), dipakai di _count_framework_indicators()
self._ind_year_framework: pd.DataFrame = None
# =========================================================================
# STEP 1: Load data dari Gold layer
# STEP 1: Load data
# =========================================================================
def load_data(self):
@@ -390,36 +456,23 @@ class FoodSecurityAggregator:
self.logger.info("STEP 1: LOAD DATA from fs_asean_gold")
self.logger.info("=" * 70)
# -----------------------------------------------------------------------
# CHANGED: sumber tabel -> fact_asean_food_security_selected
# Tabel ini sudah include: country_name, indicator_name, pillar_name,
# direction, year -> tidak perlu join ke dim_* lagi
# -----------------------------------------------------------------------
self.df = read_from_bigquery(
self.client, "fact_asean_food_security_selected", layer='gold'
)
self.logger.info(f" fact_asean_food_security_selected : {len(self.df):,} rows")
# Validasi kolom wajib yang harus sudah ada di tabel baru
required_cols = {
"country_id", "country_name",
"indicator_id", "indicator_name", "direction",
"pillar_id", "pillar_name",
"time_id", "year",
"value",
"time_id", "year", "value",
}
missing_cols = required_cols - set(self.df.columns)
if missing_cols:
raise ValueError(
f"Kolom berikut tidak ditemukan di fact_asean_food_security_selected: "
f"{missing_cols}"
f"Kolom berikut tidak ditemukan di fact_asean_food_security_selected: {missing_cols}"
)
# -----------------------------------------------------------------------
# Tidak perlu join ke dim_* lagi karena semua nama sudah ada.
# Hanya load dim_indicator untuk keperluan fallback / referensi direction
# jika ada NULL yang perlu di-fill.
# -----------------------------------------------------------------------
n_null_dir = self.df["direction"].isna().sum()
if n_null_dir > 0:
self.logger.warning(
@@ -441,61 +494,105 @@ class FoodSecurityAggregator:
)
# =========================================================================
# STEP 1b: Klasifikasi indikator ke MDGs / SDGs
# STEP 1b: Detect sdgs_start_year + assign framework per (indicator, year)
# Konsisten dengan logika di bigquery_aggraget_fact_selected_layer.py
# =========================================================================
def _classify_indicators(self):
self.logger.info("\n" + "=" * 70)
self.logger.info("STEP 1b: KLASIFIKASI INDIKATOR -> MDGs / SDGs")
self.logger.info("=" * 70)
def _detect_sdgs_start_year(self) -> int:
"""Deteksi sdgs_start_year dari kehadiran FIES di data (metode eksplisit)."""
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" [FIES explicit] sdgs_start_year = {sdgs_start}")
return sdgs_start
# Fallback: gap terbesar pada distribusi min_year
ind_min_year = (
self.df.groupby("indicator_id")["year"]
.min().reset_index()
.rename(columns={"year": "min_year"})
.min().reset_index().rename(columns={"year": "min_year"})
)
unique_years = sorted(ind_min_year["min_year"].unique())
self.logger.info(f"\n Unique min_year per indikator: {unique_years}")
if len(unique_years) == 1:
gap_threshold = unique_years[0] + 1
self.logger.info(" Hanya 1 cluster -> semua = MDGs")
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)
largest_gap_size, y_before, y_after = gaps[0]
gap_threshold = y_after
self.logger.info(f" Gap terbesar: {y_before} -> {y_after} (selisih {largest_gap_size})")
self.logger.info(" [Fallback] Hanya 1 cluster -> semua MDGs")
return int(unique_years[0]) + 9999
ind_min_year["framework"] = ind_min_year["min_year"].apply(
lambda y: "MDGs" if int(y) < gap_threshold else "SDGs"
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]
self.logger.info(f" [Fallback gap] sdgs_start_year = {y_after} (gap {y_before}->{y_after})")
return int(y_after)
def _assign_framework_labels(self):
"""
Buat lookup table _ind_year_framework: DataFrame(indicator_id, year, framework).
Aturan (identik dengan IndicatorNormAggregator._assign_framework):
- Indikator TIDAK di SDG_ONLY_KEYWORDS -> selalu "MDGs"
- Indikator DI SDG_ONLY_KEYWORDS:
year < sdgs_start_year -> "MDGs"
year >= sdgs_start_year -> "SDGs"
Juga attach kolom 'framework' ke self.df untuk dipakai _get_norm_value_df().
"""
self.logger.info("\n" + "=" * 70)
self.logger.info("STEP 1b: ASSIGN FRAMEWORK LABELS (per indicator per year)")
self.logger.info(f" sdgs_start_year = {self.sdgs_start_year}")
self.logger.info("=" * 70)
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
# Build compact lookup (unique indicator_id x year x framework)
self._ind_year_framework = (
self.df[["indicator_id", "year", "framework"]]
.drop_duplicates()
.reset_index(drop=True)
)
sdgs_rows = ind_min_year[ind_min_year["framework"] == "SDGs"]
self.sdgs_start_year = (
int(sdgs_rows["min_year"].min()) if not sdgs_rows.empty
else int(self.df["year"].max()) + 1
)
# Log distribusi
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")
self.logger.info(f" sdgs_start_year: {self.sdgs_start_year}")
self.mdgs_indicator_ids = set(
ind_min_year[ind_min_year["framework"] == "MDGs"]["indicator_id"].tolist()
# n_indicators per framework per year
ind_fw_yr = (
self._ind_year_framework
.groupby(["year", "framework"])["indicator_id"]
.nunique()
.reset_index()
.rename(columns={"indicator_id": "n_indicators"})
.sort_values(["year", "framework"])
)
self.sdgs_indicator_ids = set(
ind_min_year[ind_min_year["framework"] == "SDGs"]["indicator_id"].tolist()
self.logger.info(
f"\n {'Year':<6} {'Framework':<8} {'n_indicators'}"
)
self.logger.info(" " + "-" * 30)
for _, r in ind_fw_yr.iterrows():
self.logger.info(
f" {int(r['year']):<6} {r['framework']:<8} {int(r['n_indicators'])}"
)
self.logger.info(f" MDGs: {len(self.mdgs_indicator_ids)} indicators")
self.logger.info(f" SDGs: {len(self.sdgs_indicator_ids)} indicators")
self.df = self.df.merge(
ind_min_year[["indicator_id", "framework"]], on="indicator_id", how="left"
def _count_framework_indicators(self, year: int, framework: str) -> int:
"""
Hitung jumlah indikator unik untuk framework tertentu di tahun tertentu.
Menggunakan _ind_year_framework yang dibangun di _assign_framework_labels().
"""
mask = (
(self._ind_year_framework["year"] == year) &
(self._ind_year_framework["framework"] == framework)
)
return int(self._ind_year_framework.loc[mask, "indicator_id"].nunique())
# =========================================================================
# CORE HELPER: normalisasi raw value per indikator
@@ -504,14 +601,14 @@ class FoodSecurityAggregator:
def _get_norm_value_df(self) -> pd.DataFrame:
if "framework" not in self.df.columns:
raise ValueError(
"Kolom 'framework' tidak ada. Pastikan _classify_indicators() dipanggil lebih dulu."
"Kolom 'framework' tidak ada. Pastikan _assign_framework_labels() dipanggil lebih dulu."
)
norm_parts = []
for ind_id, grp in self.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}")
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()
@@ -537,7 +634,7 @@ class FoodSecurityAggregator:
return pd.concat(norm_parts, ignore_index=True)
# =========================================================================
# STEP 2: agg_pillar_composite -> Gold
# STEP 2: agg_pillar_composite
# =========================================================================
def calc_pillar_composite(self) -> pd.DataFrame:
@@ -595,7 +692,7 @@ class FoodSecurityAggregator:
return df
# =========================================================================
# STEP 3: agg_pillar_by_country -> Gold
# STEP 3: agg_pillar_by_country
# =========================================================================
def calc_pillar_by_country(self) -> pd.DataFrame:
@@ -648,11 +745,10 @@ class FoodSecurityAggregator:
return df
# =========================================================================
# STEP 4: agg_framework_by_country -> Gold
# STEP 4: agg_framework_by_country
# =========================================================================
def _calc_country_composite_inmemory(self) -> pd.DataFrame:
"""Hitung country composite in-memory (tidak disimpan ke BQ)."""
df_normed = self._get_norm_value_df()
df = (
df_normed
@@ -689,19 +785,22 @@ class FoodSecurityAggregator:
df_normed = self._get_norm_value_df()
parts = []
# Layer TOTAL
# TOTAL
agg_total = (
country_composite[[
"country_id", "country_name", "year",
"score_1_100", "n_indicators", "composite_score"
]]
.copy()
.rename(columns={"score_1_100": "framework_score_1_100", "composite_score": "framework_norm"})
.rename(columns={
"score_1_100" : "framework_score_1_100",
"composite_score": "framework_norm",
})
)
agg_total["framework"] = "Total"
parts.append(agg_total)
# Layer MDGs — Era pre-SDGs = Total
# MDGs pre-SDGs
pre_sdgs_rows = country_composite[country_composite["year"] < self.sdgs_start_year].copy()
if not pre_sdgs_rows.empty:
mdgs_pre = (
@@ -710,22 +809,31 @@ class FoodSecurityAggregator:
"score_1_100", "n_indicators", "composite_score"
]]
.copy()
.rename(columns={"score_1_100": "framework_score_1_100", "composite_score": "framework_norm"})
.rename(columns={
"score_1_100" : "framework_score_1_100",
"composite_score": "framework_norm",
})
)
mdgs_pre["framework"] = "MDGs"
parts.append(mdgs_pre)
# Layer MDGs — Era mixed
if self.mdgs_indicator_ids:
# MDGs mixed (year >= sdgs_start_year, hanya indikator MDGs)
mdgs_indicator_ids = set(
self._ind_year_framework[self._ind_year_framework["framework"] == "MDGs"]["indicator_id"]
)
if mdgs_indicator_ids:
df_mdgs_mixed = df_normed[
(df_normed["indicator_id"].isin(self.mdgs_indicator_ids)) &
(df_normed["indicator_id"].isin(mdgs_indicator_ids)) &
(df_normed["year"] >= self.sdgs_start_year)
].copy()
if not df_mdgs_mixed.empty:
agg_mdgs_mixed = (
df_mdgs_mixed
.groupby(["country_id", "country_name", "year"])
.agg(framework_norm=("norm_value", "mean"), n_indicators=("indicator_id", "nunique"))
.agg(
framework_norm=("norm_value", "mean"),
n_indicators =("indicator_id", "nunique"),
)
.reset_index()
)
if not NORMALIZE_FRAMEWORKS_JOINTLY:
@@ -733,17 +841,23 @@ class FoodSecurityAggregator:
agg_mdgs_mixed["framework"] = "MDGs"
parts.append(agg_mdgs_mixed)
# Layer SDGs
if self.sdgs_indicator_ids:
# SDGs
sdgs_indicator_ids = set(
self._ind_year_framework[self._ind_year_framework["framework"] == "SDGs"]["indicator_id"]
)
if sdgs_indicator_ids:
df_sdgs = df_normed[
(df_normed["indicator_id"].isin(self.sdgs_indicator_ids)) &
(df_normed["indicator_id"].isin(sdgs_indicator_ids)) &
(df_normed["year"] >= self.sdgs_start_year)
].copy()
if not df_sdgs.empty:
agg_sdgs = (
df_sdgs
.groupby(["country_id", "country_name", "year"])
.agg(framework_norm=("norm_value", "mean"), n_indicators=("indicator_id", "nunique"))
.agg(
framework_norm=("norm_value", "mean"),
n_indicators =("indicator_id", "nunique"),
)
.reset_index()
)
if not NORMALIZE_FRAMEWORKS_JOINTLY:
@@ -794,7 +908,7 @@ class FoodSecurityAggregator:
return df
# =========================================================================
# STEP 5: agg_framework_asean -> Gold
# STEP 5: agg_framework_asean (+ performance_status)
# =========================================================================
def calc_framework_asean(self) -> pd.DataFrame:
@@ -802,95 +916,128 @@ class FoodSecurityAggregator:
self.load_metadata[table_name]["start_time"] = datetime.now()
self.logger.info("\n" + "=" * 70)
self.logger.info(f"STEP 5: {table_name} -> [Gold] fs_asean_gold")
self.logger.info(f" performance_status threshold: {PERFORMANCE_THRESHOLD}")
self.logger.info("=" * 70)
df_normed = self._get_norm_value_df()
country_composite = self._calc_country_composite_inmemory()
country_norm = (
df_normed.groupby(["country_id", "country_name", "year"])["norm_value"]
.mean().reset_index().rename(columns={"norm_value": "country_norm"})
df_normed
.groupby(["country_id", "country_name", "year"])["norm_value"]
.mean().reset_index()
.rename(columns={"norm_value": "country_norm"})
)
asean_overall = (
country_norm.groupby("year")
.agg(asean_norm=("country_norm", "mean"), std_norm=("country_norm", "std"),
n_countries=("country_norm", "count"))
.agg(
asean_norm =("country_norm", "mean"),
std_norm =("country_norm", "std"),
n_countries=("country_norm", "count"),
)
.reset_index()
)
asean_overall["asean_score_1_100"] = global_minmax(asean_overall["asean_norm"])
asean_comp = (
country_composite.groupby("year")["composite_score"]
.mean().reset_index().rename(columns={"composite_score": "asean_composite"})
.mean().reset_index()
.rename(columns={"composite_score": "asean_composite"})
)
asean_overall = asean_overall.merge(asean_comp, on="year", how="left")
parts = []
# Layer TOTAL
total_cols = asean_overall[["year", "asean_score_1_100", "asean_norm", "std_norm", "n_countries"]].copy()
total_cols = total_cols.rename(columns={
# ------------------------------------------------------------------
# Helper: hitung n_indicators per framework per year dari lookup
# ------------------------------------------------------------------
def _n_ind(year_val, framework_val):
return self._count_framework_indicators(year_val, framework_val)
# TOTAL
total_cols = asean_overall[[
"year", "asean_score_1_100", "asean_norm", "std_norm", "n_countries"
]].copy().rename(columns={
"asean_score_1_100": "framework_score_1_100",
"asean_norm": "framework_norm",
"n_countries": "n_countries_with_data",
"asean_norm" : "framework_norm",
"n_countries" : "n_countries_with_data",
})
n_ind_total = df_normed.groupby("year")["indicator_id"].nunique().reset_index().rename(columns={"indicator_id": "n_indicators"})
total_cols = total_cols.merge(n_ind_total, on="year", how="left")
# n_indicators Total = semua indikator yang hadir di tahun tsb
total_cols["n_indicators"] = total_cols["year"].apply(
lambda y: int(self._ind_year_framework[
self._ind_year_framework["year"] == y
]["indicator_id"].nunique())
)
total_cols["framework"] = "Total"
parts.append(total_cols)
# Layer MDGs — pre-SDGs = Total
# MDGs pre-SDGs
pre_sdgs = asean_overall[asean_overall["year"] < self.sdgs_start_year].copy()
if not pre_sdgs.empty:
mdgs_pre = pre_sdgs[["year", "asean_score_1_100", "asean_norm", "std_norm", "n_countries"]].copy()
mdgs_pre = mdgs_pre.rename(columns={
mdgs_pre = pre_sdgs[[
"year", "asean_score_1_100", "asean_norm", "std_norm", "n_countries"
]].copy().rename(columns={
"asean_score_1_100": "framework_score_1_100",
"asean_norm": "framework_norm",
"n_countries": "n_countries_with_data",
"asean_norm" : "framework_norm",
"n_countries" : "n_countries_with_data",
})
n_ind_pre = (
df_normed[df_normed["year"] < self.sdgs_start_year]
.groupby("year")["indicator_id"].nunique()
.reset_index().rename(columns={"indicator_id": "n_indicators"})
# Pre-SDGs era: semua indikator berlabel MDGs
mdgs_pre["n_indicators"] = mdgs_pre["year"].apply(
lambda y: _n_ind(y, "MDGs")
)
mdgs_pre = mdgs_pre.merge(n_ind_pre, on="year", how="left")
mdgs_pre["framework"] = "MDGs"
parts.append(mdgs_pre)
# Layer MDGs — mixed
if self.mdgs_indicator_ids:
# MDGs mixed
mdgs_indicator_ids = set(
self._ind_year_framework[self._ind_year_framework["framework"] == "MDGs"]["indicator_id"]
)
if mdgs_indicator_ids:
df_mdgs_mixed = df_normed[
(df_normed["indicator_id"].isin(self.mdgs_indicator_ids)) &
(df_normed["indicator_id"].isin(mdgs_indicator_ids)) &
(df_normed["year"] >= self.sdgs_start_year)
].copy()
if not df_mdgs_mixed.empty:
cn = df_mdgs_mixed.groupby(["country_id", "year"])["norm_value"].mean().reset_index().rename(columns={"norm_value": "country_norm"})
cn = (
df_mdgs_mixed
.groupby(["country_id", "year"])["norm_value"].mean()
.reset_index().rename(columns={"norm_value": "country_norm"})
)
asean_mdgs = cn.groupby("year").agg(
framework_norm=("country_norm", "mean"),
std_norm=("country_norm", "std"),
n_countries_with_data=("country_id", "count"),
framework_norm =("country_norm", "mean"),
std_norm =("country_norm", "std"),
n_countries_with_data=("country_id", "count"),
).reset_index()
n_ind_mdgs = df_mdgs_mixed.groupby("year")["indicator_id"].nunique().reset_index().rename(columns={"indicator_id": "n_indicators"})
asean_mdgs = asean_mdgs.merge(n_ind_mdgs, on="year", how="left")
asean_mdgs["n_indicators"] = asean_mdgs["year"].apply(
lambda y: _n_ind(y, "MDGs")
)
if not NORMALIZE_FRAMEWORKS_JOINTLY:
asean_mdgs["framework_score_1_100"] = global_minmax(asean_mdgs["framework_norm"])
asean_mdgs["framework"] = "MDGs"
parts.append(asean_mdgs)
# Layer SDGs
if self.sdgs_indicator_ids:
# SDGs
sdgs_indicator_ids = set(
self._ind_year_framework[self._ind_year_framework["framework"] == "SDGs"]["indicator_id"]
)
if sdgs_indicator_ids:
df_sdgs = df_normed[
(df_normed["indicator_id"].isin(self.sdgs_indicator_ids)) &
(df_normed["indicator_id"].isin(sdgs_indicator_ids)) &
(df_normed["year"] >= self.sdgs_start_year)
].copy()
if not df_sdgs.empty:
cn = df_sdgs.groupby(["country_id", "year"])["norm_value"].mean().reset_index().rename(columns={"norm_value": "country_norm"})
cn = (
df_sdgs
.groupby(["country_id", "year"])["norm_value"].mean()
.reset_index().rename(columns={"norm_value": "country_norm"})
)
asean_sdgs = cn.groupby("year").agg(
framework_norm=("country_norm", "mean"),
std_norm=("country_norm", "std"),
n_countries_with_data=("country_id", "count"),
framework_norm =("country_norm", "mean"),
std_norm =("country_norm", "std"),
n_countries_with_data=("country_id", "count"),
).reset_index()
n_ind_sdgs = df_sdgs.groupby("year")["indicator_id"].nunique().reset_index().rename(columns={"indicator_id": "n_indicators"})
asean_sdgs = asean_sdgs.merge(n_ind_sdgs, on="year", how="left")
asean_sdgs["n_indicators"] = asean_sdgs["year"].apply(
lambda y: _n_ind(y, "SDGs")
)
if not NORMALIZE_FRAMEWORKS_JOINTLY:
asean_sdgs["framework_score_1_100"] = global_minmax(asean_sdgs["framework_norm"])
asean_sdgs["framework"] = "SDGs"
@@ -906,14 +1053,30 @@ class FoodSecurityAggregator:
df = check_and_dedup(df, ["framework", "year"], context=table_name, logger=self.logger)
df = add_yoy(df, ["framework"], "framework_score_1_100")
# performance_status
df["performance_status"] = df["framework_score_1_100"].apply(_performance_status)
df["year"] = df["year"].astype(int)
df["n_indicators"] = safe_int(df["n_indicators"], col_name="n_indicators", logger=self.logger)
df["n_countries_with_data"] = safe_int(df["n_countries_with_data"], col_name="n_countries_with_data", logger=self.logger)
for col in ["framework_norm", "std_norm", "framework_score_1_100"]:
df[col] = df[col].astype(float)
df["performance_status"] = df["performance_status"].astype(str)
self._validate_mdgs_equals_total(df, level="asean")
# Log performance summary
self.logger.info(f"\n performance_status summary (threshold={PERFORMANCE_THRESHOLD}):")
for fw in df["framework"].unique():
sub = df[df["framework"] == fw].sort_values("year")
for _, r in sub.iterrows():
self.logger.info(
f" {fw:<8} {int(r['year'])}: "
f"score={r['framework_score_1_100']:.2f} "
f"n_ind={int(r['n_indicators'])} "
f"-> {r['performance_status']}"
)
schema = [
bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("framework", "STRING", mode="REQUIRED"),
@@ -923,6 +1086,7 @@ class FoodSecurityAggregator:
bigquery.SchemaField("std_norm", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("framework_score_1_100", "FLOAT", mode="REQUIRED"),
bigquery.SchemaField("year_over_year_change", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("performance_status", "STRING", mode="REQUIRED"),
]
rows = load_to_bigquery(
self.client, df, table_name, layer='gold',
@@ -932,7 +1096,7 @@ class FoodSecurityAggregator:
return df
# =========================================================================
# STEP 6: agg_narrative_overview -> Gold
# STEP 6: agg_narrative_overview
# =========================================================================
def calc_narrative_overview(
@@ -952,32 +1116,31 @@ class FoodSecurityAggregator:
.reset_index(drop=True)
)
score_by_year = dict(zip(
asean_total["year"].astype(int),
asean_total["framework_score_1_100"].astype(float),
))
score_by_year = dict(zip(asean_total["year"].astype(int), asean_total["framework_score_1_100"].astype(float)))
status_by_year = dict(zip(asean_total["year"].astype(int), asean_total["performance_status"].astype(str)))
country_total = (
df_framework_by_country[df_framework_by_country["framework"] == "Total"]
.copy()
)
ind_year = self.df.drop_duplicates(subset=["indicator_id", "year", "framework"])
country_total = df_framework_by_country[df_framework_by_country["framework"] == "Total"].copy()
records = []
for _, row in asean_total.iterrows():
yr = int(row["year"])
score = float(row["framework_score_1_100"])
yoy = row["year_over_year_change"]
yoy_val = float(yoy) if pd.notna(yoy) else None
yr = int(row["year"])
score = float(row["framework_score_1_100"])
perf_status = str(row["performance_status"])
yoy = row["year_over_year_change"]
yoy_val = float(yoy) if pd.notna(yoy) else None
yr_ind = ind_year[ind_year["year"] == yr]
n_mdg = int(yr_ind[yr_ind["framework"] == "MDGs"]["indicator_id"].nunique())
n_sdg = int(yr_ind[yr_ind["framework"] == "SDGs"]["indicator_id"].nunique())
n_total_ind = int(yr_ind["indicator_id"].nunique())
# n_indicators per framework per year (dari lookup)
n_mdg = self._count_framework_indicators(yr, "MDGs")
n_sdg = self._count_framework_indicators(yr, "SDGs")
n_total_ind = int(
self._ind_year_framework[
self._ind_year_framework["year"] == yr
]["indicator_id"].nunique()
)
prev_score = score_by_year.get(yr - 1, None)
prev_score = score_by_year.get(yr - 1, None)
prev_status = status_by_year.get(yr - 1, "N/A")
yoy_pct = (
(yoy_val / prev_score * 100)
@@ -1015,20 +1178,22 @@ class FoodSecurityAggregator:
most_improved_delta = most_declined_delta = None
narrative = _build_overview_narrative(
year = yr,
n_mdg = n_mdg,
n_sdg = n_sdg,
n_total_ind = n_total_ind,
score = score,
yoy_val = yoy_val,
yoy_pct = yoy_pct,
prev_year = yr - 1,
prev_score = prev_score,
ranking_list = ranking_list,
most_improved_country = most_improved_country,
most_improved_delta = most_improved_delta,
most_declined_country = most_declined_country,
most_declined_delta = most_declined_delta,
year = yr,
n_mdg = n_mdg,
n_sdg = n_sdg,
n_total_ind = n_total_ind,
score = score,
performance_status = perf_status,
yoy_val = yoy_val,
yoy_pct = yoy_pct,
prev_year = yr - 1,
prev_score = prev_score,
prev_performance_status = prev_status,
ranking_list = ranking_list,
most_improved_country = most_improved_country,
most_improved_delta = most_improved_delta,
most_declined_country = most_declined_country,
most_declined_delta = most_declined_delta,
)
records.append({
@@ -1037,6 +1202,7 @@ class FoodSecurityAggregator:
"n_sdg_indicators": n_sdg,
"n_total_indicators": n_total_ind,
"asean_total_score": round(score, 2),
"performance_status": perf_status,
"yoy_change": yoy_val,
"yoy_change_pct": round(yoy_pct, 2) if yoy_pct is not None else None,
"country_ranking_json": country_ranking_json,
@@ -1053,6 +1219,7 @@ class FoodSecurityAggregator:
df["n_sdg_indicators"] = df["n_sdg_indicators"].astype(int)
df["n_total_indicators"] = df["n_total_indicators"].astype(int)
df["asean_total_score"] = df["asean_total_score"].astype(float)
df["performance_status"] = df["performance_status"].astype(str)
for col in ["yoy_change", "yoy_change_pct", "most_improved_delta", "most_declined_delta"]:
df[col] = pd.to_numeric(df[col], errors="coerce").astype(float)
@@ -1062,6 +1229,7 @@ class FoodSecurityAggregator:
bigquery.SchemaField("n_sdg_indicators", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("n_total_indicators", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("asean_total_score", "FLOAT", mode="REQUIRED"),
bigquery.SchemaField("performance_status", "STRING", mode="REQUIRED"),
bigquery.SchemaField("yoy_change", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("yoy_change_pct", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("country_ranking_json", "STRING", mode="REQUIRED"),
@@ -1079,7 +1247,7 @@ class FoodSecurityAggregator:
return df
# =========================================================================
# STEP 7: agg_narrative_pillar -> Gold
# STEP 7: agg_narrative_pillar
# =========================================================================
def calc_narrative_pillar(
@@ -1109,8 +1277,8 @@ class FoodSecurityAggregator:
yr_pillars_yoy = yr_pillars.dropna(subset=["year_over_year_change"])
if not yr_pillars_yoy.empty:
best_p_idx = yr_pillars_yoy["year_over_year_change"].idxmax()
worst_p_idx = yr_pillars_yoy["year_over_year_change"].idxmin()
best_p_idx = yr_pillars_yoy["year_over_year_change"].idxmax()
worst_p_idx = yr_pillars_yoy["year_over_year_change"].idxmin()
most_improved_pillar = str(yr_pillars_yoy.loc[best_p_idx, "pillar_name"])
most_improved_delta = round(float(yr_pillars_yoy.loc[best_p_idx, "year_over_year_change"]), 2)
most_declined_pillar = str(yr_pillars_yoy.loc[worst_p_idx, "pillar_name"])
@@ -1228,8 +1396,7 @@ class FoodSecurityAggregator:
"rows_loaded": rows_loaded, "status": "success", "end_time": datetime.now(),
})
log_update(self.client, "DW", table_name, "full_load", rows_loaded)
self.logger.info(f" ✓ {table_name}: {rows_loaded:,} rows → [Gold] fs_asean_gold")
self.logger.info(f" Metadata → [AUDIT] etl_logs")
self.logger.info(f" [OK] {table_name}: {rows_loaded:,} rows -> [Gold] fs_asean_gold")
def _fail(self, table_name: str, error: Exception):
self.load_metadata[table_name].update({"status": "failed", "end_time": datetime.now()})
@@ -1245,13 +1412,15 @@ class FoodSecurityAggregator:
self.logger.info("\n" + "=" * 70)
self.logger.info("FOOD SECURITY AGGREGATION — 6 TABLES -> fs_asean_gold")
self.logger.info(" Source : fact_asean_food_security_selected")
self.logger.info(" Outputs : agg_pillar_composite | agg_pillar_by_country")
self.logger.info(" agg_framework_by_country| agg_framework_asean")
self.logger.info(" agg_narrative_overview | agg_narrative_pillar")
self.logger.info(" Outputs : agg_pillar_composite | agg_pillar_by_country")
self.logger.info(" agg_framework_by_country | agg_framework_asean")
self.logger.info(" agg_narrative_overview | agg_narrative_pillar")
self.logger.info(f" Performance threshold: {PERFORMANCE_THRESHOLD} (Good/Bad)")
self.logger.info("=" * 70)
self.load_data()
self._classify_indicators()
self.sdgs_start_year = self._detect_sdgs_start_year()
self._assign_framework_labels()
df_pillar_composite = self.calc_pillar_composite()
df_pillar_by_country = self.calc_pillar_by_country()
@@ -1276,12 +1445,12 @@ class FoodSecurityAggregator:
self.logger.info(f" Durasi : {duration:.2f}s")
self.logger.info(f" Total rows : {total_rows:,}")
for tbl, meta in self.load_metadata.items():
icon = "✓" if meta["status"] == "success" else "✗"
icon = "[OK]" if meta["status"] == "success" else "[FAIL]"
self.logger.info(f" {icon} {tbl:<35} {meta['rows_loaded']:>10,}")
# =============================================================================
# AIRFLOW TASK FUNCTIONS
# AIRFLOW TASK
# =============================================================================
def run_aggregation():
@@ -1298,7 +1467,7 @@ def run_aggregation():
# =============================================================================
# MAIN EXECUTION
# MAIN
# =============================================================================
if __name__ == "__main__":
@@ -1313,6 +1482,7 @@ if __name__ == "__main__":
print("FOOD SECURITY AGGREGATION -> fs_asean_gold")
print(f" Source : fact_asean_food_security_selected")
print(f" NORMALIZE_FRAMEWORKS_JOINTLY : {NORMALIZE_FRAMEWORKS_JOINTLY}")
print(f" PERFORMANCE_THRESHOLD : {PERFORMANCE_THRESHOLD}")
print("=" * 70)
logger = setup_logging()

8
scripts/bigquery_cleaned_layer.py
View File

@@ -177,16 +177,16 @@ def standardize_country_names_asean(df: pd.DataFrame, country_column: str = 'cou
def assign_pillar(indicator_name: str) -> str:
"""
Assign pillar berdasarkan keyword indikator.
Return values: 'Availability', 'Access', 'Utilization', 'Stability', 'Other'
Return values: 'Availability', 'Access', 'Utilization', 'Stability', 'Supporting'
All ≤ 20 chars (varchar(20) constraint).
"""
if pd.isna(indicator_name):
return 'Other'
return 'Supporting'
ind = str(indicator_name).lower()
for kw in ['requirement', 'coefficient', 'losses', 'fat supply']:
if kw in ind:
return 'Other'
return 'Supporting'
if any(kw in ind for kw in [
'adequacy', 'protein supply', 'supply of protein',
@@ -215,7 +215,7 @@ def assign_pillar(indicator_name: str) -> str:
]):
return 'Utilization'
return 'Other'
return 'Supporting'
# =============================================================================

6
scripts/bigquery_dimensional_model.py
View File

@@ -350,7 +350,7 @@ class DimensionalModelLoader:
elif any(w in n for w in ['water', 'sanitation', 'infrastructure', 'rail']):
return 'Infrastructure'
else:
return 'Other'
return 'Supporting'
dim_indicator['indicator_category'] = dim_indicator['indicator_name'].apply(categorize_indicator)
dim_indicator = dim_indicator.drop_duplicates(subset=['indicator_name'], keep='first')
@@ -471,10 +471,10 @@ class DimensionalModelLoader:
try:
pillar_codes = {
'Availability': 'AVL', 'Access' : 'ACC',
'Utilization' : 'UTL', 'Stability': 'STB', 'Other': 'OTH',
'Utilization' : 'UTL', 'Stability': 'STB', 'Supporting': 'SPT',
}
pillars_data = [
{'pillar_name': p, 'pillar_code': pillar_codes.get(p, 'OTH')}
{'pillar_name': p, 'pillar_code': pillar_codes.get(p, 'SPT')}
for p in self.df_clean['pillar'].unique()
]