Fix UT and expose first loss/ by term

Author: yellowbean

absbox/client.py

@@ -53,6 +53,8 @@ class Endpoints(str, enum.Enum):
     """Run a single deal with multiple deal run scenarios endpoint"""
     RunByCombo = "runByCombo"
     """Run mulitple sensitivities"""
+    RunFirstLoss = "runByFirstLoss"
+    """Run first loss"""
     RunDate = "runDate"
     """Run Dates from a datepattern """
     Version = "version"
@@ -76,6 +78,8 @@ class RunReqType(str, enum.Enum):
     """ Single Pool With Multiple Assumptions """
     ComboSensitivity = "MultiComboReq"
     """ Multiple sensitivities """
+    FirstLoss = "FirstLossReq"
+    """ First Loss Run """
 
 
 class RunResp(int, enum.Enum):
@@ -260,9 +264,9 @@ def build_run_deal_req(self, run_type: str, deal, perfAssump=None, nonPerfAssump
 
         match Schema(str).validate(run_type):
             case "Single" | "S":
-                _nonPerfAssump = mkNonPerfAssumps({}, nonPerfAssump)
                 _deal = deal.json if hasattr(deal, "json") else deal
                 _perfAssump = earlyReturnNone(mkAssumpType, perfAssump)
+                _nonPerfAssump = mkNonPerfAssumps({}, nonPerfAssump)
                 r = mkTag((RunReqType.Single.value, [_deal, _perfAssump, _nonPerfAssump]))
             case "MultiScenarios" | "MS":
                 _nonPerfAssump = mkNonPerfAssumps({}, nonPerfAssump)
@@ -288,6 +292,11 @@ def build_run_deal_req(self, run_type: str, deal, perfAssump=None, nonPerfAssump
                 if mRunAssump == {}:
                     mRunAssump = {"Empty":{}}
                 r = mkTag((RunReqType.ComboSensitivity.value, [mDeal, mAssump, mRunAssump]))
+            case "FirstLoss" | "FL":
+                _deal = deal.json if hasattr(deal, "json") else deal
+                _perfAssump = mkAssumpType(perfAssump)
+                _nonPerfAssump = mkNonPerfAssumps({}, nonPerfAssump)
+                r = mkTag((RunReqType.FirstLoss.value, [_deal, _perfAssump, _nonPerfAssump]))
             case _:
                 raise RuntimeError(f"Failed to match run type:{run_type}")
         try:
@@ -449,7 +458,6 @@ def read_single(self, pool_resp) -> tuple:
         result = _read_cf(pool_flow['contents'][1], self.lang)
         return (result, pool_bals)
 
-
     def runPoolByScenarios(self, pool, poolAssump, rateAssump=None, read=True, debug=False) -> dict :
         """ run a pool with multiple scenario ,return result as map , with key same to pool assumption map
 
@@ -677,6 +685,43 @@ def runByCombo(self,
         else:
             return result
 
+    def runFirstLoss(self, deal, bName, poolAssump=None, runAssump=[], read=True, showWarning=True, debug=False) -> dict:
+        """run first loss with deal and pool assumptions
+
+        :param deal: a deal object
+        :type deal: Generic | SPV
+        :param poolAssump: pool performance assumption, a tuple for single run/ a dict for multi-scenario run, defaults to None
+        :type poolAssump: tuple, optional
+        :param runAssump: deal level assumption, defaults to []
+        :type runAssump: list, optional
+        :param read: flag to convert result to pandas dataframe, defaults to True
+        :type read: bool, optional
+        :param showWarning: flag to show warnings, defaults to True
+        :type showWarning: bool, optional
+        :param debug: return request text instead of sending out such request, defaults to False
+        :type debug: bool, optional
+        :return: result of run, a dict of dataframe if `read` is True.
+        :rtype: dict
+        """
+
+        if (poolAssump is None):
+            raise AbsboxError(f"❌{MsgColor.Error.value} poolAssump must be set for first loss run")
+
+        url = f"{self.url}/{Endpoints.RunFirstLoss.value}"
+
+        req = self.build_run_deal_req("FL", deal, poolAssump, runAssump)
+        if debug:
+            return req
+
+        result = self._send_req(req & lens.Json()['contents'].modify(lambda x:x+[bName]), url)
+
+        if result is None or 'error' in result or 'Left' in result:
+            leftVal = result.get("Left","")
+            raise AbsboxError(f"❌{MsgColor.Error.value}Failed to get response from run: {leftVal}")
+        # rawWarnMsg = map( lambda x:f"{MsgColor.Warning.value}{x['contents']}", filter_by_tags(result[RunResp.LogResp.value], enumVals(ValidationMsg)))
+        # if rawWarnMsg and showWarning:
+        #     console.print("Warning Message from server:\n"+"\n".join(list(rawWarnMsg)))
+        return result['Right']
 
     def runAsset(self, date, _assets, poolAssump=None, rateAssump=None
                  , pricing=None, read=True, debug=False) -> tuple:
@@ -716,8 +761,7 @@ def readResult(x):
         _rate = lmap(mkRateAssumption, rateAssump) if rateAssump else None
         _pricing = mkLiqMethod(pricing) if pricing else None
         assets = lmap(mkAssetUnion, _assets) 
-        req = json.dumps([date, assets, _assumptions, _rate, _pricing]
-                         , ensure_ascii=False)
+        req = json.dumps([date, assets, _assumptions, _rate, _pricing], ensure_ascii=False)
         if debug:
             return req
 

absbox/local/component.py

@@ -6,7 +6,7 @@
 
 import sys
 
-from absbox.validation import vDict, vList, vStr, vNum, vInt, vDate, vFloat, vBool, vTuple
+from absbox.validation import vDict, vList, vStr, vNum, vInt, vDate, vFloat, vBool, vTuple, vListOfList
 from schema import Or
 from enum import Enum
 import itertools
@@ -1726,6 +1726,8 @@ def mkAssumpDefault(x):
             return mkTag(("DefaultAtEndByRate", [vNum(r1), vNum(r2)]))
         case {"StressByCurve": [curve, assump]}:
             return mkTag(("DefaultStressByTs", [ mkRateTs(curve), mkAssumpDefault(assump)]))
+        case {"byTerm": rs}:
+            return mkTag(("DefaultByTerm", vListOfList(rs, numVal)))
         case _ :
             raise RuntimeError(f"failed to match {x}")
 
@@ -1743,6 +1745,8 @@ def mkAssumpPrepay(x):
             return mkTag(("PrepayStressByTs", [ mkRateTs(curve), mkAssumpPrepay(assump)]))
         case {"PSA": r}:
             return mkTag(("PrepaymentPSA", vNum(r)))
+        case {"byTerm": rs}:
+            return mkTag(("PrepaymentByTerm", vListOfList(rs, numVal)))
         case _ :
             raise RuntimeError(f"failed to match {x}")