expose key name prodDealsBy

absbox/deal.py

@@ -81,7 +81,7 @@ def mkDealsBy(d, m: dict)->dict:
     return {k: dataclasses.replace(d, **v) for k, v in m.items()} 
 
 
-def setDealsBy(d, *receipes: list, init=None):
+def setDealsBy(d, *receipes: list, init=None, **kwargs):
     "input a deal object, a list of tweaks( path, values) ,return an updated deal"
     if init:
         receipes = [(init & _[0], _[1]) for _ in receipes]
@@ -93,4 +93,21 @@ def setDealsBy(d, *receipes: list, init=None):
 def prodDealsBy(d, *receipes, **kwargs) -> dict:
     inflated = [[(p, _) for _ in vs] for (p, vs) in receipes]
     permus = product(*inflated)
-    return {str(v): setDealsBy(d, *v, **kwargs) for v in permus}
+    # print(list(permus)[0])
+    if kwargs.get('guessKey', False) == True:
+        return {strFromPath(v): setDealsBy(d, *v, **kwargs) for v in permus}
+    return {v: setDealsBy(d, *v, **kwargs) for v in permus}
+
+
+def setAssumpsBy(a, *receipes: list, init=None):
+    if init:
+        receipes = [(init & _[0], _[1]) for _ in receipes]
+    for (p, v) in receipes:
+        a &= p.set(v)
+    return a
+
+
+def prodAssumpsBy(a, *receipes, **kwargs):
+    inflated = [[(p, _) for _ in vs] for (p, vs) in receipes]
+    permus = product(*inflated)
+    return {str(v): setDealsBy(a, *v, **kwargs) for v in permus}

absbox/local/base.py

@@ -139,3 +139,6 @@ class DC(Enum):  # TODO need to check with HS code
     DC_30_360_ISDA = "DC_30_360_ISDA"
     DC_30_360_German = "DC_30_360_German"
     DC_30_360_US = "DC_30_360_US"
+
+#valid inspection tags
+inspectTags = ["InspectBal", "InspectBool", "InspectRate", "InspectInt"]

absbox/local/component.py

@@ -1600,7 +1600,7 @@ def mkCf(x:list):
         return None
     else:
         cfs = [mkTag(("MortgageFlow", _x+[0.0]*5+[None,None,None])) for _x in x]
-        return mkTag(("CashFlowFrame",cfs))
+        return mkTag(("CashFlowFrame", cfs))
 
 def mkPid(x):
     match x:
@@ -1670,6 +1670,7 @@ def readPricingResult(x, locale) -> dict:
 
     return pd.DataFrame.from_dict({k: v['contents'] for k, v in x.items()}, orient='index', columns=h[locale]).sort_index()
 
+
 def readPoolCf(x, lang='english'):
     r = None
     _pool_cf_header,_,expandFlag = guess_pool_flow_header(x[0], lang)
@@ -1683,6 +1684,7 @@ def readPoolCf(x, lang='english'):
     r.index.rename(pool_idx, inplace=True)    
     return r
 
+
 def readRunSummary(x, locale) -> dict:
     def filter_by_tags(xs, tags):
         tags_set = set(tags)
@@ -1715,13 +1717,13 @@ def filter_by_tags(xs, tags):
     r['status'] = pd.DataFrame(data=status_change_logs, columns=dealStatusLog[locale])
 
     # inspection variables
-    def uplift_ds(df):
+    def uplift_ds(df:pd.DataFrame) -> pd.DataFrame:
         ds_name = readTagStr(df['DealStats'].iloc[0])
         df.drop(columns=["DealStats"],inplace=True)
         df.rename(columns={"Value":ds_name},inplace=True)
         df.set_index("Date",inplace=True)
         return df
-    inspect_vars = filter_by_tags(x, ["InspectBal", "InspectBool", "InspectRate","InspectInt"])
+    inspect_vars = filter_by_tags(x, inspectTags)
     if inspect_vars:
         inspect_df = pd.DataFrame(data = [ (c['contents'][0],str(c['contents'][1]),c['contents'][2]) for c in inspect_vars ]
                                 ,columns = ["Date","DealStats","Value"])

absbox/local/generic.py

@@ -119,3 +119,7 @@ def read(resp):
         output['result'] = readRunSummary(resp[2], 'en')
         output['_deal'] = resp[0]
         return output
+
+    def __str__(self):
+        ''' Return a simplified string representation of deal object '''
+        return f"{self.name}"

absbox/local/util.py

@@ -1,17 +1,15 @@
 import pandas as pd
-import functools,json,copy
-import logging
-import itertools,re
-from enum import Enum
 import numpy as np
+import functools,json,copy,logging,re,itertools
+from enum import Enum
 from functools import reduce
 from absbox.local.base import *
 from pyspecter import query, S
 from datetime import datetime
-
 import rich
 from rich.console import Console
 from rich.json import JSON
+from lenses import lens, ui, optics
 
 console = Console()
 
@@ -340,10 +338,12 @@ def flow_by_scenario(rs, flowpath, node="col", rtn_df=True, ax=1, rnd=2):
 
 
 def positionFlow(x, m: dict, facePerPaper=100):
+    ''' Get a position bond cashflow from a run result '''
     _, _bflow = list(x['bonds'].items())[0]
     bflowHeader = _bflow.columns.to_list()
 
-    def calcBondFlow(_bflow, factor):
+    def calcBondFlow(_bflow: pd.DataFrame, factor):
+        ''' input: (bond cashflow : dataframe, factor: float) '''
         bflow = copy.deepcopy(_bflow)
         bflow[bflowHeader[0]] *= factor
         bflow[bflowHeader[1]] *= factor
@@ -353,9 +353,9 @@ def calcBondFlow(_bflow, factor):
 
     assert isinstance(m, dict), "Position info must be a map/dict"
 
-    bOrignBal = {k: x['_deal']['contents']['bonds'][k]['bndOriginInfo']['originBalance'] for k, v in m.items()}
-    bPapersPerBond = {k: v/facePerPaper for k, v in bOrignBal.items()}
-    bflowFactor = {k: (v/bPapersPerBond[k]) for k, v in m.items()}
+    bOrignBal = {bondName: x['_deal']['contents']['bonds'][bondName]['bndOriginInfo']['originBalance'] for bondName, bondPos in m.items()}
+    bPapersPerBond = {bondName: bondOrigBal/facePerPaper for (bondName, bondOrigBal) in bOrignBal.items()}
+    bflowFactor = {bondName: (bondPos/bPapersPerBond[bondName]) for (bondName, bondPos) in m.items()}
     return {bn: calcBondFlow(bf, bflowFactor[bn]) for bn, bf in x['bonds'].items() if bn in m}
 
 
@@ -388,10 +388,31 @@ def searchByFst(xs, v, defaultRtn=None):
             return x
     return defaultRtn
 
-def isMixedDeal(x:dict) -> bool :
+
+def isMixedDeal(x: dict) -> bool :
     if 'assets' in x or 'cashflow' in x:
         return False
-    assetTags = query(x, [S.MVALS,S.ALL,'assets',S.FIRST,S.FIRST])
-    if len(set(assetTags))>1:
+    assetTags = query(x, [S.MVALS, S.ALL, 'assets', S.FIRST, S.FIRST])
+    if len(set(assetTags)) > 1:
         return True
-    return False
+    return False
+
+
+def strFromPath(xs: list) -> str:
+    ps = [strFromLens(_)+f":{v}" for (_,v) in xs]
+    return "/".join(ps)
+
+
+def strFromLens(x) -> str: 
+    if isinstance(x, ui.UnboundLens) and not hasattr(x._optic, "lenses"):
+        if isinstance(x._optic, optics.true_lenses.GetitemLens):
+            return str(x._optic.key)
+        return x._optic.name
+    elif isinstance(x, ui.UnboundLens) and hasattr(x._optic, "lenses"):
+        return "-".join([strFromLens(_) for _ in x._optic.lenses])
+    elif isinstance(x, optics.true_lenses.GetitemLens):
+        return str(x.key)
+    elif isinstance(x, optics.traversals.GetZoomAttrTraversal):
+        return x.name   
+    else:
+        return str(x)