From 6a55a911129f6051f960cfb3544924bc2ca1f7fc Mon Sep 17 00:00:00 2001 From: Debby Date: Wed, 1 Apr 2026 15:46:20 +0700 Subject: [PATCH] code final --- scripts/bigquery_analytical_layer.py | 323 +++++++++++++++------------ 1 file changed, 183 insertions(+), 140 deletions(-) diff --git a/scripts/bigquery_analytical_layer.py b/scripts/bigquery_analytical_layer.py index bf1381e..8c27334 100644 --- a/scripts/bigquery_analytical_layer.py +++ b/scripts/bigquery_analytical_layer.py @@ -19,21 +19,31 @@ NORMALISASI (Step 8): - norm_value_1_100 = min-max normalisasi nilai raw per indikator, skala 1-100 - Direction-aware: lower_better diinvert sehingga nilai tinggi selalu = lebih baik - Normalisasi dilakukan GLOBAL per indikator (semua negara, semua tahun sekaligus) - sehingga nilai antar negara dan antar tahun tetap comparable -- Kolom ini memungkinkan perbandingan antar indikator yang berbeda satuan di Looker Studio -FRAMEWORK LOGIC (Per-Row, bukan per indikator): -- sdg_start_year dideteksi dari data: tahun pertama indikator FIES lengkap - di semua fixed countries (setelah Step 3-5 filter selesai) -- Proxy deteksi sdg_start_year: HANYA FIES ("food insecurity", "food insecure") - Anemia TIDAK dipakai sebagai proxy karena datanya sudah ada sebelum era SDGs -- Framework di-assign PER BARIS (per year), bukan per indikator: - * row['year'] >= sdg_start_year AND nama ada di SDG_INDICATOR_KEYWORDS -> 'SDGs' - * Selain itu -> 'MDGs' -- Ini menangani indikator "shared" (anemia, stunting, wasting, undernourishment) - yang datanya ada sebelum SDGs: - * row lama (year < sdg_start_year) -> 'MDGs' - * row baru (year >= sdg_start_year) -> 'SDGs' +FRAMEWORK LOGIC (Per-Row, threshold = sdg_start_year global): + + sdg_start_year dideteksi HANYA dari FIES ("food insecurity" / "food insecure"), + karena FIES adalah satu-satunya indikator yang murni baru di era SDGs. + Anemia, stunting, wasting, undernourishment TIDAK dipakai sebagai proxy + karena data mereka sudah ada sebelum SDGs sehingga actual_start < sdg_start. + + Framework di-assign PER BARIS menggunakan sdg_start_year global: + - Indikator ada di SDG_INDICATOR_KEYWORDS AND year >= sdg_start_year -> 'SDGs' + - Selain itu -> 'MDGs' + + Efek per kategori indikator (contoh sdg_start_year = 2016): + + Indikator shared (anemia, stunting, wasting, undernourishment): + data mulai 2013 -> year 2013, 2014, 2015 = 'MDGs' (year < 2016) + -> year 2016, 2017, ... = 'SDGs' (year >= 2016) + => SPLIT: sebagian MDGs, sebagian SDGs ✓ + + Indikator FIES (murni SDGs): + data mulai 2016 (== sdg_start_year) -> seluruh baris = 'SDGs' + => Selalu SDGs (tidak ada baris sebelum 2016) ✓ + + Indikator di luar SDG_INDICATOR_KEYWORDS: + -> selalu 'MDGs', tidak peduli tahunnya ✓ """ import pandas as pd @@ -61,16 +71,14 @@ from google.cloud import bigquery # ============================================================================= # SDG INDICATOR KEYWORDS -# Daftar nama indikator (lowercase) yang masuk SDG framework. -# Indikator ini akan di-assign 'SDGs' untuk baris dengan year >= sdg_start_year, -# dan 'MDGs' untuk baris dengan year < sdg_start_year. +# Indikator yang termasuk SDG framework (target 2.1 & 2.2). +# Framework per baris ditentukan oleh sdg_start_year global (dari FIES proxy). # ============================================================================= - SDG_INDICATOR_KEYWORDS = frozenset([ # TARGET 2.1.1 — Prevalence of undernourishment (shared: ada sebelum SDGs) "prevalence of undernourishment (percent) (3-year average)", "number of people undernourished (million) (3-year average)", - # TARGET 2.1.2 — FIES (SDGs only — murni baru di era SDGs) + # TARGET 2.1.2 — FIES (murni baru di era SDGs) "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)", @@ -91,23 +99,19 @@ SDG_INDICATOR_KEYWORDS = frozenset([ "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)", - # TARGET 2.2.3 — Anaemia (shared: data ada sebelum SDGs, listed here agar - # baris >= sdg_start_year di-assign 'SDGs') + # TARGET 2.2.3 — Anaemia (shared: ada sebelum SDGs) "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 ERA PROXY KEYWORDS -# HANYA indikator yang MURNI baru di era SDGs (FIES saja). -# Dipakai untuk mendeteksi sdg_start_year dari data. +# HANYA FIES — dipakai HANYA untuk mendeteksi sdg_start_year dari data. # -# PENTING — Anemia/anaemia TIDAK dipakai sebagai proxy: -# Data anemia sudah ada sebelum era SDGs sehingga actual_start_year-nya -# lebih awal dari sdg_start_year. Jika dipakai sebagai proxy, sdg_start_year -# akan terdeteksi terlalu awal dan seluruh baris anemia akan menjadi 'SDGs'. -# FIES adalah satu-satunya indikator yang benar-benar murni baru di era SDGs -# dan dapat dipakai sebagai penanda tahun mulainya era SDGs. +# KRITIS — anemia/stunting/wasting/undernourishment TIDAK boleh ada di sini: +# Data mereka sudah ada sebelum era SDGs sehingga actual_start_year < sdg_start_year. +# Jika dipakai sebagai proxy, sdg_start_year terdeteksi terlalu awal (misal 2013) +# sehingga seluruh baris indikator shared menjadi 'SDGs' — SALAH. # ============================================================================= _SDG_ERA_PROXY_KEYWORDS = frozenset([ "food insecurity", @@ -117,21 +121,13 @@ _SDG_ERA_PROXY_KEYWORDS = frozenset([ # ============================================================================= # THRESHOLD KONDISI (fixed absolute, skala 1-100) # ============================================================================= -# Digunakan untuk assign kondisi di analysis_layer. -# Didefinisikan di sini agar konsisten antara kedua file. -# bad : norm_value_1_100 < THRESHOLD_BAD -# good : norm_value_1_100 > THRESHOLD_GOOD -# moderate : di antara keduanya - THRESHOLD_BAD = 40.0 THRESHOLD_GOOD = 60.0 def assign_condition(norm_value_1_100: float) -> str: """ - Assign kondisi berdasarkan norm_value_1_100 (skala 1-100, sudah direction-aware). - Nilai tinggi selalu berarti lebih baik (lower_better sudah diinvert). - + Assign kondisi berdasarkan norm_value_1_100 (skala 1-100, direction-aware). Returns: 'good' / 'moderate' / 'bad' """ if pd.isna(norm_value_1_100): @@ -149,30 +145,27 @@ def assign_framework_per_row( sdg_start_year: int, ) -> str: """ - Tentukan framework (MDGs/SDGs) per BARIS (per row year), bukan per indikator. + Tentukan framework (MDGs/SDGs) per BARIS menggunakan sdg_start_year GLOBAL. - Logic: - - 'SDGs' jika KEDUA kondisi terpenuhi: - 1. Nama indikator ada di SDG_INDICATOR_KEYWORDS - 2. year (tahun baris ini) >= sdg_start_year - - 'MDGs' untuk semua kasus lain. + Rules: + 1. Indikator TIDAK ada di SDG_INDICATOR_KEYWORDS -> selalu 'MDGs' + 2. Indikator ada di SDG_INDICATOR_KEYWORDS: + - year >= sdg_start_year -> 'SDGs' + - year < sdg_start_year -> 'MDGs' - Mengapa per row, bukan per indikator? - Indikator "shared" seperti anemia, stunting, wasting, undernourishment - memiliki data yang ada SEBELUM era SDGs dimulai. Jika assign dilakukan - per indikator menggunakan actual_start_year, indikator-indikator ini - akan selalu di-assign 'MDGs' karena actual_start_year < sdg_start_year. - Dengan assign per row menggunakan year baris: - - baris lama (year < sdg_start_year) -> 'MDGs' (benar: belum era SDGs) - - baris baru (year >= sdg_start_year) -> 'SDGs' (benar: sudah era SDGs) + sdg_start_year dideteksi dari FIES (proxy murni SDGs), bukan dari + actual_start_year masing-masing indikator. Ini memastikan indikator + shared (anemia, stunting, wasting, undernourishment) yang datanya + ada sebelum SDGs tetap mendapat label 'MDGs' untuk baris sebelum + sdg_start_year dan 'SDGs' untuk baris sejak sdg_start_year. - Contoh anemia (sdg_start_year = 2016): - - row year=2013 -> 'MDGs' - - row year=2014 -> 'MDGs' - - row year=2015 -> 'MDGs' - - row year=2016 -> 'SDGs' - - row year=2017 -> 'SDGs' - - ... + Contoh (sdg_start_year = 2016): + anemia year=2013 -> 'MDGs' (ada di SDG list, tapi year < 2016) + anemia year=2015 -> 'MDGs' + anemia year=2016 -> 'SDGs' (year >= 2016) + anemia year=2023 -> 'SDGs' + FIES year=2016 -> 'SDGs' (tidak ada baris FIES sebelum 2016) + non-SDG year=any -> 'MDGs' (tidak ada di SDG_INDICATOR_KEYWORDS) """ name_lower = str(indicator_name).lower().strip() in_sdg_list = name_lower in SDG_INDICATOR_KEYWORDS @@ -187,21 +180,28 @@ def assign_framework_per_row( class AnalyticalLayerLoader: """ - Analytical Layer Loader for BigQuery + Analytical Layer Loader for BigQuery. Output kolom fact_asean_food_security_selected: country_id, country_name, indicator_id, indicator_name, direction, framework, pillar_id, pillar_name, time_id, year, value, - norm_value_1_100, <- min-max norm per indikator, skala 1-100, direction-aware + norm_value_1_100, yoy_change, yoy_pct - Catatan framework: - Framework di-assign PER BARIS (per year), sehingga indikator shared - seperti anemia dapat memiliki framework berbeda di baris yang berbeda: - - baris sebelum sdg_start_year -> 'MDGs' - - baris sejak sdg_start_year -> 'SDGs' + Framework logic (sdg_start_year global dari FIES proxy): + Indikator shared (anemia, stunting, wasting, undernourishment): + year < sdg_start_year -> 'MDGs' (misal 2013-2015) + year >= sdg_start_year -> 'SDGs' (misal 2016-2023) + => SPLIT: sebagian MDGs, sebagian SDGs ✓ + + Indikator FIES (murni SDGs): + seluruh baris -> 'SDGs' + (tidak ada data FIES sebelum sdg_start_year) ✓ + + Indikator di luar SDG_INDICATOR_KEYWORDS: + seluruh baris -> 'MDGs' ✓ """ def __init__(self, client: bigquery.Client): @@ -218,9 +218,9 @@ class AnalyticalLayerLoader: self.start_year = 2013 self.end_year = None - self.baseline_year = 2023 # hardcode per syarat dosen (tahun terlengkap) + self.baseline_year = 2023 # hardcode per syarat dosen - self.sdg_start_year = None + self.sdg_start_year = None # dideteksi HANYA dari FIES proxy di Step 6 self.pipeline_metadata = { 'source_class' : self.__class__.__name__, @@ -306,7 +306,6 @@ class AnalyticalLayerLoader: self.logger.info("STEP 2: DETERMINE YEAR BOUNDARIES") self.logger.info("=" * 80) - # Filter single years only (is_year_range == False) if 'is_year_range' in self.df_clean.columns: before = len(self.df_clean) self.df_clean = self.df_clean[self.df_clean['is_year_range'] == False].copy() @@ -314,7 +313,6 @@ class AnalyticalLayerLoader: f" Filter single years only: {before:,} -> {len(self.df_clean):,} rows" ) - # baseline_year = 2023 hardcode (syarat dosen: minimal 2023) df_baseline = self.df_clean[self.df_clean['year'] == self.baseline_year] baseline_indicator_count = df_baseline['indicator_id'].nunique() @@ -542,6 +540,7 @@ class AnalyticalLayerLoader: self.df_clean['indicator_id'].isin(valid_indicators) ].copy() + # Trim baris di bawah max_start_year per indikator self.df_clean = self.df_clean.merge( indicator_max_start[['indicator_id', 'max_start_year']], on='indicator_id', how='left' @@ -567,13 +566,16 @@ class AnalyticalLayerLoader: self.logger.info("STEP 6: DETERMINE SDG START YEAR & ASSIGN FRAMEWORK PER ROW") self.logger.info("=" * 80) self.logger.info( - " Proxy: FIES only (food insecurity/food insecure).\n" - " Anemia TIDAK dipakai sebagai proxy — datanya ada sebelum era SDGs.\n" - " Framework di-assign PER BARIS (year), bukan per indikator." + " sdg_start_year dideteksi HANYA dari FIES proxy\n" + " (food insecurity / food insecure — murni baru di era SDGs).\n" + " Anemia/stunting/wasting/undernourishment TIDAK dipakai sebagai proxy.\n\n" + " Framework per baris (threshold = sdg_start_year global):\n" + " SDG_INDICATOR_KEYWORDS + year >= sdg_start_year -> 'SDGs'\n" + " SDG_INDICATOR_KEYWORDS + year < sdg_start_year -> 'MDGs' [SPLIT]\n" + " Indikator di luar SDG_INDICATOR_KEYWORDS -> selalu 'MDGs'" ) - # actual_start_year per indikator = max(min_year per country) - # = konsisten dengan max_start_year di Step 5 + # Hitung actual_start_year per indikator (untuk logging & validasi) indicator_actual_start = ( self.df_clean .groupby(['indicator_id', 'indicator_name', 'country_id'])['year'] @@ -583,7 +585,9 @@ class AnalyticalLayerLoader: ) indicator_actual_start.columns = ['indicator_id', 'indicator_name', 'actual_start_year'] - # Deteksi sdg_start_year dari proxy SDGs-only (FIES saja, BUKAN anemia) + # ------------------------------------------------------------------ + # Deteksi sdg_start_year HANYA dari FIES proxy + # ------------------------------------------------------------------ proxy_mask = indicator_actual_start['indicator_name'].str.lower().apply( lambda n: any(kw in n for kw in _SDG_ERA_PROXY_KEYWORDS) ) @@ -591,22 +595,46 @@ class AnalyticalLayerLoader: if df_proxy.empty: raise ValueError( - "Tidak ada indikator proxy SDGs (FIES) yang lolos filter. " - "Pastikan indikator FIES (food insecurity/food insecure) ada di data." + "Tidak ada indikator FIES (food insecurity/food insecure) yang lolos filter. " + "Pastikan indikator FIES ada di data dan lolos Step 3-5." ) self.sdg_start_year = int(df_proxy['actual_start_year'].min()) - self.logger.info(f"\n sdg_start_year = {self.sdg_start_year}") - self.logger.info(f" Proxy indicators (FIES only):") + + self.logger.info(f"\n sdg_start_year = {self.sdg_start_year} (dari FIES proxy)") + self.logger.info(f" FIES proxy indicators:") for _, row in df_proxy.iterrows(): self.logger.info(f" [{int(row['actual_start_year'])}] {row['indicator_name']}") - # ---------------------------------------------------------------- - # Assign framework PER BARIS menggunakan year baris, bukan actual_start_year - # Sehingga indikator "shared" (anemia, stunting, dll) mendapat: - # - 'MDGs' untuk baris sebelum sdg_start_year - # - 'SDGs' untuk baris sejak sdg_start_year - # ---------------------------------------------------------------- + # Log indikator shared yang akan split (ada di SDG list, data mulai sebelum sdg_start_year) + shared_sdg = indicator_actual_start[ + ~proxy_mask & + indicator_actual_start['indicator_name'].str.lower().isin(SDG_INDICATOR_KEYWORDS) & + (indicator_actual_start['actual_start_year'] < self.sdg_start_year) + ] + if not shared_sdg.empty: + self.logger.info( + f"\n Indikator shared yang akan SPLIT MDGs/SDGs " + f"(data mulai < sdg_start_year={self.sdg_start_year}):" + ) + for _, row in shared_sdg.iterrows(): + n_mdgs = len(self.df_clean[ + (self.df_clean['indicator_id'] == row['indicator_id']) & + (self.df_clean['year'] < self.sdg_start_year) + ]) + n_sdgs = len(self.df_clean[ + (self.df_clean['indicator_id'] == row['indicator_id']) & + (self.df_clean['year'] >= self.sdg_start_year) + ]) + self.logger.info( + f" [actual_start={int(row['actual_start_year'])}] " + f"{row['indicator_name'][:50]} " + f"| MDGs rows: {n_mdgs:,} | SDGs rows: {n_sdgs:,}" + ) + + # ------------------------------------------------------------------ + # Assign framework PER BARIS menggunakan sdg_start_year global + # ------------------------------------------------------------------ self.df_clean['framework'] = self.df_clean.apply( lambda row: assign_framework_per_row( indicator_name = row['indicator_name'], @@ -616,9 +644,9 @@ class AnalyticalLayerLoader: axis=1 ) - # ---------------------------------------------------------------- - # Logging: ringkasan per indikator (frameworks apa yang muncul) - # ---------------------------------------------------------------- + # ------------------------------------------------------------------ + # Logging ringkasan per indikator + # ------------------------------------------------------------------ ind_fw_summary = ( self.df_clean .groupby(['indicator_id', 'indicator_name'])['framework'] @@ -634,9 +662,9 @@ class AnalyticalLayerLoader: ) self.logger.info(f"\n Framework assignment per indikator:") - self.logger.info(f" {'-'*85}") + self.logger.info(f" {'-'*90}") self.logger.info(f" {'ID':<5} {'Frameworks':<18} {'ActualStart':<13} {'Indicator Name'}") - self.logger.info(f" {'-'*85}") + self.logger.info(f" {'-'*90}") for _, row in ind_fw_summary.sort_values( ['frameworks', 'actual_start_year', 'indicator_name'] ).iterrows(): @@ -645,24 +673,48 @@ class AnalyticalLayerLoader: f"{int(row['actual_start_year']):<13} {row['indicator_name'][:48]}" ) - # Indikator dengan framework split (MDGs/SDGs) — highlight untuk validasi + # Ringkasan per kategori + mdgs_only = ind_fw_summary[ind_fw_summary['frameworks'] == 'MDGs'] + sdgs_only = ind_fw_summary[ind_fw_summary['frameworks'] == 'SDGs'] split_inds = ind_fw_summary[ind_fw_summary['frameworks'] == 'MDGs/SDGs'] + + if not mdgs_only.empty: + self.logger.info( + f"\n [MDGs only — {len(mdgs_only)} indikator] " + f"Tidak ada di SDG_INDICATOR_KEYWORDS:" + ) + for _, row in mdgs_only.iterrows(): + self.logger.info(f" - {row['indicator_name'][:65]}") + + if not sdgs_only.empty: + self.logger.info( + f"\n [SDGs only — {len(sdgs_only)} indikator] " + f"Data mulai = sdg_start_year, tidak ada baris sebelumnya:" + ) + for _, row in sdgs_only.iterrows(): + self.logger.info( + f" - [{int(row['actual_start_year'])}] {row['indicator_name'][:65]}" + ) + if not split_inds.empty: self.logger.info( - f"\n [INFO] {len(split_inds)} indikator memiliki framework split " - f"(MDGs sebelum {self.sdg_start_year}, SDGs sejak {self.sdg_start_year}):" + f"\n [SPLIT MDGs/SDGs — {len(split_inds)} indikator] " + f"Baris < {self.sdg_start_year} = MDGs | " + f"Baris >= {self.sdg_start_year} = SDGs:" ) for _, row in split_inds.iterrows(): - self.logger.info(f" - {row['indicator_name'][:60]}") + self.logger.info( + f" - [actual_start={int(row['actual_start_year'])}] " + f"{row['indicator_name'][:65]}" + ) fw_summary = self.df_clean['framework'].value_counts() self.logger.info( f"\n Ringkasan rows: " + " | ".join(f"{fw}: {cnt:,}" for fw, cnt in fw_summary.items()) ) - self.logger.info( - f"\n [OK] 'framework' ditambahkan per row — " + f"\n [OK] 'framework' ditambahkan — " f"MDGs: {(self.df_clean['framework'] == 'MDGs').sum():,} rows | " f"SDGs: {(self.df_clean['framework'] == 'SDGs').sum():,} rows" ) @@ -704,25 +756,6 @@ class AnalyticalLayerLoader: # ------------------------------------------------------------------ def calculate_norm_value(self): - """ - Hitung norm_value_1_100 per indikator — min-max normalisasi skala 1-100, - direction-aware. - - CARA KERJA: - - Normalisasi dilakukan GLOBAL per indikator (semua negara + semua tahun sekaligus) - sehingga nilai antar negara dan antar tahun tetap comparable. - - lower_better diinvert: nilai tinggi selalu = kondisi lebih baik. - Contoh: undernourishment 5% (rendah = baik) → norm tinggi setelah invert. - - Skala 1-100 (bukan 0-100) untuk menghindari nilai absolut nol di Looker Studio. - - Kolom ini memungkinkan perbandingan lintas indikator yang berbeda satuan - (persen, juta orang, dll) karena sudah dinormalisasi ke skala yang sama. - - Catatan: - - Berbeda dengan norm_value di _get_norm_value_df() di analysis_layer - yang skala 0-1 dan dipakai untuk agregasi composite score. - - norm_value_1_100 ini adalah per baris (per country per year per indicator), - untuk ditampilkan langsung di Looker Studio. - """ self.logger.info("\n" + "=" * 80) self.logger.info("STEP 8: CALCULATE NORM_VALUE_1_100 PER INDICATOR") self.logger.info("=" * 80) @@ -735,7 +768,10 @@ class AnalyticalLayerLoader: norm_parts = [] indicators = df.groupby(['indicator_id', 'indicator_name', 'direction']) - self.logger.info(f"\n {'ID':<5} {'Direction':<15} {'Invert':<8} {'Min':>10} {'Max':>10} {'Indicator Name'}") + self.logger.info( + f"\n {'ID':<5} {'Direction':<15} {'Invert':<8} " + f"{'Min':>10} {'Max':>10} {'Indicator Name'}" + ) self.logger.info(f" {'-'*90}") for (ind_id, ind_name, direction), grp in indicators: @@ -749,21 +785,17 @@ class AnalyticalLayerLoader: 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) + 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: - # Semua nilai sama → beri nilai tengah (50.5 pada skala 1-100) normed[valid_mask.values] = 50.5 else: - # Min-max ke 0-1 dulu scaled = (raw - v_min) / (v_max - v_min) - # Invert jika lower_better if do_invert: scaled = 1.0 - scaled - # Scale ke 1-100 normed[valid_mask.values] = 1.0 + scaled * 99.0 grp['norm_value_1_100'] = normed @@ -776,7 +808,6 @@ class AnalyticalLayerLoader: self.df_clean = pd.concat(norm_parts, ignore_index=True) - # Statistik ringkasan valid_norm = self.df_clean['norm_value_1_100'].notna().sum() null_norm = self.df_clean['norm_value_1_100'].isna().sum() self.logger.info(f"\n norm_value_1_100 — valid: {valid_norm:,} | null: {null_norm:,}") @@ -786,15 +817,17 @@ class AnalyticalLayerLoader: f"{self.df_clean['norm_value_1_100'].max():.2f}" ) - # Log distribusi kondisi berdasarkan threshold self.df_clean['_condition_preview'] = self.df_clean['norm_value_1_100'].apply(assign_condition) cond_dist = self.df_clean['_condition_preview'].value_counts() - self.logger.info(f"\n Distribusi kondisi (threshold: bad<{THRESHOLD_BAD}, good>{THRESHOLD_GOOD}):") + self.logger.info( + f"\n Distribusi kondisi " + f"(threshold: bad<{THRESHOLD_BAD}, good>{THRESHOLD_GOOD}):" + ) for cond, cnt in cond_dist.items(): self.logger.info(f" {cond}: {cnt:,} rows") self.df_clean = self.df_clean.drop(columns=['_condition_preview']) - self.logger.info(f"\n [OK] Kolom 'norm_value_1_100' ditambahkan ke df_clean") + self.logger.info(f"\n [OK] Kolom 'norm_value_1_100' ditambahkan") return self.df_clean # ------------------------------------------------------------------ @@ -862,7 +895,6 @@ class AnalyticalLayerLoader: 'start_year', 'end_year', 'country_count' ] - # Framework summary per indikator (bisa MDGs, SDGs, atau MDGs/SDGs split) ind_fw = ( self.df_clean .groupby('indicator_id')['framework'] @@ -963,13 +995,11 @@ class AnalyticalLayerLoader: self.logger.info(f" Total rows: {len(analytical_df):,}") - # Framework distribution per row fw_dist_rows = analytical_df['framework'].value_counts() self.logger.info(f" Framework distribution (rows):") for fw, cnt in fw_dist_rows.items(): self.logger.info(f" {fw}: {cnt:,} rows") - # Framework distribution per indikator (label) ind_fw_label = ( analytical_df .groupby('indicator_id')['framework'] @@ -1028,7 +1058,11 @@ class AnalyticalLayerLoader: 'sdg_start_year' : self.sdg_start_year, 'fixed_countries' : len(self.selected_country_ids), 'norm_scale' : '1-100 per indicator global minmax direction-aware', - 'framework_assignment' : 'per-row by year (not per-indicator)', + 'framework_assignment' : ( + f'per-row, sdg_start_year={self.sdg_start_year} global (FIES proxy only). ' + 'SDG_INDICATOR_KEYWORDS + year >= sdg_start_year -> SDGs, else MDGs. ' + 'Shared indicators (anemia/stunting/wasting/undernourishment) split MDGs/SDGs.' + ), 'sdg_proxy_keywords' : list(_SDG_ERA_PROXY_KEYWORDS), 'condition_thresholds' : { 'bad' : f'< {THRESHOLD_BAD}', @@ -1064,8 +1098,12 @@ class AnalyticalLayerLoader: self.logger.info("\n" + "=" * 80) self.logger.info("Output: fact_asean_food_security_selected -> fs_asean_gold") self.logger.info("Kolom baru: norm_value_1_100 (min-max 1-100, direction-aware)") - self.logger.info("Framework: per-row by year (shared indicators split MDGs/SDGs)") - self.logger.info(f"SDG Proxy: FIES only (food insecurity/food insecure)") + self.logger.info( + "Framework: per-row, threshold = sdg_start_year global (dari FIES proxy)\n" + " SDG_INDICATOR_KEYWORDS + year >= sdg_start_year -> 'SDGs'\n" + " SDG_INDICATOR_KEYWORDS + year < sdg_start_year -> 'MDGs' [SPLIT]\n" + " Indikator di luar SDG_INDICATOR_KEYWORDS -> selalu 'MDGs'" + ) self.logger.info(f"Condition threshold: bad<{THRESHOLD_BAD}, good>{THRESHOLD_GOOD}") self.logger.info("=" * 80) @@ -1074,10 +1112,10 @@ class AnalyticalLayerLoader: self.filter_complete_indicators_per_country() self.select_countries_with_all_pillars() self.filter_indicators_consistent_across_fixed_countries() - self.determine_sdg_start_year() # Step 6: per-row framework assignment + self.determine_sdg_start_year() self.verify_no_gaps() - self.calculate_norm_value() # Step 8: norm_value_1_100 - self.calculate_yoy() # Step 9: yoy_change, yoy_pct + self.calculate_norm_value() + self.calculate_yoy() self.analyze_indicator_availability_by_year() self.save_analytical_table() @@ -1089,7 +1127,7 @@ class AnalyticalLayerLoader: self.logger.info("=" * 80) self.logger.info(f" Duration : {duration:.2f}s") self.logger.info(f" Year Range : {self.start_year}-{self.end_year}") - self.logger.info(f" SDG Start Yr : {self.sdg_start_year}") + self.logger.info(f" SDG Start Yr : {self.sdg_start_year} (dari FIES proxy)") self.logger.info(f" Countries : {len(self.selected_country_ids)}") self.logger.info(f" Indicators : {self.df_clean['indicator_id'].nunique()}") self.logger.info(f" Rows Loaded : {self.pipeline_metadata['rows_loaded']:,}") @@ -1116,7 +1154,12 @@ if __name__ == "__main__": print("BIGQUERY ANALYTICAL LAYER - DATA FILTERING") print("Output: fact_asean_food_security_selected -> fs_asean_gold") print(f"Norm: min-max 1-100 per indicator, direction-aware") - print(f"Framework: per-row by year | SDG Proxy: FIES only") + print( + "Framework: per-row, threshold = sdg_start_year global (dari FIES proxy)\n" + " SDG_INDICATOR_KEYWORDS + year >= sdg_start_year -> SDGs\n" + " SDG_INDICATOR_KEYWORDS + year < sdg_start_year -> MDGs [SPLIT]\n" + " Indikator di luar SDG_INDICATOR_KEYWORDS -> selalu MDGs" + ) print(f"Condition threshold: bad<{THRESHOLD_BAD}, good>{THRESHOLD_GOOD}") print("=" * 80)