from __future__ import annotations import csv import hashlib import io import json import os import urllib.parse import urllib.request import xml.etree.ElementTree as ET from datetime import datetime, timezone from pathlib import Path import numpy as np import pandas as pd ROOT = Path(__file__).resolve().parent REPO = ROOT.parents[1] CACHE = REPO / "tmp" / "research-028-source-cache" DATA = ROOT / "data" CACHE.mkdir(parents=True, exist_ok=True) DATA.mkdir(parents=True, exist_ok=True) START = pd.Timestamp("2002-01-01") END = pd.Timestamp("2026-06-01") RETRIEVED_DATE = "2026-07-30" USER_AGENT = "SHWRP-2026-028 research data acquisition/1.0" CBC_STOCK_URL = ( "https://www.cbc.gov.tw/public/data/OpenData/" "%E7%B6%93%E7%A0%94%E8%99%95/EG27M01.csv" ) CBC_FX_URL = ( "https://www.cbc.gov.tw/public/data/OpenData/" "%E7%B6%93%E7%A0%94%E8%99%95/BP01M01.csv" ) FAO_URL = ( "https://www.fao.org/media/docs/worldfoodsituationlibraries/" "default-document-library/food_price_indices_data.csv" ) EIA_BRENT_URL = "https://www.eia.gov/dnav/pet/hist_xls/RBRTEm.xls" DATA_GOV_METADATA = "https://data.gov.tw/api/v2/rest/dataset/{dataset_id}" DGBAS_DATASETS = { "cpi": 6019, "import_usd": 8240, "import_twd": 8241, } def fetch_bytes(url: str) -> bytes: request = urllib.request.Request( url, headers={ "User-Agent": USER_AGENT, "Accept": "*/*", }, ) with urllib.request.urlopen(request, timeout=180) as response: return response.read() def cached_download(url: str, filename: str) -> Path: destination = CACHE / filename if destination.exists() and os.getenv("REFRESH_SOURCES") != "1": return destination payload = fetch_bytes(url) if len(payload) < 100: raise RuntimeError(f"Downloaded payload is unexpectedly small: {url}") destination.write_bytes(payload) return destination def resolve_data_gov_resource(dataset_id: int) -> tuple[str, dict]: url = DATA_GOV_METADATA.format(dataset_id=dataset_id) metadata = json.loads(fetch_bytes(url).decode("utf-8-sig")) if not metadata.get("success"): raise RuntimeError(f"data.gov.tw metadata request failed: {dataset_id}") result = metadata["result"] distributions = result.get("distribution") or [] if not distributions: raise RuntimeError(f"No downloadable resource in dataset {dataset_id}") resource_url = distributions[0]["resourceDownloadUrl"] return resource_url, { "metadata_url": url, "resource_url": resource_url, "metadata_modified": result.get("modifiedDate", ""), "license_code": result.get("license", ""), "title": result.get("title", ""), } def parse_month(value: str) -> pd.Timestamp: return pd.to_datetime(value, format="%YM%m") def numeric(series: pd.Series) -> pd.Series: return pd.to_numeric(series.replace({"-": np.nan, "…": np.nan}), errors="coerce") def read_cbc_stock(path: Path) -> pd.DataFrame: frame = pd.read_csv(path, encoding="utf-8-sig") output = pd.DataFrame( { "date": frame["月"].map(parse_month), "taiex_monthly_average": numeric(frame["加權平均股價指數-原始值"]), } ).dropna() output = output.sort_values("date").reset_index(drop=True) output["taiex_log_return"] = np.log(output["taiex_monthly_average"]).diff() output["taiex_running_peak"] = output["taiex_monthly_average"].cummax() output["taiex_drawdown_all_time"] = ( output["taiex_monthly_average"] / output["taiex_running_peak"] - 1 ) output["taiex_rolling_24m_peak"] = ( output["taiex_monthly_average"].rolling(24, min_periods=1).max() ) output["taiex_drawdown_24m"] = ( output["taiex_monthly_average"] / output["taiex_rolling_24m_peak"] - 1 ) return output def read_cbc_fx(path: Path) -> pd.DataFrame: frame = pd.read_csv(path, encoding="utf-8-sig") return ( pd.DataFrame( { "date": frame["期間"].map(parse_month), "ntd_per_usd": numeric(frame["新台幣NTD/USD"]), } ) .dropna() .sort_values("date") .reset_index(drop=True) ) def read_fao(path: Path) -> pd.DataFrame: frame = pd.read_csv(path, skiprows=2) frame = frame.rename( columns={ "Date": "date", "Food Price Index": "fao_food_index", "Cereals": "fao_cereals_index", "Oils": "fao_oils_index", } ) frame["date"] = pd.to_datetime(frame["date"], format="%Y-%m", errors="coerce") for column in ["fao_food_index", "fao_cereals_index", "fao_oils_index"]: frame[column] = numeric(frame[column]) return frame[ ["date", "fao_food_index", "fao_cereals_index", "fao_oils_index"] ].dropna(subset=["date"]) def read_eia_brent(path: Path) -> pd.DataFrame: frame = pd.read_excel(path, sheet_name="Data 1", header=2) frame = frame.rename(columns={frame.columns[0]: "date", frame.columns[1]: "brent_usd_bbl"}) frame["date"] = pd.to_datetime(frame["date"], errors="coerce").dt.to_period("M").dt.to_timestamp() frame["brent_usd_bbl"] = numeric(frame["brent_usd_bbl"]) return frame[["date", "brent_usd_bbl"]].dropna() def item_key(value: str) -> str: return value.split("(指數基期", 1)[0].strip() def read_dgbas_xml(path: Path, targets: dict[str, str]) -> pd.DataFrame: records: list[dict[str, object]] = [] reverse_targets = {label: column for column, label in targets.items()} for _, element in ET.iterparse(path, events=("end",)): if element.tag != "Obs": continue row = {child.tag: (child.text or "").strip() for child in element} key = item_key(row.get("Item", "")) if ( key in reverse_targets and row.get("TYPE") == "原始值" and row.get("FREQ") == "M" and row.get("Item_VALUE") ): records.append( { "date": parse_month(row["TIME_PERIOD"]), "series": reverse_targets[key], "value": float(row["Item_VALUE"]), } ) element.clear() if not records: raise RuntimeError(f"No requested DGBAS series found in {path.name}") long = pd.DataFrame(records) wide = long.pivot(index="date", columns="series", values="value").reset_index() wide.columns.name = None return wide def detect_crash_episodes(stock: pd.DataFrame) -> pd.DataFrame: episodes: list[dict[str, object]] = [] in_episode = False onset_index = -1 for index, row in stock.iterrows(): regime = row["taiex_drawdown_24m"] <= -0.20 if regime and not in_episode: in_episode = True onset_index = index if in_episode and (not regime or index == len(stock) - 1): end_index = index - 1 if not regime else index window = stock.loc[onset_index:end_index] trough_index = window["taiex_drawdown_24m"].idxmin() onset = stock.loc[onset_index] trough = stock.loc[trough_index] recovered = not regime recovery_date = stock.loc[index, "date"] if recovered else pd.NaT episodes.append( { "episode": len(episodes) + 1, "threshold_crossing_month": onset["date"].strftime("%Y-%m"), "trough_month": trough["date"].strftime("%Y-%m"), "trough_drawdown_pct": round(trough["taiex_drawdown_24m"] * 100, 3), "recovery_month": ( recovery_date.strftime("%Y-%m") if recovered else "" ), "months_below_minus_20pct": int(len(window)), } ) in_episode = False return pd.DataFrame(episodes) def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def write_source_log(rows: list[dict[str, str]]) -> None: with (ROOT / "source-log.csv").open( "w", encoding="utf-8-sig", newline="" ) as handle: writer = csv.DictWriter(handle, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) def main() -> None: source_paths: dict[str, Path] = {} source_paths["cbc_stock"] = cached_download(CBC_STOCK_URL, "cbc-stock-monthly.csv") source_paths["cbc_fx"] = cached_download(CBC_FX_URL, "cbc-fx-monthly.csv") source_paths["fao"] = cached_download(FAO_URL, "fao-food-price-indices.csv") source_paths["eia_brent"] = cached_download(EIA_BRENT_URL, "eia-brent-monthly.xls") data_gov_records: dict[str, dict] = {} for name, dataset_id in DGBAS_DATASETS.items(): resource_url, metadata = resolve_data_gov_resource(dataset_id) suffix = Path(urllib.parse.urlparse(resource_url).path).suffix or ".xml" source_paths[name] = cached_download(resource_url, f"dgbas-{name}{suffix}") data_gov_records[name] = metadata stock_full = read_cbc_stock(source_paths["cbc_stock"]) fx = read_cbc_fx(source_paths["cbc_fx"]) fao = read_fao(source_paths["fao"]) brent = read_eia_brent(source_paths["eia_brent"]) cpi = read_dgbas_xml( source_paths["cpi"], { "cpi_all": "總指數", "cpi_food": "一.食物類", "cpi_grain_products": "1.穀類及其製品", "cpi_rice_products": "(1)米類及其製品", }, ) import_targets = { "import_all": "總指數", "import_plant_products": "第2類植物產品", "import_cereals": "10穀類", "import_corn": "玉米", "import_soybeans": "黃豆", "import_feed": "23食品產製過程之殘渣及調製動物飼料", "import_crude_oil": "原油", } import_usd = read_dgbas_xml(source_paths["import_usd"], import_targets).rename( columns={ column: f"{column}_usd" for column in import_targets if column != "date" } ) import_twd = read_dgbas_xml(source_paths["import_twd"], import_targets).rename( columns={ column: f"{column}_twd" for column in import_targets if column != "date" } ) stock = stock_full.loc[stock_full["date"].between(START, END)].copy() panel = stock.merge(fx, on="date", how="left") for frame in [cpi, import_usd, import_twd, fao, brent]: panel = panel.merge(frame, on="date", how="left") panel = panel.loc[panel["date"].between(START, END)].sort_values("date").reset_index(drop=True) expected_months = len(pd.period_range(START, END, freq="M")) if len(panel) != expected_months: raise RuntimeError(f"Expected {expected_months} months, observed {len(panel)}") core_columns = [ "taiex_monthly_average", "ntd_per_usd", "cpi_food", "cpi_grain_products", "cpi_rice_products", "import_cereals_usd", "import_cereals_twd", "fao_food_index", "fao_cereals_index", "brent_usd_bbl", ] missing = panel[core_columns].isna().sum() if int(missing.sum()) > 0: raise RuntimeError(f"Core series contain missing values: {missing[missing > 0].to_dict()}") transform_columns = [ column for column in panel.columns if column not in { "date", "taiex_log_return", "taiex_running_peak", "taiex_drawdown_all_time", "taiex_rolling_24m_peak", "taiex_drawdown_24m", } and pd.api.types.is_numeric_dtype(panel[column]) and (panel[column] > 0).all() ] for column in transform_columns: panel[f"dlog_{column}"] = np.log(panel[column]).diff() tail_threshold = float(panel["taiex_log_return"].quantile(0.05)) panel["stock_tail_loss"] = (panel["taiex_log_return"] <= tail_threshold).astype(int) panel["stock_loss_magnitude"] = (-panel["taiex_log_return"]).clip(lower=0) panel["stock_gain_magnitude"] = panel["taiex_log_return"].clip(lower=0) panel["drawdown_regime_20pct"] = (panel["taiex_drawdown_24m"] <= -0.20).astype(int) pre_sample_stock = stock_full.loc[stock_full["date"] < START] pre_sample_regime = ( int(pre_sample_stock.iloc[-1]["taiex_drawdown_24m"] <= -0.20) if not pre_sample_stock.empty else 0 ) prior_drawdown_regime = panel["drawdown_regime_20pct"].shift() prior_drawdown_regime.iloc[0] = pre_sample_regime panel["drawdown_onset_20pct"] = ( (panel["drawdown_regime_20pct"] == 1) & (prior_drawdown_regime == 0) ).astype(int) panel["fx_depreciation"] = panel["dlog_ntd_per_usd"] panel["positive_fx_depreciation"] = panel["fx_depreciation"].clip(lower=0) panel["tail_loss_x_depreciation"] = ( panel["stock_tail_loss"] * panel["positive_fx_depreciation"] ) panel["year"] = panel["date"].dt.year panel["month"] = panel["date"].dt.month panel["date"] = panel["date"].dt.strftime("%Y-%m") panel.to_csv(DATA / "monthly-analysis-panel.csv", index=False, encoding="utf-8-sig") episodes = detect_crash_episodes(stock_full) episodes = episodes.loc[ episodes["threshold_crossing_month"].between("2002-01", "2026-06") ].reset_index(drop=True) episodes.to_csv(DATA / "drawdown-episodes.csv", index=False, encoding="utf-8-sig") source_rows = [ { "source_id": "CBC_STOCK", "source_name": "中央銀行：股票交易與股價指數月資料", "source_url": CBC_STOCK_URL, "retrieved_at": RETRIEVED_DATE, "coverage_used": "2002-01至2026-06；月平均加權股價指數", "rights_or_terms": "政府資料開放授權條款第1版；原始資料權利仍歸來源", "raw_file_publicly_redistributed": "否；公開衍生月資料與取得程式", "sha256": sha256(source_paths["cbc_stock"]), }, { "source_id": "CBC_FX", "source_name": "中央銀行：我國與主要貿易對手通貨之匯率月資料", "source_url": CBC_FX_URL, "retrieved_at": RETRIEVED_DATE, "coverage_used": "2002-01至2026-06；新臺幣NTD/USD", "rights_or_terms": "政府資料開放授權條款第1版；研究參考匯率", "raw_file_publicly_redistributed": "否；公開衍生月資料與取得程式", "sha256": sha256(source_paths["cbc_fx"]), }, { "source_id": "DGBAS_CPI", "source_name": data_gov_records["cpi"]["title"], "source_url": data_gov_records["cpi"]["metadata_url"], "retrieved_at": RETRIEVED_DATE, "coverage_used": "2002-01至2026-06；食品、穀類及其製品、米類及其製品", "rights_or_terms": "政府資料開放授權條款第1版", "raw_file_publicly_redistributed": "否；公開必要衍生月資料", "sha256": sha256(source_paths["cpi"]), }, { "source_id": "DGBAS_IMPORT_USD", "source_name": data_gov_records["import_usd"]["title"], "source_url": data_gov_records["import_usd"]["metadata_url"], "retrieved_at": RETRIEVED_DATE, "coverage_used": "2002-01至2026-06；美元計價進口穀類等指數", "rights_or_terms": "政府資料開放授權條款第1版", "raw_file_publicly_redistributed": "否；公開必要衍生月資料", "sha256": sha256(source_paths["import_usd"]), }, { "source_id": "DGBAS_IMPORT_TWD", "source_name": data_gov_records["import_twd"]["title"], "source_url": data_gov_records["import_twd"]["metadata_url"], "retrieved_at": RETRIEVED_DATE, "coverage_used": "2002-01至2026-06；新臺幣計價進口穀類等指數", "rights_or_terms": "政府資料開放授權條款第1版", "raw_file_publicly_redistributed": "否；公開必要衍生月資料", "sha256": sha256(source_paths["import_twd"]), }, { "source_id": "FAO_FFPI", "source_name": "FAO Food Price Index", "source_url": FAO_URL, "retrieved_at": RETRIEVED_DATE, "coverage_used": "2002-01至2026-06；食品、穀物及食用油指數", "rights_or_terms": "FAO Statistical Database Terms；CC BY 4.0並有第三方權利及不得暗示背書等條件", "raw_file_publicly_redistributed": "否；公開必要衍生月資料", "sha256": sha256(source_paths["fao"]), }, { "source_id": "EIA_BRENT", "source_name": "U.S. EIA Europe Brent Spot Price FOB", "source_url": EIA_BRENT_URL, "retrieved_at": RETRIEVED_DATE, "coverage_used": "2002-01至2026-06；月平均美元／桶", "rights_or_terms": "美國政府公有領域資料；標示EIA來源", "raw_file_publicly_redistributed": "否；公開必要衍生月資料", "sha256": sha256(source_paths["eia_brent"]), }, ] write_source_log(source_rows) summary = { "report_number": "SHWRP-2026-028", "retrieved_at": RETRIEVED_DATE, "generated_at_utc": datetime.now(timezone.utc).isoformat(), "sample_start": panel["date"].min(), "sample_end": panel["date"].max(), "months": int(len(panel)), "source_count": len(source_rows), "core_missing_values": int(missing.sum()), "tail_loss_threshold_log_return": tail_threshold, "tail_loss_months": int(panel["stock_tail_loss"].sum()), "drawdown_episodes": int(len(episodes)), "source_cache_public": False, "public_panel": "data/monthly-analysis-panel.csv", } (DATA / "acquisition-summary.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8", ) print(json.dumps(summary, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()