Sesi 5: Machine Learning Integration

from decimal import Decimal from typing import Dict class ProductionScoringEngine: """ ENGINE SKOR KREDIT PRODUKSI (NON-SIMULASI). Metodologi: Weighted Scorecard (Tabel Bobot). Pendekatan ini bersifat deterministik (input A selalu menghasilkan output B), memudahkan penjelasan ke nasabah dan audit internal. """ # Konstanta Bobot (Weights) BASE_SCORE = 300 MAX_SCORE = 850 # Threshold Segmen (Sesuai manual Bagian 1) SEGMENT_HIGH_RISK_THRESHOLD = 550 SEGMENT_MED_RISK_THRESHOLD = 700 def calculate_score(self, monthly_income: float, assets: float, loan_amount: float, tenor_months: int) -> Dict: """ Menghitung Skor Kredit berdasarkan data finansial nyata. """ # --- 1. Perhitungan Angsuran (Flat Rate Syariah) --- # Menggunakan margin standar 15% per tahun untuk simulasi kapasitas bayar margin_rate = 0.15 total_margin = loan_amount * margin_rate total_debt = loan_amount + total_margin monthly_installment = total_debt / tenor_months # --- 2. Variabel Penilaian (Assessment Variables) --- # A. Debt Service Ratio (DSR) = Angsuran / Pendapatan # Standar OJK: DSR aman di bawah 35-40% if monthly_income <= 0: dsr_score = -200 else: dsr = monthly_installment / monthly_income if dsr <= 0.30: dsr_score = 150 # Sangat Sehat elif dsr <= 0.40: dsr_score = 100 # Sehat elif dsr <= 0.50: dsr_score = 0 # Netral else: dsr_score = -150 # Berisiko # B. Asset Coverage Ratio (ACR) = Total Aset / Plafond if loan_amount <= 0: acr_score = 0 else: acr = assets / loan_amount if acr >= 1.2: acr_score = 100 # Aset > Pinjaman elif acr >= 0.8: acr_score = 50 else: acr_score = -50 # C. Tenor Risk if tenor_months <= 12: tenor_score = 50 elif tenor_months <= 24: tenor_score = 20 elif tenor_months <= 36: tenor_score = -20 else: tenor_score = -50 # --- 3. Agregasi Skor --- final_score = self.BASE_SCORE + dsr_score + acr_score + tenor_score final_score = max(self.BASE_SCORE, min(self.MAX_SCORE, int(final_score))) # --- 4. Probabilitas Gagal Bayar (Deterministik) --- if final_score >= 750: pd = 0.01 # 1% elif final_score >= 650: pd = 0.05 # 5% elif final_score >= 550: pd = 0.15 # 15% else: pd = 0.40 # 40% # --- 5. Segmentasi --- if final_score >= self.SEGMENT_MED_RISK_THRESHOLD: segment = "low" elif final_score >= self.SEGMENT_HIGH_RISK_THRESHOLD: segment = "medium" else: segment = "high" return { "credit_score": final_score, "default_probability": float(pd), "segment": segment, "dsr": round(monthly_installment / monthly_income if monthly_income > 0 else 0, 2), "monthly_installment": round(monthly_installment, 2), "assessment": { "dsr_score": dsr_score, "acr_score": acr_score, "tenor_score": tenor_score } } # Instance Engine untuk dipanggil oleh API scoring_engine = ProductionScoringEngine()

from pydantic import BaseModel from typing import List, Dict, Any class PredictionResult(BaseModel): customer_id: int credit_score: int default_probability: float segment: str recommendation: str class BulkPredictionRequest(BaseModel): file_path: str # Path ke file yang sudah diupload di sesi 4 class BulkPredictionResponse(BaseModel): total_rows: int high_risk_count: int results: List[Dict[str, Any]]

from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session import pandas as pd import os from db import get_db from app.models.loan import Loan from app.models.customer import Customer from app.schemas.prediction import PredictionResult, BulkPredictionRequest, BulkPredictionResponse from app.ml.predictor import scoring_engine router = APIRouter() @router.post("/individual/{loan_id}", response_model=PredictionResult) def predict_individual( loan_id: int, db: Session = Depends(get_db) ): """ Melakukan scoring pada 1 pembiayaan yang sudah ada. Mengupdate data scoring di tabel Loan. """ # 1. Ambil Data Loan & Customer loan = db.query(Loan).filter(Loan.id == loan_id).first() if not loan: raise HTTPException(status_code=404, detail="Loan not found") customer = db.query(Customer).filter(Customer.id == loan.customer_id).first() if not customer: raise HTTPException(status_code=404, detail="Customer not found") # 2. Siapkan Data untuk Model income = float(customer.monthly_income or 0) assets = float(customer.assets_value or 0) amount = float(loan.principal_amount) tenor = loan.tenor_months # 3. Jalankan Engine result = scoring_engine.calculate_score(income, assets, amount, tenor) # 4. Update Database (Loan Table) loan.credit_score = result['credit_score'] loan.default_probability = result['default_probability'] loan.segment = result['segment'] db.commit() db.refresh(loan) return { "customer_id": customer.id, "credit_score": result['credit_score'], "default_probability": result['default_probability'], "segment": result['segment'], "recommendation": f"Segment {result['segment'].upper()} - DSR: {result['dsr']}" } @router.post("/bulk", response_model=BulkPredictionResponse) def predict_bulk( request: BulkPredictionRequest ): """ Membaca file CSV/XLSX yang diupload dan melakukan prediksi massal. Mengembalikan ringkasan risiko portofolio. """ file_path = request.file_path # Validasi file ada if not os.path.exists(file_path): raise HTTPException(status_code=404, detail="File not found on server") # Baca File try: if file_path.endswith('.csv'): df = pd.read_csv(file_path) else: df = pd.read_excel(file_path) except Exception as e: raise HTTPException(status_code=400, detail=f"Failed to read file: {str(e)}") # Mapping Kolom (Pastikan file upload punya kolom ini, atau disesuaikan) # Disini kita asumsikan file punya kolom: 'monthly_income', 'assets', 'loan_amount', 'tenor' # Jika kolom berbeda, user perlu mapping di frontend atau rename kolom di Excel. results = [] high_risk_count = 0 for index, row in df.iterrows(): try: income = float(row.get('monthly_income', 0)) assets = float(row.get('assets_value', 0)) amount = float(row.get('loan_amount', 0)) tenor = int(row.get('tenor_months', 12)) pred = scoring_engine.calculate_score(income, assets, amount, tenor) if pred['segment'] == 'high': high_risk_count += 1 results.append({ "row_index": index, "score": pred['credit_score'], "segment": pred['segment'] }) except Exception as e: continue # Lewati baris error return { "total_rows": len(results), "high_risk_count": high_risk_count, "results": results }

from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from db import engine import config # Import Routers from app.api.auth import router as auth_router from app.api.customers import router as cust_router from app.api.loans import router as loan_router from app.api.files import router as file_router from app.api.predictions import router as pred_router # NEW app = FastAPI(title=config.get_settings().APP_NAME, version="1.0.0") # CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Include Routes app.include_router(auth_router, prefix="/api/v1/auth", tags=["Authentication"]) app.include_router(cust_router, prefix="/api/v1/customers", tags=["Customers"]) app.include_router(loan_router, prefix="/api/v1/loans", tags=["Loans"]) app.include_router(file_router, prefix="/api/v1/files", tags=["Files & Uploads"]) app.include_router(pred_router, prefix="/api/v1/predictions", tags=["Machine Learning (AI)"]) # NEW @app.on_event("startup") def on_startup(): try: with engine.connect() as conn: print("✅ DB Connected!") import os os.makedirs("/www/wwwroot/app-collection/ao/uploads", exist_ok=True) print("✅ Directories checked.") except Exception as e: print(f"❌ Error: {e}") @app.get("/") def root(): return {"status": "BPRS API Running", "session": "5 - AI Integration Active"}