from abc import ABC, abstractmethod from datetime import date from pathlib import Path from typing import Dict import numpy as np import pandas as pd import pymysql from pymysql import MySQLError class DataUnavailableError(Exception): pass class PriceRepository(ABC): @abstractmethod def get_series(self, commodity: str, market: str) -> pd.Series: raise NotImplementedError @abstractmethod def last_data_date(self, commodity: str, market: str) -> date: raise NotImplementedError class CsvPriceRepository(PriceRepository): FILE_PATTERNS = { "rawit": ("*rawit*.csv",), "merah": ("*cabe_merah_besar*.csv", "*merah_besar*.csv"), "keriting": ("*merah_keriting*.csv",), } def __init__(self, data_dir: Path, unit_scale: float = 1.0) -> None: self.data_dir = data_dir self.unit_scale = unit_scale self._frames: Dict[str, pd.DataFrame] = {} def _find_file(self, commodity: str) -> Path: candidates = [self.data_dir / f"{commodity}.csv", self.data_dir / f"data_{commodity}.csv"] for pattern in self.FILE_PATTERNS.get(commodity, (f"*{commodity}*.csv",)): candidates.extend(sorted(self.data_dir.rglob(pattern))) if self.data_dir.exists() else None for candidate in candidates: if candidate.is_file(): return candidate raise DataUnavailableError( f"File CSV untuk komoditas {commodity} tidak ditemukan di {self.data_dir}." ) def _load(self, commodity: str) -> pd.DataFrame: if commodity in self._frames: return self._frames[commodity] path = self._find_file(commodity) frame = pd.read_csv(path) date_column = next( (column for column in frame.columns if str(column).strip().lower() in {"date", "tanggal", "datetime"}), None, ) if date_column is None: raise DataUnavailableError( f"CSV {path.name} harus memiliki kolom date atau tanggal." ) frame[date_column] = pd.to_datetime(frame[date_column], errors="coerce") frame = frame.dropna(subset=[date_column]).set_index(date_column).sort_index() if not frame.empty: frame = frame[~frame.index.duplicated(keep="last")] frame = frame.reindex(pd.date_range(frame.index.min(), frame.index.max(), freq="D")) self._frames[commodity] = frame return frame @staticmethod def _market_column(frame: pd.DataFrame, market: str) -> str: aliases = {market, market.replace("pasar_", "Pasar_")} for column in frame.columns: normalized = str(column).strip().lower().replace(" ", "_") if normalized in {alias.lower() for alias in aliases}: return column raise DataUnavailableError(f"Kolom harga untuk pasar {market} tidak ditemukan.") def get_series(self, commodity: str, market: str) -> pd.Series: frame = self._load(commodity) column = self._market_column(frame, market) series = pd.to_numeric(frame[column], errors="coerce").replace(0, np.nan) series = series * self.unit_scale series = series.interpolate(method="linear", limit=2, limit_area="inside") return series.astype(float) def last_data_date(self, commodity: str, market: str) -> date: series = self.get_series(commodity, market).dropna() if series.empty: raise DataUnavailableError( f"Tidak ada data valid untuk {commodity}/{market}." ) return series.index[-1].date() class MySQLPriceRepository(PriceRepository): def __init__(self, config) -> None: self.config = config def _connect(self): return pymysql.connect( host=self.config["MYSQL_HOST"], port=self.config["MYSQL_PORT"], database=self.config["MYSQL_DATABASE"], user=self.config["MYSQL_USER"], password=self.config["MYSQL_PASSWORD"], connect_timeout=self.config["MYSQL_CONNECT_TIMEOUT"], ) def _load(self, commodity: str, market: str) -> pd.Series: connection = None cursor = None try: connection = self._connect() cursor = connection.cursor() cursor.execute( """ SELECT ph.price_date, ph.price FROM price_history AS ph INNER JOIN commodities AS c ON c.id = ph.commodity_id INNER JOIN markets AS m ON m.id = ph.market_id WHERE c.slug = %s AND m.slug = %s ORDER BY ph.price_date ASC """, (commodity, market), ) rows = cursor.fetchall() except MySQLError as error: raise DataUnavailableError(f"Gagal membaca database MySQL: {error}") from error finally: if cursor is not None: cursor.close() if connection is not None: connection.close() if not rows: raise DataUnavailableError( f"Tidak ada data harga untuk {commodity}/{market} di database." ) frame = pd.DataFrame(rows, columns=["date", "price"]) frame["date"] = pd.to_datetime(frame["date"], errors="coerce") frame["price"] = pd.to_numeric(frame["price"], errors="coerce") frame = frame.dropna(subset=["date"]).set_index("date").sort_index() frame = frame[~frame.index.duplicated(keep="last")] frame["price"] = frame["price"].replace(0, np.nan) frame["price"] = frame["price"].interpolate(method="linear", limit=2, limit_area="inside") return frame["price"].astype(float) def get_series(self, commodity: str, market: str) -> pd.Series: return self._load(commodity, market) def last_data_date(self, commodity: str, market: str) -> date: series = self.get_series(commodity, market).dropna() if series.empty: raise DataUnavailableError( f"Tidak ada data valid untuk {commodity}/{market} di database." ) return series.index[-1].date()