DigiFi/digifi-BankToProductCore
1
1
Fork 1
Code Issues Pull Requests 2 Releases Activity

[add]: Loan table extention

Browse Source
This commit is contained in:
VivianDee
2025-04-24 18:29:38 +01:00
parent 7d691db7a5
commit c895cc36e0
5 changed files with 165 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/services/base_service.py
+72
View File
@@ -60,3 +60,75 @@ class BaseService:
def async_send_to_kafka(cls, loan_data, request_id, topic):
KafkaIntegration.send_loan_request(loan_data = loan_data, request_id = request_id, topic = topic)
KafkaIntegration.flush()
@classmethod
def calculate_charges(cls, offer, amount):
"""
Calculates and returns the charges for the given offer and amount.
Args:
offer (Offer): The offer object that contains the charges.
amount (float): The requested loan amount.
Returns:
dict: A dictionary containing the calculated charges.
"""
if not offer or not offer.charges:
logger.error(f"No charges found for offer ID {offer.id}")
return {"error": "No charges found for the offer"}
loan_charges = offer.charges
interest = cls.get_charge_detail(loan_charges, "INTEREST", amount)
management = cls.get_charge_detail(loan_charges, "MGTFEE", amount)
insurance = cls.get_charge_detail(loan_charges, "INSURANCE", amount)
vat = cls.get_charge_detail(loan_charges, "VAT", amount)
# Up-front payment: (principal + only those fees due immediately i.e due_days == 0)
upfront_payment = amount + sum(
amount * charge.percent / 100
for charge in loan_charges
if charge.due == 0
)
# Total amount: (principal + all fees)
total_amount = amount + sum(
amount * charge.percent / 100
for charge in loan_charges
)
# Calculate the installment amount
tenor = offer.tenor
installment_amount = total_amount / tenor
return {
"interest": interest,
"management": management,
"insurance": insurance,
"vat": vat,
"upfront_payment": upfront_payment,
"total_amount": total_amount,
"installment_amount": installment_amount
}
@classmethod
def get_charge_detail(cls, charges, code, amount):
"""
Get details for a specific charge code from a list of loan charges.
Returns default values if not found.
"""
for charge in charges:
if charge.code == code:
return {
"rate": charge.percent,
"fee": amount * charge.percent / 100,
"due_days": charge.due,
}
return {"rate": 0, "fee": 0, "due_days": 0}
app/api/services/provide_loan.py
+15 -2
View File
@@ -36,6 +36,8 @@ class ProvideLoanService(BaseService):
collection_type = validated_data.get('collectionType')
transaction_id = validated_data.get('transactionId')
offer_id = validated_data.get('offerId')
amount = validated_data.get("requestedAmount")
product_id = validated_data.get("productId")
customer = Customer.is_valid_customer(customer_id)
@@ -48,8 +50,7 @@ class ProvideLoanService(BaseService):
return jsonify({
"message": "Invalid Offer."
}), 400
# Log Transaction
transaction = ProvideLoanService.log_transaction(validated_data=validated_data)
@@ -58,6 +59,15 @@ class ProvideLoanService(BaseService):
return jsonify({
"message": "Failed to log transaction."
}), 400
db.session.flush()
charges = ProvideLoanService.calculate_charges(offer, amount)
upfront_fee = charges["upfront_payment"]
repayment_amount = charges["total_amount"]
installment_amount = charges["installment_amount"]
# Save the loan details
@@ -69,6 +79,9 @@ class ProvideLoanService(BaseService):
collection_type = collection_type,
transaction_id = validated_data.get('transactionId'),
initial_loan_amount = validated_data.get('requestedAmount'),
upfront_fee = upfront_fee,
repayment_amount = repayment_amount,
installment_amount = installment_amount,
status= LoanStatus.ACTIVE
)
app/api/services/select_offer.py
+9 -54
View File
@@ -51,38 +51,16 @@ class SelectOfferService(BaseService):
# Get the offer by product ID
offer = Offer.get_offer_by_product_id(product_id)
if not offer:
logger.error(f"Offer with product ID {product_id} not found")
return jsonify({"message": "Offer not found"}), 404
# Get the loan charges for the offer
loan_charges = offer.charges
if not loan_charges:
logger.error(f"No charges found for offer ID {offer.id}")
return jsonify({"message": "No charges found for the offer"}), 404
logger.error(f"{loan_charges}")
db.session.flush()
interest = SelectOfferService.get_charge_detail(loan_charges, "INTEREST", amount)
management = SelectOfferService.get_charge_detail(loan_charges, "MGTFEE", amount)
insurance = SelectOfferService.get_charge_detail(loan_charges, "INSURANCE", amount)
vat = SelectOfferService.get_charge_detail(loan_charges, "VAT", amount)
# Up-front payment: (principal + only those fees due immediately i.e due_days == 0)
upfront_payment = amount + sum(
amount * charge.percent / 100
for charge in loan_charges
if charge.due == 0
)
# Total amount (principal + all fees)
total_amount = amount + sum(
amount * charge.percent / 100
for charge in loan_charges
)
charges = SelectOfferService.calculate_charges(offer, amount)
upfront_payment = charges["upfront_payment"]
total_amount = charges["total_amount"]
installment_amount = charges["installment_amount"]
interest = charges["interest"]
management = charges["management"]
insurance = charges["insurance"]
vat = charges["vat"]
# Calculate the repayment dates
@@ -97,8 +75,6 @@ class SelectOfferService(BaseService):
for i in range(months)
]
# Calculate the installment amount
installment_amount = total_amount / tenor
offers = [
@@ -150,25 +126,4 @@ class SelectOfferService(BaseService):
logger.error(f"An error occurred: {str(e)}", exc_info=True)
db.session.rollback()
return jsonify({"message": "Internal Server Error"}), 500
@staticmethod
def get_charge_detail(charges, code, amount):
"""
Get details for a specific charge code from a list of loan charges.
Returns default values if not found.
"""
for charge in charges:
if charge.code == code:
return {
"rate": charge.percent,
"fee": amount * charge.percent / 100,
"due_days": charge.due,
}
return {"rate": 0, "fee": 0, "due_days": 0}
app/models/loan.py
+20 -2
View File
@@ -25,6 +25,9 @@ class Loan(db.Model):
initial_loan_amount = db.Column(db.Float, nullable=False)
default_penalty_fee = db.Column(db.Float, default=0)
continuous_fee = db.Column(db.Float, default=0)
upfront_fee = db.Column(db.Float, nullable=True, default=0.0)
repayment_amount = db.Column(db.Float, nullable=True, default=0.0)
installment_amount = db.Column(db.Float, nullable=True, default=0.0)
status = db.Column(db.String(20), default='pending')
due_date = db.Column(db.DateTime, nullable=True)
created_at = db.Column(db.DateTime, default=datetime.now(timezone.utc))
@@ -45,8 +48,20 @@ class Loan(db.Model):
)
@classmethod
def create_loan(cls, customer_id, account_id, offer_id, product_id, initial_loan_amount, collection_type, transaction_id, status='pending'):
def create_loan(
cls,
customer_id,
account_id,
offer_id,
product_id,
initial_loan_amount,
collection_type,
transaction_id,
upfront_fee,
repayment_amount,
installment_amount,
status="pending",
):
# Check if customer exists
customer = Customer.is_valid_customer(customer_id)
if not customer:
@@ -64,6 +79,9 @@ class Loan(db.Model):
transaction_id = transaction_id,
initial_loan_amount = initial_loan_amount,
current_loan_amount = initial_loan_amount,
upfront_fee = upfront_fee,
repayment_amount = repayment_amount,
installment_amount = installment_amount,
due_date=now,
status = status
)
app/models/rac_checks.py
+49 -8
View File
@@ -1,12 +1,53 @@
from datetime import datetime, timezone
from app.extensions import db
from sqlalchemy.orm import relationship
from sqlalchemy.dialects.postgresql import UUID
from uuid import uuid4
from sqlalchemy.types import JSON
class RACCheck(Base):
__tablename__ = "rac_checks"
class RACCheck(db.Model):
__tablename__ = 'rac_checks'
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
transaction_id = Column(UUID, ForeignKey('transactions.id'), nullable=False)
customer_id = Column(String, nullable=False)
account_id = Column(String, nullable=False)
rac_response = Column(JSON, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
id = db.Column(db.String, primary_key=False)
transaction_id = db.Column(db.String(50), nullable=False)
customer_id = db.Column(db.String, nullable=False)
account_id = db.Column(db.String, nullable=False)
rac_response = db.Column(db.JSON, nullable=False)
created_at = db.Column(db.DateTime, default=datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=datetime.now(timezone.utc), onupdate=datetime.now(timezone.utc))
@classmethod
def get_all_rac_checks(cls):
"""
Return all RAC checks in dictionary format.
"""
rac_checks = cls.query.all()
if not rac_checks:
raise ValueError("No available RAC checks")
return rac_checks
@classmethod
def get_rac_check_by_id(cls, check_id):
"""
Return a RAC check by its ID.
"""
rac_check = cls.query.filter_by(id=check_id).first()
if not rac_check:
raise ValueError(f"RAC Check with ID {check_id} not found")
return rac_check
def to_dict(self):
return {
"id": str(self.id),
"transactionId": str(self.transaction_id),
"customerId": self.customer_id,
"accountId": self.account_id,
"racResponse": self.rac_response,
"createdAt": self.created_at.isoformat(),
"updatedAt": self.updated_at.isoformat() if self.updated_at else None
}
def __repr__(self):
return f'<RACCheck {self.id}>'