# ⚡ EPIAŞ Veri Entegrasyon ve Analiz Platformu

Türkiye elektrik piyasası verilerini uçtan uca işleyen, üretim ortamına hazır (GCE üzerinde canlı çalışan) bir veri mühendisliği projesi.

**[📁 Veri Kaynağı: EPIAŞ Şeffaflık Platformu](https://seffaflik.epias.com.tr/)** | **[📊 Dashboard: Enerji Analitiği (Streamlit)](https://epiasdataplatform.streamlit.app/)**

---

## 🏗️ Mimari

EPIAŞ API & Open-Meteo (Hava Durumu) API & TCMB EVDS (Döviz Kuru)
    │
    ▼
┌─────────────────────────────────────────────────────────────┐
│         Apache Airflow (GCE VM, Docker Compose ile)          │
│   Saatlik medallion pipeline + ayrı ML/backfill DAG'ları     │
└─────────────────────────┬─────────────────────────────────────┘
                          │
                          ▼
              ┌───────────────────────┐
              │   Google Cloud Storage │
              │   🥉 Bronze Katman     │
              │   Ham Parquet dosyalar │
              └───────────┬───────────┘
                          │
                          ▼
              ┌───────────────────────────────┐
              │   Dataproc Serverless          │
              │   🥈 Silver Katman              │
              │   Pool'lanmış Spark batch'leri  │
              │   (DataprocCreateBatchOperator) │
              └───────────┬───────────────────┘
                          │
                          ▼
              ┌───────────────────────┐
              │    Google BigQuery     │
              │  External tablolar +   │
              │  dbt build (staging/   │
              │  marts, --fail-fast)   │
              │   🥇 Gold Katman        │
              └───────────┬───────────┘
                          │
              ┌───────────┴────────────┐
              ▼                        ▼
    ┌───────────────────┐    ┌──────────────────────┐
    │  SMF Tahminleme    │    │      Streamlit        │
    │  CatBoost→XGBoost  │───▶│      Dashboard        │
    │  Eğitim: haftalık  │    │   (dashboard.py)      │
    │  Inference: saatlik│    │  Streamlit Community  │
    │  + Trading Sinyali │    │  Cloud                │
    └───────────────────┘    └──────────────────────┘

Hata durumunda: Airflow, Gmail SMTP üzerinden `mehmetkarahanc@gmail.com` adresine e-posta ile uyarı gönderir (ADR-0004).

---

## 🛰️ Veri Akışı (Pipeline)

1. **Ingestion (Bronze):** `src/epias_client.py`, `src/weather_client.py` ve `src/fx_client.py` ile veriler çekilir; ana DAG (`dags/epias_dag.py`, DAG id: `epias_medallion_pipeline_v3`) **saatlik** (`schedule_interval="0 * * * *"`) çalışarak GCS'e (`gs://epias-data-lake/bronze/...`) ham Parquet olarak yazar. Kaynak tanımlarının tek doğrusu `dags/epias_sources.py`'dir (method, GCS yolu, backfill/daily uygunluğu).
2. **Transformation (Silver):** Silver katmanı artık container içinde çalışan Spark yerine **Dataproc Serverless** üzerinde çalışıyor (`DataprocCreateBatchOperator`, ADR-0002). Kaynaklar `DATAPROC_POOL_SIZE=2` grup halinde konsolide ediliyor (ADR-0003) — cold-start maliyetini (~3-4 dk) 24 kez yerine 2 kez ödemek için.
3. **Warehouse (Gold):** Silver veriler BigQuery'e köprülenir (`src/load_to_bigquery.py`) ve `epias_dbt/` altındaki dbt modelleri `dbt build --fail-fast` ile staging → marts katmanlarını üretir; hatalı bir model, üzerine kurulu mart'ların o saat build edilmesini durdurur (ADR-0004).
4. **SMF Tahminleme (ML):** Sistem Marjinal Fiyatı (SMF) için 2 aşamalı model: Aşama 1 `CatBoostClassifier` ile sistem yönünü (`system_direction`: enerji açığı/fazlası) tahmin eder, Aşama 2 `XGBRegressor` bu olasılıkları ek feature olarak kullanıp `smf_try` fiyatını tahmin eder (Aşama 2 eğitiminde sızıntıyı önlemek için out-of-fold olasılıklar kullanılır).
   - `dags/epias_smf_training_weekly.py` (`epias_smf_training_weekly`, pazartesi 04:00 UTC): `src/smf_trainer.py` ile Optuna + TimeSeriesSplit tabanlı tam yeniden eğitim; model artifact'ı GCS'e yazılır.
   - `dags/smf_inference_dag.py` (`smf_hourly_inference`, saatlik): `src/smf_inference.py` GCS'ten modeli yükler; backtest (`gold_smf_predictions`), ileriye dönük tahmin (`gold_smf_forward_predictions`) ve sabit 7 saat öncesi snapshot'ı (`gold_smf_forward_snapshot_7h`) BigQuery'e yazar. Feature'lar `src/smf_features.py` ve `mart_smf_lag_features` / `mart_smf_forward_features` mart'larından gelir.
   - `src/smf_trading_signal.py` (ADR-08): DUY Madde 28 dengesizlik formülü altında (açık: `GREATEST(PTF,SMF)×1.03`, fazla: `LEAST(PTF,SMF)×0.97`) "GİP'te şimdi kapat" ile "SMF'de dengesizliği kabul et" arasındaki beklenen maliyeti, tahmin belirsizliğini de hesaba katarak (Jensen eşitsizliği) saatlik olarak hesaplar. Bu bir **karar destek** sistemidir; emir iletmez.
   - Doğrulama araçları: `src/smf_walk_forward.py`, `src/smf_direction_model_comparison.py`; sonuçları `models/*.csv` altındadır (ADR-09: önce doğrula, sonra iterasyon yap — persistence/GİP-GÖP spread/blend denemeleri net fayda göstermediği için devreye alınmadı).
5. **Şirket Bazlı GÖP Aktivitesi:** `dags/epias_gop_company_activity_dag.py` (`epias_gop_company_activity_daily`, günlük 11:30 UTC) her organizasyon için ayrı istek gerektirdiği (~1600 org, ~20 dk) için saatlik ana DAG'a konmadı (ADR-0007).
6. **Backfill & Pilot:** `dags/epias_backfill_dag.py` (`epias_historical_backfill`, sadece manuel tetik) geçmiş veriyi günlük parçalar halinde doldurur (API bazı kaynaklarda haftalık istekleri sessizce tek güne kırptığı için gün-gün backfill zorunlu); `dags/epias_dataproc_pilot_dag.py` Dataproc Serverless geçişinin pilot/test DAG'ıdır. `src/silver_lookback_fix.py` geç güncellenen kayıtlar (ör. outages UMM/Arıza) için Silver'da lookback düzeltmesi yapar.
7. **Analytics:** `src/data_quality_check.py` veri kalitesi kontrollerini çalıştırır (ayrıca dbt testleri her staging modeli için de var); Streamlit dashboard'u (`dashboard.py`, 2000+ satır, 16 sayfa — Executive Summary, Fiyat Analizi, GÖP/GİP/KGÜP analizleri, Arz-Talep, **SMF Tahmin & ML**, **Trading Sinyalleri**, Hidrolik & Baraj, Vardiya Optimizasyonu vb.) BigQuery `epias_gold` dataset'ini sorgulayarak yönetici özeti sunar.

---

## 🚀 Kurulum ve Başlatma

Bu platform şu anda bir GCE VM'inde (`epias-airflow-host`) canlı çalışıyor. Yerelde ayağa kaldırmak için:

1. Repo'yu klonlayın ve ana dizine gidin:
   cd epias_data_platform

2. Kimlik bilgilerini yerleştirin:
   - `credentials/gcp-key.json` (GCP servis hesabı anahtarı — BigQuery, GCS, Dataproc erişimi için)
   - `.env` dosyasında `GMAIL_APP_PASSWORD` (Airflow email_on_failure alerting için, ADR-0004 — normal hesap şifresi değil, 2FA'lı bir Gmail App Password)

3. Python bağımlılıklarını yükleyin (yerel geliştirme/dashboard için):
   pip install -r requirements.txt

4. Airflow'u (webserver + scheduler + Postgres) inşa edin ve başlatın:
   docker-compose up -d --build

   Not: Silver katman Dataproc Serverless'te çalıştığı için container içinde artık Java/Spark kurulu değil (Dockerfile, ADR-0002) — bu container yalnızca Dataproc batch submit eder.

5. Dashboard'u yerel olarak çalıştırmak için:
   streamlit run dashboard.py

---

## 📁 Proje Yapısı

epias_data_platform/
├── dags/
│   ├── epias_dag.py                  # Ana Airflow DAG — epias_medallion_pipeline_v3 (saatlik)
│   ├── epias_sources.py              # Kaynak/tablo tanımları (tek doğru kaynak) + Dataproc config
│   ├── epias_backfill_dag.py         # Geçmiş veri backfill DAG'ı (gün-gün, manuel)
│   ├── epias_dataproc_pilot_dag.py   # Dataproc Serverless geçiş pilot DAG'ı
│   ├── epias_gop_company_activity_dag.py # Günlük şirket bazlı GÖP aktivitesi (ADR-0007)
│   ├── epias_smf_training_weekly.py  # Haftalık SMF model eğitimi
│   └── smf_inference_dag.py          # Saatlik SMF tahmini (smf_hourly_inference)
├── src/
│   ├── epias_client.py               # EPIAŞ API Wrapper
│   ├── weather_client.py             # Open-Meteo API Wrapper
│   ├── fx_client.py                  # TCMB EVDS döviz kuru API Wrapper
│   ├── load_to_bigquery.py           # Silver → BigQuery köprüleme
│   ├── silver_lookback_fix.py        # Silver lookback düzeltmesi (geç güncellenen kayıtlar)
│   ├── data_quality_check.py         # Veri kalitesi kontrolleri
│   ├── smf_features.py               # SMF feature engineering
│   ├── smf_trainer.py                # 2 aşamalı SMF eğitimi (CatBoost → XGBoost)
│   ├── smf_inference.py              # SMF tahmin (backtest + forward + 7h snapshot)
│   ├── smf_trading_signal.py         # Olasılıksal EV trading sinyali (ADR-08)
│   ├── smf_walk_forward.py           # Walk-forward doğrulama
│   ├── smf_direction_model_comparison.py # Yön sınıflandırıcı model karşılaştırması
│   └── config.py                     # Ortak konfigürasyon
├── spark_jobs/
│   ├── bronze_to_silver_*.py         # 25+ Spark ETL süreci — artık Dataproc
│   │                                  #  Serverless'e submit ediliyor (ADR-0002)
│   ├── spark_utils.py                # Ortak Spark yardımcıları
│   └── gcs-connector-hadoop3-*.jar   # GCS bağlantı köprüsü
├── credentials/
│   └── gcp-key.json                  # Google Cloud anahtarı
├── epias_dbt/
│   └── models/                       # BigQuery dbt modelleri — staging + marts,
│                                      #  her staging modeli için dbt test coverage
├── scripts/                          # backfill_fx_rates.py, cleanup_silver_overlap.py,
│                                      #  fix_silver_parquet_types.py gibi tek seferlik/yardımcı script'ler
├── docs/
│   └── epias_openapi.yaml            # EPIAŞ API OpenAPI şeması
├── infra/
│   └── gcs-lifecycle-policy.json     # GCS lifecycle (bronze/ → Nearline/Coldline)
├── models/                           # SMF walk-forward / ablation / karşılaştırma sonuç CSV'leri
├── plans/                            # ADR'ler (01–09): tekilleştirme, Dataproc geçişi ve konsolidasyonu,
│                                      #  veri kalitesi, forward-forecast kalitesi, şirket bazlı aktivite,
│                                      #  SMF trading sinyali ve doğrulama metodolojisi
├── tests/                            # Birim/entegrasyon testleri
├── dashboard.py                      # Streamlit yönetici özeti (çok sayfalı)
├── Dockerfile                        # Airflow imajı (Spark kaldırıldı, sadece Dataproc submit)
└── docker-compose.yml                # Postgres + Airflow webserver/scheduler (Spark master/worker YOK)

---

## 🔑 Teknik Kararlar

Neden LocalExecutor ve Postgres kullanıldı? Varsayılan SequentialExecutor yerine LocalExecutor kullanılarak görevlerin paralel çalışması sağlandı ve tüm meta veriler Postgres'te kalıcı hale getirildi.

Neden Medallion Mimarisi? Ham veriyi (Bronze) hiç silmeden tutmak, her dönüşümü ayrı katmanda yapmak hata ayıklamayı ve yeniden işlemeyi kolaylaştırır.

**ADR-0002 — Neden Dataproc Serverless'e geçildi?** Her zaman açık duran docker-compose spark-master/spark-worker container'ları (GCE `e2-medium` üzerinde) kaynakları sürekli tüketiyordu. Silver katman artık pay-per-batch Dataproc Serverless'te çalışıyor; container artık Java/Spark içermiyor, sadece `DataprocCreateBatchOperator` ile batch submit ediyor.

**ADR-0003 — Neden kaynaklar pool'lanıyor?** Her kaynak için ayrı Dataproc batch'i başlatmak, cold-start maliyetini (~3-4 dk) kaynak sayısı kadar (24 kez) ödetiyordu. `DATAPROC_POOL_SIZE=2` ile kaynaklar gruplanıp paylaşılan Spark session'larında çalıştırılıyor — Silver katman süresi ~40 dk'dan ~4 dk'ya indi, bu da saatlik schedule'ı mümkün kıldı.

**Neden saatlik (hourly) pipeline'a geçildi?** EPIAŞ çoğu kaynağı saat saat yayınlıyor; ADR-0003'ün konsolidasyonu sayesinde tüm pipeline saatin içine rahatça sığdığından, günlük yerine saatlik çekim yapılarak veri gün içinde ilerledikçe yeni saatler otomatik yakalanıyor.

**ADR-0004 — Neden `dbt build --fail-fast` ve email alerting eklendi?** 2026-07-25/26'daki Silver dedup incelemesinde ~12 tablonun aylardır sessizce duplike olduğu, çünkü `schema.yml`'deki testlerin hiç çalıştırılmadığı ortaya çıktı. Ayrı bir `run_dbt_tests` task'ı bu *tespit* açığını kapattı ama *önleme* açığını kapatmadı — `dbt run` zaten (muhtemelen bozuk) veriyi yazmıştı. `dbt build --fail-fast`, her modeli kendi testiyle bağımlılık sırasında çalıştırıp ilk gerçek hatada durur; ayrıca `on_failure_callback` artık sadece log atmak yerine Gmail SMTP üzerinden gerçek bir e-posta gönderiyor (`email_on_failure`).

Neden tüm silver Spark job'ları numerik kolonları FLOAT64'e cast ediyor? Bronze API yanıtları bazen tamsayı, bazen ondalık dönebiliyor; bu şema kaymasını (schema drift) önlemek için tüm numerik kolonlar FLOAT64'e sabitlendi.

dbt Path ve Kütüphane bağımlılıkları: Python 3.8 uyumluluğu için `dbt-core < 1.8.0` sürümü sabitlendi ve ENV PATH ayarı ile dbt'nin Airflow içinden sorunsuz tetiklenmesi sağlandı.

**ML kütüphane sürümleri pinli:** `xgboost==2.1.4`, `scikit-learn==1.3.2`, `catboost==1.2.10` (Dockerfile). Pinsiz kurulumda image her rebuild'de PyPI'daki en güncel sürüme kayıyor, VM eğitim ortamından ayrışıp sessizce yanlış tahmin üretiyordu (xgboost 3.x / scikit-learn 1.4+ Python 3.8'i desteklemiyor).

**Zaman aşımı korumaları:** Ana DAG'da `execution_timeout` (30 dk) ve `dagrun_timeout` (1 saat) tanımlı; takılı kalan bir task'ın sessizce tüm saatlik akışı bloke etmesi (zombie-task) engellenir. SMF inference'ta ayrıca freshness watchdog ve zorla çıkış (force-exit) var.

**UTC tarih kesme hatası sınıfı:** Staging modellerinde `CAST(ts AS DATE)` UTC'de keserek TRT'nin ilk 3 saatini bir önceki güne yazıyordu; 13 staging modelde düzeltildi. Yeni staging modelleri bu kalıba karşı kontrol edilmelidir.

**Neden SMF tahmini karar destek, işlem değil?** Model gerçek ama gürültülü bir avantaj sağlıyor (yön doğruluğu ~%70, MASE ~0.77, naif T-24h'yi geçiyor); bu yüzden çıktı emir iletmek yerine trader'a beklenen maliyet karşılaştırması sunuyor (ADR-08, ADR-09).

**Maliyet:** BigQuery maliyetinin büyük kısmı, incremental olmayan mart'ların `stg_idm_transactions`'ı her çalışmada tam taramasından geliyordu; bu kalıp birkaç mart'ta düzeltildi. Yeni mart'lar bu açıdan gözden geçirilmelidir.

---

## 🔗 Servisler ve Erişim

- Airflow UI: localhost:8080 (GCE üzerinde: VM'in dinamik IP'si — bkz. firewall notu)
- Streamlit Dashboard: canlı ortamda [epiasdataplatform.streamlit.app](https://epiasdataplatform.streamlit.app/) (Streamlit Community Cloud); yerelde localhost:8501 (`streamlit run dashboard.py` veya `run_dashboard_local.bat`)

Not: GCE VM'in IP'si dinamik olduğundan firewall kurallarını buna göre güncellemek gerekebilir.

---

## ✉️ İletişim

Mehmet Karahan - Veri Analisti
mehmetkarahanc@gmail.com

---

### 💡 Not
Bu platform, enerji maliyetlerini optimize etmek ve piyasa risklerini öngörmek için tasarlanmıştır. Veri akışı sırasında "Forbidden 403" hatası alınması durumunda GCP faturalandırma hesabı kontrol edilmelidir. Gün öncesi piyasası (pricing/dam_clearing/price_ind_bid) kaynakları, EPIAŞ'ın kendi yayın kesim saatinden önce boş sonuç dönebilir — bu artık hata değil, `allow_empty=True` ile beklenen bir durum olarak ele alınıyor.
</content>