expose AssetLevelCashflow

yellowbean · yellowbean · commit dca5b8bbe367 · 2025-06-30T23:33:03.000+08:00
diff --git a/absbox/client.py b/absbox/client.py
@@ -5,6 +5,7 @@
 import requests
 from rich.console import Console
 import toolz as tz
+from functools import partial
 from lenses import lens
 
 from requests.exceptions import ConnectionError, ReadTimeout
@@ -193,7 +194,7 @@ def __post_init__(self) -> None:
         self.session = requests.Session()
         console.print(f"✅Connected, local lib:{'.'.join(self.version)}, server:{'.'.join(engine_version)}")
 
-    def build_run_deal_req(self, run_type, deal, perfAssump=None, nonPerfAssump=[]) -> str:
+    def build_run_deal_req(self, run_type, deal, perfAssump=None, nonPerfAssump=[], rtn = []) -> str:
         """build run deal requests: (single run, multi-scenario run, multi-struct run) 2
 
         :meta private:
@@ -217,22 +218,22 @@ def build_run_deal_req(self, run_type, deal, perfAssump=None, 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]))
+                r = mkTag((RunReqType.Single.value, [rtn, _deal, _perfAssump, _nonPerfAssump]))
             case "MultiScenarios" | "MS":
                 _nonPerfAssump = mkNonPerfAssumps({}, nonPerfAssump)
                 _deal = deal.json if hasattr(deal, "json") else deal
                 mAssump = mapValsBy(perfAssump, mkAssumpType)
-                r = mkTag((RunReqType.MultiScenarios.value, [_deal, mAssump, _nonPerfAssump]))
+                r = mkTag((RunReqType.MultiScenarios.value, [rtn, _deal, mAssump, _nonPerfAssump]))
             case "MultiStructs" | "MD" :
                 _nonPerfAssump = mkNonPerfAssumps({}, nonPerfAssump)
                 mDeal = {k: v.json if hasattr(v, "json") else v for k, v in deal.items()}
                 _perfAssump = mkAssumpType(perfAssump)
-                r = mkTag((RunReqType.MultiStructs.value, [mDeal, _perfAssump, _nonPerfAssump]))
+                r = mkTag((RunReqType.MultiStructs.value, [rtn, mDeal, _perfAssump, _nonPerfAssump]))
             case "MultiRunScenarios" | "MRS" if isinstance(nonPerfAssump,dict):
                 _deal = deal.json if hasattr(deal, "json") else deal
                 _perfAssump = earlyReturnNone(mkAssumpType, perfAssump)
                 mRunAssump = mapValsBy(nonPerfAssump, lambda x: mkNonPerfAssumps({}, x))
-                r = mkTag((RunReqType.MultiRunScenarios.value, [_deal, _perfAssump, mRunAssump]))
+                r = mkTag((RunReqType.MultiRunScenarios.value, [rtn, _deal, _perfAssump, mRunAssump]))
             case "ComboSensitivity" | "CS" if isinstance(nonPerfAssump,dict) and isinstance(perfAssump,dict) and isinstance(deal,dict):
                 mDeal = {k: v.json if hasattr(v, "json") else v for k, v in deal.items()}
                 mAssump = mapValsBy(perfAssump, mkAssumpType)
@@ -241,7 +242,7 @@ def build_run_deal_req(self, run_type, deal, perfAssump=None, nonPerfAssump=[])
                 mRunAssump = mapValsBy(nonPerfAssump, lambda x: mkNonPerfAssumps({}, x))
                 if mRunAssump == {}:
                     mRunAssump = {"Empty":{}}
-                r = mkTag((RunReqType.ComboSensitivity.value, [mDeal, mAssump, mRunAssump]))
+                r = mkTag((RunReqType.ComboSensitivity.value, [rtn, mDeal, mAssump, mRunAssump]))
             case ("RootFinder", tweak, stop):
                 _deal = deal.json if hasattr(deal, "json") else deal
                 _perfAssump = earlyReturnNone(mkAssumpType, perfAssump)
@@ -269,7 +270,7 @@ def build_run_deal_req(self, run_type, deal, perfAssump=None, nonPerfAssump=[])
 
         
 
-    def build_pool_req(self, pool, poolAssump, rateAssumps, isMultiScenario=False) -> str:
+    def build_pool_req(self, pool, poolAssump, rateAssumps, isMultiScenario=False, breakdown = False) -> str:
         """build pool run request: (single run, multi-scenario run)
 
         :meta private:
@@ -298,9 +299,9 @@ def buildPoolType(p) -> dict:
                 return mkTag((assetTag, mkPoolType(assetDate, _p, False)))
 
         if not isMultiScenario:
-            r = mkTag((RunReqType.SinglePool.value, [buildPoolType(pool), mkAssumpType(poolAssump), _rateAssump]))
+            r = mkTag((RunReqType.SinglePool.value, [breakdown, buildPoolType(pool), mkAssumpType(poolAssump), _rateAssump]))
         else:
-            r = mkTag((RunReqType.MultiPoolScenarios.value, [buildPoolType(pool), mapValsBy(poolAssump, mkAssumpType), _rateAssump]))
+            r = mkTag((RunReqType.MultiPoolScenarios.value, [breakdown, buildPoolType(pool), mapValsBy(poolAssump, mkAssumpType), _rateAssump]))
 
         return json.dumps(r, ensure_ascii=False)
 
@@ -320,6 +321,7 @@ def run(self, deal,
             runAssump=[],
             read=True,
             showWarning=True,
+            rtn = [],
             debug=False) -> dict:
         """ run deal with pool and deal run assumptions
 
@@ -348,7 +350,7 @@ def run(self, deal,
         url = f"{self.url}/{Endpoints.RunDeal.value}"
 
         # construct request
-        req = self.build_run_deal_req("Single", deal, poolAssump, runAssump)
+        req = self.build_run_deal_req("Single", deal, poolAssump, runAssump, rtn=rtn)
         if debug:
             return req
 
@@ -418,19 +420,28 @@ def runByScenarios(self, deal,
         else:
             return result
 
-    def read_single(self, pool_resp) -> tuple:
+    def read_single(self, breakdown, pool_resp) -> tuple:
         """ read pool run response from engine and convert to dataframe
 
         :param pool_resp: (pool raw cashflow, pool statistics)
         :type pool_resp: tuple
         :return: (pool Cashflow in dataFrame, pool statistics)
         :rtype: tuple
         """
-        (pool_flow, pool_bals) = pool_resp
-        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 :
+        ((pool_flow, pool_bals), pool_breakdown_flow) = pool_resp
+        result = _read_cf(pool_flow['contents'][1], self.lang)
+        if not breakdown:
+            return {"flow":result, "stat":pool_bals}
+        else:
+            assert pool_breakdown_flow is not None, "Breakdown flow is None"
+            assert len(pool_breakdown_flow)>0, "Breakdown flow is empty"
+            return {"flow":result, "stat":pool_bals
+                    ,"breakdown": [ {"flow": _read_cf(_[0]['contents'][1], self.lang), "stat":_[1]}
+                                    for _ in pool_breakdown_flow  ]
+                    }
+
+    def runPoolByScenarios(self, pool, poolAssump, rateAssump=None, read=True, breakdown = False,debug=False) -> dict :
         """ run a pool with multiple scenario ,return result as map , with key same to pool assumption map
 
         :param pool: pool map
@@ -462,10 +473,10 @@ def runPoolByScenarios(self, pool, poolAssump, rateAssump=None, read=True, debug
         result = tz.valmap(lambda x:x['Right'] ,result)
 
         if read:
-            return result & lens.Values().Values().modify(self.read_single)
+            return result & lens.Values().Values().modify(partial(self.read_single, breakdown))
         return result
 
-    def runPool(self, pool, poolAssump=None, rateAssump=None, read=True, debug=False, **kwargs) -> tuple:
+    def runPool(self, pool, poolAssump=None, rateAssump=None, read=True, debug=False, breakdown = False, **kwargs) -> tuple:
         """perform pool run with pool and rate assumptions
 
         :param pool: a pool object
@@ -487,7 +498,7 @@ def runPool(self, pool, poolAssump=None, rateAssump=None, read=True, debug=False
 
         url = f"{self.url}/{Endpoints.RunPool.value}"
 
-        req = self.build_pool_req(pool, poolAssump, rateAssump, isMultiScenario=False)
+        req = self.build_pool_req(pool, poolAssump, rateAssump, isMultiScenario=False, breakdown = breakdown)
 
         if debug:
             return req
@@ -497,10 +508,11 @@ def runPool(self, pool, poolAssump=None, rateAssump=None, read=True, debug=False
         if result is None or 'error' in result or 'Left' in result:
             leftVal = result.get("Left","")
             raise AbsboxError(f"❌ Failed to get response from run: {leftVal}")
+
         result = result['Right']
-        
+
         if read:
-            return result & lens.Values().modify(self.read_single)
+            return result & lens.Values().modify(partial(self.read_single, breakdown))
         else:
             return result
 
diff --git a/absbox/local/component.py b/absbox/local/component.py
@@ -304,7 +304,7 @@ def mkDs(x):
                 if pNames:
                     return mkTag(("PoolScheduleCfPv", [mkLiqMethod(pricingMethod), lmap(mkPid,pNames)]))
                 return mkTag(("PoolScheduleCfPv", [mkLiqMethod(pricingMethod), None]))
-            case ("资产池加权利差", pNames) | ("PoolWaSpread", pNames):
+            case ("资产池加权利差", pNames) | ("poolWaSpread", pNames):
                 if pNames:
                     return mkTag(("PoolWaSpread", lmap(mkPid,pNames)))
                 return mkTag(("PoolWaSpread", None))
@@ -1967,17 +1967,17 @@ def mkFieldRule(z):
 def mkAssumpType(x):
     ''' make assumps either on pool level or asset level '''
     match x:
-        case ("Pool", p, d, f):
+        case ("Pool", p, d, f) | ("pool", p, d, f):
             return mkTag(("PoolLevel",mkPDF(p, d, f)))
-        case ("ByIndex", *ps):
+        case ("ByIndex", *ps) | ("byIndex", *ps):
             return mkTag(("ByIndex",[ [idx, mkPDF(a,b,c)] for (idx,(a,b,c)) in ps ]))
-        case ("ByObligor",*rules):
+        case ("ByObligor",*rules) | ("byObligor",*rules):
             return mkTag(("ByObligor",[mkObligorStrategy(r) for r in rules]))
-        case ("ByName", assumpMap):
+        case ("ByName", assumpMap) | ("byName", assumpMap):
             return mkTag(("ByName",{f"PoolName:{k}":mkPDF(*v) for k,v in assumpMap.items()}))
-        case ("ByPoolId", assumpMap):
+        case ("ByPoolId", assumpMap) | ("byPoolId", assumpMap):
             return mkTag(("ByPoolId",{f"PoolName:{k}": mkAssumpType(v) for k,v in assumpMap.items()}))
-        case ("ByDealName", assumpMap):
+        case ("ByDealName", assumpMap) | ("byDealName", assumpMap):
             return mkTag(("ByDealName",{k:(mkAssumpType(perfAssump),mkNonPerfAssumps({},nonPerfAssump)) 
                                         for k,(perfAssump,nonPerfAssump) in assumpMap.items()}))
         case None:
@@ -2020,18 +2020,6 @@ def mkRevolvingPool(x):
 
 
 def mkPoolType(assetDate, x, mixedFlag) -> dict:
-    # try:
-    #     if 'assets' in x or "清单" in x or "归集表" in x:
-    #         return mkTag(("MultiPool" ,{"PoolConsol":mkPoolComp(vDate(assetDate), x, False)}))
-    #     elif 'deals' in x and isinstance(x['deals'],dict):
-    #         return mkTag(("ResecDeal",{f"{dealObj.json['contents']['name']}:{bn}:{sd}:{str(pct)}": \
-    #                                     {"deal":dealObj.json['contents'],"future":None,"futureScheduleCf":None,"issuanceStat":None}\
-    #                                       for ((bn,pct,sd),dealObj) in x['deals'].items()} ))
-    #     else:
-    #         return mkTag(("MultiPool" ,{f"PoolName:{k}":mkPoolComp(vDate(assetDate),v,mixedFlag) for (k,v) in x.items()}))
-    # except Exception as e:
-    #     print("Error in mk pool type",e)
-
     match x:
         case {"assets": y} | {"清单": y} | {"归集表": y}: 
             return mkTag(("MultiPool" ,{"PoolConsol":mkPoolComp(vDate(assetDate), x, False)}))
@@ -2055,7 +2043,8 @@ def mkPoolComp(asOfDate, x, mixFlag) -> dict:
         , "issuanceStat": tz.pipe(getValWithKs(x, ["issuanceStat", "统计", "发行", "Issuance"],defaultReturn={})
                                 , lambda y: updateKs(y, validCutoffFields)  
                                 )
-        , "futureCf":mkCf(getValWithKs(x, ['cashflow', '现金流归集表', '归集表'], []))
+        , "futureCf":[mkCf(getValWithKs(x, ['cashflow', '现金流归集表', '归集表'], [])),None]
+        , "futureScheduleCf" : [mkCf([]), None]
         , "extendPeriods":mkDatePattern(getValWithKs(x, ['extendBy'], "MonthEnd"))}
     return r
 
@@ -2152,7 +2141,7 @@ def mkLedger(n: str, x: dict=None):
 def mkCf(x:list):
     """ Make project cashflow ( Mortgage Only ) """
     if len(x) == 0:
-        return None
+        return mkTag(("CashFlowFrame", [[0,"1900-01-01",None],[]]))
     else:
         cfs = [mkTag(("MortgageFlow", _x+[0.0]*5+[None,None,None])) for _x in x]
         return mkTag(("CashFlowFrame", [[0,"1900-01-01",None],cfs]))
diff --git a/absbox/local/generic.py b/absbox/local/generic.py
@@ -145,7 +145,8 @@ def read(resp):
                 output['pool']['flow'] = readPoolCf(deal_content['pool']['contents']['futureCf']['contents'])
         elif deal_content['pool']['tag']=='MultiPool':
             poolMap = deal_content['pool']['contents']
-            output['pool']['flow'] = tz.valmap(lambda v: readPoolCf(v['futureCf']['contents']) if (not v['futureCf'] is None) else pd.DataFrame(), poolMap)
+            output['pool']['flow'] = tz.valmap(lambda v: readPoolCf(v['futureCf'][0]['contents']) if (not v['futureCf'] is None) else pd.DataFrame(), poolMap)
+            output['pool']['breakdown'] = tz.valmap(lambda v: list(tz.map(readPoolCf, v['futureCf'][1] & lens.Each()['contents'].collect() )) if (not v['futureCf'][1] is None) else [], poolMap)
         elif deal_content['pool']['tag']=='ResecDeal':
             poolMap = deal_content['pool']['contents']
             output['pool']['flow'] = {tz.get([1,2,4],k.split(":")): readPoolCf(v['futureCf']['contents']) for (k,v) in poolMap.items() }
diff --git a/absbox/local/util.py b/absbox/local/util.py
@@ -323,6 +323,7 @@ def _read_cf(x, lang):
     ''' read cashflow from a list , and set index to date'''
     if x == []:
         return []
+    assert isinstance(x, list), f"Input is not a list but {type(x)}"
     flow_header, idx, expandFlag = guess_pool_flow_header(x[0], lang)
     result = None
     try:
@@ -331,7 +332,7 @@ def _read_cf(x, lang):
         else:
             result = pd.DataFrame([_['contents'] for _ in x], columns=flow_header)
     except ValueError as e:
-        print(e)
+        logging.error(f"{e}")
         logging.error(f"Failed to match header:{flow_header} with {result}")
         return False
     result.set_index(idx, inplace=True)
