Sesi 6: Monitoring & Payments

from sqlalchemy.orm import Session from datetime import date, timedelta from app.models.loan import Payment, Loan, Notification from decimal import Decimal def pay_installment(db: Session, payment_id: int, amount_paid: Decimal): """ Membayar satu jadwal angsuran. """ payment = db.query(Payment).filter(Payment.id == payment_id).first() if not payment: raise ValueError("Payment schedule not found") if payment.status == "default": raise ValueError("Cannot pay a defaulted loan without restructure.") # Update Data Pembayaran payment.amount_paid += amount_paid payment.paid_date = date.today() # Simpan tanggal bayar hari ini # Validasi Status if payment.amount_paid >= payment.amount_due: payment.status = "on_time" # Jika lunas, anggap on_time untuk record ini else: # Jika bayar sebagian, status tetap late atau on_time tergantung tanggal jatuh tempo pass db.commit() db.refresh(payment) return payment def check_and_flag_late_payments(db: Session): """ Fungsi ini seharusnya dijalankan oleh Cron Job setiap hari. Ia memeriksa pembayaran yang belum dibayar dan tanggalnya sudah lewat. """ today = date.today() # Cari pembayaran yang belum dibayar (amount_paid = 0) dan tanggal jatuh tempo kemarin/sebelumnya late_payments = db.query(Payment).filter( Payment.schedule_date < today, Payment.amount_paid == 0, Payment.status == "on_time" # Hanya update yang masih on_time ).all() for p in late_payments: # Update status menjadi late p.status = "late" # Buat Notifikasi di Database notif = Notification( loan_id=p.loan_id, type="late_alert", message=f"Angsuran tgl {p.schedule_date} belum dibayar.", sent_to="system", # Di production, ini bisa diambil dari loan.customer.phone status="pending" ) db.add(notif) db.commit() return len(late_payments) def get_portfolio_summary(db: Session): """ Menghitung statistik untuk Dashboard. """ # Total Pinjaman Aktif total_loans = db.query(Loan).filter(Loan.status == "active").count() # Total Portofolio (Principal Amount) from sqlalchemy import func total_principal = db.query(func.sum(Loan.principal_amount)).filter(Loan.status == "active").scalar() or 0 # Distribusi Risiko (Dari tabel loans) low_risk = db.query(Loan).filter(Loan.segment == "low", Loan.status == "active").count() med_risk = db.query(Loan).filter(Loan.segment == "medium", Loan.status == "active").count() high_risk = db.query(Loan).filter(Loan.segment == "high", Loan.status == "active").count() # Rasio Kelalaian (Delinquency) = Pembayaran Late / Total Pembayaran jatuh tempo # Untuk simplifikasi dashboard, kita hitung jumlah payment yang status 'late' late_payments_count = db.query(Payment).filter(Payment.status == "late").count() due_payments_count = db.query(Payment).filter(Payment.schedule_date <= date.today()).count() delinquency_rate = (late_payments_count / due_payments_count * 100) if due_payments_count > 0 else 0 return { "total_active_loans": total_loans, "total_principal": float(total_principal), "risk_distribution": { "low": low_risk, "medium": med_risk, "high": high_risk }, "delinquency_rate": round(delinquency_rate, 2) }

from sqlalchemy import Column, BigInteger, Integer, String, Text, DateTime, ForeignKey, Enum from app.models.base import TimestampMixin, Base class NotificationType(str, enum.Enum): late_alert = "late_alert" default_alert = "default_alert" general = "general" class Notification(Base, TimestampMixin): __tablename__ = "notifications" id = Column(BigInteger, primary_key=True, index=True) loan_id = Column(BigInteger, ForeignKey("loans.id"), nullable=False) type = Column(Enum(NotificationType), default=NotificationType.general) message = Column(String(255), nullable=False) sent_to = Column(String(150)) # Bisa no HP atau Email sent_at = Column(DateTime, nullable=True) status = Column(String(20), default="pending") # pending, sent, failed

from pydantic import BaseModel from datetime import date from decimal import Decimal class PaymentResponse(BaseModel): id: int loan_id: int schedule_date: date amount_due: Decimal amount_paid: Decimal status: str class Config: from_attributes = True class PayRequest(BaseModel): amount: Decimal

from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from typing import List from db import get_db from app.models.loan import Payment from app.schemas.payment import PaymentResponse, PayRequest from app.services.payment_service import pay_installment, check_and_flag_late_payments router = APIRouter() @router.get("/{loan_id}", response_model=List[PaymentResponse]) def get_loan_schedule(loan_id: int, db: Session = Depends(get_db)): """Melihat jadwal pembayaran suatu pembiayaan.""" payments = db.query(Payment).filter(Payment.loan_id == loan_id).order_by(Payment.schedule_date).all() return payments @router.post("/{payment_id}/pay", response_model=PaymentResponse) def make_payment( payment_id: int, req: PayRequest, db: Session = Depends(get_db) ): """Membayar angsuran tertentu.""" try: payment = pay_installment(db, payment_id, req.amount) return payment except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) @router.post("/system/check-late") def run_daily_checker(db: Session = Depends(get_db)): """ Endpoint Manual untuk Cron Job. Biasanya dipanggil otomatis setiap hari jam 00:01. """ count = check_and_flag_late_payments(db) return {"message": f"Checked and flagged {count} late payments."}

from fastapi import APIRouter, Depends from sqlalchemy.orm import Session from db import get_db from app.services.payment_service import get_portfolio_summary router = APIRouter() @router.get("/dashboard") def dashboard_stats(db: Session = Depends(get_db)): """ Endpoint Dashboard Utama. Mengembalikan ringkasan portofolio dan risiko. """ stats = get_portfolio_summary(db) return stats @router.get("/early-warning") def early_warning_list(db: Session = Depends(get_db)): """ Early Warning System (EWS). Menampilkan daftar pinjaman berisiko tinggi (High Risk Score) DAN memiliki pembayaran yang terlambat. """ from app.models.loan import Loan, LoanStatus from sqlalchemy import or_, and_ # Logika EWS: Status High Risk OR (Medium Risk AND Late Payment exists) high_risk_loans = db.query(Loan).filter( Loan.segment == "high", Loan.status == LoanStatus.active ).all() # Konversi ke JSON sederhana warnings = [] for loan in high_risk_loans: warnings.append({ "loan_id": loan.id, "customer_name": loan.customer.name if loan.customer else "Unknown", "segment": loan.segment, "default_probability": float(loan.default_probability) if loan.default_probability else 0, "reason": "High ML Score" }) return { "alert_count": len(warnings), "alerts": warnings }

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 from app.api.payments import router as pay_router # NEW from app.api.monitoring import router as mon_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)"]) app.include_router(pay_router, prefix="/api/v1/payments", tags=["Payments & Installments"]) # NEW app.include_router(mon_router, prefix="/api/v1/monitoring", tags=["Monitoring Dashboard"]) # 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("✅ System Ready.") except Exception as e: print(f"❌ Error: {e}") @app.get("/") def root(): return {"status": "BPRS API Running", "session": "6 - Monitoring Active"}