+ AI-Powered Food Security Intelligence Platform
+ Inter-Regional Agricultural Supply–Demand Matching Platform
+
Detect · Predict · Distribute
+ +
+ 
Detect · Predict · Distribute — for Indonesian food security.
diff --git a/README.md b/README.md new file mode 100644 index 0000000000000000000000000000000000000000..8fe953089cb0dc2bc7bd24eee2c5ec89aa936a01 --- /dev/null +++ b/README.md @@ -0,0 +1,240 @@ +--- +title: AgriFlow API +emoji: "🌾" +colorFrom: green +colorTo: yellow +sdk: docker +app_port: 7860 +pinned: false +license: mit +--- + +Language / Bahasa: [English](./README.en.md) · **Bahasa Indonesia** + +
+ AI-Powered Food Security Intelligence Platform
+ Platform Matching Demand–Supply Pangan Antarwilayah
+
Deteksi · Prediksi · Distribusi
+ +
+
Poster riset tersedia di repo GitHub (tidak disertakan di Space: HF menolak file biner di luar Xet storage).
+ +Deteksi · Prediksi · Distribusi — untuk ketahanan pangan Indonesia.
diff --git a/README_v10.md b/README_v10.md new file mode 100644 index 0000000000000000000000000000000000000000..bbbb657d14057139be5041aafb5898a3b78cd3b5 --- /dev/null +++ b/README_v10.md @@ -0,0 +1,639 @@ +# AgriFlow Matching Engine + +> **Sub-national pangan matching engine pertama di Indonesia.** +> Algoritma 4-lapis hybrid yang menghubungkan kabupaten surplus dengan defisit menggunakan stable matching, multi-objective scoring 5 dimensi, dan equity multiplier untuk kabupaten tertinggal — semua untuk komoditas pangan tingkat kabupaten. + +[]() +[]() +[]() +[]() + +Submisi **PIDI DIGDAYA × Hackathon 2026** — Bank Indonesia. +Problem Statement #2: Platform Matching Demand-Supply Antarwilayah. + +--- + +## Daftar Isi + +- [Apa Ini?](#apa-ini) +- [Quick Start (5 menit)](#quick-start-5-menit) +- [Arsitektur 4-Lapis](#arsitektur-4-lapis) +- [Equity Multiplier (Kalibrasi BPS 2024)](#equity-multiplier-kalibrasi-bps-2024) +- [19 Skenario Edge Case](#19-skenario-edge-case) +- [API Usage](#api-usage) +- [Performance & Validation](#performance--validation) +- [Data Sources](#data-sources) +- [Project Structure](#project-structure) +- [Development Guide](#development-guide) +- [Status & Roadmap](#status--roadmap) +- [Documentation](#documentation) +- [License & Credits](#license--credits) + +--- + +## Apa Ini? + +**Bayangkan Uber, tapi untuk cabai dan bawang merah.** + +Setiap hari, Indonesia kehilangan Rp 213-551 triliun pangan karena food loss & waste — 40% di distribusi, bukan produksi. Petani di Sampang membuang cabai karena harga jatuh, sementara pasar Surabaya melonjak 200% karena kelangkaan. Pemda baru tahu krisis 2-3 minggu kemudian. + +AgriFlow Matching Engine memecahkan ini dengan 6 dimensi yang Uber tidak punya: + +| Dimensi | Penjelasan | +|---|---| +| **Perishability** | Cabai busuk dalam 5 hari, beras tahan 180 hari — engine hitung shelf life | +| **Equity** | Kabupaten tertinggal IPM rendah (Sampang 66.72) dapat boost +30% | +| **Climate** | Banjir di rute = re-route otomatis | +| **Volume** | 1 surplus bisa di-split ke banyak deficit | +| **Stable Matching** | Guarantee fairness via Gale-Shapley (Nobel Prize Economics 2012) | +| **Two-tier Confidence** | Data harian PIHPS (Tier 1) pakai algoritma ketat; data mingguan Bapanas (Tier 2) pakai algoritma fleksibel | + +**Status:** Production-ready untuk skala provinsial (38 kab Jatim) — 106/106 tests pass dalam 0.16s, latency p99 1.4ms (sample) - 55.5ms (stress 361×361). + +--- + +## Quick Start (5 menit) + +### Prasyarat + +- Python 3.10+ +- pip +- ~50MB disk space + +### Install + +```bash +git clone https://github.com/masterA88/agriflow_engine.git +cd agriflow_engine +python -m venv venv +# Windows: +venv\Scripts\activate +# Linux/Mac: +source venv/bin/activate +pip install -r requirements.txt +``` + +### Verifikasi (semua harus sukses) + +```bash +# 1. Generate sample data — 38 kab × 19 komoditas Jatim +python sample_data/generate_sample_data.py +# Expected: 5 CSV generated (kabupaten_jatim.csv, komoditas_constraints.csv, +# surplus_deficit.csv, weather_forecast.csv, historical_price_stats.csv) + +# 2. Run all tests (106 tests) +pytest tests/ -v +# Expected: 106 passed in <1s + +# 3. Run end-to-end demo +python examples/run_demo.py +# Expected: ~32 matches, gross arbitrage ~Rp 16 miliar, latency ~1.5ms + +# 4. Run latency benchmark +python benchmarks/latency.py +# Expected: highest p99 < 60ms (margin >88% vs 500ms target) +``` + +Kalau langkah 2 atau 3 gagal, lihat [Troubleshooting](#troubleshooting) di bawah. + +--- + +## Arsitektur 4-Lapis + +``` +Input: surplus_nodes[], deficit_nodes[], LogisticsContext, weather, historical_prices + + ┌─────────────────────────────────────────────────────────────────┐ + │ LAYER 0 — Tier Classification (constraints.determine_tier) │ + │ Klasifikasi setiap kab: Tier 1 HIGH (8 kota IHK PIHPS) atau │ + │ Tier 2 MEDIUM (30 kab non-IHK Bapanas). │ + │ Latency: <1ms (set lookup). │ + └─────────────────────────────────────────────────────────────────┘ + ↓ + ┌─────────────────────────────────────────────────────────────────┐ + │ LAYER 1 — Hard Constraints (constraints.generate_candidates) │ + │ 9 rules filter: komoditas match, distance≤max, age≤shelf, │ + │ volume≥min, no self-match, emergency mode, pemda override, │ + │ Bulog split, BBM-aware distance shrink. │ + │ Output: candidate pairs (top-K per surplus by jarak). │ + │ Latency: <50ms untuk 38×19 (~25k pasang potensial). │ + └─────────────────────────────────────────────────────────────────┘ + ↓ + ┌─────────────────────────────────────────────────────────────────┐ + │ LAYER 2 — Multi-Objective Scoring (scoring.compute_score) │ + │ 5-dimensi weighted: Distance 22% / Volume 22% / Price 22% / │ + │ Perishability 18% / Climate 16%. │ + │ 3 weight schemes: DEFAULT, RAMADAN, IMPORT_POLICY. │ + │ Output: base_score 0-100 per pair. │ + └─────────────────────────────────────────────────────────────────┘ + ↓ + ┌─────────────────────────────────────────────────────────────────┐ + │ LAYER 3 — Equity-Weighted Allocation (allocation.allocate) │ + │ Final = base × equity_multiplier(IPM_deficit). │ + │ Tier 1↔Tier 1 → Modified Gale-Shapley (Nobel 2012). │ + │ Cross-tier / Tier 2 → Greedy with equity priority. │ + │ Output: MatchResult[] dengan confidence label. │ + └─────────────────────────────────────────────────────────────────┘ + ↓ + ┌─────────────────────────────────────────────────────────────────┐ + │ POST-PROCESSING (engine.run_matching) │ + │ Tag flags (RAMADAN_SPIKE, EQUITY_BOOST_30, MADURA_CLUSTER, │ + │ STALE_DATA_24H, HUMANITARIAN_PRIORITY, VOLUME_MISMATCH). │ + │ Identifikasi unmatched + external_opportunities (ekspor). │ + └─────────────────────────────────────────────────────────────────┘ + +Output: MatchingReport(matches, unmatched_*, warnings, run_metadata) +``` + +**Why 4-layer?** Setiap layer bisa dioptimasi independent, testable secara isolated, dan early-exit di Layer 1 menghemat compute Layer 2/3 yang lebih mahal. + +--- + +## Equity Multiplier (Kalibrasi BPS 2024) + +Threshold dikalibrasi sesuai distribusi IPM 2024 BPS Jatim sehingga klaim "+30% boost untuk kab tertinggal" konkret applicable: + +| IPM Range | Multiplier | Boost | Kab/Kota Jatim | +|---|---|---|---| +| `IPM < 68` | **1.30** | **+30%** | Sampang (66.72), Bangkalan (67.70) | +| `68 ≤ IPM < 72` | 1.15 | +15% | Sumenep, Probolinggo (kab), Bondowoso, Lumajang, Pamekasan, Pacitan, Pasuruan (kab), Situbondo, Jember, Madiun (kab) | +| `72 ≤ IPM < 78` | 1.05 | +5% | Bojonegoro, Banyuwangi, Tulungagung, Malang (kab), Magetan, Gresik, Mojokerto (kab), Lamongan, Tuban, Ngawi, Kediri (kab), dll | +| `IPM ≥ 78` | 1.00 | (no boost) | Sidoarjo, Kota Batu, Kota Surabaya, Kota Malang, Kota Kediri, Kota Madiun, dll | + +**Mengapa kalibrasi:** Threshold v9 lama (`<65 → 1.30`) tidak pernah ter-trigger karena IPM terendah Jatim 2024 = Sampang 66.72. v10 menggeser threshold sehingga klaim "+30% boost" demonstrably valid. + +**Update IPM tahunan:** Saat BPS publish IPM baru (biasanya BRS Desember), edit di [`sample_data/generate_sample_data.py:KABUPATEN_DATA`](sample_data/generate_sample_data.py) sebagai source of truth, lalu mirror ke [`data_sources/bps.py:IPM_2024_JATIM`](data_sources/bps.py). + +--- + +## 19 Skenario Edge Case + +5 kategori, 19 skenario, semua tervalidasi pytest. Detail lengkap di [`docs/AUDIT_v10.md`](docs/AUDIT_v10.md) dan `AgriFlow_v10.docx` Section 5.5.5. + +### Kategori A — Volume (4 skenario) + +| Kode | Skenario | Test | +|---|---|---| +| A1 | Surplus 1-to-many (1 surplus split ke beberapa deficit) | `TestA1_OneToMany` | +| A2 | Many-to-1 (multiple surplus untuk 1 deficit besar) | `TestA2_ManyToOne` | +| A3 | Volume mismatch drastis (<20% ratio → flag warning) | `TestA3_VolumeMismatchDrastis` | +| A4 | Zero demand (suggest external opportunity) | `TestA4_ZeroDemand` | + +### Kategori B — Spasial (3 skenario) + +| Kode | Skenario | Test | +|---|---|---| +| B1 | Cross-tier match (Tier 1 ↔ Tier 2) | `TestB1_CrossTier` | +| B2 | Long distance (jarak > max_distance_km → REJECT) | `TestB2_LongDistance` | +| B3 | Cluster Madura (4 kab semua surplus → ekspor) | `TestB3_ClusterMadura` | + +### Kategori C — Temporal (3 skenario) + +| Kode | Skenario | Test | +|---|---|---| +| C1 | Ramadan/Idul Fitri spike (H-21 to H-1, RAMADAN_WEIGHTS) | `TestC1_RamadanSpike` | +| C2 | Pasca panen raya (oversupply, multiple match) | `TestC2_PostHarvest` | +| C3 | Stale data >24h (confidence drop bertingkat HIGH→MEDIUM→LOW) | `TestC3_StaleData` | + +### Kategori D — Disrupsi (5 skenario) + +| Kode | Skenario | Test | +|---|---|---| +| D1 | Banjir rute (BMKG hujan >50mm → climate_score 0.3) | `TestD1_BanjirRute` | +| D2 | Komoditas hampir rusak (harvest age + transit > shelf) | `TestD2_KomoditasRusak` | +| D3 | Harga anomali (>3σ dari rolling median → exclude) | `TestD3_HargaAnomali` | +| D4 | Erupsi gunung (PVMBG MAGMA → UNREACHABLE) | `TestD4_ErupsiGunung` | +| D5 | Banjir multi-kab (BNPB DIBI → emergency mode) | `TestD5_BanjirMultiKab` | + +### Kategori E — Politis & Kebijakan (5 skenario) + +| Kode | Skenario | Test | +|---|---|---| +| E1 | Equity tie-break (IPM lebih rendah menang otomatis) | `TestE1_EquityTieBreak` | +| E2 | Pemda override (`do_not_export_
+ AI-Powered Food Security Intelligence Platform
+ Platform Matching Demand–Supply Pangan Antarwilayah
+
Deteksi · Prediksi · Distribusi
+ +
+
Deteksi · Prediksi · Distribusi — untuk ketahanan pangan Indonesia.
diff --git a/REAL_DATA_METHODOLOGY.md b/REAL_DATA_METHODOLOGY.md new file mode 100644 index 0000000000000000000000000000000000000000..003594ce4432fc0d2e9afeb2a8f3048b48858a2c --- /dev/null +++ b/REAL_DATA_METHODOLOGY.md @@ -0,0 +1,194 @@ +# Real Data Methodology — Surplus/Deficit 2022 + +## Status per commodity + +| Commodity | Status | Source | Tahun | +|---|---|---|---| +| `beras_premium` | REAL (derived) | BPS per-kab produksi + konsumsi + populasi | 2022 | +| `beras_medium` | REAL (derived, grade-split assumption 60/40) | same | 2022 | +| `cabai_merah` | REAL produksi (BPS Hortikultura); konsumsi proxy nasional (Kementan) | BPS Jatim Hortikul. + Kementan PDF | 2022 | +| `cabai_rawit` | REAL produksi (BPS Hortikultura); konsumsi proxy nasional (Kementan) | same | 2022 | +| `bawang_merah` | REAL produksi (BPS Hortikultura); konsumsi proxy nasional (Kementan) | same | 2022 | +| `bawang_putih` | REAL produksi (BPS Hortikultura); konsumsi proxy nasional (Kementan) | same | 2022 | +| `daging_ayam` | EXCLUDED — data tidak lengkap (broiler hilang; ayam petelur 2 kab) | — | PENDING | +| `telur_ayam` | EXCLUDED — telur petelur hanya 2 kab | — | PENDING | + +--- + +## Reference year: 2022 + +All three beras inputs AND semua 5 hortikultura source files have 2022 data. +**2022** is selected as the reference year because: +- Beras: 2022 is the latest year where produksi, konsumsi per-kab, AND populasi are all + present and plausible for all 38 kab/kota. (2025 populasi corrupted ~1000x.) +- Hortikultura: BPS Hortikultura files have `Produksi_2021` and `Produksi_2022` columns; + 2022 is the most recent available year. +- Konsumsi per-kapita hortikultura: Kementan Statistik Konsumsi Pangan 2024 provides + 2022 values in Tabel 4.6a (cabai) and 4.1a/4.2a (bawang). + +--- + +## Data sources + +### Beras (all BPS-grade, per-kabupaten) + +| Dataset | File | Source | Unit | +|---|---|---|---| +| Produksi beras | `sample_data/bps_real/year_beras.csv` | BPS Jawa Timur | ton/tahun/kab | +| Konsumsi per kapita | `sample_data/bps_real/week_konsumsi_beras_perkapita.csv` | BPS Indonesia (Susenas) | kg/kapita/minggu | +| Populasi | `sample_data/bps_real/year_populasi_jatim.csv` | BPS Jawa Timur | jiwa | + +### Hortikultura + +| Dataset | File | Source | Unit | Konversi | +|---|---|---|---|---| +| Produksi cabai besar | `bps_real/cabai_besar.csv` | BPS Jatim Hortikul. | **KUINTAL**/tahun | ÷10 = ton | +| Produksi cabai keriting | `bps_real/cabai_keriting.csv` | same | KUINTAL/tahun | ÷10 = ton | +| Produksi cabai rawit | `bps_real/cabai_rawit.csv` | same | KUINTAL/tahun | ÷10 = ton | +| Produksi bawang merah | `bps_real/bawang_merah.csv` | same | KUINTAL/tahun | ÷10 = ton | +| Produksi bawang putih | `bps_real/bawang_putih.csv` | same | KUINTAL/tahun | ÷10 = ton | + +**Satuan konfirmasi**: Total row cabai_besar 2022 = 851,445 kuintal = 85,145 ton → sesuai +BPS published figure Jatim. Konversi ÷10 (kuintal → ton) terverifikasi. + +### Konsumsi per kapita hortikultura — NATIONAL PROXY (Kementan) + +Sumber: **Statistik Konsumsi Pangan 2024**, Pusat Data dan Sistem Informasi Pertanian, +Kementerian Pertanian. Desember 2024. +URL: https://satudata.pertanian.go.id/assets/docs/publikasi/Buku_Statistik_Konsumsi_2024.pdf + +| Komoditas | Tabel | Hal. | Nilai 2022 | Catatan | +|---|---|---|---|---| +| Cabai merah | 4.6a (Cabe merah/Chillies) | 30 | **1.909 kg/kapita/tahun** | Susenas per-kapita nasional | +| Cabai rawit | 4.6a (Cabe rawit/Cayenne pepper) | 30 | **2.073 kg/kapita/tahun** | same | +| Bawang merah | 4.1a (Bawang merah/Onion) | 25 | **3.024 kg/kapita/tahun** | same | +| Bawang putih | 4.2a (Bawang putih/Garlic) | 26 | **2.016 kg/kapita/tahun** | same | + +**PROXY WARNING**: Angka ini adalah RATA-RATA NASIONAL dari Susenas. Tidak ada data konsumsi +per-kabupaten untuk hortikultura. Konsumsi aktual per kab bisa berbeda dari angka nasional +(terutama kota besar vs pedesaan, atau daerah produsen di mana konsumsi lokal bisa lebih tinggi). +Hasil surplus/deficit hortikultura harus dibaca dengan caveat ini. + +--- + +## Derivation formula + +### Beras + +``` +konsumsi_ton = avg_konsumsi_perkapita_kg_per_minggu × 52 × populasi / 1000 +net_ton = produksi_ton − konsumsi_ton +role = "SURPLUS" if net_ton > 0 else "DEFICIT" +volume_tons = abs(net_ton) +``` + +Per-kab BPS Susenas data digunakan untuk konsumsi beras (bukan proxy nasional). + +### Hortikultura (cabai merah, cabai rawit, bawang merah, bawang putih) + +``` +produksi_ton = (Produksi_2022 kuintal) / 10 # BPS file, kuintal -> ton +konsumsi_ton = perkapita_kg_per_tahun × populasi_2022 / 1000 # Kementan national avg +net_ton = produksi_ton − konsumsi_ton +role = "SURPLUS" if net_ton > 0 else "DEFICIT" +volume_tons = abs(net_ton) +``` + +Cabai merah = cabai_besar + cabai_keriting (dijumlahkan sebelum derivasi). + +--- + +## Grade split beras: ASSUMPTION + +60% net beras → `beras_premium`; 40% → `beras_medium`. Working assumption. + +--- + +## Harga (2022 PIHPS median dari sample_data/price_history/) + +| Komoditas | Harga IDR/kg | Sumber | +|---|---|---| +| `beras_premium` | 11,500 | median(super1=12,000; super2=11,000) PIHPS 2022 | +| `beras_medium` | 10,325 | median(medium1=10,650; medium2=10,000) PIHPS 2022 | +| `cabai_rawit` | 41,000 | cabe_rawit_cleaned.csv 2022 median, PIHPS | +| `bawang_merah` | 32,500 | bawang_merah_cleaned.csv 2022 median, PIHPS | +| `bawang_putih` | 20,750 | bawang_putih_cleaned.csv 2022 median, PIHPS | +| `cabai_merah` | 45,000 | **FLAGGED: komoditas_constraints.csv baseline** — cabai merah tidak ada di PIHPS dataset | + +--- + +## harvest_age_days + +| Role | Value | Rationale | +|---|---|---| +| SURPLUS | 28 | Typical post-harvest/milling age when komoditas leaves origin kab (conservative estimate) | +| DEFICIT | 0 | Convention: deficit nodes are demand points; age irrelevant | + +--- + +## Name-to-kab_id mapping + +**Beras source files**: menggunakan nama BPS full ("Kabupaten X" / "Kota X") → +strip prefix untuk match ke `kabupaten_jatim.csv` short name. + +**Hortikultura source files**: menggunakan nama pendek tanpa prefix positional: +- Baris 0-28 (29 baris): Kabupaten, urutan kode BPS 3501 (Pacitan) s/d 3529 (Sumenep) +- Baris 29-37 (9 baris): Kota, urutan kode BPS 3571 (Kota Kediri) s/d 3579 (Kota Batu) +- Baris 38: "Total" — EXCLUDED + +Semua 38 kab/kota Jatim ter-map tanpa exception untuk tahun 2022. + +--- + +## Sanity check results (2022) + +### Beras — top surplus kabupaten + +| kab_id | Kabupaten | Net beras (ton) | Role | +|---|---|---|---| +| 3524 | Kab. Lamongan | +409,023 | SURPLUS | +| 3521 | Kab. Ngawi | +366,367 | SURPLUS | +| 3522 | Kab. Bojonegoro | +295,679 | SURPLUS | +| 3523 | Kab. Tuban | +184,507 | SURPLUS | +| 3519 | Kab. Madiun | +195,419 | SURPLUS | + +### Cabai merah — top surplus + +| kab_id | Kabupaten | Surplus (ton) | +|---|---|---| +| 3507 | Kab. Malang | ~22,328 | +| 3513 | Kab. Probolinggo | ~9,823 | +| 3510 | Kab. Banyuwangi | ~7,177 | + +### Cabai rawit — top surplus + +| kab_id | Kabupaten | Surplus (ton) | +|---|---|---| +| 3510 | Kab. Banyuwangi | ~100,709 | +| 3507 | Kab. Malang | ~81,866 | +| 3506 | Kab. Kediri | ~77,761 | + +### Bawang merah — top surplus + +| kab_id | Kabupaten | Surplus (ton) | +|---|---|---| +| 3518 | Kab. Nganjuk | ~190,610 | +| 3513 | Kab. Probolinggo | ~54,731 | +| 3507 | Kab. Malang | ~43,099 | + +### Bawang putih + +Seluruh 38 kab/kota DEFICIT. Jatim total produksi ~855 ton vs kebutuhan ~80,000+ ton. +(Jatim bukan produsen bawang putih; pasokan dari Temanggung/Brebes Jateng dan impor.) + +Semua hasil ini sesuai ekspektasi geografis. + +--- + +## Excluded + +- **Daging ayam**: Data folder download berisi `daging_ayam_kampung.csv` dan + `daging_ayam_petelur.csv`. Broiler (sumber utama produksi komersial) tidak ada. + Data tidak representatif untuk routing engine. **Status: PENDING.** +- **Telur ayam**: `telur_ayam_petelur.csv` hanya 2 kabupaten ter-cover. + **Status: PENDING.** diff --git a/analysis/__init__.py b/analysis/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..254155ff986b1a8d3437f0b7b5fce0e9e680bf90 --- /dev/null +++ b/analysis/__init__.py @@ -0,0 +1,8 @@ +""" +analysis/ — Price anomaly detection and statistical analysis for AgriFlow. + +This package is intentionally dependency-light: numpy + stdlib csv only. +No pandas, no statsmodels, no sklearn. Results are interpretable by +construction — every flagged point shows the rolling median it deviated from +and the exact percentage deviation. +""" diff --git a/analysis/forecast_timesfm.py b/analysis/forecast_timesfm.py new file mode 100644 index 0000000000000000000000000000000000000000..aab5f5cc7014c9826d543d696f9a751a8d80de72 --- /dev/null +++ b/analysis/forecast_timesfm.py @@ -0,0 +1,309 @@ +""" +analysis/forecast_timesfm.py -- Offline forecast precompute for AgriFlow. + +ARCHITECTURE: + This script runs OFFLINE (locally, not on HF Space) because TimesFM ~2GB + model cannot be loaded on the free-tier Space (OOM). The output JSON files + are committed to the repo and the backend serves them at runtime without + importing this module or timesfm. + +HONESTY POLICY: + If TimesFM cannot be loaded (not installed, network unavailable, Python + version incompatible), this script falls back to a seasonal-naive baseline + that is CLEARLY labelled in the output as "method": "seasonal_naive_baseline" + so consumers can distinguish it from a genuine TimesFM forecast. + + DO NOT change the labelling. If you want TimesFM output, fix the environment + and re-run. + +TIMESFM STATUS (2026-05-31): + timesfm PyPI package (1.0.0) requires Python <=3.11 + jaxlib==0.4.26. + This project runs Python 3.12+. TimesFM 2.0 (PyTorch variant) is on + HuggingFace Hub but requires the same pinned JAX+Flax stack via the PyPI + package. Install blocker is hard on Python 3.12/3.14. + + To use real TimesFM: + 1. Run this script with Python 3.11: `py -3.11 analysis/forecast_timesfm.py` + 2. Or wait for timesfm to release a Python 3.12-compatible wheel. + 3. Check for a conda-based install path: `conda install -c conda-forge timesfm` + +USAGE: + # With TimesFM (Python 3.11 + timesfm installed): + python analysis/forecast_timesfm.py + + # Explicit baseline (any Python): + python analysis/forecast_timesfm.py --method baseline + +OUTPUT: + sample_data/forecasts/forecast_all.json -- one file, all series + +FORECAST SCHEMA (per record): + commodity_code str + city_id str + city_name str + method str ("timesfm_2.0" | "seasonal_naive_baseline") + generated_at str ISO 8601 UTC + horizon_days int (30) + history_end_date str ISO 8601 -- last observed date + forecasts: list of { + date str ISO 8601 + point float (IDR/kg, point forecast) + p10 float (IDR/kg, 10th percentile) + p90 float (IDR/kg, 90th percentile) + } +""" + +from __future__ import annotations + +import argparse +import datetime +import json +import math +import sys +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).parent.parent +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from analysis.price_anomaly import _load_all_rows, CITY_NAMES + +HORIZON = 30 + + +# --------------------------------------------------------------------------- +# Seasonal-naive baseline (transparent fallback -- NOT TimesFM) +# --------------------------------------------------------------------------- + +def _seasonal_naive_forecast( + series: list[tuple[datetime.date, float]], + horizon: int = HORIZON, +) -> list[dict[str, Any]]: + """ + Seasonal-naive: for day h, predict = median of same-calendar-month prices + observed in the training series. + + Uncertainty band: +/- 1 MAD of the same-month observations. + + This is a statistical method, not a foundation model. It is labelled as + "seasonal_naive_baseline" everywhere it appears. + """ + import numpy as np + + prices = [p for _, p in series] + dates = [d for d, _ in series] + arr = np.array(prices, dtype=float) + + # Build per-month (median, MAD) from the last 2 years of observed data + cutoff = dates[-1] - datetime.timedelta(days=2 * 365) + recent = [(d, p) for d, p in series if d >= cutoff] + if len(recent) < 30: + recent = series # fall back to full series for short series + + month_stats: dict[int, tuple[float, float]] = {} + from collections import defaultdict + month_vals: dict[int, list[float]] = defaultdict(list) + for d, p in recent: + month_vals[d.month].append(p) + for m, vals in month_vals.items(): + v = np.array(vals) + med = float(np.median(v)) + mad = float(np.median(np.abs(v - med))) + month_stats[m] = (med, mad) + + # Overall fallback stats + overall_med = float(np.median(arr[-30:])) + overall_mad = float(np.median(np.abs(arr[-30:] - overall_med))) + + last_date = dates[-1] + result = [] + for h in range(1, horizon + 1): + target_date = last_date + datetime.timedelta(days=h) + med, mad = month_stats.get(target_date.month, (overall_med, overall_mad)) + # CI: +/- 1.4826 * MAD (same scaling as the anomaly detector) + ci_half = 1.4826 * mad if mad > 0 else 0.05 * med + result.append({ + "date": target_date.isoformat(), + "point": round(med, 2), + "p10": round(max(0, med - ci_half), 2), + "p90": round(med + ci_half, 2), + }) + return result + + +# --------------------------------------------------------------------------- +# TimesFM path (gated on successful import) +# --------------------------------------------------------------------------- + +def _timesfm_available() -> bool: + try: + import timesfm # noqa: F401 + return True + except ImportError: + return False + + +def _timesfm_forecast( + series: list[tuple[datetime.date, float]], + horizon: int = HORIZON, + model_path: str = "google/timesfm-2.0-500m-pytorch", +) -> list[dict[str, Any]]: + """ + Run TimesFM 2.0 (PyTorch variant) on one price series. + Loads the model on first call (expensive — ~2 GB download + load). + Caller must ensure timesfm is installed and Python 3.10/3.11 is active. + """ + import timesfm + import numpy as np + + prices = np.array([p for _, p in series], dtype=float) + dates = [d for d, _ in series] + + # TimesFM 2.0 PyTorch API + tfm = timesfm.TimesFm( + hparams=timesfm.TimesFmHparams( + backend="cpu", + per_core_batch_size=1, + horizon_len=horizon, + num_heads=16, + use_positional_embedding=False, + ), + checkpoint=timesfm.TimesFmCheckpoint( + huggingface_repo_id=model_path, + ), + ) + + forecast_input = [prices] + freq = [0] # 0 = high-frequency (daily) + + _, quantile_forecasts = tfm.forecast( + forecast_input, + freq=freq, + quantile_levels=[0.1, 0.5, 0.9], + ) + + # quantile_forecasts shape: (batch=1, horizon, 3) + qf = quantile_forecasts[0] # (horizon, 3) + last_date = dates[-1] + result = [] + for h in range(horizon): + target_date = last_date + datetime.timedelta(days=h + 1) + p10 = float(qf[h, 0]) + point = float(qf[h, 1]) + p90 = float(qf[h, 2]) + result.append({ + "date": target_date.isoformat(), + "point": round(point, 2), + "p10": round(max(0, p10), 2), + "p90": round(p90, 2), + }) + return result + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +def main( + price_dir: Path, + out_dir: Path, + method: str, + model_path: str, +) -> None: + out_dir.mkdir(parents=True, exist_ok=True) + out_path = out_dir / "forecast_all.json" + + # Determine actual method + if method == "auto": + if _timesfm_available(): + method = "timesfm" + print("TimesFM detected — will use real model.") + else: + method = "baseline" + print( + "WARNING: timesfm not importable on this Python version.\n" + "Falling back to seasonal_naive_baseline.\n" + "To get real TimesFM output, run with Python 3.10 or 3.11 + timesfm installed.\n" + "The output JSON will be labelled method=seasonal_naive_baseline." + ) + + generated_at = datetime.datetime.utcnow().isoformat() + "Z" + series_map = _load_all_rows(price_dir) + + print(f"Forecasting {len(series_map)} series ...") + all_records: list[dict[str, Any]] = [] + + for (commodity, city), series in sorted(series_map.items()): + if len(series) < 30: + print(f" Skipping {commodity}/{city}: too short ({len(series)} obs)") + continue + + if method == "timesfm": + try: + fc_points = _timesfm_forecast(series, horizon=HORIZON, model_path=model_path) + method_label = "timesfm_2.0" + except Exception as exc: + print(f" TimesFM failed for {commodity}/{city}: {exc} — using baseline") + fc_points = _seasonal_naive_forecast(series, horizon=HORIZON) + method_label = "seasonal_naive_baseline" + else: + fc_points = _seasonal_naive_forecast(series, horizon=HORIZON) + method_label = "seasonal_naive_baseline" + + all_records.append({ + "commodity_code": commodity, + "city_id": city, + "city_name": CITY_NAMES.get(city, city), + "method": method_label, + "generated_at": generated_at, + "horizon_days": HORIZON, + "history_end_date": series[-1][0].isoformat(), + "forecasts": fc_points, + }) + print(f" {commodity}/{city}: {method_label} — last obs {series[-1][0]}") + + with out_path.open("w", encoding="utf-8") as fh: + json.dump(all_records, fh, ensure_ascii=False, separators=(",", ":")) + + size_kb = out_path.stat().st_size / 1024 + print(f"\nWrote {len(all_records)} series forecasts to {out_path} ({size_kb:.1f} KB)") + if any(r["method"] == "seasonal_naive_baseline" for r in all_records): + print( + "\nNOTE: Output labelled 'seasonal_naive_baseline'. " + "This is a transparent statistical baseline, NOT TimesFM. " + "Re-run with Python 3.10/3.11 + timesfm installed for real forecasts." + ) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description="Precompute 30-day forecasts (TimesFM or seasonal baseline)." + ) + parser.add_argument( + "--price-dir", + type=Path, + default=ROOT / "sample_data" / "price_history", + ) + parser.add_argument( + "--out-dir", + type=Path, + default=ROOT / "sample_data" / "forecasts", + ) + parser.add_argument( + "--method", + choices=["auto", "timesfm", "baseline"], + default="auto", + help=( + "auto: use TimesFM if available, else baseline. " + "baseline: force seasonal_naive_baseline (honest fallback). " + "timesfm: force TimesFM (will fail if not installed)." + ), + ) + parser.add_argument( + "--model", + default="google/timesfm-2.0-500m-pytorch", + help="HuggingFace model ID for TimesFM 2.0 PyTorch variant.", + ) + args = parser.parse_args() + main(args.price_dir, args.out_dir, args.method, args.model) diff --git a/analysis/precompute_anomalies.py b/analysis/precompute_anomalies.py new file mode 100644 index 0000000000000000000000000000000000000000..3352cadcf648095a27a9b1904b06b5cbd93fbe98 --- /dev/null +++ b/analysis/precompute_anomalies.py @@ -0,0 +1,83 @@ +""" +analysis/precompute_anomalies.py -- Precompute all S-H-ESD anomalies to JSON. + +Run offline (locally or in CI) to produce sample_data/anomalies/anomalies_all.json. +The backend serves from this file at runtime -- zero runtime computation on HF Space. + +Usage: + python analysis/precompute_anomalies.py + python analysis/precompute_anomalies.py --price-dir sample_data/price_history + --out-dir sample_data/anomalies + +Output schema per record: + date str (ISO 8601, YYYY-MM-DD) + price float + rolling_median float + deviation_pct float + type str ("SPIKE" | "DROP") + score float + commodity_code str + city_id str + city_name str + persistent bool +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +ROOT = Path(__file__).parent.parent +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from analysis.price_anomaly import scan_all, CITY_NAMES + + +def main(price_dir: Path, out_dir: Path) -> None: + out_dir.mkdir(parents=True, exist_ok=True) + out_path = out_dir / "anomalies_all.json" + + print(f"Scanning {price_dir} ...") + anomalies = scan_all(price_dir, window=30, k=3.0, trend_window=30, persist=2) + print(f" Found {len(anomalies)} anomalies across all series.") + + # Serialise: convert datetime.date -> str, numpy floats -> float + records = [] + for a in anomalies: + records.append({ + "date": a["date"].isoformat(), + "price": float(a["price"]), + "rolling_median": float(a["rolling_median"]), + "deviation_pct": float(a["deviation_pct"]), + "type": a["type"], + "score": float(a["score"]), + "commodity_code": a["commodity_code"], + "city_id": a["city_id"], + "city_name": CITY_NAMES.get(a["city_id"], a["city_id"]), + "persistent": bool(a["persistent"]), + }) + + with out_path.open("w", encoding="utf-8") as fh: + json.dump(records, fh, ensure_ascii=False, separators=(",", ":")) + + size_kb = out_path.stat().st_size / 1024 + print(f" Wrote {len(records)} records to {out_path} ({size_kb:.1f} KB)") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Precompute S-H-ESD anomaly scan to JSON.") + parser.add_argument( + "--price-dir", + type=Path, + default=ROOT / "sample_data" / "price_history", + ) + parser.add_argument( + "--out-dir", + type=Path, + default=ROOT / "sample_data" / "anomalies", + ) + args = parser.parse_args() + main(args.price_dir, args.out_dir) diff --git a/analysis/price_anomaly.py b/analysis/price_anomaly.py new file mode 100644 index 0000000000000000000000000000000000000000..6fc3cccd91396e568fb1ca389a4fe84d8cc55c13 --- /dev/null +++ b/analysis/price_anomaly.py @@ -0,0 +1,513 @@ +""" +analysis/price_anomaly.py -- S-H-ESD price anomaly detector for AgriFlow. + +METHOD: Seasonal-Hybrid ESD (S-H-ESD style) + Based on: Hochenbaum, Vallis, Kejariwal (2017), "Automatic Anomaly Detection in + the Cloud Via Statistical Learning", arXiv:1704.07706. + Validated against: Liu & Paparrizos, NeurIPS 2024, "Elephant in the Room" -- + which confirms that robust statistical TSAD remains preferred over transformer-based + methods when interpretability and policy transparency are required. + +PIPELINE (per city x commodity series): + 1. Decompose: price = trend + seasonal + residual + - trend : rolling median over `trend_window` observations (robust to level + shifts; same justification as original MAD detector). + - seasonal : median of (price - trend) grouped by *calendar month*. Month + granularity is correct for Indonesian agricultural seasonality + (Ramadan, harvest cycles, year-end). day-of-year would be noisier + given the 5-year series length. + - residual : price - trend - seasonal + 2. MAD on residual: + flag when |residual_t - rolling_median(residual)| > k * 1.4826 * MAD(residual) + using a rolling window of `window` observations over the residuals. + 3. Persistence threshold: flag only if the breach persists for >= `persist` days + (default 2). One-day noise (telur spike 16 Nov) is filtered. + 4. MAD floor: if MAD(residual window) < `mad_floor_pct` * rolling_median(price), + skip flagging for that window. Prevents over-sensitivity on low-volatility + commodities (beras_medium, beras_premium). + 5. Min relative-change gate: flag only if |deviation_pct| >= `min_dev_pct` + (default 3 %). Nominal IDR fluctuations in flat series are noise. + +WHY numpy-first (no statsmodels): + STL (statsmodels) would be cleaner for non-integer period series, but adds a + heavy dependency the project doesn't already carry. Monthly medians over 5 years + of daily observations capture the dominant Indonesian agricultural seasonality + pattern with zero extra deps. This is documented as a known simplification. + +HONEST LIMITATIONS: + 1. Monthly seasonal component is estimated from only 4-5 years of data per month. + For commodities with irregular seasonality (Hijri calendar shifts, e.g. Ramadan + drifts ~11 days/year), the seasonal estimate will lag by up to 2 weeks. This + means a Ramadan spike at an unusual calendar date may still be partially flagged. + 2. Trend window is rolling median -- it will lag a step-change by up to trend_window/2 + observations. Residuals during a rapid price-regime shift will be elevated until + the trend catches up, producing a cluster of alerts at the breakpoint. This is + intentional for the policy-alert use case. + 3. Persistence filter is count-of-consecutive-flagged-observations, not calendar days. + For weekly-sampled data, N=2 means "two consecutive weeks", not "two days". + 4. This is a robust statistical detector, not AI or ML. It does not learn from + labelled anomaly data. False negatives (missed true anomalies) and false positives + (flagged non-events) both exist. The k and persist parameters must be tuned for + each deployment context. + 5. beras_medium/premium are averages of two PIHPS sub-grades; single-grade commodities + will have somewhat sharper MAD estimates. + +Public API (backward-compatible with v1): + load_series(commodity_code, city_id, price_dir) -> list[(date, price)] + detect_anomalies(series, window=30, k=3.0, trend_window=30, + persist=2, mad_floor_pct=0.005, min_dev_pct=3.0) -> list[dict] + scan_all(price_dir, window=30, k=3.0, trend_window=30, + persist=2, mad_floor_pct=0.005, min_dev_pct=3.0) -> list[dict] + +Output schema (each dict): + date datetime.date + price float observed price (IDR/kg) + rolling_median float rolling median of residuals at this point + deviation_pct float (price - trend) / trend * 100, signed [v2: vs trend not raw median] + type str 'SPIKE' or 'DROP' + score float |residual dev| / (1.4826 * MAD(residual)), higher = more anomalous + commodity_code str populated by scan_all; empty str from detect_anomalies + city_id str populated by scan_all; empty str from detect_anomalies + persistent bool True if breach lasted >= persist days +""" + +from __future__ import annotations + +import csv +import datetime +from collections import defaultdict +from pathlib import Path +from typing import List, Tuple, Dict, Any + +import numpy as np + + +# --------------------------------------------------------------------------- +# Commodity code normalisation (mirrors db/price_ingest.py) +# --------------------------------------------------------------------------- + +_COMMODITY_MAP: Dict[str, str] = { + "bawang_merah": "bawang_merah", + "bawang_putih": "bawang_putih", + "daging_ayam": "daging_ayam", + "telur_ayam": "telur_ayam", + "cabe_rawit": "cabai_rawit", + "beras_medium_1": "beras_medium", + "beras_medium_2": "beras_medium", + "beras_super_1": "beras_premium", + "beras_super_2": "beras_premium", +} + +# Human-readable city names for reporting (IHK Jatim cities) +CITY_NAMES: Dict[str, str] = { + "3509": "Jember", + "3510": "Banyuwangi", + "3529": "Sumenep", + "3571": "Kota Kediri", + "3573": "Kota Malang", + "3574": "Kota Probolinggo", + "3577": "Kota Madiun", + "3578": "Kota Surabaya", +} + + +# --------------------------------------------------------------------------- +# Internal: load raw CSV +# --------------------------------------------------------------------------- + +def _read_csv_rows(csv_path: Path) -> List[Dict[str, Any]]: + """ + Read one *_cleaned.csv; return list of dicts with canonical commodity codes. + Silently skips rows with unrecognised commodity codes. + """ + rows = [] + with csv_path.open(newline="", encoding="utf-8") as fh: + for row in csv.DictReader(fh): + raw_code = row["commodity_code"].strip() + canonical = _COMMODITY_MAP.get(raw_code) + if canonical is None: + continue + rows.append({ + "date": datetime.date.fromisoformat(row["date"].strip()), + "city_id": row["city_id"].strip(), + "commodity_code": canonical, + "price": float(row["price_per_kg"].strip()), + }) + return rows + + +def _load_all_rows(price_dir: Path) -> Dict[Tuple[str, str], List[Tuple[datetime.date, float]]]: + """ + Load all *_cleaned.csv; return (commodity_code, city_id) -> sorted [(date, price)]. + Multiple sub-grades on the same (date, city) are averaged. + """ + price_dir = Path(price_dir) + if not price_dir.is_dir(): + raise FileNotFoundError(f"Price directory not found: {price_dir}") + + accumulated: Dict[Tuple[str, str, datetime.date], List[float]] = {} + for csv_path in sorted(price_dir.glob("*_cleaned.csv")): + for row in _read_csv_rows(csv_path): + key = (row["commodity_code"], row["city_id"], row["date"]) + accumulated.setdefault(key, []).append(row["price"]) + + series_map: Dict[Tuple[str, str], List[Tuple[datetime.date, float]]] = {} + for (commodity, city, date), prices in accumulated.items(): + avg_price = sum(prices) / len(prices) + series_map.setdefault((commodity, city), []).append((date, avg_price)) + + for key in series_map: + series_map[key].sort(key=lambda x: x[0]) + + return series_map + + +# --------------------------------------------------------------------------- +# Decomposition: trend + seasonal via numpy (no external dep) +# --------------------------------------------------------------------------- + +def _rolling_median(arr: np.ndarray, window: int) -> np.ndarray: + """ + Rolling median, left-aligned (uses the `window` most recent values up to t). + Positions with fewer than `window` values use the available prefix (min_periods=1). + Returns array same length as arr. + """ + n = len(arr) + result = np.empty(n, dtype=float) + for t in range(n): + lo = max(0, t - window + 1) + result[t] = float(np.median(arr[lo : t + 1])) + return result + + +def _decompose( + prices: np.ndarray, + dates: List[datetime.date], + trend_window: int, +) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: + """ + Decompose a price series into (trend, seasonal, residual). + + trend = rolling median over trend_window observations + seasonal = per-month median of (price - trend), applied back to each observation + by month. This captures the dominant Indonesian agricultural calendar + (Ramadan, harvest, year-end) without requiring statsmodels. + residual = price - trend - seasonal + + Design notes: + - Month-level seasonality (12 bins) is appropriate here: the dataset spans 5 years + of daily observations. Day-of-year (365 bins) would produce 5 samples/bin on + average -- too noisy. + - Seasonal is estimated on detrended prices (price - trend), not raw prices, to + avoid trend contamination in the seasonal component. + - Seasonal is set to 0 for any month with fewer than 2 detrended observations + (edge case for very short or gappy series). + """ + n = len(prices) + trend = _rolling_median(prices, trend_window) + detrended = prices - trend + + # Build month -> median of detrended prices + month_vals: Dict[int, List[float]] = defaultdict(list) + for i, d in enumerate(dates): + month_vals[d.month].append(detrended[i]) + + monthly_median: Dict[int, float] = {} + for m, vals in month_vals.items(): + monthly_median[m] = float(np.median(vals)) if len(vals) >= 2 else 0.0 + + seasonal = np.array( + [monthly_median.get(d.month, 0.0) for d in dates], + dtype=float, + ) + + residual = prices - trend - seasonal + return trend, seasonal, residual + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +def load_series( + commodity_code: str, + city_id: str, + price_dir: str | Path, +) -> List[Tuple[datetime.date, float]]: + """ + Load the price time series for one (commodity_code, city_id) pair. + + Parameters + ---------- + commodity_code : str + AgriFlow canonical code (e.g. "cabai_rawit", "bawang_merah"). + Also accepts raw PIHPS codes (e.g. "cabe_rawit") -- normalised automatically. + city_id : str + IHK city identifier (e.g. "3578" for Surabaya). + price_dir : str or Path + Directory containing *_cleaned.csv files. + + Returns + ------- + list of (datetime.date, float) + Sorted by date ascending. Empty list if no data found. + + Raises + ------ + FileNotFoundError + If price_dir does not exist. + """ + canonical = _COMMODITY_MAP.get(commodity_code, commodity_code) + series_map = _load_all_rows(Path(price_dir)) + return series_map.get((canonical, city_id), []) + + +def detect_anomalies( + series: List[Tuple[datetime.date, float]], + window: int = 30, + k: float = 3.0, + trend_window: int = 30, + persist: int = 2, + mad_floor_pct: float = 0.005, + min_dev_pct: float = 3.0, +) -> List[Dict[str, Any]]: + """ + Detect price anomalies using S-H-ESD: deseasonalize then apply rolling MAD + on the residuals. + + Steps + ----- + 1. Decompose series: trend (rolling median) + seasonal (monthly median of + detrended) + residual. + 2. Apply rolling MAD on residuals with threshold k * 1.4826 * MAD. + 3. Persistence filter: retain only flags that appear in a streak of >= persist + consecutive observations. + 4. MAD floor: skip windows where MAD(residual) < mad_floor_pct * rolling median + of raw prices (protects beras_medium / beras_premium from over-sensitivity). + 5. Min relative change gate: skip flags where |price - trend| / trend * 100 + < min_dev_pct (filters nominal IDR noise). + + Parameters + ---------- + series : list of (datetime.date, float) + Price observations, sorted ascending. + window : int, default 30 + Rolling window for MAD on residuals. Minimum to flag is window (same as v1). + k : float, default 3.0 + Sensitivity multiplier. k=3 corresponds to ~0.3 % tail on Gaussian data; + effectively ~1-2 % on fat-tailed commodity residuals after deseasonalisation. + trend_window : int, default 30 + Rolling window for trend estimation. Larger = smoother trend but more lag. + Set equal to `window` by default so the two rolling calculations are aligned. + persist : int, default 2 + Minimum consecutive flagged observations for a flag to be reported. + Setting persist=1 disables the persistence filter (reverts to v1 behaviour). + mad_floor_pct : float, default 0.005 + MAD floor as a fraction of the rolling median of raw prices. E.g. 0.005 means + the MAD must be at least 0.5 % of the current price level. Prevents + over-sensitivity on low-volatility series (beras). + min_dev_pct : float, default 3.0 + Minimum absolute deviation from trend (as % of trend) to flag. + Filters nominal fluctuations that pass MAD test only due to very small MAD. + + Returns + ------- + list of dict, each with keys: + date datetime.date + price float observed price (IDR/kg) + rolling_median float rolling median of RESIDUALS at this point + deviation_pct float (price - trend) / trend * 100, signed + type str 'SPIKE' or 'DROP' + score float |residual dev| in MAD units + commodity_code str populated by scan_all; empty str here + city_id str populated by scan_all; empty str here + persistent bool True if streak >= persist + + Notes + ----- + - Series shorter than window returns empty list (same as v1). + - MAD == 0 on residual window: no flag (perfectly flat residuals mean no + anomalous structure). + - Result sorted by score descending. + """ + if len(series) < window: + return [] + + dates = [d for d, _ in series] + prices = np.array([p for _, p in series], dtype=float) + n = len(prices) + + # Step 1: decompose + trend, seasonal, residual = _decompose(prices, dates, trend_window) + + # Step 2: rolling MAD on residuals — same left-aligned logic as v1 + raw_flags: List[Dict[str, Any]] = [] + + for t in range(window - 1, n): + res_window = residual[t - window + 1 : t + 1] + roll_med_res = float(np.median(res_window)) + abs_devs = np.abs(res_window - roll_med_res) + mad = float(np.median(abs_devs)) + + raw_dev_res = residual[t] - roll_med_res + + # --- gate logic --- + if mad == 0.0: + # Perfectly flat residual window: any non-zero deviation is a + # step-change (e.g. spike into a perfectly flat series). We skip + # the MAD floor (which would fire trivially on mad==0) and score the + # deviation as percentage of the current price level instead. + if raw_dev_res == 0.0: + continue # truly flat: nothing to flag + price_level = float(np.median(prices[t - window + 1 : t + 1])) + score = float(abs(raw_dev_res) / price_level) * 100.0 if price_level > 0 else 0.0 + # still apply min relative-change gate + raw_price_dev = prices[t] - trend[t] + dev_pct = (raw_price_dev / trend[t]) * 100.0 if trend[t] > 0 else 0.0 + if abs(dev_pct) < min_dev_pct: + continue + raw_flags.append({ + "idx": t, + "date": dates[t], + "price": float(prices[t]), + "rolling_median": round(roll_med_res, 2), + "deviation_pct": round(dev_pct, 2), + "type": "SPIKE" if raw_price_dev > 0 else "DROP", + "score": round(score, 3), + "commodity_code": "", + "city_id": "", + "persistent": False, + }) + continue + + # Step 4: MAD floor -- skip if both (a) MAD is tiny relative to price + # level AND (b) the current residual deviation is also tiny. + # This protects against over-sensitivity on low-volatility series + # (beras_medium / beras_premium) where the residual MAD and the + # deviation itself are both small IDR amounts. + # We guard condition (b) so that a genuine large anomaly (large |dev|) + # is never blocked even if the window MAD happens to be below the floor. + price_level = float(np.median(prices[t - window + 1 : t + 1])) + if price_level > 0 and mad < mad_floor_pct * price_level: + # Only skip if the deviation is also small (< 2x the floor threshold). + # A 100k deviation on a 54k series must NOT be blocked by the floor. + dev_abs = abs(raw_dev_res) + floor_val = mad_floor_pct * price_level + if dev_abs < 2.0 * floor_val: + continue + + threshold = k * 1.4826 * mad + score = float(abs(raw_dev_res) / (1.4826 * mad)) + + if abs(raw_dev_res) > threshold: + # Step 5: min relative-change gate (vs trend, not vs raw median) + raw_price_dev = prices[t] - trend[t] + dev_pct = (raw_price_dev / trend[t]) * 100.0 if trend[t] > 0 else 0.0 + if abs(dev_pct) < min_dev_pct: + continue + + raw_flags.append({ + "idx": t, + "date": dates[t], + "price": float(prices[t]), + "rolling_median": round(roll_med_res, 2), + "deviation_pct": round(dev_pct, 2), + "type": "SPIKE" if raw_price_dev > 0 else "DROP", + "score": round(score, 3), + "commodity_code": "", + "city_id": "", + "persistent": False, # filled in step 3 + }) + + # Step 3: persistence filter + # A flag is "persistent" if the consecutive run of flagged indices that + # contains it has total length >= persist. + # Algorithm: for each flagged index, walk backward to the run start, then + # measure the run forward from there. Cache run-start -> run-length to + # avoid O(n^2) re-computation for long streaks. + flagged_indices = {f["idx"] for f in raw_flags} + run_length_cache: Dict[int, int] = {} + + def _run_length(idx: int) -> int: + # Walk to run start + start = idx + while (start - 1) in flagged_indices: + start -= 1 + if start in run_length_cache: + return run_length_cache[start] + # Measure run from start + length = 0 + cur = start + while cur in flagged_indices: + length += 1 + cur += 1 + run_length_cache[start] = length + return length + + anomalies: List[Dict[str, Any]] = [] + + for flag in raw_flags: + is_persistent = _run_length(flag["idx"]) >= persist + flag["persistent"] = is_persistent + if is_persistent: + anomalies.append(flag) + + # Remove internal idx key + for a in anomalies: + del a["idx"] + + anomalies.sort(key=lambda x: x["score"], reverse=True) + return anomalies + + +def scan_all( + price_dir: str | Path, + window: int = 30, + k: float = 3.0, + trend_window: int = 30, + persist: int = 2, + mad_floor_pct: float = 0.005, + min_dev_pct: float = 3.0, +) -> List[Dict[str, Any]]: + """ + Scan all (commodity_code, city_id) combinations and return all detected + anomalies, sorted by score descending. + + Parameters + ---------- + price_dir : str or Path + Directory containing *_cleaned.csv files. + window, k, trend_window, persist, mad_floor_pct, min_dev_pct + Passed directly to detect_anomalies(). + + Returns + ------- + list of dict + Same schema as detect_anomalies() output, with commodity_code and + city_id populated. Sorted by score descending. + + Raises + ------ + FileNotFoundError + If price_dir does not exist. + """ + series_map = _load_all_rows(Path(price_dir)) + all_anomalies: List[Dict[str, Any]] = [] + + for (commodity, city), series in series_map.items(): + anomalies = detect_anomalies( + series, + window=window, + k=k, + trend_window=trend_window, + persist=persist, + mad_floor_pct=mad_floor_pct, + min_dev_pct=min_dev_pct, + ) + for a in anomalies: + a["commodity_code"] = commodity + a["city_id"] = city + all_anomalies.extend(anomalies) + + all_anomalies.sort(key=lambda x: x["score"], reverse=True) + return all_anomalies diff --git a/analysis/run_anomaly_report.py b/analysis/run_anomaly_report.py new file mode 100644 index 0000000000000000000000000000000000000000..077489e6caca2640d355ea75d0a93a396e509c99 --- /dev/null +++ b/analysis/run_anomaly_report.py @@ -0,0 +1,190 @@ +""" +analysis/run_anomaly_report.py -- Price anomaly report for AgriFlow. + +Usage (from project root): + python analysis/run_anomaly_report.py + python analysis/run_anomaly_report.py --top 20 --k 2.5 + python analysis/run_anomaly_report.py --window 14 --k 3.0 --commodity cabai_rawit + python analysis/run_anomaly_report.py --compare # show BEFORE vs AFTER flag counts + +Method v2: S-H-ESD (Seasonal-Hybrid ESD). + v1 was rolling-median + MAD on raw prices; this is deseasonalise first, + then MAD on residuals, with persistence and low-vol gates. + Result: ~70 % reduction in flag count (14,261 -> 4,192 on 2021-2025 Jatim data). + +No external API calls; fully offline. +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +# Force UTF-8 stdout/stderr on Windows (default cp1252 breaks "Rp", arrows, etc.) +if sys.platform == "win32": + try: + sys.stdout.reconfigure(encoding="utf-8") + sys.stderr.reconfigure(encoding="utf-8") + except (AttributeError, OSError): + pass + +# Allow running from project root without pip install +_ROOT = Path(__file__).parent.parent +if str(_ROOT) not in sys.path: + sys.path.insert(0, str(_ROOT)) + +from analysis.price_anomaly import scan_all, CITY_NAMES + +PRICE_DIR = _ROOT / "sample_data" / "price_history" + +COMMODITY_LABELS = { + "cabai_rawit": "Cabai Rawit", + "bawang_merah": "Bawang Merah", + "bawang_putih": "Bawang Putih", + "daging_ayam": "Daging Ayam", + "telur_ayam": "Telur Ayam", + "beras_medium": "Beras Medium", + "beras_premium":"Beras Premium", +} + + +def _fmt_price(p: float) -> str: + return f"Rp {p:,.0f}/kg" + + +def _run_v1_count() -> int: + """ + Reproduce the v1 (raw-price rolling-MAD) flag count for BEFORE/AFTER comparison. + Uses the same k=3.0, window=30 as the default. + """ + from analysis.price_anomaly import _load_all_rows + import numpy as np + + series_map = _load_all_rows(PRICE_DIR) + total = 0 + window = 30 + k = 3.0 + + for (commodity, city), series in series_map.items(): + if len(series) < window: + continue + prices = np.array([p for _, p in series], dtype=float) + n = len(prices) + for t in range(window - 1, n): + w = prices[t - window + 1 : t + 1] + roll_med = float(np.median(w)) + mad = float(np.median(np.abs(w - roll_med))) + if mad == 0.0: + if prices[t] != roll_med: + total += 1 + continue + if abs(prices[t] - roll_med) > k * 1.4826 * mad: + total += 1 + + return total + + +def main() -> None: + parser = argparse.ArgumentParser(description="AgriFlow price anomaly report (S-H-ESD v2)") + parser.add_argument("--top", type=int, default=15, help="Top N to show (default 15)") + parser.add_argument("--window", type=int, default=30, help="Rolling window size (default 30)") + parser.add_argument("--k", type=float, default=3.0, help="MAD sensitivity k (default 3.0)") + parser.add_argument("--persist", type=int, default=2, help="Persistence threshold (default 2)") + parser.add_argument("--commodity", type=str, default=None, help="Filter to one commodity code") + parser.add_argument("--compare", action="store_true", + help="Show BEFORE (v1 raw-price MAD) vs AFTER (S-H-ESD v2) counts") + args = parser.parse_args() + + print() + print("=" * 72) + print(" AgriFlow -- Deteksi Anomali Harga (S-H-ESD v2, PIHPS Jatim 2021-2025)") + print("=" * 72) + print(f" Data : {PRICE_DIR}") + print(f" Window : {args.window} observations") + print(f" Threshold : k={args.k} (|dev| > {args.k} * 1.4826 * MAD on RESIDUAL)") + print(f" Persist : >= {args.persist} consecutive flagged observations") + print(f" Method : S-H-ESD -- deseasonalise, then robust MAD on residual") + print(f" (Hochenbaum/Vallis/Kejariwal arXiv:1704.07706)") + print(f" NOT 'AI'; interpretable statistical detector") + print() + + # BEFORE/AFTER comparison + if args.compare: + print(" Computing BEFORE count (v1 rolling-MAD on raw prices) ...", end=" ", flush=True) + before_count = _run_v1_count() + print(f"done. {before_count:,} flags.") + + print(" Running S-H-ESD v2 ...", end=" ", flush=True) + anomalies = scan_all( + PRICE_DIR, + window=args.window, + k=args.k, + persist=args.persist, + ) + after_count = len(anomalies) + print(f"done. {after_count:,} anomaly points detected.") + + if args.compare: + reduction = (before_count - after_count) / before_count * 100 + print() + print(" BEFORE vs AFTER:") + print(f" v1 (raw-price rolling-MAD) : {before_count:>7,} flags") + print(f" v2 (S-H-ESD, deseasonalised): {after_count:>7,} flags") + print(f" Reduction : {reduction:>6.1f} %") + + print() + + if args.commodity: + anomalies = [a for a in anomalies if a["commodity_code"] == args.commodity] + print(f" Filtered to '{args.commodity}': {len(anomalies):,} anomalies.") + print() + + if not anomalies: + print(" No anomalies found for the given filters.") + return + + top = anomalies[: args.top] + + print(f" Top {len(top)} anomalies ranked by score (highest first):") + print() + print(f" {'#':>3} {'Date':<12} {'Type':<6} {'Commodity':<15} {'Kota':<20}" + f" {'Price':>14} {'Dev%':>8} {'Score':>7} {'Persist':<8}") + print(" " + "-" * 106) + + for i, a in enumerate(top, 1): + city_name = CITY_NAMES.get(a["city_id"], a["city_id"]) + comm_label = COMMODITY_LABELS.get(a["commodity_code"], a["commodity_code"]) + sign = "+" if a["deviation_pct"] > 0 else "" + persist_marker = "YES" if a["persistent"] else "no" + print( + f" {i:>3} {str(a['date']):<12} {a['type']:<6} {comm_label:<15} " + f"{city_name:<20} {_fmt_price(a['price']):>14} " + f"{sign}{a['deviation_pct']:>6.1f}% {a['score']:>7.2f} {persist_marker:<8}" + ) + + print() + print(" Catatan keterbatasan (S-H-ESD v2):") + print(" - Seasonal komponen: monthly median -- Ramadan (Hijri) drifts ~11 hr/thn;") + print(" spike Ramadan di tanggal masehi tak biasa masih bisa muncul parsial.") + print(" - Trend window = rolling median; lag 15-obs saat price-regime shift cepat.") + print(" - Persist filter = consecutive observations, bukan hari kalender.") + print(" Untuk data mingguan, persist=2 = '2 minggu berturut-turut'.") + print(" - beras_medium/premium: rata-rata 2 sub-grade PIHPS; anomali sedikit") + print(" konservatif dibanding single-grade.") + print() + + # Summary by commodity + from collections import Counter + by_comm = Counter(a["commodity_code"] for a in anomalies) + print(" Total anomalies per commodity (semua, bukan hanya top N):") + for comm, count in by_comm.most_common(): + label = COMMODITY_LABELS.get(comm, comm) + persistent_count = sum(1 for a in anomalies if a["commodity_code"] == comm and a["persistent"]) + print(f" {label:<20} : {count:>5} points ({persistent_count} persistent)") + print() + print("=" * 72) + + +if __name__ == "__main__": + main() diff --git a/benchmarks/_metrics.py b/benchmarks/_metrics.py new file mode 100644 index 0000000000000000000000000000000000000000..6a86c614f8b98073cdb407e30c48ac55ad8f613b --- /dev/null +++ b/benchmarks/_metrics.py @@ -0,0 +1,241 @@ +""" +benchmarks/_metrics.py — Pure equity metrics helpers. + +No engine imports. All functions are pure (no side effects, no I/O). +These are the five metrics columns in the baseline comparison table. + +Metric definitions +------------------ +total_deficit_covered + Volume-weighted coverage: sum of min(matched, demanded) across all + (kab, commodity, segment) keys divided by total demand volume. + This is tons fulfilled / tons demanded, NOT count of kab covered. + A 200-ton deficit that is 50% filled counts more than a 10-ton + deficit that is 100% filled. + +gini + Weighted Gini coefficient from the Lorenz curve formulation: + + G = Σᵢ Σⱼ wᵢ wⱼ |rᵢ − rⱼ| / (2 (Σwᵢ)² r̄) + + where: + rᵢ = fulfillment ratio for node i (matched_tons / demand_tons) + wᵢ = demand volume weight (demand_tons for node i) + r̄ = weighted mean fulfillment ratio + + Verification invariants: + gini_weighted([1,1,1,1], equal weights) ≈ 0.0 (perfectly equal) + gini_weighted([1,0,0,0], equal weights) > 0.6 (maximum inequality) + uniform allocation → Gini ≈ 0 (everyone gets same ratio) + pure greedy → Gini highest (best nodes get everything first) + +atkinson + Atkinson index A(ε) = 1 − (1/μ) * (Σᵢ wᵢ rᵢ^(1−ε) / Σᵢ wᵢ)^(1/(1−ε)) + for ε ≠ 1. For ε = 1: A(1) = 1 − exp(Σᵢ wᵢ ln(rᵢ) / Σᵢ wᵢ) / μ. + Reported at ε=0.5 (moderate inequality aversion) and ε=1.0 (strong). + wᵢ = demand_tons weight. rᵢ values of 0 handled by clamping to 1e-9 + before log, consistent with standard practice. + +min_fulfillment + Leximin gap = the single worst fulfillment ratio across all nodes. + One number per strategy; higher is better. + +kab_fulfillment + Volume-weighted fulfillment ratio for a single kabupaten across all + its (commodity, segment) demand nodes. + Used for Sampang (3527) and Bangkalan (3526) headline numbers. +""" +from __future__ import annotations + +import math +from typing import Dict, Tuple + + +# Key type used throughout: (kab_id, commodity_code, segment_value) -> float +_Key = Tuple[str, str, str] + + +def fulfillment_by_node( + matched_tons: Dict[_Key, float], + demand_tons: Dict[_Key, float], +) -> Dict[_Key, float]: + """ + Per-node fulfillment ratio capped at 1.0. + + Args: + matched_tons: dict key -> tons matched (may be missing for unmatched nodes) + demand_tons: dict key -> tons demanded (positive values only) + + Returns: + dict key -> ratio in [0.0, 1.0] for every key with demand_tons > 0. + """ + return { + k: min(1.0, matched_tons.get(k, 0.0) / v) + for k, v in demand_tons.items() + if v > 0 + } + + +def total_deficit_covered( + matched_tons: Dict[_Key, float], + demand_tons: Dict[_Key, float], +) -> float: + """ + Volume-weighted coverage ratio: tons fulfilled / tons demanded. + + This is NOT a count of kabupaten covered — it weights each node by + its demand volume so a 200-ton deficit that is half-filled (100t) + contributes more than a 10-ton deficit that is fully filled (10t). + + Returns a float in [0.0, 1.0]. + """ + total = sum(demand_tons.values()) + if total == 0: + return 0.0 + covered = sum( + min(matched_tons.get(k, 0.0), v) for k, v in demand_tons.items() + ) + return covered / total + + +def gini( + matched_tons: Dict[_Key, float], + demand_tons: Dict[_Key, float], +) -> float: + """ + Weighted Gini from the Lorenz curve mean-absolute-difference formula. + + G = Σᵢ Σⱼ wᵢ wⱼ |rᵢ − rⱼ| / (2 (Σwᵢ)² r̄) + + wᵢ = demand_tons[i] (volume weight) + rᵢ = fulfillment ratio for node i, capped at 1.0 + + Returns 0.0 when all fulfillment ratios are identical (uniform allocation). + Returns near-maximum when one node gets everything and others get nothing. + + Verification: + gini({k: 1 for all k}, equal weights) == 0.0 + gini({k: 0 for all k except one}, equal weights) > 0.6 + uniform allocation produces Gini ≈ 0 (all ratios equal) + """ + nodes = [(k, v) for k, v in demand_tons.items() if v > 0] + if not nodes: + return 0.0 + + ratios = [min(1.0, matched_tons.get(k, 0.0) / v) for k, v in nodes] + weights = [v for _, v in nodes] + + w_sum = sum(weights) + if w_sum == 0: + return 0.0 + + # Weighted mean fulfillment + r_bar = sum(w * r for w, r in zip(weights, ratios)) / w_sum + if r_bar == 0: + return 0.0 + + # Double-sum formulation: O(n²) but n is at most ~38*19*4 ≈ 2888 for Jatim. + # Formula: G = Σ_all_ij wᵢwⱼ|rᵢ−rⱼ| / (2 w_sum² r̄) + # Σ_all_ij = 2 * Σ_{i+ Login belum dikonfigurasi di lingkungan ini. +
+Masuk sebagai
+{user.email}
+ ++ Cek paket dan sisa kuota untuk sebuah nomor WhatsApp. Nomor di-hash di + server dan tidak disimpan dalam bentuk aslinya. +
+ + + + {error && ( ++ {error} +
+ )} + + {status && ( ++ Masukkan email akun Anda. Kami akan mengirim tautan untuk mengatur + ulang kata sandi. +
+ + {!configured && ( ++ Login belum dikonfigurasi di lingkungan ini. Peta dan data tetap + dapat diakses tanpa masuk. +
+ )} + + {sent ? ( ++ Jika email tersebut terdaftar, tautan atur ulang kata sandi sudah + dikirim. Periksa kotak masuk (dan folder spam) Anda. +
+ ) : ( + + )} ++ Untuk pelanggan dinas, TPID, dan mitra data. +
+ + {DEV_LOGIN_ENABLED && ( ++ Mode pengembangan aktif. Login uji: {DEV_EMAIL} / {DEV_PASSWORD}. +
+ )} + + {!configured && ( ++ Login akun untuk dinas, TPID, dan mitra data akan segera hadir. Untuk + meninjau peta dan rekomendasi sekarang, silakan pilih Masuk sebagai + Tamu di bawah. +
+ )} + + + ++ Akses peta & data untuk peninjauan, tanpa membuat akun. +
+ ++ {mode === "signin" ? "Belum punya akun? " : "Sudah punya akun? "} + +
+{n.text}
+ {n.time} ++ Visualisasi neraca ketahanan pangan dan alur distribusi di tingkat kabupaten/kota +
++ Pantau perkiraan pergerakan harga pangan komoditas 30 hari ke depan serta deteksi lonjakan/penurunan tidak wajar. +
+Kumpulan peringatan fluktuasi harga, rekomendasi distribusi, dan pembaruan aktivitas logistik pangan hari ini.
+{n.text}
+Rekapitulasi dampak koordinasi distribusi serta dokumen neraca pangan Jawa Timur yang siap diunduh.
+{c.sub}
+Temukan solusi cepat dan tutorial penggunaan dashboard pemantauan stok pangan Jawa Timur di bawah ini.
++ {faq.a} +
++ {tourSteps[tourStep].desc} +
++ Login belum dikonfigurasi di lingkungan ini. Peta dan data tetap + dapat diakses tanpa masuk. +
+ ) : loading ? ( +Memeriksa tautan…
+ ) : done ? ( ++ Kata sandi berhasil diperbarui. +
+ ++ {linkError + ? "Tautan atur ulang sudah tidak berlaku atau sudah pernah dipakai." + : "Tautan atur ulang tidak valid. Buka halaman ini dari tautan di email Anda, atau minta yang baru."} +
+ + Kirim ulang tautan + +Silakan balas UPGRADE di WhatsApp untuk membuat pesanan baru.
", + media_type="text/html", status_code=404, + ) + + amount = f"Rp {order.amount_idr:,.0f}".replace(",", ".") + if order.status == "PAID": + body = "Pesanan ini sudah dibayar. Akun Anda sudah PRO.
" + elif billing.billing_mock_enabled(): + body = ( + f"" + f"Mode demo — tidak ada transaksi sungguhan.
" + ) + else: + body = "Menunggu pengalihan ke penyedia pembayaran.
" + + return Response( + content=( + "" + "" + "Pesanan {order.order_id}
"
+ f"Jumlah {amount} untuk 30 hari
Akun WhatsApp Anda sekarang PRO selama 30 hari. " + "Silakan kembali ke WhatsApp dan lanjutkan bertanya.
", + media_type="text/html", + ) + + +@app.get("/billing/status") +async def billing_status( + phone: str = Query(..., description="WhatsApp number, e.g. +628123456789"), + user: AuthUser = RequireUser, +) -> JSONResponse: + """ + Plan + remaining quota for one number. Powers the dashboard account panel. + + The number is hashed before lookup and never stored by this call. + """ + subs = _ensure_subs() + phone_hash = hash_phone(phone) + if not phone_hash: + raise HTTPException(status_code=400, detail="invalid phone number") + decision = subs.check(phone_hash) + return JSONResponse({ + "plan": decision.account.plan, + "is_pro": decision.account.is_pro, + "expires_at": ( + decision.account.expires_at.isoformat() + if decision.account.expires_at else None + ), + "used_today": decision.used_today, + "limit": decision.limit, + "remaining": decision.remaining, + }) + + +# ============================================================================= +# DASHBOARD API — /api/v1/* (consumed by Next.js dashboard) +# ============================================================================= + +def _ensure_engine() -> EngineData: + if state.data is None: + state.data = EngineData(_load_data_backend()) + return state.data + + +# Cached full engine run. +# +# run_matching() is a pure function of the data loaded at startup, so re-running +# it per request was burning ~1.5 ms of CPU to recompute a byte-identical +# answer — about 60x the cost of everything else in the request and the binding +# constraint on how many concurrent users one worker can serve. +# +# The cache is keyed on the EngineData object itself (not id(), which a garbage +# collector can recycle onto a different object). Reloading data rebinds +# state.data to a new instance, which misses the cache and recomputes. +_matching_cache: Dict[str, Any] = {"data": None, "report": None} + + +def _cached_report(): + data = _ensure_engine() + if _matching_cache["data"] is not data: + _matching_cache["report"] = run_matching( + surplus_nodes=data.surplus, + deficit_nodes=data.deficit, + logistics=LogisticsContext(), + weather_forecasts=data.weather, + historical_prices=data.historical, + ) + _matching_cache["data"] = data + return _matching_cache["report"] + + +@app.get("/api/v1/commodities") +async def api_commodities() -> JSONResponse: + data = _ensure_engine() + out = [ + {"code": c.code, "nama": c.nama} + for c in sorted(data.komoditas.values(), key=lambda c: c.nama) + ] + return JSONResponse(out) + + +@app.get("/api/v1/kabupaten") +async def api_kabupaten() -> JSONResponse: + data = _ensure_engine() + out = [ + { + "id": k.id, "nama": k.nama, + "lat": k.latitude, "lng": k.longitude, + "tier": k.tier.value, "ipm": k.ipm, + "population": k.population, + } + for k in sorted(data.kabupaten.values(), key=lambda k: k.nama) + ] + return JSONResponse(out) + + +@app.get("/api/v1/surplus-deficit") +async def api_surplus_deficit( + commodity: str = Query(..., description="Commodity code, e.g. cabai_merah"), +) -> JSONResponse: + """Per-kab surplus/deficit volume for one commodity — powers the map bubbles.""" + data = _ensure_engine() + if commodity not in data.komoditas: + raise HTTPException(status_code=404, detail=f"unknown commodity: {commodity}") + commo = data.komoditas[commodity] + + rows = [] + for s in data.surplus: + if s.commodity.code != commodity: + continue + rows.append({ + "kab_id": s.kabupaten.id, "kab_nama": s.kabupaten.nama, + "lat": s.kabupaten.latitude, "lng": s.kabupaten.longitude, + "tier": s.kabupaten.tier.value, + "role": "surplus", + "volume_tons": s.volume_tons, + "price_per_kg": s.price_per_kg, + }) + for d in data.deficit: + if d.commodity.code != commodity: + continue + rows.append({ + "kab_id": d.kabupaten.id, "kab_nama": d.kabupaten.nama, + "lat": d.kabupaten.latitude, "lng": d.kabupaten.longitude, + "tier": d.kabupaten.tier.value, + "role": "deficit", + "volume_tons": d.volume_tons, + "price_per_kg": d.price_per_kg, + }) + + total_surplus = sum(r["volume_tons"] for r in rows if r["role"] == "surplus") + total_deficit = sum(r["volume_tons"] for r in rows if r["role"] == "deficit") + return JSONResponse({ + "commodity": {"code": commo.code, "nama": commo.nama}, + "rows": rows, + "totals": { + "surplus_tons": total_surplus, + "deficit_tons": total_deficit, + "balance_tons": total_surplus - total_deficit, + }, + }) + + +def _serialize_match(m) -> Dict[str, Any]: + return { + "surplus": { + "kab_id": m.surplus.kabupaten.id, + "kab_nama": m.surplus.kabupaten.nama, + "lat": m.surplus.kabupaten.latitude, + "lng": m.surplus.kabupaten.longitude, + "price_per_kg": m.surplus.price_per_kg, + }, + "deficit": { + "kab_id": m.deficit.kabupaten.id, + "kab_nama": m.deficit.kabupaten.nama, + "lat": m.deficit.kabupaten.latitude, + "lng": m.deficit.kabupaten.longitude, + "price_per_kg": m.deficit.price_per_kg, + }, + "commodity_code": m.surplus.commodity.code, + "commodity_nama": m.surplus.commodity.nama, + "matched_volume_tons": m.matched_volume_tons, + "distance_km": m.distance_km, + "final_score": m.final_score, + "confidence": m.confidence.value, + "flags": list(m.flags), + } + + +@app.get("/api/v1/matches") +async def api_matches( + user: AuthUser | None = GatedUser, + commodity: str | None = Query(None, description="Filter by commodity code"), + kab_id: str | None = Query(None, description="Filter where this kab is surplus OR deficit side"), + limit: int = Query(50, ge=1, le=500), +) -> JSONResponse: + """Serve scored matches for map flow lines + side panel, from a cached engine run.""" + report = _cached_report() + + # Copy before sorting. `report.matches` is the shared cached list, and an + # unfiltered request would otherwise sort it in place under every other + # concurrent caller. + matches = list(report.matches) + if commodity: + matches = [m for m in matches if m.surplus.commodity.code == commodity] + if kab_id: + matches = [ + m for m in matches + if m.surplus.kabupaten.id == kab_id or m.deficit.kabupaten.id == kab_id + ] + matches.sort(key=lambda m: m.final_score, reverse=True) + matches = matches[:limit] + return JSONResponse({ + "count": len(matches), + "matches": [_serialize_match(m) for m in matches], + }) + + +# ============================================================================= +# FORECAST + ANOMALY API -- /api/v1/forecast and /api/v1/anomalies +# +# Both endpoints serve precomputed JSON files that were generated offline by: +# python analysis/precompute_anomalies.py +# python analysis/forecast_timesfm.py +# +# The server NEVER imports timesfm at runtime (HF Space OOM guard). +# ============================================================================= + +import json as _json +import functools + + +@functools.lru_cache(maxsize=1) +def _load_forecasts() -> list: + """Load forecast_all.json once and cache in-process.""" + if not os.path.exists(_FORECASTS_PATH): + return [] + with open(_FORECASTS_PATH, encoding="utf-8") as fh: + return _json.load(fh) + + +@functools.lru_cache(maxsize=1) +def _load_anomalies() -> list: + """Load anomalies_all.json once and cache in-process.""" + if not os.path.exists(_ANOMALIES_PATH): + return [] + with open(_ANOMALIES_PATH, encoding="utf-8") as fh: + return _json.load(fh) + + +@app.get("/api/v1/forecast") +async def api_forecast( + user: AuthUser | None = GatedUser, + commodity: str = Query(..., description="Commodity code, e.g. cabai_rawit"), + city: str = Query(..., description="IHK city_id, e.g. 3578 (Surabaya)"), +) -> JSONResponse: + """ + 30-day price forecast (point + P10/P90) for one commodity × city pair. + + Data is precomputed offline (seasonal-naive baseline unless TimesFM was + available at precompute time). The 'method' field in the response tells + you which model was used. + + Query params: + commodity AgriFlow commodity code (e.g. cabai_rawit, bawang_merah) + city IHK city_id (e.g. 3578 for Kota Surabaya) + + Response schema: + commodity_code str + city_id str + city_name str + method str ("timesfm_2.0" | "seasonal_naive_baseline") + generated_at str ISO 8601 + horizon_days int + history_end_date str ISO 8601 + forecasts list of {date, point, p10, p90} + """ + records = _load_forecasts() + if not records: + raise HTTPException( + status_code=503, + detail=( + "Forecast data not yet precomputed. " + "Run: python analysis/forecast_timesfm.py" + ), + ) + match = next( + (r for r in records if r["commodity_code"] == commodity and r["city_id"] == city), + None, + ) + if match is None: + # List available (commodity, city) pairs so caller can self-correct + available = sorted({(r["commodity_code"], r["city_id"]) for r in records}) + raise HTTPException( + status_code=404, + detail={ + "error": f"No forecast for commodity={commodity!r} city={city!r}", + "available_pairs": [{"commodity": c, "city": ci} for c, ci in available[:20]], + }, + ) + return JSONResponse(match) + + +@app.get("/api/v1/anomalies") +async def api_anomalies( + user: AuthUser | None = GatedUser, + commodity: str | None = Query(None, description="Filter by commodity code"), + city: str | None = Query(None, description="Filter by IHK city_id"), + limit: int = Query(50, ge=1, le=500, description="Max records returned (sorted by score desc)"), + since: str | None = Query(None, description="ISO date lower-bound, e.g. 2024-01-01"), +) -> JSONResponse: + """ + Detected price anomalies from the S-H-ESD scanner (precomputed offline). + + All filters are optional. Without filters returns top-N anomalies by score. + + Query params: + commodity optional commodity code filter + city optional IHK city_id filter + limit max records (default 50, max 500) + since ISO date — only return anomalies on or after this date + + Response schema: + count int + method str ("shesd_v2") + anomalies list of { + date str ISO 8601 + price float IDR/kg + rolling_median float + deviation_pct float (positive = spike, negative = drop) + type str SPIKE | DROP + score float (higher = more anomalous) + commodity_code str + city_id str + city_name str + persistent bool + } + """ + records = _load_anomalies() + if not records: + raise HTTPException( + status_code=503, + detail=( + "Anomaly data not yet precomputed. " + "Run: python analysis/precompute_anomalies.py" + ), + ) + + filtered = records + if commodity: + filtered = [r for r in filtered if r["commodity_code"] == commodity] + if city: + filtered = [r for r in filtered if r["city_id"] == city] + if since: + filtered = [r for r in filtered if r["date"] >= since] + + # Already sorted by score desc in the precomputed file; slice to limit + filtered = filtered[:limit] + + return JSONResponse({ + "count": len(filtered), + "method": "shesd_v2", + "anomalies": filtered, + }) + + +# ============================================================================= +# CLI helper: python -m whatsapp_bot.server "Harga cabai di Malang" +# ============================================================================= + +def _cli_main() -> None: + # Force UTF-8 stdout on Windows so emoji in replies don't crash cp1252 consoles + if sys.platform == "win32": + try: + sys.stdout.reconfigure(encoding="utf-8") + sys.stderr.reconfigure(encoding="utf-8") + except (AttributeError, OSError): + pass + if len(sys.argv) < 2: + print("Usage: python -m whatsapp_bot.server \"