from decimal import Decimal, getcontext from datetime import datetime, date from calendar import isleap from typing import Callable, Dict import uuid getcontext().prec = 28 # { # "computation_type": "INTEREST_ACCRUAL", # "input": { # "start_date": "2026-01-01", # "end_date": "2026-06-30", # "notional": "1000000", # "rate": { # "value": "0.08", # "type": "ANNUAL" # }, # "day_count_convention": "ACT/360", # "compounding_method": "SIMPLE" # } # } # result = FinancialComputationService.execute(payload) # interest = Decimal(result["result"]["interest_amount"]) # interest = currency.round(interest) # ============================================================ # VERSIONING # ============================================================ ENGINE_VERSION = "1.0.0" # ============================================================ # REGISTRY (PLUGIN SYSTEM) # ============================================================ DAY_COUNT_REGISTRY: Dict[str, Callable] = {} COMPOUNDING_REGISTRY: Dict[str, Callable] = {} def register_day_count(name: str): def decorator(func): DAY_COUNT_REGISTRY[name] = func return func return decorator def register_compounding(name: str): def decorator(func): COMPOUNDING_REGISTRY[name] = func return func return decorator # ============================================================ # DOMAIN – DAY COUNT CONVENTIONS # ============================================================ @register_day_count("ACT/360") def act_360(start: date, end: date) -> Decimal: return Decimal((end - start).days) / Decimal(360) @register_day_count("ACT/365F") def act_365f(start: date, end: date) -> Decimal: return Decimal((end - start).days) / Decimal(365) @register_day_count("ACT/ACT_ISDA") def act_act_isda(start: date, end: date) -> Decimal: total = Decimal("0") current = start while current < end: year_end = date(current.year, 12, 31) next_year = date(current.year + 1, 1, 1) period_end = min(next_year, end) days = (period_end - current).days year_days = 366 if isleap(current.year) else 365 total += Decimal(days) / Decimal(year_days) current = period_end return total @register_day_count("30E/360") def thirty_e_360(start: date, end: date) -> Decimal: d1 = min(start.day, 30) d2 = min(end.day, 30) days = ( (end.year - start.year) * 360 + (end.month - start.month) * 30 + (d2 - d1) ) return Decimal(days) / Decimal(360) # ============================================================ # DOMAIN – COMPOUNDING STRATEGIES # ============================================================ @register_compounding("SIMPLE") def simple(rate: Decimal, yf: Decimal) -> Decimal: return rate * yf @register_compounding("ANNUAL") def annual(rate: Decimal, yf: Decimal) -> Decimal: return (Decimal(1) + rate) ** yf - Decimal(1) @register_compounding("MONTHLY") def monthly(rate: Decimal, yf: Decimal) -> Decimal: monthly_rate = rate / Decimal(12) months = yf * Decimal(12) return (Decimal(1) + monthly_rate) ** months - Decimal(1) @register_compounding("CONTINUOUS") def continuous(rate: Decimal, yf: Decimal) -> Decimal: from math import exp return Decimal(exp(float(rate * yf))) - Decimal(1) # ============================================================ # DOMAIN – INTEREST COMPUTATION OBJECT # ============================================================ class InterestComputation: def __init__( self, start_date: date, end_date: date, notional: Decimal, rate_value: Decimal, rate_type: str, day_count: str, compounding: str, ): self.start_date = start_date self.end_date = end_date self.notional = notional self.rate_value = rate_value self.rate_type = rate_type self.day_count = day_count self.compounding = compounding def compute(self): if self.end_date <= self.start_date: raise ValueError("end_date must be after start_date") if self.day_count not in DAY_COUNT_REGISTRY: raise ValueError("Unsupported day count convention") if self.compounding not in COMPOUNDING_REGISTRY: raise ValueError("Unsupported compounding method") yf = DAY_COUNT_REGISTRY[self.day_count]( self.start_date, self.end_date ) # Normalize rate to annual if self.rate_type == "MONTHLY": annual_rate = self.rate_value * Decimal(12) else: annual_rate = self.rate_value factor = COMPOUNDING_REGISTRY[self.compounding]( annual_rate, yf ) interest_amount = self.notional * factor return { "year_fraction": yf, "interest_factor": factor, "interest_amount": interest_amount, } # ============================================================ # APPLICATION LAYER – JSON SERVICE (Camunda Ready) # ============================================================ class FinancialComputationService: @staticmethod def execute(payload: dict) -> dict: """ Stateless JSON entrypoint. Compatible Camunda / REST / Tryton bridge. """ try: request_id = str(uuid.uuid4()) input_data = payload["input"] start_date = datetime.strptime( input_data["start_date"], "%Y-%m-%d" ).date() end_date = datetime.strptime( input_data["end_date"], "%Y-%m-%d" ).date() notional = Decimal(input_data["notional"]) rate_value = Decimal(input_data["rate"]["value"]) rate_type = input_data["rate"]["type"].upper() day_count = input_data["day_count_convention"] compounding = input_data["compounding_method"] computation = InterestComputation( start_date=start_date, end_date=end_date, notional=notional, rate_value=rate_value, rate_type=rate_type, day_count=day_count, compounding=compounding, ) result = computation.compute() return { "metadata": { "engine_version": ENGINE_VERSION, "request_id": request_id, "timestamp": datetime.utcnow().isoformat() + "Z" }, "result": { "year_fraction": str(result["year_fraction"]), "interest_factor": str(result["interest_factor"]), "interest_amount": str(result["interest_amount"]), }, "explainability": { "formula": "Interest = Notional × Factor", "factor_definition": f"{compounding} compounding applied to annualized rate", "day_count_used": day_count }, "status": "SUCCESS" } except Exception as e: return { "status": "ERROR", "message": str(e), "engine_version": ENGINE_VERSION }