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

add condition at framework and edit narrative indicator

Browse Source
This commit is contained in:
Debby
2026-07-16 09:57:16 +07:00
parent 989d3445a0
commit d21fac69aa
2 changed files with 239 additions and 311 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/bigquery_aggraget_fact_selected_layer.py
+183 -308
View File
@@ -435,47 +435,41 @@ def _detect_consistency(df_ind: pd.DataFrame, lower_better: bool) -> tuple:
# =============================================================================
# NARRATIVE BUILDER
# NARRATIVE BUILDER — PER INDICATOR PER YEAR (1 pillar, 1 tahun)
# =============================================================================
#
# Granularity agg_narrative_indicator SEKARANG per (indicator_id, year), bukan
# lagi per indicator_id gabungan seluruh tahun. Setiap baris narasi menjelaskan
# posisi 1 indikator dalam 1 pillar, pada 1 tahun tertentu: nilai regional,
# skor ternormalisasi, peringkat indikator tsb di antara indikator lain dalam
# pillar yang sama pada tahun yang sama, perubahan YoY, serta negara terbaik/
# terlemah untuk indikator tsb pada tahun tersebut.
# =============================================================================
def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tuple:
ind_id = int(row["indicator_id"])
def _build_narrative_per_indicator_year(row: pd.Series) -> tuple:
"""
row diharapkan datang dari baris ASEAN (country_id=0) pada agg_indicator_norm,
sudah digabung dengan kolom rank_in_pillar_year, n_indicators_in_pillar_year,
country_best, country_worst (dihitung dari negara asli untuk indicator+year yang sama).
"""
ind_name_en = str(row["indicator_name"]).strip()
ind_name_id = str(row.get("indicator_name_id", ind_name_en)).strip()
unit = str(row["unit"]).strip() if row["unit"] else ""
direction = str(row["direction"]).strip()
pillar_en = str(row["pillar_name"]).strip()
pillar_id_ = get_pillar_name_id(pillar_en)
framework = str(row["framework"]).strip()
year_min = int(row["year_min"])
year_max = int(row["year_max"])
lower_better = _is_lower_better(direction)
year = int(row["year"])
# Gunakan hanya negara asli (bukan ASEAN) untuk analisa tren/gap/konsistensi
df_ind = df_full[
(df_full["indicator_id"] == ind_id) &
(df_full["country_id"] != ASEAN_COUNTRY_ID)
].copy()
value = row.get("value", np.nan)
score = row.get("norm_score_1_100", np.nan)
yoy = row.get("yoy_value", np.nan)
rank = row.get("rank_in_pillar_year", np.nan)
n_ind = row.get("n_indicators_in_pillar_year", np.nan)
if df_ind.empty:
na_en = f"{ind_name_en} ({framework}, {pillar_en}): Insufficient data for analysis."
na_id = f"{ind_name_id} ({framework}, {pillar_id_}): Data tidak cukup untuk dianalisis."
return na_en, na_id
asean_avg_by_year = (
df_ind.groupby("year")["value"].mean().dropna()
)
trend_label = _detect_trend(asean_avg_by_year, lower_better)
gap_label = _detect_gap_trend(df_ind, lower_better)
anomaly_year, anomaly_dir = _detect_anomaly_year(asean_avg_by_year)
best_country_en, worst_country_en, is_consistent = _detect_consistency(df_ind, lower_better)
best_country_id = get_country_name_id(best_country_en) if best_country_en else None
worst_country_id = get_country_name_id(worst_country_en) if worst_country_en else None
avg_first = row.get("avg_value_first", np.nan)
avg_last = row.get("avg_value_last", np.nan)
best_country_en = row.get("country_best")
worst_country_en = row.get("country_worst")
best_country_id = row.get("country_best_id")
worst_country_id = row.get("country_worst_id")
def fmt(v):
if pd.isna(v):
@@ -487,65 +481,65 @@ def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tup
sentences_en = []
sentences_id = []
s1_en = f"{ind_name_en} ({framework}, {pillar_en}, {year_min}-{year_max}):"
s1_id = f"{ind_name_id} ({framework}, {pillar_id_}, {year_min}-{year_max}):"
perf_word_en = "good" if (pd.notna(score) and score >= _PERFORMANCE_THRESHOLD) else "below target"
perf_word_id = "baik" if (pd.notna(score) and score >= _PERFORMANCE_THRESHOLD) else "di bawah target"
if pd.notna(rank) and pd.notna(n_ind) and int(n_ind) > 0:
rank_i = int(rank)
n_i = int(n_ind)
suffix = {1: "st", 2: "nd", 3: "rd"}.get(rank_i, "th")
s1_en = (
f"In {year}, {ind_name_en} ({framework}, {pillar_en}) recorded a regional average of "
f"{fmt(value)}, ranking {rank_i}{suffix} out of {n_i} indicators within the {pillar_en} "
f"pillar for that year, with a normalized score of {fmt(score)} ({perf_word_en})."
)
s1_id = (
f"Pada tahun {year}, {ind_name_id} ({framework}, {pillar_id_}) mencatat rata-rata "
f"regional sebesar {fmt(value)}, menempati peringkat {rank_i} dari {n_i} indikator "
f"dalam pilar {pillar_id_} pada tahun tersebut, dengan skor ternormalisasi "
f"{fmt(score)} ({perf_word_id})."
)
else:
s1_en = (
f"{ind_name_en} ({framework}, {pillar_en}, {year}): regional average was {fmt(value)}, "
f"with a normalized score of {fmt(score)} ({perf_word_en})."
)
s1_id = (
f"{ind_name_id} ({framework}, {pillar_id_}, {year}): rata-rata regional sebesar "
f"{fmt(value)}, dengan skor ternormalisasi {fmt(score)} ({perf_word_id})."
)
sentences_en.append(s1_en)
sentences_id.append(s1_id)
trend_map_en = {
"improving_consistent": f"Regional average improved consistently from {fmt(avg_first)} to {fmt(avg_last)}.",
"improving_slowing": f"Regional average improved from {fmt(avg_first)} to {fmt(avg_last)}, though the pace slowed in recent years.",
"deteriorating": f"Regional average worsened from {fmt(avg_first)} to {fmt(avg_last)} over the period.",
"fluctuating": f"Regional average fluctuated between {fmt(avg_first)} and {fmt(avg_last)} with no clear trend.",
"insufficient_data": f"Trend analysis is limited due to sparse data.",
}
trend_map_id = {
"improving_consistent": f"Rata-rata regional membaik secara konsisten dari {fmt(avg_first)} menjadi {fmt(avg_last)}.",
"improving_slowing": f"Rata-rata regional membaik dari {fmt(avg_first)} menjadi {fmt(avg_last)}, namun lajunya melambat dalam beberapa tahun terakhir.",
"deteriorating": f"Rata-rata regional memburuk dari {fmt(avg_first)} menjadi {fmt(avg_last)} sepanjang periode.",
"fluctuating": f"Rata-rata regional berfluktuasi antara {fmt(avg_first)} dan {fmt(avg_last)} tanpa tren yang jelas.",
"insufficient_data": f"Analisis tren terbatas karena data yang tersedia tidak cukup.",
}
sentences_en.append(trend_map_en.get(trend_label, ""))
sentences_id.append(trend_map_id.get(trend_label, ""))
if gap_label == "widening":
sentences_en.append("Disparity among ASEAN countries has widened over time, indicating unequal progress.")
sentences_id.append("Kesenjangan antar negara ASEAN melebar seiring waktu, menunjukkan kemajuan yang tidak merata.")
elif gap_label == "narrowing":
sentences_en.append("Disparity among ASEAN countries has narrowed, suggesting more balanced regional progress.")
sentences_id.append("Kesenjangan antar negara ASEAN menyempit, mengindikasikan kemajuan regional yang lebih merata.")
elif gap_label == "stable":
sentences_en.append("The gap among ASEAN countries remained relatively stable throughout the period.")
sentences_id.append("Kesenjangan antar negara ASEAN relatif stabil sepanjang periode.")
if anomaly_year is not None:
if anomaly_dir == "drop":
sentences_en.append(f"A notable decline was recorded in {anomaly_year}, which stood out from the overall pattern.")
sentences_id.append(f"Penurunan signifikan tercatat pada tahun {anomaly_year}, yang menyimpang dari pola keseluruhan.")
elif anomaly_dir == "rise":
sentences_en.append(f"A sharp improvement was observed in {anomaly_year}, standing out from the overall pattern.")
sentences_id.append(f"Peningkatan tajam tercatat pada tahun {anomaly_year}, yang menyimpang dari pola keseluruhan.")
if best_country_en and worst_country_en:
if is_consistent:
sentences_en.append(
f"{best_country_en} consistently performed above the regional average, "
f"while {worst_country_en} consistently lagged behind."
)
sentences_id.append(
f"{best_country_id} secara konsisten berada di atas rata-rata regional, "
f"sementara {worst_country_id} secara konsisten tertinggal."
)
if pd.notna(yoy):
if abs(yoy) < 1e-9:
s2_en = "This value was unchanged compared to the previous year."
s2_id = "Nilai ini tidak berubah dibandingkan tahun sebelumnya."
elif yoy > 0:
s2_en = f"This represents an increase of {fmt(abs(yoy))} from the previous year."
s2_id = f"Ini menunjukkan kenaikan sebesar {fmt(abs(yoy))} dari tahun sebelumnya."
else:
sentences_en.append(
f"Overall, {best_country_en} showed the best performance, "
f"while {worst_country_en} had the weakest results across the period."
)
sentences_id.append(
f"Secara keseluruhan, {best_country_id} menunjukkan performa terbaik, "
f"sementara {worst_country_id} memiliki hasil terlemah sepanjang periode."
)
s2_en = f"This represents a decrease of {fmt(abs(yoy))} from the previous year."
s2_id = f"Ini menunjukkan penurunan sebesar {fmt(abs(yoy))} dari tahun sebelumnya."
sentences_en.append(s2_en)
sentences_id.append(s2_id)
else:
s2_en = "Year-over-year comparison is not available for this year."
s2_id = "Perbandingan tahun-ke-tahun tidak tersedia untuk tahun ini."
sentences_en.append(s2_en)
sentences_id.append(s2_id)
if best_country_en and worst_country_en and best_country_en != worst_country_en:
s3_en = (
f"Among ASEAN countries in {year}, {best_country_en} recorded the best performance "
f"for this indicator, while {worst_country_en} recorded the weakest."
)
s3_id = (
f"Di antara negara ASEAN pada tahun {year}, {best_country_id} mencatat performa "
f"terbaik untuk indikator ini, sementara {worst_country_id} mencatat performa terlemah."
)
sentences_en.append(s3_en)
sentences_id.append(s3_id)
narrative_en = " ".join(s for s in sentences_en if s)
narrative_id = " ".join(s for s in sentences_id if s)
@@ -959,259 +953,138 @@ class IndicatorNormAggregator:
return rows_loaded
# =========================================================================
# STEP 11: agg_narrative_indicator (per indicator, ASEAN summary sebagai kolom)
# STEP 11: agg_narrative_indicator (per indicator_id PER YEAR, 1 pillar 1 tahun)
# =========================================================================
def _build_narrative_table(self, df_final: pd.DataFrame):
self.logger.info("\n" + "=" * 80)
self.logger.info("STEP 11: agg_narrative_indicator")
self.logger.info(" Granularity: per indicator_id")
self.logger.info(" ASEAN data: digunakan untuk asean_avg_value_first/last")
self.logger.info(" Granularity: per indicator_id PER YEAR (1 pillar, 1 tahun)")
self.logger.info(" Narasi menjelaskan posisi indikator dalam pillar-nya, per tahun")
self.logger.info("=" * 80)
# Negara asli saja untuk analisa statistik
df_real = df_final[df_final["country_id"] != ASEAN_COUNTRY_ID]
df_asean = df_final[df_final["country_id"] == ASEAN_COUNTRY_ID]
# Negara asli saja untuk analisa negara terbaik/terlemah per tahun
df_real = df_final[df_final["country_id"] != ASEAN_COUNTRY_ID].copy()
# Baris ASEAN (regional average) menjadi basis nilai per indicator per year
df_asean = df_final[df_final["country_id"] == ASEAN_COUNTRY_ID].copy()
# ---- Statistik per indikator (negara asli) ----
df_yr = (
if df_asean.empty:
self.logger.warning(" [WARNING] Tidak ada baris ASEAN; agg_narrative_indicator kosong.")
df_asean = df_final.copy()
# ---- Rank indikator dalam pillar yang sama, pada tahun yang sama ----
# (dihitung dari norm_score_1_100 baris ASEAN/regional; skor sudah
# searah -- semakin tinggi semakin baik -- untuk semua indikator)
df_asean["rank_in_pillar_year"] = (
df_asean.groupby(["pillar_id", "year"])["norm_score_1_100"]
.rank(method="min", ascending=False)
)
df_asean["n_indicators_in_pillar_year"] = (
df_asean.groupby(["pillar_id", "year"])["indicator_id"]
.transform("nunique")
)
# ---- Negara terbaik / terlemah per indikator, per tahun (negara asli) ----
def _best_worst(g: pd.DataFrame) -> pd.Series:
g_valid = g[g["norm_score_1_100"].notna()]
if g_valid.empty:
return pd.Series({"country_best": None, "country_worst": None})
best_row = g_valid.loc[g_valid["norm_score_1_100"].idxmax()]
worst_row = g_valid.loc[g_valid["norm_score_1_100"].idxmin()]
return pd.Series({
"country_best" : best_row["country_name"],
"country_worst": worst_row["country_name"],
})
country_stats = (
df_real.groupby(["indicator_id", "year"])
.agg(
avg_value =("value", "mean"),
avg_norm_score =("norm_score_1_100", "mean"),
n_countries_yr =("country_id", "nunique"),
)
.apply(_best_worst)
.reset_index()
)
df_first = (
df_yr.sort_values("year").groupby("indicator_id").first().reset_index()
[["indicator_id", "year", "avg_value"]]
.rename(columns={"year": "year_min", "avg_value": "avg_value_first"})
country_stats["country_best_id"] = country_stats["country_best"].apply(
lambda x: get_country_name_id(x) if pd.notna(x) and x is not None else None
)
df_last = (
df_yr.sort_values("year").groupby("indicator_id").last().reset_index()
[["indicator_id", "year", "avg_value"]]
.rename(columns={"year": "year_max", "avg_value": "avg_value_last"})
)
df_score_avg = (
df_yr.groupby("indicator_id")
.agg(avg_norm_score_1_100=("avg_norm_score", "mean"))
.reset_index()
)
df_nc = (
df_real.groupby("indicator_id")["country_id"]
.nunique().reset_index()
.rename(columns={"country_id": "n_countries"})
country_stats["country_worst_id"] = country_stats["country_worst"].apply(
lambda x: get_country_name_id(x) if pd.notna(x) and x is not None else None
)
# ASEAN avg per indikator
df_asean_yr = (
df_asean.groupby(["indicator_id", "year"])
.agg(asean_avg_value=("value", "mean"))
.reset_index()
)
df_asean_first = (
df_asean_yr.sort_values("year").groupby("indicator_id").first().reset_index()
[["indicator_id", "asean_avg_value"]]
.rename(columns={"asean_avg_value": "asean_avg_value_first"})
)
df_asean_last = (
df_asean_yr.sort_values("year").groupby("indicator_id").last().reset_index()
[["indicator_id", "asean_avg_value"]]
.rename(columns={"asean_avg_value": "asean_avg_value_last"})
)
# ---- Gabung ----
df_agg = df_asean.merge(country_stats, on=["indicator_id", "year"], how="left")
# YoY stats (negara asli)
dir_map = (
df_real[["indicator_id", "direction"]]
.drop_duplicates(subset=["indicator_id"])
.set_index("indicator_id")["direction"]
.to_dict()
)
yoy_parts = []
for ind_id, grp in df_yr.groupby("indicator_id"):
grp = grp.sort_values("year").copy()
grp["prev_avg"] = grp["avg_value"].shift(1)
grp["yoy"] = np.where(
grp["avg_value"].notna() & grp["prev_avg"].notna(),
grp["avg_value"] - grp["prev_avg"],
np.nan,
)
grp = grp.drop(columns=["prev_avg"])
yoy_parts.append(grp)
df_yr = pd.concat(yoy_parts, ignore_index=True)
def _is_positive_yoy(ind_id, yoy_val):
if pd.isna(yoy_val):
return False
lb = _is_lower_better(dir_map.get(ind_id, "positive"))
return (yoy_val < 0) if lb else (yoy_val > 0)
yoy_stats = []
for ind_id, grp in df_yr.groupby("indicator_id"):
grp_yoy = grp[grp["yoy"].notna()].copy()
lb = _is_lower_better(dir_map.get(ind_id, "positive"))
n_total = len(grp_yoy)
n_positive = int(sum(_is_positive_yoy(ind_id, v) for v in grp_yoy["yoy"]))
if n_total > 0:
idx_best = grp_yoy["yoy"].idxmin() if lb else grp_yoy["yoy"].idxmax()
best_row = grp_yoy.loc[idx_best]
best_yoy_from = best_row["year"] - 1
best_yoy_to = best_row["year"]
else:
best_yoy_from = np.nan
best_yoy_to = np.nan
yoy_stats.append({
"indicator_id" : ind_id,
"n_yoy_total" : n_total,
"n_yoy_positive": n_positive,
"best_yoy_from" : best_yoy_from,
"best_yoy_to" : best_yoy_to,
})
df_yoy_stats = pd.DataFrame(yoy_stats)
# Country best/worst
df_country_avg = (
df_real.groupby(["indicator_id", "country_id", "country_name"])
.agg(country_avg_value=("value", "mean"))
.reset_index()
)
country_stats = []
for ind_id, grp in df_country_avg.groupby("indicator_id"):
lb = _is_lower_better(dir_map.get(ind_id, "positive"))
if lb:
worst_row = grp.loc[grp["country_avg_value"].idxmax()]
best_row = grp.loc[grp["country_avg_value"].idxmin()]
else:
worst_row = grp.loc[grp["country_avg_value"].idxmin()]
best_row = grp.loc[grp["country_avg_value"].idxmax()]
country_stats.append({
"indicator_id" : ind_id,
"country_worst" : worst_row["country_name"],
"country_best" : best_row["country_name"],
"country_worst_id": get_country_name_id(worst_row["country_name"]),
"country_best_id" : get_country_name_id(best_row["country_name"]),
})
df_country_stats = pd.DataFrame(country_stats)
# Dim cols
dim_cols = [
"indicator_name", "indicator_name_id",
"unit", "direction",
"pillar_name", "pillar_name_id",
"framework",
]
df_dim = df_real[["indicator_id"] + dim_cols].drop_duplicates(subset=["indicator_id"])
# Merge semua
df_agg = (
df_dim
.merge(df_first, on="indicator_id", how="left")
.merge(df_last, on="indicator_id", how="left")
.merge(df_score_avg, on="indicator_id", how="left")
.merge(df_nc, on="indicator_id", how="left")
.merge(df_yoy_stats, on="indicator_id", how="left")
.merge(df_country_stats, on="indicator_id", how="left")
.merge(df_asean_first, on="indicator_id", how="left")
.merge(df_asean_last, on="indicator_id", how="left")
)
# Performance
df_agg["performance"] = pd.NA
has_score = df_agg["avg_norm_score_1_100"].notna()
df_agg.loc[has_score & (df_agg["avg_norm_score_1_100"] >= _PERFORMANCE_THRESHOLD), "performance"] = "Good"
df_agg.loc[has_score & (df_agg["avg_norm_score_1_100"] < _PERFORMANCE_THRESHOLD), "performance"] = "Bad"
# Build narrative
# ---- Build narrative per baris (per indicator_id per year) ----
narratives_en = []
narratives_id = []
for _, row in df_agg.iterrows():
n_en, n_id = _build_narrative_per_indicator(row, df_final)
n_en, n_id = _build_narrative_per_indicator_year(row)
narratives_en.append(n_en)
narratives_id.append(n_id)
df_agg["narrative_en"] = narratives_en
df_agg["narrative_id"] = narratives_id
# Output
# ---- Output ----
out = df_agg[[
"year",
"indicator_id", "indicator_name", "indicator_name_id",
"unit", "direction",
"pillar_name", "pillar_name_id",
"pillar_id", "pillar_name", "pillar_name_id",
"framework",
"year_min", "year_max", "n_countries",
"avg_value_first", "avg_value_last",
"asean_avg_value_first", "asean_avg_value_last",
"avg_norm_score_1_100", "performance",
"n_yoy_total", "n_yoy_positive",
"best_yoy_from", "best_yoy_to",
"country_worst", "country_best",
"country_worst_id", "country_best_id",
"value", "norm_score_1_100", "performance",
"yoy_value",
"rank_in_pillar_year", "n_indicators_in_pillar_year",
"country_best", "country_worst",
"country_best_id", "country_worst_id",
"narrative_en", "narrative_id",
]].copy()
out = out.sort_values(["pillar_name", "indicator_name"]).reset_index(drop=True)
out = out.sort_values(["year", "pillar_name", "rank_in_pillar_year", "indicator_name"]).reset_index(drop=True)
out["indicator_id"] = out["indicator_id"].astype(int)
out["indicator_name"] = out["indicator_name"].astype(str)
out["indicator_name_id"] = out["indicator_name_id"].astype(str)
out["unit"] = out["unit"].fillna("").astype(str)
out["direction"] = out["direction"].astype(str)
out["pillar_name"] = out["pillar_name"].astype(str)
out["pillar_name_id"] = out["pillar_name_id"].astype(str)
out["framework"] = out["framework"].astype(str)
out["year_min"] = out["year_min"].astype(int)
out["year_max"] = out["year_max"].astype(int)
out["n_countries"] = out["n_countries"].astype(int)
out["avg_value_first"] = pd.to_numeric(out["avg_value_first"], errors="coerce").astype(float)
out["avg_value_last"] = pd.to_numeric(out["avg_value_last"], errors="coerce").astype(float)
out["asean_avg_value_first"]= pd.to_numeric(out["asean_avg_value_first"], errors="coerce").astype(float)
out["asean_avg_value_last"] = pd.to_numeric(out["asean_avg_value_last"], errors="coerce").astype(float)
out["avg_norm_score_1_100"] = pd.to_numeric(out["avg_norm_score_1_100"], errors="coerce").astype(float)
out["performance"] = out["performance"].astype(str).replace("nan", pd.NA).astype("string")
out["n_yoy_total"] = pd.to_numeric(out["n_yoy_total"], errors="coerce").astype("Int64")
out["n_yoy_positive"] = pd.to_numeric(out["n_yoy_positive"], errors="coerce").astype("Int64")
out["best_yoy_from"] = pd.to_numeric(out["best_yoy_from"], errors="coerce").astype("Int64")
out["best_yoy_to"] = pd.to_numeric(out["best_yoy_to"], errors="coerce").astype("Int64")
out["country_worst"] = out["country_worst"].astype(str).replace("nan", pd.NA).astype("string")
out["country_best"] = out["country_best"].astype(str).replace("nan", pd.NA).astype("string")
out["country_worst_id"] = out["country_worst_id"].astype(str).replace("nan", pd.NA).astype("string")
out["country_best_id"] = out["country_best_id"].astype(str).replace("nan", pd.NA).astype("string")
out["narrative_en"] = out["narrative_en"].astype(str)
out["narrative_id"] = out["narrative_id"].astype(str)
out["year"] = out["year"].astype(int)
out["indicator_id"] = out["indicator_id"].astype(int)
out["indicator_name"] = out["indicator_name"].astype(str)
out["indicator_name_id"] = out["indicator_name_id"].astype(str)
out["unit"] = out["unit"].fillna("").astype(str)
out["direction"] = out["direction"].astype(str)
out["pillar_id"] = out["pillar_id"].astype(int)
out["pillar_name"] = out["pillar_name"].astype(str)
out["pillar_name_id"] = out["pillar_name_id"].astype(str)
out["framework"] = out["framework"].astype(str)
out["value"] = pd.to_numeric(out["value"], errors="coerce").astype(float)
out["norm_score_1_100"] = pd.to_numeric(out["norm_score_1_100"], errors="coerce").astype(float)
out["performance"] = out["performance"].astype(str).replace("nan", pd.NA).astype("string")
out["yoy_value"] = pd.to_numeric(out["yoy_value"], errors="coerce").astype(float)
out["rank_in_pillar_year"] = pd.to_numeric(out["rank_in_pillar_year"], errors="coerce").astype("Int64")
out["n_indicators_in_pillar_year"] = pd.to_numeric(out["n_indicators_in_pillar_year"], errors="coerce").astype("Int64")
out["country_best"] = out["country_best"].astype(str).replace("nan", pd.NA).astype("string")
out["country_worst"] = out["country_worst"].astype(str).replace("nan", pd.NA).astype("string")
out["country_best_id"] = out["country_best_id"].astype(str).replace("nan", pd.NA).astype("string")
out["country_worst_id"] = out["country_worst_id"].astype(str).replace("nan", pd.NA).astype("string")
out["narrative_en"] = out["narrative_en"].astype(str)
out["narrative_id"] = out["narrative_id"].astype(str)
schema = [
bigquery.SchemaField("indicator_id", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("indicator_name", "STRING", mode="REQUIRED"),
bigquery.SchemaField("indicator_name_id", "STRING", mode="NULLABLE"),
bigquery.SchemaField("unit", "STRING", mode="NULLABLE"),
bigquery.SchemaField("direction", "STRING", mode="REQUIRED"),
bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"),
bigquery.SchemaField("pillar_name_id", "STRING", mode="NULLABLE"),
bigquery.SchemaField("framework", "STRING", mode="REQUIRED"),
bigquery.SchemaField("year_min", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("year_max", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("n_countries", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("avg_value_first", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("avg_value_last", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("asean_avg_value_first", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("asean_avg_value_last", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("avg_norm_score_1_100", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("performance", "STRING", mode="NULLABLE"),
bigquery.SchemaField("n_yoy_total", "INTEGER", mode="NULLABLE"),
bigquery.SchemaField("n_yoy_positive", "INTEGER", mode="NULLABLE"),
bigquery.SchemaField("best_yoy_from", "INTEGER", mode="NULLABLE"),
bigquery.SchemaField("best_yoy_to", "INTEGER", mode="NULLABLE"),
bigquery.SchemaField("country_worst", "STRING", mode="NULLABLE"),
bigquery.SchemaField("country_best", "STRING", mode="NULLABLE"),
bigquery.SchemaField("country_worst_id", "STRING", mode="NULLABLE"),
bigquery.SchemaField("country_best_id", "STRING", mode="NULLABLE"),
bigquery.SchemaField("narrative_en", "STRING", mode="NULLABLE"),
bigquery.SchemaField("narrative_id", "STRING", mode="NULLABLE"),
bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("indicator_id", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("indicator_name", "STRING", mode="REQUIRED"),
bigquery.SchemaField("indicator_name_id", "STRING", mode="NULLABLE"),
bigquery.SchemaField("unit", "STRING", mode="NULLABLE"),
bigquery.SchemaField("direction", "STRING", mode="REQUIRED"),
bigquery.SchemaField("pillar_id", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"),
bigquery.SchemaField("pillar_name_id", "STRING", mode="NULLABLE"),
bigquery.SchemaField("framework", "STRING", mode="REQUIRED"),
bigquery.SchemaField("value", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("norm_score_1_100", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("performance", "STRING", mode="NULLABLE"),
bigquery.SchemaField("yoy_value", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("rank_in_pillar_year", "INTEGER", mode="NULLABLE"),
bigquery.SchemaField("n_indicators_in_pillar_year", "INTEGER", mode="NULLABLE"),
bigquery.SchemaField("country_best", "STRING", mode="NULLABLE"),
bigquery.SchemaField("country_worst", "STRING", mode="NULLABLE"),
bigquery.SchemaField("country_best_id", "STRING", mode="NULLABLE"),
bigquery.SchemaField("country_worst_id", "STRING", mode="NULLABLE"),
bigquery.SchemaField("narrative_en", "STRING", mode="NULLABLE"),
bigquery.SchemaField("narrative_id", "STRING", mode="NULLABLE"),
]
rows_loaded = load_to_bigquery(
@@ -1234,15 +1107,16 @@ class IndicatorNormAggregator:
"rows_loaded" : rows_loaded,
"completeness_pct" : 100.0,
"config_snapshot" : json.dumps({
"granularity" : "indicator_id only",
"granularity" : "indicator_id x year (1 pillar, 1 tahun)",
"narrative_style" : "interpretive, plain text, bilingual EN/ID",
"asean_columns" : ["asean_avg_value_first", "asean_avg_value_last"],
"architecture" : "ASEAN rows included in agg_indicator_norm",
"rank_basis" : "norm_score_1_100 within same pillar_id and year",
"architecture" : "Built from ASEAN rows (country_id=0) in agg_indicator_norm",
"pillar_change" : "Renamed to Food Other; all pillars use 'Food ' prefix",
}),
"validation_metrics" : json.dumps({
"total_rows" : rows_loaded,
"n_indicators": int(out["indicator_id"].nunique()),
"n_years" : int(out["year"].nunique()),
}),
}
save_etl_metadata(self.client, metadata)
@@ -1260,6 +1134,7 @@ class IndicatorNormAggregator:
self.logger.info("INDICATOR NORM AGGREGATION")
self.logger.info(" ASEAN rows ditambahkan ke agg_indicator_norm (country_id=0)")
self.logger.info(" Rename Food Other; all pillars use Food prefix")
self.logger.info(" agg_narrative_indicator: granularity per indicator per year (1 pillar, 1 tahun)")
self.logger.info("=" * 80)
self.load_data()
scripts/bigquery_aggregate_layer.py
+56 -3
View File
@@ -25,6 +25,11 @@ KONDISI PILAR (pillar_condition_en / pillar_condition_id):
Kolom tambahan di agg_pillar_by_country untuk mendeskripsikan kondisi
tiap pilar per negara per tahun secara kontekstual dan kuantitatif.
KONDISI FRAMEWORK (framework_condition_en / framework_condition_id):
Kolom tambahan di agg_framework_by_country untuk mendeskripsikan kondisi
tiap framework (MDGs/SDGs/Total) per negara per tahun, mengikuti tier
dan referensi yang sama dengan kondisi pilar.
Landasan teori:
1. FAO & CFS (1996 World Food Summit; CFS Reform Document 2009):
Definisi 4 pilar ketahanan pangan dan makna substantif masing-masing.
@@ -997,7 +1002,10 @@ class FoodSecurityAggregator:
"asean_country_id" : ASEAN_COUNTRY_ID,
"pillar_change" : "Sustainability renamed to Food Other, all pillars prefixed with Food",
"architecture" : "ASEAN merged into country tables (country_id=0)",
"condition_column" : "pillar_condition_en/id added to agg_pillar_by_country",
"condition_column" : (
"pillar_condition_en/id added to agg_pillar_by_country; "
"framework_condition_en/id added to agg_framework_by_country"
),
"condition_reference" : (
"GFSI 2022 (Economist Impact) score tiers >= 75/60/40/20; "
"IPC Technical Manual 2019; FAO/CFS 4-pillar framework 1996/2009; "
@@ -1187,7 +1195,7 @@ class FoodSecurityAggregator:
return df
# =========================================================================
# STEP 3: agg_framework_by_country (termasuk ASEAN)
# STEP 3: agg_framework_by_country (termasuk ASEAN + kolom kondisi)
# =========================================================================
def calc_framework_by_country(self) -> pd.DataFrame:
@@ -1196,6 +1204,7 @@ class FoodSecurityAggregator:
self.logger.info("\n" + "=" * 70)
self.logger.info(f"STEP 3: {table_name} -> [Gold] fs_asean_gold")
self.logger.info(" Termasuk baris ASEAN (country_id=0)")
self.logger.info(" Kolom baru: framework_condition_en, framework_condition_id")
self.logger.info("=" * 70)
try:
@@ -1313,6 +1322,31 @@ class FoodSecurityAggregator:
df = check_and_dedup(df, ["country_id", "framework", "year"], context=table_name, logger=self.logger)
# -----------------------------------------------------------------
# TAMBAHAN: kolom kondisi framework (tier + deskripsi dipisah)
# Dibangkitkan SETELAH framework_score_1_100 tersedia dan SETELAH
# dedup, mengikuti pola yang sama dengan pillar_condition di
# agg_pillar_by_country. Referensi tier: GFSI 2022 (Economist
# Impact); IPC 2019; FAO/CFS 1996/2009; FAO SOFI 2024.
#
# Kolom yang dihasilkan:
# framework_condition_en — label tier EN, e.g. "At Risk"
# framework_condition_id — label tier ID, e.g. "Berisiko"
# framework_condition_desc_en — deskripsi kontekstual EN
# framework_condition_desc_id — deskripsi kontekstual ID
# -----------------------------------------------------------------
fw_conditions = df.apply(
lambda row: get_framework_condition(
row["framework"],
row["framework_score_1_100"]
),
axis=1
)
df["framework_condition_en"] = fw_conditions.apply(lambda x: x[0])
df["framework_condition_id"] = fw_conditions.apply(lambda x: x[1])
df["framework_condition_desc_en"] = fw_conditions.apply(lambda x: x[2])
df["framework_condition_desc_id"] = fw_conditions.apply(lambda x: x[3])
country_mask = df["country_id"] != ASEAN_COUNTRY_ID
df.loc[country_mask, "rank_in_framework_year"] = (
df[country_mask]
@@ -1331,6 +1365,18 @@ class FoodSecurityAggregator:
df["framework_norm"] = df["framework_norm"].astype(float)
df["framework_score_1_100"] = df["framework_score_1_100"].astype(float)
df["country_name_id"] = df["country_name_id"].astype(str)
df["framework_condition_en"] = df["framework_condition_en"].astype(str)
df["framework_condition_id"] = df["framework_condition_id"].astype(str)
df["framework_condition_desc_en"] = df["framework_condition_desc_en"].astype(str)
df["framework_condition_desc_id"] = df["framework_condition_desc_id"].astype(str)
self.logger.info(f"\n Total rows: {len(df):,}")
# Log distribusi kondisi untuk QA
self.logger.info("\n Distribusi framework_condition_en (sample):")
fw_cond_dist = df["framework_condition_en"].value_counts().head(10)
for cond, cnt in fw_cond_dist.items():
self.logger.info(f" {cnt:>6,} {cond}")
schema = [
bigquery.SchemaField("country_id", "INTEGER", mode="REQUIRED"),
@@ -1343,6 +1389,13 @@ class FoodSecurityAggregator:
bigquery.SchemaField("framework_score_1_100", "FLOAT", mode="REQUIRED"),
bigquery.SchemaField("rank_in_framework_year", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("year_over_year_change", "FLOAT", mode="NULLABLE"),
# --- KOLOM KONDISI (sama pola dengan agg_pillar_by_country) ---
# Tier label (GFSI 2022): Secure / Adequate / Moderate / At Risk / Critical
bigquery.SchemaField("framework_condition_en", "STRING", mode="REQUIRED"),
bigquery.SchemaField("framework_condition_id", "STRING", mode="REQUIRED"),
# Deskripsi kontekstual per framework (MDGs/SDGs/Total)
bigquery.SchemaField("framework_condition_desc_en", "STRING", mode="REQUIRED"),
bigquery.SchemaField("framework_condition_desc_id", "STRING", mode="REQUIRED"),
]
rows = load_to_bigquery(
self.client, df, table_name, layer='gold',
@@ -1600,7 +1653,7 @@ class FoodSecurityAggregator:
self.logger.info("\n" + "=" * 70)
self.logger.info("FOOD SECURITY AGGREGATION — 3 TABLES -> fs_asean_gold")
self.logger.info(" ASEAN aggregate DIGABUNG ke tabel yang sama (country_id=0)")
self.logger.info(" Kolom baru : pillar_condition_en, pillar_condition_id")
self.logger.info(" Kolom baru : pillar_condition_en/id, framework_condition_en/id")
self.logger.info(f" Performance threshold: {PERFORMANCE_THRESHOLD}")
self.logger.info(f" Condition tiers (GFSI 2022): >=75 Secure | >=60 Adequate |")
self.logger.info(f" >=40 Moderate | >=20 At Risk | <20 Critical")