diff --git a/scripts/bigquery_aggregate_layer.py b/scripts/bigquery_aggregate_layer.py index c5c5f9e..edfaced 100644 --- a/scripts/bigquery_aggregate_layer.py +++ b/scripts/bigquery_aggregate_layer.py @@ -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() diff --git a/scripts/bigquery_cleaned_layer.py b/scripts/bigquery_cleaned_layer.py index 0d112fc..8e698af 100644 --- a/scripts/bigquery_cleaned_layer.py +++ b/scripts/bigquery_cleaned_layer.py @@ -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' # ============================================================================= diff --git a/scripts/bigquery_dimensional_model.py b/scripts/bigquery_dimensional_model.py index a5e665c..ba8d14c 100644 --- a/scripts/bigquery_dimensional_model.py +++ b/scripts/bigquery_dimensional_model.py @@ -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() ]