awal bukanlah akhir

.gitignore

@@ -0,0 +1,13 @@
+__pycache__/
+article/
+docs/
+output/
+validation_output/
+venv/
+backtesting.py
+data_exploration.ipynb
+main.py
+optimize_data.py
+WORKFLOW.md
+data/
+hf_cache/

app.py

@@ -0,0 +1,335 @@
+import logging
+import os
+import sys
+from pathlib import Path
+
+import gradio as gr
+import pandas as pd
+from typing import Optional, Tuple
+
+from config import Config
+from data_processor import DataProcessor
+from evaluator import Evaluator
+from prophet_predictor import ProphetPredictor
+from utils import setup_logging
+
+setup_logging("INFO")
+logger = logging.getLogger("GradioApp")
+
+_processor: Optional[DataProcessor] = None
+_predictor: Optional[ProphetPredictor] = None
+_config: Optional[Config] = None
+_df_enrollment: Optional[pd.DataFrame] = None
+_elective_codes: Optional[set] = None
+_backtest_metrics: Optional[dict] = None
+
+
+def initialize_system():
+    """Initialize the prediction system (called once at startup)."""
+    global _processor, _predictor, _config, _df_enrollment, _elective_codes, _backtest_metrics
+
+    try:
+        logger.info("Initializing prediction system...")
+        _config = Config()
+
+        _processor = DataProcessor(_config)
+        _df_enrollment, _elective_codes = _processor.load_and_process()
+
+        _predictor = ProphetPredictor(_config)
+        _predictor.train_student_population_model(
+            _processor.raw_data["students_yearly"]
+        )
+
+        logger.info("✓ System initialized successfully")
+        return True
+    except Exception as e:
+        logger.error(f"Failed to initialize system: {e}", exc_info=True)
+        return False
+
+
+def generate_predictions(year: int, semester: int) -> Tuple[str, Optional[pd.DataFrame], Optional[pd.DataFrame]]:
+    """
+    Generate enrollment predictions for a given year and semester.
+
+    Args:
+        year: Target year (e.g., 2025)
+        semester: Target semester (1 = Ganjil/Odd, 2 = Genap/Even)
+
+    Returns:
+        Tuple of (summary_text, recommendations_df, all_predictions_df)
+    """
+    global _processor, _predictor, _config, _df_enrollment, _elective_codes, _backtest_metrics
+
+    try:
+        if semester not in [1, 2]:
+            return "❌ Error: Semester must be 1 (Ganjil) or 2 (Genap)", None, None
+
+        if year < 2020 or year > 2030:
+            return "❌ Error: Year must be between 2020 and 2030", None, None
+
+        if _config is None or _predictor is None or _processor is None or _df_enrollment is None or _elective_codes is None:
+            return "❌ Error: System not initialized. Please restart the app.", None, None
+
+        logger.info(f"Generating predictions for {year} Semester {semester}...")
+
+        _config.prediction.PREDICT_YEAR = year
+        _config.prediction.PREDICT_SEMESTER = semester
+
+        if _backtest_metrics is None:
+            logger.info("Running backtest for the first time...")
+            evaluator = Evaluator(_config)
+            backtest_results = evaluator.run_backtest(_df_enrollment, _predictor)
+
+            if backtest_results is None or len(backtest_results) == 0:
+                logger.warning("Backtest returned no results, using defaults")
+                _backtest_metrics = {'mae': 0, 'rmse': 0}
+            else:
+                _backtest_metrics = evaluator.generate_metrics(backtest_results)
+                if _backtest_metrics is None:
+                    logger.warning("Metrics calculation failed, using defaults")
+                    _backtest_metrics = {'mae': 0, 'rmse': 0}
+        else:
+            logger.info("Using cached backtest metrics")
+
+        metrics = _backtest_metrics
+
+        predictions = _predictor.generate_batch_predictions(
+            _df_enrollment,
+            _processor.raw_data["courses"],
+            _elective_codes,
+            year,
+            semester,
+        )
+
+        recommended = predictions[predictions["recommendation"] == "BUKA"].copy()
+
+        semester_name = "Ganjil (Odd)" if semester == 1 else "Genap (Even)"
+        summary = f"""
+## 📊 Prediction Summary for {year} Semester {semester_name}
+
+### Model Performance (Backtest)
+- **Mean Absolute Error (MAE)**: {metrics['mae']:.2f} students
+- **Root Mean Squared Error (RMSE)**: {metrics['rmse']:.2f} students
+
+### Recommendations
+- **Courses to Open**: {len(recommended)}
+- **Total Seats Needed**: {int(recommended['recommended_quota'].sum()) if not recommended.empty else 0}
+- **Estimated Students**: {int(recommended['predicted_enrollment'].sum()) if not recommended.empty else 0}
+
+### Top Course
+"""
+
+        if not recommended.empty:
+            top_course = recommended.iloc[0]
+            summary += f"- **{top_course['nama_mk']}** ({top_course['kode_mk']})\n"
+            summary += f"  - Predicted: {top_course['predicted_enrollment']:.0f} students\n"
+            summary += f"  - Recommended Quota: {top_course['recommended_quota']:.0f} seats"
+        else:
+            summary += "- No courses recommended to open"
+
+        if not recommended.empty:
+            recommended_display = recommended[[
+                'kode_mk', 'nama_mk', 'predicted_enrollment',
+                'recommended_quota', 'strategy'
+            ]].copy()
+            recommended_display.columns = [
+                'Course Code', 'Course Name', 'Predicted Students',
+                'Recommended Quota', 'Prediction Strategy'
+            ]
+            recommended_display['Predicted Students'] = recommended_display['Predicted Students'].round(1)
+            recommended_display['Recommended Quota'] = recommended_display['Recommended Quota'].astype(int)
+            recommended_display = recommended_display.sort_values('Predicted Students', ascending=False)
+        else:
+            recommended_display = pd.DataFrame()
+
+        # All predictions
+        all_predictions_display = predictions[[
+            'kode_mk', 'nama_mk', 'predicted_enrollment',
+            'recommended_quota', 'recommendation', 'strategy'
+        ]].copy()
+        all_predictions_display.columns = [
+            'Course Code', 'Course Name', 'Predicted Students',
+            'Recommended Quota', 'Recommendation', 'Strategy'
+        ]
+        all_predictions_display['Predicted Students'] = all_predictions_display['Predicted Students'].round(1)
+        all_predictions_display['Recommended Quota'] = all_predictions_display['Recommended Quota'].astype(int)
+        all_predictions_display = all_predictions_display.sort_values('Predicted Students', ascending=False)
+
+        logger.info(f"✓ Predictions generated successfully")
+        return summary, recommended_display, all_predictions_display
+
+    except Exception as e:
+        error_msg = f"❌ Error generating predictions: {str(e)}"
+        logger.error(error_msg, exc_info=True)
+        return error_msg, None, None
+
+
+def get_data_info() -> str:
+    """Get information about the loaded dataset."""
+    global _processor, _config
+
+    try:
+        if _processor is None or _config is None:
+            return "❌ System not initialized"
+
+        courses = _processor.raw_data.get("courses")
+        students = _processor.raw_data.get("students_yearly")
+
+        if courses is None or students is None:
+            return "❌ Data not loaded"
+
+        # Get elective courses
+        elective_courses = courses[courses["kategori_mk"] == "P"]
+
+        info = f"""
+## 📁 Dataset Information
+
+### Course Catalog
+- **Total Courses**: {len(courses)}
+- **Elective Courses**: {len(elective_courses)}
+- **Mandatory Courses**: {len(courses) - len(elective_courses)}
+
+### Student Population
+- **Years Available**: {students['thn'].min()} - {students['thn'].max()}
+- **Total Records**: {len(students)}
+
+### Data Source
+- File: `{_config.data.FILE_PATH}`
+- Last Updated: October 8, 2025
+"""
+        return info
+
+    except Exception as e:
+        return f"❌ Error getting data info: {str(e)}"
+
+
+# Initialize system at startup
+logger.info("Starting Gradio app...")
+init_success = initialize_system()
+
+if not init_success:
+    logger.error("Failed to initialize system. App may not work correctly.")
+
+# Create Gradio Interface
+with gr.Blocks(title="SKS Enrollment Predictor") as demo:
+
+    # Show disclaimer banner if using demo data
+    if os.getenv("DEMO_MODE", "false").lower() == "true":
+        gr.Markdown(
+            """
+            <div style='padding: 15px; background-color: #fff3cd; border-left: 5px solid #ffc107; margin-bottom: 20px;'>
+            <h3 style='margin-top: 0; color: #856404;'>⚠️ Demo Version - Anonymized Data</h3>
+            <p style='margin-bottom: 0; color: #856404;'>
+            This demonstration uses <strong>anonymized enrollment data</strong> to protect student privacy.
+            All predictions and functionality are identical to the production version.
+            </p>
+            <details style='margin-top: 10px;'>
+            <summary style='cursor: pointer; color: #856404;'><strong>Changes made for demo:</strong></summary>
+            <ul style='color: #856404;'>
+                <li>Student IDs replaced with anonymous codes (STU000001, STU000002, ...)</li>
+                <li>Population counts have ±3% random noise added</li>
+                <li>Course information and enrollment patterns fully preserved</li>
+            </ul>
+            </details>
+            </div>
+            """,
+            sanitize_html=False
+        )
+
+    with gr.Tabs():
+        with gr.Tab("Generate Predictions"):
+
+            with gr.Row():
+                with gr.Column(scale=1):
+                    year_input = gr.Number(
+                        label="Target Year",
+                        value=2025,
+                        precision=0,
+                        minimum=2020,
+                        maximum=2030,
+                        info="Masukkan tahun yang ingin diprediksi"
+                    )
+
+                    semester_input = gr.Radio(
+                        choices=[1, 2],
+                        label="Semester",
+                        value=2,
+                        info="1 = Ganjil, 2 = Genap"
+                    )
+
+                    predict_btn = gr.Button(
+                        "Generate Predictions",
+                        variant="primary",
+                        size="lg"
+                    )
+
+                with gr.Column(scale=2):
+                    summary_output = gr.Markdown(
+                        label="Summary",
+                        value="Click 'Generate Predictions' to start"
+                    )
+
+            gr.Markdown("### Recommended Courses to Open")
+            recommended_output = gr.Dataframe(
+                label="Courses Recommended to Open",
+                wrap=True,
+                interactive=False
+            )
+
+            with gr.Accordion("View All Predictions", open=False):
+                all_predictions_output = gr.Dataframe(
+                    label="All Elective Courses",
+                    wrap=True,
+                    interactive=False
+                )
+
+        with gr.Tab("Data Information"):
+            gr.Markdown(
+            )
+
+            data_info_btn = gr.Button("Refresh Data Info", variant="secondary")
+            data_info_output = gr.Markdown()
+
+            data_info_btn.click(
+                fn=get_data_info,
+                inputs=[],
+                outputs=data_info_output
+            )
+
+            demo.load(fn=get_data_info, inputs=[], outputs=data_info_output)
+
+
+    predict_btn.click(
+        fn=generate_predictions,
+        inputs=[year_input, semester_input],
+        outputs=[summary_output, recommended_output, all_predictions_output]
+    )
+
+    # Footer
+    if os.getenv("DEMO_MODE", "false").lower() == "true":
+        gr.Markdown(
+            """
+            ---
+            <div style='text-align: center; color: #666; font-size: 0.9em;'>
+            📊 Demo Version with Anonymized Data | For Educational Purposes
+            </div>
+            """
+        )
+    else:
+        gr.Markdown(
+            """
+            ---
+            <div style='text-align: center; color: #666; font-size: 0.9em;'>
+            🔒 Private & Confidential | For Authorized Use Only
+            </div>
+            """
+        )
+
+# Launch the app
+if __name__ == "__main__":
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=False,
+        show_error=True
+    )

config.py

@@ -0,0 +1,170 @@
+from dataclasses import dataclass, field
+from typing import Dict, List
+import os
+
+# Import data loader for private HF dataset support
+try:
+    from data_loader import load_data_file
+    DATA_LOADER_AVAILABLE = True
+except ImportError:
+    DATA_LOADER_AVAILABLE = False
+    def load_data_file() -> str:
+        """Fallback if data_loader not available."""
+        return "data/optimized_data.xlsx"
+
+
+def _get_data_file_path() -> str:
+    """
+    Get data file path based on environment.
+
+    Priority:
+    1. If HF_TOKEN set: Load from private HF dataset (muhalwan/optimized_data_mhs)
+    2. If DEMO_MODE=true: Use demo_data.xlsx (anonymized)
+    3. Otherwise: Use local optimized_data.xlsx
+    """
+    if os.getenv("HF_TOKEN"):
+        return load_data_file()  # Loads from HF dataset if HF_TOKEN is set
+    elif os.getenv("DEMO_MODE", "false").lower() == "true":
+        return "data/demo_data.xlsx"
+    else:
+        return "data/optimized_data.xlsx"
+
+
+@dataclass
+class DataConfig:
+    """Data source configuration and validation rules."""
+
+    # Data file path - automatically determined based on environment
+    FILE_PATH: str = field(default_factory=_get_data_file_path)
+
+    # Sheet mappings
+    SHEET_COURSES: str = "tabel1_data_matkul"
+    SHEET_OFFERINGS: str = "tabel2_data_matkul_dibuka"
+    SHEET_STUDENTS_YEARLY: str = "tabel3_data_mahasiswa_per_tahun"
+    SHEET_STUDENTS_INDIVIDUAL: str = "tabel4_data_individu_mahasiswa"
+
+    # Standardization
+    OFFERINGS_RENAME: Dict[str, str] = field(
+        default_factory=lambda: {"tahun": "thn", "semester": "smt"}
+    )
+
+    # Elective Course Identification
+    # IMPORTANT: Elective courses are identified by kategori_mk = 'P' in tabel1
+    # Mandatory/Required courses have kategori_mk = 'W'
+    ELECTIVE_CATEGORY: str = "P"
+    MANDATORY_CATEGORY: str = "W"
+
+    # Valid category values (will be normalized to uppercase)
+    VALID_CATEGORIES: List[str] = field(default_factory=lambda: ["P", "W"])
+
+
+@dataclass
+class ModelConfig:
+    """Prophet model hyperparameters and prediction strategies."""
+
+    # Prophet Hyperparameters
+    GROWTH_MODE: str = "logistic"
+    CHANGEPOINT_SCALE: float = 0.01
+    SEASONALITY_MODE: str = "multiplicative"
+    YEARLY_SEASONALITY: bool = True
+    FALLBACK_DEFAULT: int = 20
+
+    # Prediction safety limits
+    # Maximum multiplier of historical max enrollment before flagging as unrealistic
+    SANITY_CHECK_MAX_MULTIPLIER: float = 3.0
+    # Minimum historical data points required for reliable prediction
+    MIN_HISTORY_POINTS: int = 3
+
+
+@dataclass
+class PredictionConfig:
+    """Business logic for predictions."""
+
+    PREDICT_YEAR: int = 2025
+    PREDICT_SEMESTER: int = 2
+
+    # Buffer Calculations
+    BUFFER_PERCENT: float = 0.20
+    MIN_QUOTA_OPEN: int = 25
+    MIN_PREDICT_THRESHOLD: int = 15
+
+    MAX_CAPACITY_MULTIPLIER: float = 2.0
+    ABSOLUTE_MAX_STUDENTS: int = 400
+
+    SEMESTER_TO_MONTH: Dict[int, str] = field(
+        default_factory=lambda: {
+            1: "09-01",
+            2: "03-01",
+        }
+    )
+
+
+@dataclass
+class OutputConfig:
+    """Output settings."""
+
+    OUTPUT_DIR: str = "output"
+    LOG_LEVEL: str = "INFO"
+    TOP_N_DISPLAY: int = 30
+
+
+@dataclass
+class BacktestConfig:
+    """Backtest settings and validation."""
+
+    START_YEAR: int = 2010
+    END_YEAR: int = 2024
+    VERBOSE: bool = True
+
+    # Minimum elective enrollments required for backtesting
+    MIN_ELECTIVE_ENROLLMENTS: int = 1
+    # Minimum unique courses required for backtesting
+    MIN_UNIQUE_COURSES: int = 1
+
+
+class Config:
+    """
+    Master Config Object.
+
+    ELECTIVE COURSE DEFINITION:
+    ---------------------------
+    Elective courses are identified by kategori_mk = 'P' in tabel1_data_matkul.
+    This is the ONLY source of truth for course categories.
+
+    Examples of elective courses (kategori_mk = 'P'):
+    - EF234607: Keamanan Aplikasi
+    - EF234613: Game Edukasi dan Simulasi
+    - UG234922: Kebudayaan dan Kebangsaan
+    - IW184301: Sistem Basis Data
+    - KI series: Various computer science electives
+
+    Mandatory courses have kategori_mk = 'W' (Wajib).
+
+    DATA REQUIREMENTS FOR BACKTESTING:
+    -----------------------------------
+    To backtest a semester, you need:
+    1. Course catalog (tabel1) with kategori_mk properly set
+    2. ACTUAL student enrollments (tabel4) for that semester
+    3. At least one elective course with enrollments
+
+    Note: Course offerings (tabel2) alone are NOT sufficient for backtesting.
+    You must have actual enrollment data (tabel4) to validate predictions.
+    """
+
+    def __init__(self):
+        self.data: DataConfig = DataConfig()
+        self.model: ModelConfig = ModelConfig()
+        self.prediction: PredictionConfig = PredictionConfig()
+        self.output: OutputConfig = OutputConfig()
+        self.backtest: BacktestConfig = BacktestConfig()
+
+    def get_prediction_target_name(self) -> str:
+        sem = "Ganjil" if self.prediction.PREDICT_SEMESTER == 1 else "Genap"
+        return f"{self.prediction.PREDICT_YEAR} Semester {sem}"
+
+    def get_elective_filter_description(self) -> str:
+        """Get human-readable description of elective identification."""
+        return f"kategori_mk = '{self.data.ELECTIVE_CATEGORY}' in {self.data.SHEET_COURSES}"
+
+
+default_config = Config()

data_loader.py

@@ -0,0 +1,88 @@
+import os
+import logging
+from pathlib import Path
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+def load_data_file() -> str:
+    hf_token = os.getenv("HF_TOKEN")
+
+    if hf_token:
+        try:
+            from huggingface_hub import hf_hub_download
+
+            logger.info("🔐 Loading data from private Hugging Face dataset...")
+            logger.info("   Dataset: muhalwan/optimized_data_mhs")
+
+            file_path = hf_hub_download(
+                repo_id="muhalwan/optimized_data_mhs",
+                filename="optimized_data.xlsx",
+                repo_type="dataset",
+                token=hf_token,
+                cache_dir="./hf_cache"
+            )
+
+            logger.info(f"✓ Data loaded successfully from HF dataset")
+            logger.info(f"   Cached at: {file_path}")
+            return file_path
+
+        except ImportError:
+            logger.error("huggingface_hub not installed. Install with: pip install huggingface_hub")
+            raise
+
+        except Exception as e:
+            logger.error(f"Failed to download from HF dataset: {e}")
+            logger.error("Falling back to local file if available...")
+
+    local_path = "data/optimized_data.xlsx"
+
+    if Path(local_path).exists():
+        logger.info(f"📁 Loading data from local file: {local_path}")
+        return local_path
+
+    error_msg = (
+        "No data file found!\n"
+        "Options:\n"
+        "1. Set HF_TOKEN environment variable to load from private dataset\n"
+        "2. Place optimized_data.xlsx in data/ folder for local development\n"
+    )
+    logger.error(error_msg)
+    raise FileNotFoundError(error_msg)
+
+
+def get_data_source_info() -> dict:
+    hf_token = os.getenv("HF_TOKEN")
+    local_exists = Path("data/optimized_data.xlsx").exists()
+
+    return {
+        "hf_token_available": bool(hf_token),
+        "local_file_available": local_exists,
+        "will_use_hf_dataset": bool(hf_token),
+        "will_use_local": not hf_token and local_exists,
+        "dataset_repo": "muhalwan/optimized_data_mhs" if hf_token else None,
+        "local_path": "data/optimized_data.xlsx" if local_exists else None
+    }
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+
+    print("=" * 80)
+    print("Data Source Information")
+    print("=" * 80)
+
+    info = get_data_source_info()
+    for key, value in info.items():
+        print(f"  {key}: {value}")
+
+    print("\n" + "=" * 80)
+    print("Attempting to load data...")
+    print("=" * 80)
+
+    try:
+        file_path = load_data_file()
+        print(f"\n✓ Success! Data file: {file_path}")
+    except Exception as e:
+        print(f"\n✗ Failed: {e}")

data_processor.py

@@ -0,0 +1,350 @@
+import logging
+from typing import Dict, Set, Tuple
+
+import numpy as np
+import pandas as pd
+
+from config import Config
+
+logger = logging.getLogger(__name__)
+
+
+class DataProcessor:
+    def __init__(self, config: Config):
+        self.config = config
+        self.raw_data: Dict[str, pd.DataFrame] = {}
+        self.processed_data: pd.DataFrame = pd.DataFrame()
+        self.elective_codes: Set[str] = set()
+
+    def load_and_process(self) -> Tuple[pd.DataFrame, Set[str]]:
+        self._load_excel()
+        self._validate_raw_data()
+        return self._preprocess()
+
+    def _load_excel(self):
+        logger.info(f"Loading {self.config.data.FILE_PATH}...")
+        try:
+            sheets = pd.read_excel(self.config.data.FILE_PATH, sheet_name=None)
+            self.raw_data = {
+                "courses": sheets[self.config.data.SHEET_COURSES],
+                "offerings": sheets[self.config.data.SHEET_OFFERINGS],
+                "students_yearly": sheets[self.config.data.SHEET_STUDENTS_YEARLY],
+                "students_ind": sheets[self.config.data.SHEET_STUDENTS_INDIVIDUAL],
+            }
+        except Exception as e:
+            logger.error(f"Failed to load Excel: {e}")
+            raise
+
+    def _validate_raw_data(self):
+        """Validate required columns and log data quality metrics."""
+        req_cols = {
+            "courses": ["kode_mk", "kategori_mk"],
+            "students_ind": ["kode_mk", "thn", "smt", "kode_mhs"],
+            "students_yearly": ["thn", "smt", "jumlah_aktif"],
+        }
+
+        for key, cols in req_cols.items():
+            if not all(col in self.raw_data[key].columns for col in cols):
+                raise ValueError(f"Missing columns in {key}: {cols}")
+
+        # Log data quality metrics
+        self._log_data_quality()
+
+    def _log_data_quality(self):
+        """Log data quality metrics for monitoring."""
+        courses_df = self.raw_data["courses"]
+        students_df = self.raw_data["students_ind"]
+
+        logger.info("=" * 60)
+        logger.info("Data Quality Report:")
+        logger.info(f"  Courses (tabel1): {len(courses_df)} records")
+        logger.info(f"    - Unique courses: {courses_df['kode_mk'].nunique()}")
+        logger.info(
+            f"    - Duplicates: {len(courses_df) - courses_df['kode_mk'].nunique()}"
+        )
+        logger.info(f"  Students (tabel4): {len(students_df)} records")
+        logger.info(f"    - Unique students: {students_df['kode_mhs'].nunique()}")
+        logger.info("=" * 60)
+
+    def _clean_courses_data(self, courses: pd.DataFrame) -> pd.DataFrame:
+        """
+        Clean and standardize course catalog data.
+
+        Cleaning steps:
+        1. Remove exact duplicates
+        2. Standardize kategori_mk values (uppercase, strip whitespace)
+        3. Remove courses with invalid/missing data
+        4. Keep first occurrence for duplicate course codes
+        5. Validate kategori_mk values
+        """
+        initial_count = len(courses)
+
+        # Step 1: Remove exact duplicate rows
+        courses = courses.drop_duplicates()
+        if len(courses) < initial_count:
+            logger.info(
+                f"  Removed {initial_count - len(courses)} exact duplicate rows"
+            )
+
+        # Step 2: Standardize kategori_mk
+        courses["kategori_mk"] = (
+            courses["kategori_mk"]
+            .astype(str)
+            .str.upper()
+            .str.strip()
+            .replace("", np.nan)
+        )
+
+        # Step 3: Remove rows with missing critical data
+        before_dropna = len(courses)
+        courses = courses.dropna(subset=["kode_mk", "kategori_mk"])
+        if len(courses) < before_dropna:
+            logger.info(
+                f"  Removed {before_dropna - len(courses)} rows with missing kode_mk or kategori_mk"
+            )
+
+        # Step 4: Validate kategori_mk values (should be P or W)
+        valid_categories = {"P", "W"}
+        invalid_mask = ~courses["kategori_mk"].isin(valid_categories)
+        if invalid_mask.any():
+            invalid_cats = courses[invalid_mask]["kategori_mk"].unique()
+            logger.warning(
+                f"  Found {invalid_mask.sum()} courses with invalid categories: {invalid_cats}"
+            )
+            logger.warning("  Keeping only valid categories (P, W)")
+            courses = courses[~invalid_mask]
+
+        # Step 5: Remove duplicate course codes (keep first)
+        before_dedup = len(courses)
+        courses = courses.drop_duplicates(subset="kode_mk", keep="first")
+        if len(courses) < before_dedup:
+            logger.info(
+                f"  Removed {before_dedup - len(courses)} duplicate course codes (kept first occurrence)"
+            )
+
+        logger.info(f"  Final course catalog: {len(courses)} unique courses")
+
+        return courses
+
+    def _clean_students_data(self, students: pd.DataFrame) -> pd.DataFrame:
+        """
+        Clean and validate student enrollment data.
+
+        Cleaning steps:
+        1. Remove rows with missing critical data
+        2. Standardize data types
+        3. Remove invalid year/semester values
+        4. Remove duplicate enrollment records
+        """
+        initial_count = len(students)
+
+        # Step 1: Remove rows with missing critical data
+        students = students.dropna(subset=["kode_mk", "thn", "smt", "kode_mhs"])
+        if len(students) < initial_count:
+            logger.info(
+                f"  Removed {initial_count - len(students)} rows with missing critical data"
+            )
+
+        # Step 2: Ensure correct data types
+        students["thn"] = pd.to_numeric(students["thn"], errors="coerce")
+        students["smt"] = pd.to_numeric(students["smt"], errors="coerce")
+
+        # Step 3: Remove rows with invalid year/semester after conversion
+        before_invalid = len(students)
+        students = students.dropna(subset=["thn", "smt"])
+        if len(students) < before_invalid:
+            logger.info(
+                f"  Removed {before_invalid - len(students)} rows with invalid year/semester values"
+            )
+
+        # Step 4: Validate semester values (should be 1, 2, or 3)
+        valid_semesters = {1, 2, 3}
+        invalid_sem = ~students["smt"].isin(valid_semesters)
+        if invalid_sem.any():
+            logger.warning(
+                f"  Found {invalid_sem.sum()} records with invalid semester values"
+            )
+            students = students[~invalid_sem]
+
+        # Step 5: Validate year range (reasonable academic years)
+        current_year = pd.Timestamp.now().year
+        invalid_year = (students["thn"] < 2000) | (students["thn"] > current_year + 1)
+        if invalid_year.any():
+            logger.warning(
+                f"  Found {invalid_year.sum()} records with unreasonable year values"
+            )
+            students = students[~invalid_year]
+
+        # Step 6: Remove exact duplicate enrollments (same student, course, semester)
+        before_dedup = len(students)
+        students = students.drop_duplicates(
+            subset=["kode_mhs", "kode_mk", "thn", "smt"], keep="first"
+        )
+        if len(students) < before_dedup:
+            logger.info(
+                f"  Removed {before_dedup - len(students)} duplicate enrollment records"
+            )
+
+        logger.info(f"  Final enrollment records: {len(students)}")
+
+        return students
+
+    def _clean_yearly_population(self, yearly_pop: pd.DataFrame) -> pd.DataFrame:
+        """
+        Clean and validate yearly student population data.
+
+        Cleaning steps:
+        1. Remove duplicates
+        2. Validate and fill missing population data
+        3. Ensure chronological order
+        """
+        # Remove duplicate year-semester combinations
+        before_dedup = len(yearly_pop)
+        yearly_pop = yearly_pop.drop_duplicates(subset=["thn", "smt"], keep="first")
+        if len(yearly_pop) < before_dedup:
+            logger.info(
+                f"  Removed {before_dedup - len(yearly_pop)} duplicate year-semester records"
+            )
+
+        # Ensure jumlah_aktif is numeric and positive
+        yearly_pop["jumlah_aktif"] = pd.to_numeric(
+            yearly_pop["jumlah_aktif"], errors="coerce"
+        )
+
+        # Replace zero or negative values with NaN (will be filled later)
+        yearly_pop.loc[yearly_pop["jumlah_aktif"] <= 0, "jumlah_aktif"] = np.nan
+
+        # Sort by year and semester
+        yearly_pop = yearly_pop.sort_values(["thn", "smt"]).reset_index(drop=True)
+
+        logger.info(f"  Yearly population records: {len(yearly_pop)}")
+
+        return yearly_pop
+
+    def _preprocess(self) -> Tuple[pd.DataFrame, Set[str]]:
+        """Clean, merge, and aggregate data with comprehensive cleaning."""
+        logger.info("Preprocessing data...")
+        logger.info("-" * 60)
+
+        # Step 1: Clean course catalog
+        logger.info("Step 1: Cleaning course catalog...")
+        courses = self._clean_courses_data(self.raw_data["courses"].copy())
+
+        # Step 2: Identify elective courses
+        elective_category = self.config.data.ELECTIVE_CATEGORY
+        self.elective_codes = set(
+            courses[courses["kategori_mk"] == elective_category]["kode_mk"]
+        )
+        logger.info(f"Step 2: Identified {len(self.elective_codes)} elective courses")
+
+        if len(self.elective_codes) == 0:
+            logger.warning(
+                f"No elective courses found! Check if kategori_mk = '{elective_category}' exists in data."
+            )
+            logger.warning(
+                f"Elective identification rule: {self.config.get_elective_filter_description()}"
+            )
+            return pd.DataFrame(), set()
+
+        # Step 3: Clean student enrollment data
+        logger.info("Step 3: Cleaning student enrollment data...")
+        students = self._clean_students_data(self.raw_data["students_ind"].copy())
+
+        # Step 4: Filter for elective courses only
+        students = students[students["kode_mk"].isin(self.elective_codes)]
+        logger.info(f"Step 4: Filtered to {len(students)} elective enrollment records")
+
+        if len(students) == 0:
+            logger.warning("No enrollment data found for elective courses!")
+            return pd.DataFrame(), self.elective_codes
+
+        # Step 5: Aggregate enrollment by course-semester
+        logger.info("Step 5: Aggregating enrollment data...")
+        enrollment = (
+            students.groupby(["kode_mk", "thn", "smt"])["kode_mhs"]
+            .nunique()
+            .reset_index(name="enrollment")
+        )
+        logger.info(f"  Created {len(enrollment)} course-semester enrollment records")
+
+        # Step 6: Clean yearly population data
+        logger.info("Step 6: Cleaning yearly population data...")
+        yearly_pop = self._clean_yearly_population(
+            self.raw_data["students_yearly"][["thn", "smt", "jumlah_aktif"]].copy()
+        )
+
+        # Step 7: Merge enrollment with population data
+        logger.info("Step 7: Merging enrollment with population data...")
+        df = enrollment.merge(yearly_pop, on=["thn", "smt"], how="left")
+
+        # Step 8: Handle missing population data
+        missing_pop = df["jumlah_aktif"].isna().sum()
+        if missing_pop > 0:
+            logger.warning(
+                f"  {missing_pop} records missing population data - filling with interpolation"
+            )
+            df["jumlah_aktif"] = df["jumlah_aktif"].ffill().bfill()
+
+            # If still missing, use a reasonable default
+            if df["jumlah_aktif"].isna().any():
+                default_pop = 500  # Reasonable default student population
+                logger.warning(
+                    f"  Some population data still missing - using default: {default_pop}"
+                )
+                df["jumlah_aktif"] = df["jumlah_aktif"].fillna(default_pop)
+
+        # Step 9: Validate enrollment data
+        logger.info("Step 8: Validating final enrollment data...")
+        df = self._validate_enrollment_data(df)
+
+        # Step 10: Sort and finalize
+        df = df.sort_values(["kode_mk", "thn", "smt"]).reset_index(drop=True)
+        self.processed_data = df
+
+        logger.info("-" * 60)
+        logger.info(
+            f"✓ Preprocessing complete. {len(df)} enrollment records generated."
+        )
+        logger.info(f"✓ Year range: {df['thn'].min():.0f} - {df['thn'].max():.0f}")
+        logger.info(f"✓ Courses with data: {df['kode_mk'].nunique()}")
+        logger.info("-" * 60)
+
+        return df, self.elective_codes
+
+    def _validate_enrollment_data(self, df: pd.DataFrame) -> pd.DataFrame:
+        """
+        Validate and clean the final enrollment dataset.
+
+        Checks:
+        1. Remove records with zero enrollment
+        2. Check for outliers
+        3. Validate population data
+        """
+        initial_count = len(df)
+
+        # Remove zero enrollments
+        df = df[df["enrollment"] > 0]
+        if len(df) < initial_count:
+            logger.info(
+                f"  Removed {initial_count - len(df)} records with zero enrollment"
+            )
+
+        # Check for extreme outliers in enrollment
+        for course in df["kode_mk"].unique():
+            course_data = df[df["kode_mk"] == course]["enrollment"]
+            if len(course_data) > 1:
+                q75, q25 = course_data.quantile([0.75, 0.25])
+                iqr = q75 - q25
+                upper_bound = q75 + (3 * iqr)  # Using 3*IQR for outliers
+
+                outliers = course_data > upper_bound
+                if outliers.any():
+                    logger.debug(
+                        f"  Course {course} has {outliers.sum()} potential outliers (keeping them)"
+                    )
+
+        # Ensure population is reasonable
+        if (df["jumlah_aktif"] < 50).any():
+            logger.warning("  Some semesters have very low student population (<50)")
+
+        return df

data_validator.py

@@ -0,0 +1,467 @@
+"""
+Data Validation Utility
+
+Provides pre-flight checks and data quality validation for the enrollment prediction system.
+This module validates data availability, quality, and completeness before processing.
+"""
+
+import logging
+from dataclasses import dataclass
+from typing import Dict, List, Optional, Tuple
+
+import pandas as pd
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ValidationResult:
+    """Result of a validation check."""
+
+    passed: bool
+    message: str
+    severity: str = "INFO"  # INFO, WARNING, ERROR
+    details: Optional[Dict] = None
+
+
+@dataclass
+class SemesterDataStatus:
+    """Status of data availability for a specific semester."""
+
+    year: int
+    semester: int
+    has_offerings: bool
+    has_enrollments: bool
+    has_elective_enrollments: bool
+    total_enrollments: int
+    elective_enrollments: int
+    elective_courses: List[str]
+
+
+class DataValidator:
+    """Validates data quality and availability for the enrollment prediction system."""
+
+    def __init__(self, file_path: str):
+        """
+        Initialize the validator.
+
+        Args:
+            file_path: Path to the Excel data file
+        """
+        self.file_path = file_path
+        self.validation_results: List[ValidationResult] = []
+
+    def validate_all(self) -> Tuple[bool, List[ValidationResult]]:
+        """
+        Run all validation checks.
+
+        Returns:
+            Tuple of (all_passed, list of validation results)
+        """
+        logger.info("Running comprehensive data validation...")
+
+        # Load raw data
+        try:
+            self.raw_data = self._load_raw_data()
+        except Exception as e:
+            self.validation_results.append(
+                ValidationResult(
+                    passed=False,
+                    message=f"Failed to load data: {str(e)}",
+                    severity="ERROR",
+                )
+            )
+            return False, self.validation_results
+
+        # Run validation checks
+        self._validate_file_structure()
+        self._validate_course_catalog()
+        self._validate_elective_courses()
+        self._validate_enrollment_data()
+        self._validate_population_data()
+
+        # Overall result
+        all_passed = all(
+            r.passed for r in self.validation_results if r.severity == "ERROR"
+        )
+
+        return all_passed, self.validation_results
+
+    def check_semester_data_availability(
+        self, year: int, semester: int
+    ) -> SemesterDataStatus:
+        """
+        Check data availability for a specific semester.
+
+        Args:
+            year: Academic year
+            semester: Semester (1 or 2)
+
+        Returns:
+            SemesterDataStatus object with detailed availability info
+        """
+        if not hasattr(self, "raw_data"):
+            self.raw_data = self._load_raw_data()
+
+        # Check course offerings (tabel2)
+        offerings = self.raw_data["offerings"]
+        has_offerings = (
+            len(
+                offerings[
+                    (offerings["tahun"] == year) & (offerings["semester"] == semester)
+                ]
+            )
+            > 0
+        )
+
+        # Check enrollments (tabel4)
+        students = self.raw_data["students"]
+        semester_enrollments = students[
+            (students["thn"] == year) & (students["smt"] == semester)
+        ]
+        has_enrollments = len(semester_enrollments) > 0
+
+        # Check elective enrollments
+        elective_codes = self._get_elective_codes()
+        elective_enrollments = semester_enrollments[
+            semester_enrollments["kode_mk"].isin(elective_codes)
+        ]
+        has_elective_enrollments = len(elective_enrollments) > 0
+
+        # Get elective courses for this semester
+        elective_courses = []
+        if has_elective_enrollments:
+            elective_courses = (
+                elective_enrollments.groupby("kode_mk")["kode_mhs"]
+                .nunique()
+                .sort_values(ascending=False)
+                .to_dict()
+            )
+
+        return SemesterDataStatus(
+            year=year,
+            semester=semester,
+            has_offerings=has_offerings,
+            has_enrollments=has_enrollments,
+            has_elective_enrollments=has_elective_enrollments,
+            total_enrollments=len(semester_enrollments),
+            elective_enrollments=len(elective_enrollments),
+            elective_courses=elective_courses,
+        )
+
+    def get_available_semesters_for_backtesting(self) -> List[Tuple[int, int]]:
+        """
+        Get list of semesters that have elective enrollment data (suitable for backtesting).
+
+        Returns:
+            List of (year, semester) tuples
+        """
+        if not hasattr(self, "raw_data"):
+            self.raw_data = self._load_raw_data()
+
+        students = self.raw_data["students"]
+        elective_codes = self._get_elective_codes()
+
+        # Filter to elective enrollments only
+        elective_students = students[students["kode_mk"].isin(elective_codes)]
+
+        # Get unique year-semester combinations
+        available = (
+            elective_students.groupby(["thn", "smt"]).size().reset_index(name="count")
+        )
+        available = available[available["count"] > 0]
+
+        semesters = [
+            (int(row["thn"]), int(row["smt"])) for _, row in available.iterrows()
+        ]
+        semesters.sort(reverse=True)  # Most recent first
+
+        return semesters
+
+    def print_validation_summary(self):
+        """Print a summary of validation results."""
+        if not self.validation_results:
+            print("\nWARNING: No validation has been run yet.")
+            return
+
+        print("\n" + "=" * 80)
+        print("DATA VALIDATION SUMMARY")
+        print("=" * 80)
+
+        errors = [r for r in self.validation_results if r.severity == "ERROR"]
+        warnings = [r for r in self.validation_results if r.severity == "WARNING"]
+        info = [r for r in self.validation_results if r.severity == "INFO"]
+
+        if errors:
+            print(f"\nERROR ({len(errors)}):")
+            for result in errors:
+                print(f"   - {result.message}")
+
+        if warnings:
+            print(f"\nWARNING ({len(warnings)}):")
+            for result in warnings:
+                print(f"   - {result.message}")
+
+        if info:
+            print(f"\nINFO ({len(info)}):")
+            for result in info:
+                print(f"   - {result.message}")
+
+        print("\n" + "=" * 80)
+        if not errors:
+            print("VALIDATION PASSED - Data is ready for processing")
+        else:
+            print("VALIDATION FAILED - Please fix errors before proceeding")
+        print("=" * 80)
+
+    def _load_raw_data(self) -> Dict[str, pd.DataFrame]:
+        """Load raw data from Excel file."""
+        logger.info(f"Loading data from {self.file_path}...")
+
+        return {
+            "courses": pd.read_excel(self.file_path, sheet_name="tabel1_data_matkul"),
+            "offerings": pd.read_excel(
+                self.file_path, sheet_name="tabel2_data_matkul_dibuka"
+            ),
+            "population": pd.read_excel(
+                self.file_path, sheet_name="tabel3_data_mahasiswa_per_tahun"
+            ),
+            "students": pd.read_excel(
+                self.file_path, sheet_name="tabel4_data_individu_mahasiswa"
+            ),
+        }
+
+    def _validate_file_structure(self):
+        """Validate that all required sheets and columns exist."""
+        required_sheets = {
+            "courses": ["kode_mk", "nama_mk", "kategori_mk"],
+            "offerings": ["kode_mk", "tahun", "semester"],
+            "students": ["kode_mk", "kode_mhs", "thn", "smt"],
+            "population": ["jumlah_aktif"],  # tahun_ajaran and semester may vary
+        }
+
+        for sheet_name, required_cols in required_sheets.items():
+            df = self.raw_data.get(sheet_name)
+            if df is None:
+                self.validation_results.append(
+                    ValidationResult(
+                        passed=False,
+                        message=f"Sheet '{sheet_name}' not found",
+                        severity="ERROR",
+                    )
+                )
+                continue
+
+            missing_cols = [col for col in required_cols if col not in df.columns]
+            if missing_cols:
+                self.validation_results.append(
+                    ValidationResult(
+                        passed=False,
+                        message=f"Missing columns in {sheet_name}: {missing_cols}",
+                        severity="ERROR",
+                    )
+                )
+            else:
+                self.validation_results.append(
+                    ValidationResult(
+                        passed=True,
+                        message=f"Sheet '{sheet_name}' has all required columns",
+                        severity="INFO",
+                    )
+                )
+
+    def _validate_course_catalog(self):
+        """Validate course catalog (tabel1)."""
+        courses = self.raw_data["courses"]
+
+        # Check for duplicates
+        total_records = len(courses)
+        unique_courses = courses["kode_mk"].nunique()
+        duplicate_count = total_records - unique_courses
+
+        if duplicate_count > 0:
+            self.validation_results.append(
+                ValidationResult(
+                    passed=True,
+                    message=f"Course catalog has {duplicate_count:,} duplicate records (will be cleaned)",
+                    severity="WARNING",
+                    details={"total": total_records, "unique": unique_courses},
+                )
+            )
+
+        # Check for category consistency
+        categories = courses["kategori_mk"].unique()
+        non_standard = [c for c in categories if c not in ["W", "P"]]
+        if non_standard:
+            self.validation_results.append(
+                ValidationResult(
+                    passed=True,
+                    message=f"Non-standard categories found: {non_standard} (will be normalized)",
+                    severity="WARNING",
+                )
+            )
+
+    def _validate_elective_courses(self):
+        """Validate elective course identification."""
+        courses = self.raw_data["courses"]
+
+        # Clean and identify electives
+        courses_clean = courses.drop_duplicates(subset="kode_mk").copy()
+        courses_clean["kategori_mk"] = (
+            courses_clean["kategori_mk"].astype(str).str.upper().str.strip()
+        )
+
+        electives = courses_clean[courses_clean["kategori_mk"] == "P"]
+        elective_count = len(electives)
+
+        if elective_count == 0:
+            self.validation_results.append(
+                ValidationResult(
+                    passed=False,
+                    message="No elective courses found (kategori_mk = 'P')",
+                    severity="ERROR",
+                )
+            )
+        else:
+            self.validation_results.append(
+                ValidationResult(
+                    passed=True,
+                    message=f"Found {elective_count} elective courses",
+                    severity="INFO",
+                    details={"electives": electives["kode_mk"].tolist()},
+                )
+            )
+
+    def _validate_enrollment_data(self):
+        """Validate student enrollment data (tabel4)."""
+        students = self.raw_data["students"]
+
+        # Check for missing critical data
+        critical_fields = ["kode_mk", "kode_mhs", "thn", "smt"]
+        missing_data = students[critical_fields].isnull().any(axis=1).sum()
+
+        if missing_data > 0:
+            self.validation_results.append(
+                ValidationResult(
+                    passed=True,
+                    message=f"{missing_data} enrollment records have missing data (will be cleaned)",
+                    severity="WARNING",
+                )
+            )
+
+        # Check for duplicates
+        duplicate_enrollments = students.duplicated(
+            subset=["kode_mhs", "kode_mk", "thn", "smt"]
+        ).sum()
+
+        if duplicate_enrollments > 0:
+            self.validation_results.append(
+                ValidationResult(
+                    passed=True,
+                    message=f"{duplicate_enrollments:,} duplicate enrollment records (will be cleaned)",
+                    severity="WARNING",
+                )
+            )
+
+        # Check year range
+        min_year = students["thn"].min()
+        max_year = students["thn"].max()
+
+        self.validation_results.append(
+            ValidationResult(
+                passed=True,
+                message=f"Enrollment data spans {int(min_year)} to {int(max_year)}",
+                severity="INFO",
+            )
+        )
+
+    def _validate_population_data(self):
+        """Validate yearly population data (tabel3)."""
+        population = self.raw_data["population"]
+
+        if len(population) == 0:
+            self.validation_results.append(
+                ValidationResult(
+                    passed=False,
+                    message="No population data found",
+                    severity="ERROR",
+                )
+            )
+            return
+
+        # Check for required fields (note: actual columns are tahun_ajaran/semester, not in sheet_name definition)
+        if "jumlah_aktif" in population.columns:
+            min_pop = population["jumlah_aktif"].min()
+            max_pop = population["jumlah_aktif"].max()
+
+            self.validation_results.append(
+                ValidationResult(
+                    passed=True,
+                    message=f"Population data: {len(population)} records, range {int(min_pop)}-{int(max_pop)} students",
+                    severity="INFO",
+                )
+            )
+        else:
+            self.validation_results.append(
+                ValidationResult(
+                    passed=False,
+                    message="Population data missing 'jumlah_aktif' column",
+                    severity="ERROR",
+                )
+            )
+
+    def _get_elective_codes(self) -> set:
+        """Get set of elective course codes."""
+        courses = self.raw_data["courses"]
+        courses_clean = courses.drop_duplicates(subset="kode_mk").copy()
+        courses_clean["kategori_mk"] = (
+            courses_clean["kategori_mk"].astype(str).str.upper().str.strip()
+        )
+        return set(courses_clean[courses_clean["kategori_mk"] == "P"]["kode_mk"])
+
+
+if __name__ == "__main__":
+    # Example usage
+    logging.basicConfig(
+        level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+    )
+
+    validator = DataValidator(
+        "data/Data Perkuliahan Mahasiswa untuk Penelitian (8 Oktober 2025).xlsx"
+    )
+
+    # Run validation
+    passed, results = validator.validate_all()
+    validator.print_validation_summary()
+
+    # Check specific semesters
+    print("\n" + "=" * 80)
+    print("SEMESTER DATA AVAILABILITY")
+    print("=" * 80)
+
+    for year, semester in [(2024, 2), (2025, 1)]:
+        status = validator.check_semester_data_availability(year, semester)
+        print(f"\n{year} Semester {semester}:")
+        print(f"  Offerings: {'Yes' if status.has_offerings else 'No'}")
+        print(
+            f"  Enrollments: {'Yes' if status.has_enrollments else 'No'} ({status.total_enrollments} records)"
+        )
+        print(
+            f"  Elective Enrollments: {'Yes' if status.has_elective_enrollments else 'No'} ({status.elective_enrollments} records)"
+        )
+        if status.elective_courses:
+            print(f"  Elective courses: {len(status.elective_courses)}")
+            for code, count in list(status.elective_courses.items())[:5]:
+                print(f"    - {code}: {count} students")
+
+    # Show available semesters for backtesting
+    print("\n" + "=" * 80)
+    print("SEMESTERS AVAILABLE FOR BACKTESTING")
+    print("=" * 80)
+    available = validator.get_available_semesters_for_backtesting()
+    if available:
+        print(f"\nFound {len(available)} semesters with elective enrollment data:")
+        for year, sem in available:
+            print(f"  • {year} Semester {sem}")
+    else:
+        print("\nERROR: No semesters with elective enrollment data found!")

evaluator.py

@@ -0,0 +1,103 @@
+import logging
+import os
+from pathlib import Path
+
+import matplotlib.pyplot as plt
+import numpy as np
+import pandas as pd
+import seaborn as sns
+from sklearn.metrics import mean_absolute_error, mean_squared_error
+
+from config import Config
+
+logger = logging.getLogger(__name__)
+
+
+class Evaluator:
+    def __init__(self, config: Config):
+        self.config = config
+
+    def run_backtest(self, full_data: pd.DataFrame, predictor):
+        """Simulate past semesters to check accuracy."""
+        logger.info("Starting Backtest...")
+        results = []
+
+        start_year: int = self.config.backtest.START_YEAR
+        end_year: int = self.config.backtest.END_YEAR
+
+        for year in range(start_year, end_year + 1):
+            for smt in [1, 2]:
+                target_mask = (full_data["thn"] == year) & (full_data["smt"] == smt)
+                test_set = full_data[target_mask]
+
+                if test_set.empty:
+                    continue
+
+                train_set = full_data[
+                    (full_data["thn"] < year)
+                    | ((full_data["thn"] == year) & (full_data["smt"] < smt))
+                ]
+
+                try:
+                    pop_est = predictor.get_student_forecast(year, smt)
+                except Exception:
+                    pop_est = test_set["jumlah_aktif"].mean()
+
+                for _, row in test_set.iterrows():
+                    pred = predictor.predict_course(
+                        row["kode_mk"], train_set, year, smt, pop_est
+                    )
+
+                    results.append(
+                        {
+                            "year": year,
+                            "semester": smt,
+                            "kode_mk": row["kode_mk"],
+                            "actual": row["enrollment"],
+                            "predicted": pred["val"],
+                            "strategy": pred["strategy"],
+                            "error": abs(row["enrollment"] - pred["val"]),
+                        }
+                    )
+
+        return pd.DataFrame(results)
+
+    def generate_metrics(self, results: pd.DataFrame):
+        """Calculate and log performance metrics."""
+        results["error"] = abs(results["predicted"] - results["actual"])
+
+        mae = mean_absolute_error(results["actual"], results["predicted"])
+        rmse = np.sqrt(mean_squared_error(results["actual"], results["predicted"]))
+
+        logger.info("\n" + "=" * 40)
+        logger.info("BACKTEST METRICS")
+        logger.info("=" * 40)
+        logger.info(f"Overall MAE:  {mae:.2f}")
+        logger.info(f"Overall RMSE: {rmse:.2f}")
+
+        logger.info("\nPerformance by Strategy:")
+        strat_perf = results.groupby("strategy")["error"].mean()
+        logger.info(strat_perf.to_string())
+
+        self._plot_results(results)
+
+        return {
+            'mae': mae,
+            'rmse': rmse
+        }
+
+    def _plot_results(self, df):
+        """Generate simple Actual vs Predicted scatter plot."""
+        Path(self.config.output.OUTPUT_DIR).mkdir(parents=True, exist_ok=True)
+
+        plt.figure(figsize=(10, 6))
+        sns.scatterplot(
+            data=df, x="actual", y="predicted", hue="strategy", style="strategy"
+        )
+
+        limit = max(df["actual"].max(), df["predicted"].max())
+        plt.plot([0, limit], [0, limit], "r--", alpha=0.5)
+
+        plt.title("Actual vs Predicted Enrollment")
+        plt.savefig(f"{self.config.output.OUTPUT_DIR}/backtest_scatter.png")
+        plt.close()

prophet_predictor.py

@@ -0,0 +1,253 @@
+import logging
+from typing import Optional
+
+import numpy as np
+import pandas as pd
+from prophet import Prophet
+
+from config import Config
+
+logger = logging.getLogger(__name__)
+
+
+class ProphetPredictor:
+    def __init__(self, config: Config):
+        self.config = config
+        self.student_model: Optional[Prophet] = None
+
+    def train_student_population_model(self, population_data: pd.DataFrame):
+        df = population_data.copy()
+        df["ds"] = pd.to_datetime(
+            df["thn"].astype(str)
+            + "-"
+            + df["smt"].map(self.config.prediction.SEMESTER_TO_MONTH)
+        )
+        df["y"] = df["jumlah_aktif"]
+
+        self.student_model = Prophet(daily_seasonality=False, weekly_seasonality=False)  # type: ignore[arg-type]
+        self.student_model.fit(df)
+        logger.info("Student population model trained.")
+
+    def get_student_forecast(self, year: int, semester: int) -> float:
+        assert self.student_model is not None, "Student model must be trained first"
+        target_date = pd.to_datetime(
+            f"{year}-{self.config.prediction.SEMESTER_TO_MONTH[semester]}"
+        )
+        future = pd.DataFrame({"ds": [target_date]})
+        forecast = self.student_model.predict(future)
+        return max(forecast["yhat"].values[0], 100)
+
+    def predict_course(
+        self,
+        course_code: str,
+        df_history: pd.DataFrame,
+        target_year: int,
+        target_smt: int,
+        student_pop: float,
+    ) -> dict:
+        hist = df_history[
+            (df_history["kode_mk"] == course_code) &
+            (df_history["smt"] == target_smt)
+        ].sort_values(["thn", "smt"])
+
+        if len(hist) == 0:
+            return {
+                "val": self.config.model.FALLBACK_DEFAULT,
+                "strategy": "cold_start",
+                "confidence": "low",
+            }
+
+        return self._predict_prophet_logistic(
+            hist, target_year, target_smt, student_pop
+        )
+
+    def _predict_prophet_logistic(
+        self, hist: pd.DataFrame, year: int, smt: int, pop: float
+    ) -> dict:
+        df = hist.copy()
+        df["ds"] = pd.to_datetime(
+            df["thn"].astype(int).astype(str)
+            + "-"
+            + df["smt"].astype(int).map(self.config.prediction.SEMESTER_TO_MONTH)
+        )
+        df["y"] = df["enrollment"]
+
+        if df["y"].notna().sum() < 2:
+            return {
+                "val": hist["enrollment"].mean(),
+                "strategy": "fallback_mean",
+                "confidence": "medium",
+            }
+
+        if "jumlah_aktif" not in df.columns:
+            logger.warning(
+                "jumlah_aktif column missing from historical data - cannot use regressor"
+            )
+            return {
+                "val": hist["enrollment"].mean(),
+                "strategy": "fallback_mean",
+                "confidence": "low",
+            }
+
+        hist_max = df["y"].max()
+        hist_mean = df["y"].mean()
+
+        cap_value = min(
+            hist_max * self.config.prediction.MAX_CAPACITY_MULTIPLIER,
+            self.config.prediction.ABSOLUTE_MAX_STUDENTS,
+        )
+
+        df["cap"] = cap_value
+        df["floor"] = 0
+
+        try:
+            m = Prophet(
+                growth=self.config.model.GROWTH_MODE,
+                changepoint_prior_scale=self.config.model.CHANGEPOINT_SCALE,
+                seasonality_mode=self.config.model.SEASONALITY_MODE,
+                daily_seasonality=False,  # type: ignore[arg-type]
+                weekly_seasonality=False,  # type: ignore[arg-type]
+            )
+
+            m.add_regressor("jumlah_aktif", mode="multiplicative")
+            m.fit(df[["ds", "y", "cap", "floor", "jumlah_aktif"]])
+
+            future_date = pd.to_datetime(
+                f"{year}-{self.config.prediction.SEMESTER_TO_MONTH[smt]}"
+            )
+
+            future = pd.DataFrame(
+                {
+                    "ds": [future_date],
+                    "cap": [cap_value],
+                    "floor": [0],
+                    "jumlah_aktif": [pop],
+                }
+            )
+
+            forecast = m.predict(future)
+            raw_pred = forecast["yhat"].values[0]
+
+            if (
+                raw_pred < 0
+                or not np.isfinite(raw_pred)
+                or raw_pred > hist_max * 5
+                or raw_pred > cap_value * 2
+            ):
+                logger.warning(
+                    f"Prophet prediction ({raw_pred:.1f}) unrealistic. "
+                    f"Using trend-based fallback. (hist_max={hist_max}, cap={cap_value})"
+                )
+                if len(df) >= 3:
+                    recent_trend = df["y"].tail(3).mean()
+                    pop_growth_factor = pop / df["jumlah_aktif"].mean()
+                    growth_factor = min(
+                        max(pop_growth_factor, 0.8), 1.3
+                    )
+                    pred = recent_trend * growth_factor
+                else:
+                    pop_growth_factor = pop / df["jumlah_aktif"].mean()
+                    pred = hist_mean * min(max(pop_growth_factor, 0.8), 1.3)
+
+                pred = min(max(pred, 0), cap_value)
+
+                return {
+                    "val": pred,
+                    "strategy": "trend_fallback",
+                    "confidence": "medium",
+                }
+
+            pred = min(max(0, raw_pred), cap_value)
+
+            return {
+                "val": pred,
+                "strategy": "prophet_logistic",
+                "confidence": "high",
+            }
+
+        except Exception as e:
+            logger.warning(f"Prophet failed for course. Error: {e}. Using fallback.")
+            return {
+                "val": hist["enrollment"].mean(),
+                "strategy": "fallback_mean",
+                "confidence": "medium",
+            }
+
+    def generate_batch_predictions(
+        self,
+        full_data: pd.DataFrame,
+        course_metadata: pd.DataFrame,
+        electives: set,
+        year: int,
+        smt: int,
+    ):
+        """Generate predictions for all courses."""
+        student_pop = self.get_student_forecast(year, smt)
+        results = []
+
+        logger.info(
+            f"Predicting for {len(electives)} courses (Pop: {int(student_pop)})..."
+        )
+
+        for code in electives:
+            meta = course_metadata[course_metadata["kode_mk"] == code].iloc[0]
+
+            pred_result = self.predict_course(code, full_data, year, smt, student_pop)
+            pred_val = pred_result["val"]
+
+            rec_quota = int(
+                np.ceil(pred_val * (1 + self.config.prediction.BUFFER_PERCENT))
+            )
+            rec_quota = max(rec_quota, self.config.prediction.MIN_QUOTA_OPEN)
+
+            status = (
+                "BUKA"
+                if pred_val >= self.config.prediction.MIN_PREDICT_THRESHOLD
+                else "TUTUP"
+            )
+
+            results.append(
+                {
+                    "kode_mk": code,
+                    "nama_mk": meta["nama_mk"],
+                    "sks": meta["sks_mk"],
+                    "predicted_enrollment": round(pred_val, 1),
+                    "recommended_quota": rec_quota if status == "BUKA" else 0,
+                    "recommendation": status,
+                    "strategy": pred_result["strategy"],
+                    "confidence": pred_result["confidence"],
+                    "classes_est": int(np.ceil(rec_quota / 40))
+                    if status == "BUKA"
+                    else 0,
+                }
+            )
+
+        return pd.DataFrame(results).sort_values(
+            "predicted_enrollment", ascending=False
+        )
+
+    def predict_course_enrollment(
+        self,
+        course_code: str,
+        train_data: pd.DataFrame,
+        test_year: int,
+        test_semester: int,
+        test_student_count: float,
+    ) -> tuple[float, str]:
+        result = self.predict_course(
+            course_code=course_code,
+            df_history=train_data,
+            target_year=test_year,
+            target_smt=test_semester,
+            student_pop=test_student_count,
+        )
+
+        val = float(result["val"])
+        if not np.isfinite(val):
+            val = self.config.model.FALLBACK_DEFAULT
+            strategy = "fallback_default"
+        else:
+            val = max(0.0, val)
+            strategy = result["strategy"]
+
+        return val, strategy

requirements.txt

@@ -0,0 +1,10 @@
+pandas>=2.0.0
+numpy>=1.24.0
+openpyxl>=3.1.0
+prophet>=1.1.5
+scikit-learn>=1.3.0
+matplotlib>=3.7.0
+seaborn>=0.12.0
+gradio>=4.0.0
+python-dateutil>=2.8.2
+huggingface_hub>=0.20.0

utils.py

@@ -0,0 +1,23 @@
+import logging
+import os
+from pathlib import Path
+
+import pandas as pd
+
+
+def setup_logging(level: str):
+    logging.basicConfig(
+        level=level,
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+        datefmt="%H:%M:%S",
+    )
+
+    logging.getLogger("prophet").setLevel(logging.WARNING)
+    logging.getLogger("cmdstanpy").setLevel(logging.WARNING)
+
+
+def save_excel(df: pd.DataFrame, filename: str, output_dir: str):
+    Path(output_dir).mkdir(exist_ok=True, parents=True)
+    path = os.path.join(output_dir, filename)
+    df.to_excel(path, index=False)
+    logging.info(f"Results saved to: {path}")