""" BIGQUERY ANALYTICAL LAYER - DATA FILTERING fact_asean_food_security_selected disimpan di fs_asean_gold (layer='gold') Filtering Order: 1. Load data (single years only) 2. Determine year boundaries (2013 - auto-detected end year, baseline=2023 per syarat dosen) 3. Filter complete indicators PER COUNTRY (auto-detect start year, no gaps) 4. Filter countries with ALL pillars (FIXED SET) 5. Filter indicators with consistent presence across FIXED countries 6. Determine SDG start year & assign framework (MDGs/SDGs) per ROW per year 7. Verify no gaps 8. Calculate norm_value_1_100 per indicator per country (min-max, direction-aware) 9. Calculate YoY per indicator per country 10. Analyze indicator availability by year 11. Save analytical table 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) 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 import numpy as np from datetime import datetime import logging from typing import Dict, List import json import sys if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8') from scripts.bigquery_config import get_bigquery_client, CONFIG, get_table_id from scripts.bigquery_helpers import ( log_update, load_to_bigquery, read_from_bigquery, setup_logging, truncate_table, save_etl_metadata, ) from google.cloud import bigquery # ============================================================================= # SDG INDICATOR KEYWORDS # 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 (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)", "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)", # TARGET 2.2.1 — Stunting (shared: ada sebelum SDGs) "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)", # TARGET 2.2.2 — Wasting & Overweight (shared: ada sebelum SDGs) "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)", # 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 FIES — dipakai HANYA untuk mendeteksi sdg_start_year dari data. # # 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", "food insecure", ]) # ============================================================================= # THRESHOLD KONDISI (fixed absolute, skala 1-100) # ============================================================================= 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, direction-aware). Returns: 'good' / 'moderate' / 'bad' """ if pd.isna(norm_value_1_100): return None if norm_value_1_100 > THRESHOLD_GOOD: return 'good' if norm_value_1_100 < THRESHOLD_BAD: return 'bad' return 'moderate' def assign_framework_per_row( indicator_name: str, year: int, sdg_start_year: int, ) -> str: """ Tentukan framework (MDGs/SDGs) per BARIS menggunakan sdg_start_year GLOBAL. 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' 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 (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 if in_sdg_list and int(year) >= sdg_start_year: return 'SDGs' return 'MDGs' # ============================================================================= # ANALYTICAL LAYER CLASS # ============================================================================= class AnalyticalLayerLoader: """ 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, yoy_change, yoy_pct 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): self.client = client self.logger = logging.getLogger(self.__class__.__name__) self.logger.propagate = False self.df_clean = None self.df_indicator = None self.df_country = None self.df_pillar = None self.selected_country_ids = None self.start_year = 2013 self.end_year = None self.baseline_year = 2023 # hardcode per syarat dosen self.sdg_start_year = None # dideteksi HANYA dari FIES proxy di Step 6 self.pipeline_metadata = { 'source_class' : self.__class__.__name__, 'start_time' : None, 'end_time' : None, 'duration_seconds' : None, 'rows_fetched' : 0, 'rows_transformed' : 0, 'rows_loaded' : 0, 'validation_metrics': {} } self.pipeline_start = None self.pipeline_end = None # ------------------------------------------------------------------ # STEP 1: LOAD SOURCE DATA # ------------------------------------------------------------------ def load_source_data(self): self.logger.info("\n" + "=" * 80) self.logger.info("STEP 1: LOADING SOURCE DATA from fs_asean_gold") self.logger.info("=" * 80) try: query = f""" SELECT f.country_id, c.country_name, f.indicator_id, i.indicator_name, i.direction, f.pillar_id, p.pillar_name, f.time_id, t.year, t.start_year, t.end_year, t.is_year_range, f.value, f.source_id FROM `{get_table_id('fact_food_security', layer='gold')}` f JOIN `{get_table_id('dim_country', layer='gold')}` c ON f.country_id = c.country_id JOIN `{get_table_id('dim_indicator', layer='gold')}` i ON f.indicator_id = i.indicator_id JOIN `{get_table_id('dim_pillar', layer='gold')}` p ON f.pillar_id = p.pillar_id JOIN `{get_table_id('dim_time', layer='gold')}` t ON f.time_id = t.time_id """ self.logger.info("Loading fact table with dimensions...") self.df_clean = self.client.query(query).result().to_dataframe( create_bqstorage_client=False ) self.logger.info(f" Loaded: {len(self.df_clean):,} rows") if 'is_year_range' in self.df_clean.columns: yr = self.df_clean['is_year_range'].value_counts() self.logger.info( f" Single years: {yr.get(False, 0):,} | " f"Year ranges: {yr.get(True, 0):,}" ) self.df_indicator = read_from_bigquery(self.client, 'dim_indicator', layer='gold') self.df_country = read_from_bigquery(self.client, 'dim_country', layer='gold') self.df_pillar = read_from_bigquery(self.client, 'dim_pillar', layer='gold') self.logger.info(f" Indicators: {len(self.df_indicator)}") self.logger.info(f" Countries: {len(self.df_country)}") self.logger.info(f" Pillars: {len(self.df_pillar)}") self.pipeline_metadata['rows_fetched'] = len(self.df_clean) return True except Exception as e: self.logger.error(f"Error loading source data: {e}") raise # ------------------------------------------------------------------ # STEP 2: DETERMINE YEAR BOUNDARIES # ------------------------------------------------------------------ def determine_year_boundaries(self): self.logger.info("\n" + "=" * 80) self.logger.info("STEP 2: DETERMINE YEAR BOUNDARIES") self.logger.info("=" * 80) 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() self.logger.info( f" Filter single years only: {before:,} -> {len(self.df_clean):,} rows" ) df_baseline = self.df_clean[self.df_clean['year'] == self.baseline_year] baseline_indicator_count = df_baseline['indicator_id'].nunique() self.logger.info(f"\n Baseline year (hardcode, syarat dosen): {self.baseline_year}") self.logger.info(f" Baseline indicator count: {baseline_indicator_count}") years_sorted = sorted(self.df_clean['year'].unique(), reverse=True) selected_end_year = None self.logger.info(f"\n Scanning end_year (>= {self.baseline_year}):") for year in years_sorted: if year >= self.baseline_year: df_year = self.df_clean[self.df_clean['year'] == year] year_indicator_count = df_year['indicator_id'].nunique() status = "OK" if year_indicator_count >= baseline_indicator_count else "X" self.logger.info(f" [{status}] Year {int(year)}: {year_indicator_count} indicators") if year_indicator_count >= baseline_indicator_count and selected_end_year is None: selected_end_year = int(year) if selected_end_year is None: selected_end_year = self.baseline_year self.logger.warning(f" [!] Fallback to baseline: {selected_end_year}") else: self.logger.info(f"\n [OK] Selected end year: {selected_end_year}") self.end_year = selected_end_year original_count = len(self.df_clean) self.df_clean = self.df_clean[ (self.df_clean['year'] >= self.start_year) & (self.df_clean['year'] <= self.end_year) ].copy() self.logger.info(f"\n Filtering {self.start_year}-{self.end_year}:") self.logger.info(f" Rows before: {original_count:,}") self.logger.info(f" Rows after : {len(self.df_clean):,}") return self.df_clean # ------------------------------------------------------------------ # STEP 3: FILTER COMPLETE INDICATORS PER COUNTRY # ------------------------------------------------------------------ def filter_complete_indicators_per_country(self): self.logger.info("\n" + "=" * 80) self.logger.info("STEP 3: FILTER COMPLETE INDICATORS PER COUNTRY (NO GAPS)") self.logger.info("=" * 80) grouped = self.df_clean.groupby([ 'country_id', 'country_name', 'indicator_id', 'indicator_name', 'pillar_id', 'pillar_name' ]) valid_combinations = [] removed_combinations = [] for (country_id, country_name, indicator_id, indicator_name, pillar_id, pillar_name), group in grouped: years_present = sorted(group['year'].unique()) start_year = int(min(years_present)) end_year_actual = int(max(years_present)) expected_years = list(range(start_year, self.end_year + 1)) missing_years = [y for y in expected_years if y not in years_present] has_gap = len(missing_years) > 0 is_complete = ( end_year_actual >= self.end_year and not has_gap and (self.end_year - start_year) >= 4 ) if is_complete: valid_combinations.append({'country_id': country_id, 'indicator_id': indicator_id}) else: reasons = [] if end_year_actual < self.end_year: reasons.append(f"ends {end_year_actual}") if has_gap: gap_str = str(missing_years[:3])[1:-1] if len(missing_years) > 3: gap_str += "..." reasons.append(f"gap:{gap_str}") if (self.end_year - start_year) < 4: reasons.append(f"span={self.end_year - start_year}") removed_combinations.append({ 'country_name' : country_name, 'indicator_name': indicator_name, 'reasons' : ", ".join(reasons) }) self.logger.info(f"\n [+] Valid: {len(valid_combinations):,}") self.logger.info(f" [-] Removed: {len(removed_combinations):,}") df_valid = pd.DataFrame(valid_combinations) df_valid['key'] = ( df_valid['country_id'].astype(str) + '_' + df_valid['indicator_id'].astype(str) ) self.df_clean['key'] = ( self.df_clean['country_id'].astype(str) + '_' + self.df_clean['indicator_id'].astype(str) ) original_count = len(self.df_clean) self.df_clean = self.df_clean[self.df_clean['key'].isin(df_valid['key'])].copy() self.df_clean = self.df_clean.drop('key', axis=1) self.logger.info(f"\n Rows before: {original_count:,}") self.logger.info(f" Rows after: {len(self.df_clean):,}") self.logger.info(f" Countries: {self.df_clean['country_id'].nunique()}") self.logger.info(f" Indicators: {self.df_clean['indicator_id'].nunique()}") return self.df_clean # ------------------------------------------------------------------ # STEP 4: SELECT COUNTRIES WITH ALL PILLARS # ------------------------------------------------------------------ def select_countries_with_all_pillars(self): self.logger.info("\n" + "=" * 80) self.logger.info("STEP 4: SELECT COUNTRIES WITH ALL PILLARS (FIXED SET)") self.logger.info("=" * 80) total_pillars = self.df_clean['pillar_id'].nunique() country_pillar_count = self.df_clean.groupby(['country_id', 'country_name']).agg({ 'pillar_id' : 'nunique', 'indicator_id': 'nunique', 'year' : lambda x: f"{int(x.min())}-{int(x.max())}" }).reset_index() country_pillar_count.columns = [ 'country_id', 'country_name', 'pillar_count', 'indicator_count', 'year_range' ] for _, row in country_pillar_count.sort_values('pillar_count', ascending=False).iterrows(): status = "[+] KEEP" if row['pillar_count'] == total_pillars else "[-] REMOVE" self.logger.info( f" {status:<12} {row['country_name']:25s} " f"{row['pillar_count']}/{total_pillars} pillars" ) selected_countries = country_pillar_count[ country_pillar_count['pillar_count'] == total_pillars ] self.selected_country_ids = selected_countries['country_id'].tolist() self.logger.info(f"\n FIXED SET: {len(self.selected_country_ids)} countries") original_count = len(self.df_clean) self.df_clean = self.df_clean[ self.df_clean['country_id'].isin(self.selected_country_ids) ].copy() self.logger.info(f" Rows before: {original_count:,}") self.logger.info(f" Rows after: {len(self.df_clean):,}") return self.df_clean # ------------------------------------------------------------------ # STEP 5: FILTER INDICATORS CONSISTENT ACROSS FIXED COUNTRIES # ------------------------------------------------------------------ def filter_indicators_consistent_across_fixed_countries(self): self.logger.info("\n" + "=" * 80) self.logger.info("STEP 5: FILTER INDICATORS WITH CONSISTENT PRESENCE") self.logger.info("=" * 80) indicator_country_start = self.df_clean.groupby([ 'indicator_id', 'indicator_name', 'country_id' ])['year'].min().reset_index() indicator_country_start.columns = [ 'indicator_id', 'indicator_name', 'country_id', 'start_year' ] indicator_max_start = indicator_country_start.groupby([ 'indicator_id', 'indicator_name' ])['start_year'].max().reset_index() indicator_max_start.columns = ['indicator_id', 'indicator_name', 'max_start_year'] valid_indicators = [] removed_indicators = [] for _, ind_row in indicator_max_start.iterrows(): indicator_id = ind_row['indicator_id'] indicator_name = ind_row['indicator_name'] max_start = int(ind_row['max_start_year']) span = self.end_year - max_start if span < 4: removed_indicators.append({ 'indicator_name': indicator_name, 'reason' : f"span={span} < 4" }) continue expected_years = list(range(max_start, self.end_year + 1)) ind_data = self.df_clean[self.df_clean['indicator_id'] == indicator_id] all_years_complete = True problematic_years = [] for year in expected_years: country_count = ind_data[ind_data['year'] == year]['country_id'].nunique() if country_count < len(self.selected_country_ids): all_years_complete = False problematic_years.append(f"{int(year)}({country_count})") if all_years_complete: valid_indicators.append(indicator_id) else: removed_indicators.append({ 'indicator_name': indicator_name, 'reason' : f"missing countries in years: {', '.join(problematic_years[:5])}" }) self.logger.info(f"\n [+] Valid: {len(valid_indicators)}") self.logger.info(f" [-] Removed: {len(removed_indicators)}") if removed_indicators: self.logger.info(f"\n Removed indicators:") for item in removed_indicators: self.logger.info(f" [-] {item['indicator_name'][:60]} | {item['reason']}") if not valid_indicators: raise ValueError("No valid indicators found after filtering!") original_count = len(self.df_clean) self.df_clean = self.df_clean[ 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' ) self.df_clean = self.df_clean[ self.df_clean['year'] >= self.df_clean['max_start_year'] ].copy() self.df_clean = self.df_clean.drop('max_start_year', axis=1) self.logger.info(f"\n Rows before: {original_count:,}") self.logger.info(f" Rows after: {len(self.df_clean):,}") self.logger.info(f" Countries: {self.df_clean['country_id'].nunique()}") self.logger.info(f" Indicators: {self.df_clean['indicator_id'].nunique()}") self.logger.info(f" Pillars: {self.df_clean['pillar_id'].nunique()}") return self.df_clean # ------------------------------------------------------------------ # STEP 6: DETERMINE SDG START YEAR & ASSIGN FRAMEWORK PER ROW # ------------------------------------------------------------------ def determine_sdg_start_year(self): self.logger.info("\n" + "=" * 80) self.logger.info("STEP 6: DETERMINE SDG START YEAR & ASSIGN FRAMEWORK PER ROW") self.logger.info("=" * 80) self.logger.info( " 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'" ) # Hitung actual_start_year per indikator (untuk logging & validasi) indicator_actual_start = ( self.df_clean .groupby(['indicator_id', 'indicator_name', 'country_id'])['year'] .min().reset_index() .groupby(['indicator_id', 'indicator_name'])['year'] .max().reset_index() ) indicator_actual_start.columns = ['indicator_id', 'indicator_name', 'actual_start_year'] # ------------------------------------------------------------------ # 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) ) df_proxy = indicator_actual_start[proxy_mask] if df_proxy.empty: raise ValueError( "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} (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']}") # 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'], year = int(row['year']), sdg_start_year = self.sdg_start_year, ), axis=1 ) # ------------------------------------------------------------------ # Logging ringkasan per indikator # ------------------------------------------------------------------ ind_fw_summary = ( self.df_clean .groupby(['indicator_id', 'indicator_name'])['framework'] .unique() .reset_index() ) ind_fw_summary['frameworks'] = ind_fw_summary['framework'].apply( lambda x: '/'.join(sorted(x)) ) ind_fw_summary = ind_fw_summary.merge( indicator_actual_start[['indicator_id', 'actual_start_year']], on='indicator_id', how='left' ) self.logger.info(f"\n Framework assignment per indikator:") self.logger.info(f" {'-'*90}") self.logger.info(f" {'ID':<5} {'Frameworks':<18} {'ActualStart':<13} {'Indicator Name'}") self.logger.info(f" {'-'*90}") for _, row in ind_fw_summary.sort_values( ['frameworks', 'actual_start_year', 'indicator_name'] ).iterrows(): self.logger.info( f" {int(row['indicator_id']):<5} {row['frameworks']:<18} " f"{int(row['actual_start_year']):<13} {row['indicator_name'][:48]}" ) # 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 [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" - [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 — " f"MDGs: {(self.df_clean['framework'] == 'MDGs').sum():,} rows | " f"SDGs: {(self.df_clean['framework'] == 'SDGs').sum():,} rows" ) return self.df_clean # ------------------------------------------------------------------ # STEP 7: VERIFY NO GAPS # ------------------------------------------------------------------ def verify_no_gaps(self): self.logger.info("\n" + "=" * 80) self.logger.info("STEP 7: VERIFY NO GAPS") self.logger.info("=" * 80) expected_countries = len(self.selected_country_ids) verification = self.df_clean.groupby( ['indicator_id', 'year'] )['country_id'].nunique().reset_index() verification.columns = ['indicator_id', 'year', 'country_count'] all_good = (verification['country_count'] == expected_countries).all() if all_good: self.logger.info( f" VERIFICATION PASSED — all combinations have {expected_countries} countries" ) else: bad = verification[verification['country_count'] != expected_countries] for _, row in bad.head(10).iterrows(): self.logger.error( f" Indicator {int(row['indicator_id'])}, Year {int(row['year'])}: " f"{int(row['country_count'])} countries (expected {expected_countries})" ) raise ValueError("Gap verification failed!") return True # ------------------------------------------------------------------ # STEP 8: CALCULATE NORM_VALUE_1_100 PER INDICATOR # ------------------------------------------------------------------ def calculate_norm_value(self): self.logger.info("\n" + "=" * 80) self.logger.info("STEP 8: CALCULATE NORM_VALUE_1_100 PER INDICATOR") self.logger.info("=" * 80) DIRECTION_INVERT = frozenset({ "negative", "lower_better", "lower_is_better", "inverse", "neg", }) df = self.df_clean.copy() norm_parts = [] indicators = df.groupby(['indicator_id', 'indicator_name', 'direction']) 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: grp = grp.copy() do_invert = str(direction).lower().strip() in DIRECTION_INVERT valid_mask = grp['value'].notna() n_valid = valid_mask.sum() if n_valid < 2: grp['norm_value_1_100'] = np.nan norm_parts.append(grp) continue raw = grp.loc[valid_mask, 'value'].values v_min = raw.min() v_max = raw.max() normed = np.full(len(grp), np.nan) if v_min == v_max: normed[valid_mask.values] = 50.5 else: scaled = (raw - v_min) / (v_max - v_min) if do_invert: scaled = 1.0 - scaled normed[valid_mask.values] = 1.0 + scaled * 99.0 grp['norm_value_1_100'] = normed self.logger.info( f" {int(ind_id):<5} {direction:<15} {'YES' if do_invert else 'no':<8} " f"{v_min:>10.3f} {v_max:>10.3f} {ind_name[:45]}" ) norm_parts.append(grp) self.df_clean = pd.concat(norm_parts, ignore_index=True) 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:,}") self.logger.info( f" Range aktual: " f"{self.df_clean['norm_value_1_100'].min():.2f} - " f"{self.df_clean['norm_value_1_100'].max():.2f}" ) 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 " 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") return self.df_clean # ------------------------------------------------------------------ # STEP 9: CALCULATE YOY # ------------------------------------------------------------------ def calculate_yoy(self): self.logger.info("\n" + "=" * 80) self.logger.info("STEP 9: CALCULATE YEAR-OVER-YEAR (YoY) PER INDICATOR PER COUNTRY") self.logger.info("=" * 80) df = self.df_clean.sort_values(['country_id', 'indicator_id', 'year']).copy() df['value_prev'] = df.groupby(['country_id', 'indicator_id'])['value'].shift(1) df['yoy_change'] = df['value'] - df['value_prev'] df['yoy_pct'] = np.where( df['value_prev'].notna() & (df['value_prev'] != 0), (df['yoy_change'] / df['value_prev'].abs()) * 100, np.nan ) df = df.drop(columns=['value_prev']) total_rows = len(df) valid_yoy = df['yoy_pct'].notna().sum() null_yoy = df['yoy_pct'].isna().sum() self.logger.info(f" Total rows : {total_rows:,}") self.logger.info(f" YoY calculated : {valid_yoy:,}") self.logger.info(f" YoY NULL (base yr): {null_yoy:,}") self.df_clean = df self.logger.info(f" [OK] Kolom 'yoy_change', 'yoy_pct' ditambahkan") return self.df_clean # ------------------------------------------------------------------ # STEP 10: ANALYZE INDICATOR AVAILABILITY BY YEAR # ------------------------------------------------------------------ def analyze_indicator_availability_by_year(self): self.logger.info("\n" + "=" * 80) self.logger.info("STEP 10: ANALYZE INDICATOR AVAILABILITY BY YEAR") self.logger.info("=" * 80) year_stats = self.df_clean.groupby('year').agg({ 'indicator_id': 'nunique', 'country_id' : 'nunique' }).reset_index() year_stats.columns = ['year', 'indicator_count', 'country_count'] self.logger.info(f"\n{'Year':<8} {'Indicators':<15} {'Countries':<12} {'Rows'}") self.logger.info("-" * 50) for _, row in year_stats.iterrows(): year = int(row['year']) row_count = len(self.df_clean[self.df_clean['year'] == year]) self.logger.info( f"{year:<8} {int(row['indicator_count']):<15} " f"{int(row['country_count']):<12} {row_count:,}" ) indicator_details = self.df_clean.groupby([ 'indicator_id', 'indicator_name', 'pillar_name', 'direction' ]).agg({'year': ['min', 'max'], 'country_id': 'nunique'}).reset_index() indicator_details.columns = [ 'indicator_id', 'indicator_name', 'pillar_name', 'direction', 'start_year', 'end_year', 'country_count' ] ind_fw = ( self.df_clean .groupby('indicator_id')['framework'] .unique() .reset_index() ) ind_fw['framework_label'] = ind_fw['framework'].apply( lambda x: '/'.join(sorted(x)) ) indicator_details = indicator_details.merge( ind_fw[['indicator_id', 'framework_label']], on='indicator_id', how='left' ) indicator_details['year_range'] = ( indicator_details['start_year'].astype(int).astype(str) + '-' + indicator_details['end_year'].astype(int).astype(str) ) indicator_details = indicator_details.sort_values( ['framework_label', 'pillar_name', 'start_year', 'indicator_name'] ) self.logger.info(f"\nTotal Indicators: {len(indicator_details)}") self.logger.info(f"Framework breakdown (per indicator label):") for fw, count in indicator_details.groupby('framework_label').size().items(): self.logger.info(f" {fw}: {count} indicators") self.logger.info(f"\n{'-'*115}") self.logger.info( f"{'ID':<5} {'Indicator Name':<55} {'Pillar':<15} " f"{'Framework':<15} {'Years':<12} {'Dir':<8} {'Countries'}" ) self.logger.info(f"{'-'*115}") for _, row in indicator_details.iterrows(): direction = 'higher+' if row['direction'] == 'higher_better' else 'lower-' self.logger.info( f"{int(row['indicator_id']):<5} {row['indicator_name'][:52]:<55} " f"{row['pillar_name'][:13]:<15} {row['framework_label']:<15} " f"{row['year_range']:<12} {direction:<8} {int(row['country_count'])}" ) return year_stats # ------------------------------------------------------------------ # STEP 11: SAVE ANALYTICAL TABLE # ------------------------------------------------------------------ def save_analytical_table(self): table_name = 'fact_asean_food_security_selected' self.logger.info("\n" + "=" * 80) self.logger.info(f"STEP 11: SAVE TO [DW/Gold] {table_name} -> fs_asean_gold") self.logger.info("=" * 80) try: if 'framework' not in self.df_clean.columns: raise ValueError("Kolom 'framework' tidak ada. Pastikan Step 6 sudah dijalankan.") if 'norm_value_1_100' not in self.df_clean.columns: raise ValueError("Kolom 'norm_value_1_100' tidak ada. Pastikan Step 8 sudah dijalankan.") if 'yoy_change' not in self.df_clean.columns: raise ValueError("Kolom 'yoy_change' tidak ada. Pastikan Step 9 sudah dijalankan.") analytical_df = self.df_clean[[ 'country_id', 'country_name', 'indicator_id', 'indicator_name', 'direction', 'framework', 'pillar_id', 'pillar_name', 'time_id', 'year', 'value', 'norm_value_1_100', 'yoy_change', 'yoy_pct', ]].copy() analytical_df = analytical_df.sort_values( ['year', 'country_name', 'pillar_name', 'indicator_name'] ).reset_index(drop=True) analytical_df['country_id'] = analytical_df['country_id'].astype(int) analytical_df['country_name'] = analytical_df['country_name'].astype(str) analytical_df['indicator_id'] = analytical_df['indicator_id'].astype(int) analytical_df['indicator_name'] = analytical_df['indicator_name'].astype(str) analytical_df['direction'] = analytical_df['direction'].astype(str) analytical_df['framework'] = analytical_df['framework'].astype(str) analytical_df['pillar_id'] = analytical_df['pillar_id'].astype(int) analytical_df['pillar_name'] = analytical_df['pillar_name'].astype(str) analytical_df['time_id'] = analytical_df['time_id'].astype(int) analytical_df['year'] = analytical_df['year'].astype(int) analytical_df['value'] = analytical_df['value'].astype(float) analytical_df['norm_value_1_100'] = analytical_df['norm_value_1_100'].astype(float) analytical_df['yoy_change'] = analytical_df['yoy_change'].astype(float) analytical_df['yoy_pct'] = analytical_df['yoy_pct'].astype(float) self.logger.info(f" Total rows: {len(analytical_df):,}") 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") ind_fw_label = ( analytical_df .groupby('indicator_id')['framework'] .unique() .apply(lambda x: '/'.join(sorted(x))) .value_counts() ) self.logger.info(f" Framework distribution (per indicator label):") for fw, cnt in ind_fw_label.items(): self.logger.info(f" {fw}: {cnt} indicators") self.logger.info( f" norm_value_1_100 range: " f"{analytical_df['norm_value_1_100'].min():.2f} - " f"{analytical_df['norm_value_1_100'].max():.2f}" ) schema = [ bigquery.SchemaField("country_id", "INTEGER", mode="REQUIRED"), bigquery.SchemaField("country_name", "STRING", mode="REQUIRED"), bigquery.SchemaField("indicator_id", "INTEGER", mode="REQUIRED"), bigquery.SchemaField("indicator_name", "STRING", mode="REQUIRED"), bigquery.SchemaField("direction", "STRING", mode="REQUIRED"), bigquery.SchemaField("framework", "STRING", mode="REQUIRED"), bigquery.SchemaField("pillar_id", "INTEGER", mode="REQUIRED"), bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"), bigquery.SchemaField("time_id", "INTEGER", mode="REQUIRED"), bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"), bigquery.SchemaField("value", "FLOAT", mode="REQUIRED"), bigquery.SchemaField("norm_value_1_100", "FLOAT", mode="NULLABLE"), bigquery.SchemaField("yoy_change", "FLOAT", mode="NULLABLE"), bigquery.SchemaField("yoy_pct", "FLOAT", mode="NULLABLE"), ] rows_loaded = load_to_bigquery( self.client, analytical_df, table_name, layer='gold', write_disposition="WRITE_TRUNCATE", schema=schema ) self.pipeline_metadata['rows_loaded'] = rows_loaded log_update(self.client, 'DW', table_name, 'full_load', rows_loaded) metadata = { 'source_class' : self.__class__.__name__, 'table_name' : table_name, 'execution_timestamp': self.pipeline_start, 'duration_seconds' : (datetime.now() - self.pipeline_start).total_seconds(), 'rows_fetched' : self.pipeline_metadata['rows_fetched'], 'rows_transformed' : rows_loaded, 'rows_loaded' : rows_loaded, 'completeness_pct' : 100.0, 'config_snapshot' : json.dumps({ 'start_year' : self.start_year, 'end_year' : self.end_year, 'baseline_year' : self.baseline_year, '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' : ( 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}', 'moderate': f'{THRESHOLD_BAD}-{THRESHOLD_GOOD}', 'good' : f'> {THRESHOLD_GOOD}', }, }), 'validation_metrics' : json.dumps({ 'fixed_countries' : len(self.selected_country_ids), 'total_indicators' : int(self.df_clean['indicator_id'].nunique()), 'sdg_start_year' : self.sdg_start_year, 'framework_dist_rows' : fw_dist_rows.to_dict(), 'framework_dist_inds' : ind_fw_label.to_dict(), }) } save_etl_metadata(self.client, metadata) self.logger.info(f" [OK] {table_name}: {rows_loaded:,} rows -> fs_asean_gold") return rows_loaded except Exception as e: self.logger.error(f"Error saving: {e}") raise # ------------------------------------------------------------------ # RUN # ------------------------------------------------------------------ def run(self): self.pipeline_start = datetime.now() self.pipeline_metadata['start_time'] = self.pipeline_start self.logger.info("\n" + "=" * 80) self.logger.info("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, 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) self.load_source_data() self.determine_year_boundaries() self.filter_complete_indicators_per_country() self.select_countries_with_all_pillars() self.filter_indicators_consistent_across_fixed_countries() self.determine_sdg_start_year() self.verify_no_gaps() self.calculate_norm_value() self.calculate_yoy() self.analyze_indicator_availability_by_year() self.save_analytical_table() self.pipeline_end = datetime.now() duration = (self.pipeline_end - self.pipeline_start).total_seconds() self.logger.info("\n" + "=" * 80) self.logger.info("COMPLETED") self.logger.info("=" * 80) self.logger.info(f" Duration : {duration:.2f}s") self.logger.info(f" Year Range : {self.start_year}-{self.end_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']:,}") # ============================================================================= # AIRFLOW TASK FUNCTION # ============================================================================= def run_analytical_layer(): from scripts.bigquery_config import get_bigquery_client client = get_bigquery_client() loader = AnalyticalLayerLoader(client) loader.run() print(f"Analytical layer loaded: {loader.pipeline_metadata['rows_loaded']:,} rows") # ============================================================================= # MAIN EXECUTION # ============================================================================= if __name__ == "__main__": print("=" * 80) 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( "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) logger = setup_logging() client = get_bigquery_client() loader = AnalyticalLayerLoader(client) loader.run() print("\n" + "=" * 80) print("[OK] COMPLETED") print(f" SDG Start Year : {loader.sdg_start_year}") print(f" Rows Loaded : {loader.pipeline_metadata['rows_loaded']:,}") print("=" * 80)