PK!'%&pynetrees/__init__.py__all__ = ["Decision", "Event", "Transition", "EndGame", "Node", "Solver", "Evaluator"] class Holder: def clear(self): d = self.__dict__ for k in d.keys(): d[k] = None class NodeHolder(Holder): def __init__(self) -> None: self.cashflowDistribution = None self.valueDistribution = None self.strategicValue = None self.propagatedCashflow = None self.propagatedValue = None self.choice = None self.failures = [] self.deadEnd = False self.probability = None class TransitionHolder(Holder): def __init__(self) -> None: self.cashflowDistribution = None self.valueDistribution = None self.strategicValue = None self.rejected = False from .decision import Decision from .endgame import EndGame from .evaluator import Evaluator from .event import Event from .node import Node from .solver import Solver from .transition import Transition PK!ypynetrees/cashflow.pyfrom collections import Sequence, Mapping from functools import partial from numbers import Real import pandas as pd def create(*args, freq=None): if len(args) == 0: return pd.Series() if len(args) == 1: arg = args[0] if isinstance(arg, Real): args = [arg] else: args = arg if isinstance(args, Sequence): return pd.Series(args) elif isinstance(args, Mapping): keys = args.keys() if not keys: return pd.Series() types = set(type(k) for k in keys) if len(types) != 1: raise ValueError("Index types differ: {}".format(" != ".join(sorted(types)))) indexType = types.pop() if indexType in (int, pd.Period, pd.Timestamp): return pd.Series(args) if indexType == str: rv = pd.Series(args) rv.index = pd.PeriodIndex(rv.index, freq=freq) return rv raise ValueError("") createDays = partial(create, freq="D") createMonths = partial(create, freq="M") createYears = partial(create, freq="Y") def combineCashflows(cashflows): cashflows = iter(cashflows) combined = next(cashflows) # type: pd.Series while True: try: cf = next(cashflows) # type: pd.Series except StopIteration: break else: combined = combined.add(cf, fill_value=0) return combined def indexAnnuity(base, count, amount): return pd.Series(amount, index=range(base, base + count)) def periodAnnuity(freq, base, count, amount): return pd.Series(amount, index=pd.period_range(base, periods=count, freq=freq)) annuityDays = partial(periodAnnuity, "D") annuityMonths = partial(periodAnnuity, "M") annuityYears = partial(periodAnnuity, "Y") PK!VVpynetrees/decision.pyfrom .node import Node class Decision(Node): TYPE_NAME = "Decision" def typeName(self): return Decision.TYPE_NAME def computePossibilities(self, solver): for t in self.transitions: t.target.computePossibilities(solver) for t in self.transitions: if t.target.results.deadEnd: t.results.rejected = True ts = [t for t in self.transitions if not t.results.rejected] if not any(ts): self.results.deadEnd = True return idx, payout = solver.strategy.selectBestStrategicValue(t.target.results.strategicValue for t in ts) choice = ts[idx] self.results.choice = choice self.results.cashflowDistribution = choice.target.results.cashflowDistribution self.results.valueDistribution = choice.target.results.valueDistribution self.results.strategicValue = choice.target.results.strategicValue def propagateEndgameDistributions(self, currentProbability): super().propagateEndgameDistributions(currentProbability) for t in self.transitions: if t is self.results.choice: t.target.propagateEndgameDistributions(currentProbability) def clone(self): return Decision(self.name, [t.clone() for t in self.transitions]) # from .solver import Solver PK!pynetrees/endgame.pyfrom .node import Node class EndGame(Node): TYPE_NAME = "EndGame" DEFAULT_NAME = "Done" def __init__(self, name: str = None, placeholder: bool = False): super().__init__(name or EndGame.DEFAULT_NAME) self.placeholder = placeholder def typeName(self): return EndGame.TYPE_NAME def cashflow(self): return self.results.propagatedCashflow def propagateCashflows(self, solver, current): self.results.propagatedCashflow = current self.results.propagatedValue = solver.cashflowToValue.actualize(current) def computePossibilities(self, solver: "Solver"): self.results.cashflowDistribution = ((1, self.results.propagatedCashflow),) self.results.valueDistribution = ((1, self.results.propagatedValue),) self.results.strategicValue = solver.strategy.computeStrategicValue(self.results.valueDistribution) failures = solver.getFailingLimits(self) if failures: self.results.failures = failures self.results.deadEnd = True def clone(self): rv = EndGame(self.name, self.placeholder) # from .solver import Solver PK!KMLuupynetrees/evaluator.pyimport numpy as np import pandas as pd class Evaluator: def __init__(self, solver, setters, getters): self.getters = getters self.setters = setters self.solver = solver self.nX = len(setters) self.nY = len(getters) def evaluate(self, X): for i in range(self.nX): self.setters[i](X[i]) self.solver.solve() rv = list() for i in range(self.nY): rv.append(self.getters[i]()) return rv def evaluateMany(self, dataFrame: pd.DataFrame) -> pd.DataFrame: nrow, ncol = dataFrame.shape if ncol != self.nX: raise ValueError("Got {} input columns, expected {}".format(ncol, self.nX, self.nY)) dataFrame = dataFrame.copy() for i in range(self.nY): dataFrame["{}".format(i)] = np.nan for r in range(nrow): for c in range(self.nX): self.setters[c](dataFrame.iloc[r, c]) self.solver.solve() for c in range(self.nY): dataFrame.iloc[r, c + self.nX] = self.getters[c]() return dataFrame PK!61H/ / pynetrees/event.pyfrom .node import Node class Event(Node): TYPE_NAME = "Event" def typeName(self): return Event.TYPE_NAME def computePossibilities(self, solver): ts = self.transitions # type: list[Transition] # Check probabilities and patch if we have a complement # This could be optimised out to an init step. Keeping it separated from actual calculation probs = [t.probability for t in ts] nnprobs = [p for p in probs if p is not None] nNones = len(probs) - len(nnprobs) if nNones > 1: raise ValueError("Cannot have more than one event without explicit probability") totalProb = sum(nnprobs) if nNones == 1: if totalProb > 1: raise ValueError("Cannot have a complement a probabily exceeding 100%") pComplement = 1 - totalProb totalProb = 1 else: pComplement = None self.results.cashflowDistribution = [] self.results.valueDistribution = [] for t in ts: # type: Transition transitionProb = t.probability if transitionProb is None: transitionProb = pComplement else: transitionProb /= totalProb t.results.probability = transitionProb t.target.computePossibilities(solver) if t.target.results.deadEnd: self.results.deadEnd = True else: for prob, outcome in t.target.results.cashflowDistribution: self.results.cashflowDistribution.append((transitionProb * prob, outcome)) for prob, outcome in t.target.results.valueDistribution: self.results.valueDistribution.append((transitionProb * prob, outcome)) if not self.results.deadEnd: self.results.strategicValue = solver.strategy.computeStrategicValue(self.results.valueDistribution) failures = solver.getFailingLimits(self) if failures: self.results.failures = failures self.results.deadEnd = True else: self.results.cashflowDistribution = [] self.results.valueDistribution = [] def propagateEndgameDistributions(self, currentProbability): super().propagateEndgameDistributions(currentProbability) for t in self.transitions: t.target.propagateEndgameDistributions(currentProbability * t.results.probability) def clone(self): return Event(self.name, [t.clone() for t in self.transitions]) from .transition import Transition PK!f22pynetrees/jupyter.pyimport graphviz def showTree(root: "Node", prune=False, discard=True, *args, **kwargs): root.createPlaceholders() eng = GraphvizEngine(root, *args, **kwargs) graph = eng.render("svg", prune, discard) return graphviz.Source(graph) from .node import Node from .render import GraphvizEngine PK!pynetrees/limits.pyfrom collections import namedtuple from functools import partial from numbers import Real import pandas as pd Limit = namedtuple("Limit", ("name", "predicate")) def failIfUnderThreshold(threshold, value): diff = value - threshold return diff < 0, abs(diff) def failIfOverThreshold(threshold, value): diff = value - threshold return diff > 0, abs(diff) def valueUnderThreshold(threshold: Real): return partial(failIfUnderThreshold, threshold) def cashflowRollsumUnderThreshold(threshold: Real): def predicate(cf: pd.Series): value = cf.expanding().sum().min() return failIfUnderThreshold(threshold, value) return predicate def expectedTTROverThreshold(threshold: Real): def predicate(cfs: list((Real, pd.Series))): value = rpExpected((prob, TTR_ACTUALIZER.actualize(cf)) for prob, cf in cfs) return failIfOverThreshold(threshold, value) return predicate def valueVarianceOverThreshold(threshold: Real): def predicate(values: list((Real, Real))): value = rpStandardDeviation(values) return failIfOverThreshold(threshold, value) return predicate from .strategy import * from .valueactualizer import TTR_ACTUALIZER PK!Ypynetrees/node.pyfrom abc import ABCMeta, abstractmethod from typing import Sequence class Node(metaclass=ABCMeta): def __init__(self, name: str, transitions: "Sequence[Transition]" = None): self.name = name self.transitions = transitions or [] # type: Sequence[Transition] self.results = NodeHolder() @abstractmethod def typeName(self): pass def __str__(self) -> str: return "{}: {}".format(self.typeName(), self.name) def createPlaceholders(self): rv = 0 for t in self.transitions: rv += t.createPlaceHolders() return rv def propagateCashflows(self, solver, current): for t in self.transitions: t.propagateCashflows(current, solver) def getNodesFlat(self): yield self for t in self.transitions: if t.target is not None: yield from t.target.getNodesFlat() def transit(self, *transitionNames) -> "Transition": if not transitionNames: raise ValueError("Need at least one transition name") target = self for name in transitionNames: try: trans = next(t for t in target.transitions if t.name == name) except StopIteration: raise KeyError("Node {} has no transition called {}".format(target.name, name)) target = trans.target return trans @abstractmethod def computePossibilities(self, solver: "Solver"): pass def propagateEndgameDistributions(self, currentProbability): self.results.probability = currentProbability @abstractmethod def clone(self): pass from .transition import Transition from . import NodeHolder PK!`pynetrees/render.pyimport graphviz import pandas as pd from . import Decision from . import Node, Event, EndGame DISCARDED_COLOR = "lightgrey" REJECTED_COLOR = "red" NODE_PREFIXES = {Decision:"d", Event:"ev", EndGame:"eg", } COMMON_NODE_ATTRIBUTES = {"style":"filled", "fontcolor":"white" } NODE_ATTRIBUTES = { Decision:{"shape":"box", "fillcolor":"green"}, Event:{"shape":"oval", "fillcolor":"orange"}, EndGame:{"shape":"doubleoctagon", "fillcolor":"blue"}, } class GraphvizEngine: def __init__(self, root: "Node", cashflowFormat="{:,.2f}", strategicValueFormat="{:,.2f}", transitionProbabilityFormat="{:.2%}", computedProbabilityFormat="{:.2%}") -> None: super().__init__() self.computedProbabilityFormat = computedProbabilityFormat self.transitionProbabilityFormat = transitionProbabilityFormat self.strategicValueFormat = strategicValueFormat self.cashflowFormat = cashflowFormat self.format = format self.nodeNumber = None # type: int self.root = root # type: Node def getNextName(self, node): self.nodeNumber += 1 prefix = NODE_PREFIXES[type(node)] return "{}{}".format(prefix, self.nodeNumber) def render(self, format, prune=False, discard=True): self.nodeNumber = 0 self.graph = graphviz.Digraph(format=format, graph_attr={"rankdir":"LR"}) self.addNode(self.root, False, prune, discard) return self.graph def addNode(self, node: "Node", discarded, prune, discard): name = self.getNextName(node) attr = dict(COMMON_NODE_ATTRIBUTES, **NODE_ATTRIBUTES[type(node)]) if node.results.deadEnd: attr["fillcolor"] = REJECTED_COLOR elif discard and discarded: attr["fillcolor"] = DISCARDED_COLOR nodeLabel = node.name if node.results.strategicValue is not None: nodeLabel += ("\nR$ = " + self.strategicValueFormat).format(node.results.strategicValue) if node.results.probability is not None: nodeLabel += "\n(P= {})".format(self.computedProbabilityFormat).format(node.results.probability) self.graph.node(name=name, label=nodeLabel, **attr) for trans in node.transitions: discartTrans = discarded or (isinstance(node, Decision) and (node.results.choice is not trans)) if prune and discartTrans: continue transitionToDeadend = trans.results.rejected or trans.target.results.deadEnd tname = self.addNode(trans.target, discartTrans or transitionToDeadend, prune, discard) edgeLabel = trans.name if trans.payout is not None: edgeLabel += "\n$= " + self.formatCashflow(trans.payout) if trans.probability is not None: edgeLabel += "\n(P= {})".format(self.transitionProbabilityFormat).format(trans.probability) if transitionToDeadend: color = REJECTED_COLOR elif discard and discartTrans: color = DISCARDED_COLOR else: color = "black" self.graph.edge(name, tname, edgeLabel, color=color) return name def formatSerie(self, serie: pd.Series): def formatOne(index): at = index ammount = serie[index] return "{}:{}".format(at, self.cashflowFormat.format(ammount)) if len(serie) == 0: return None elif len(serie) == 1: return formatOne(serie.index[0]) else: if len(serie) > 2: swallowed = "(...+{}...)\n".format(len(serie) - 2) else: swallowed = "" return "{}\n{}{}".format( formatOne(serie.index[0]), swallowed, formatOne(serie.index[-1]), ) def formatCashflow(self, cashflow): if isinstance(cashflow, pd.Series): return self.formatSerie(cashflow) else: return self.cashflowFormat.format(cashflow) PK!;pynetrees/sensitivity.pyfrom collections import namedtuple from itertools import product from numbers import Real import matplotlib.pyplot as plt import pandas as pd import numpy as np import seaborn as sb IMPACT_COLORS_DIRECT = ("red", "green") IMPACT_COLORS_REVERSED = list(reversed(IMPACT_COLORS_DIRECT)) LOW = "Low" HIGH = "High" EXTREMUM_NAMES = [LOW, HIGH] Variable = namedtuple("Variable", ["name", "setter", "base", "domain"]) Output = namedtuple("Output", "name getter".split()) class SensitivityAnalysis: def __init__(self, solver: "Solver", variables: "list(Variable)", outputs: "list(Output)"): self.outputs = outputs self.variables = variables self.solver = solver self.variableNames = [v.name for v in self.variables] self.outputNames = [out.name for out in self.outputs] self.outputGetters = [out.getter for out in self.outputs] self.extremums = pd.DataFrame( index=[v.name for v in self.variables], columns=(pd.MultiIndex.from_product((self.outputNames, EXTREMUM_NAMES))) ) for var in self.variables: var.setter(var.base) self.solver.solve() self.baseValues = pd.DataFrame(index=self.outputNames, columns=["base"]) for out in self.outputs: self.baseValues.loc[out.name, "base"] = out.getter() self.individualResponses = dict() for var in self.variables: # type: Variable X = pd.DataFrame(var.domain) evaluator = Evaluator(self.solver, [var.setter], self.outputGetters) Y = evaluator.evaluateMany(X) Y.columns = [var.name] + self.outputNames self.individualResponses[var.name] = Y mins = Y.min() maxs = Y.max() for on in self.outputNames: self.extremums.loc[var.name, (on, LOW)] = mins[on] self.extremums.loc[var.name, (on, HIGH)] = maxs[on] var.setter(var.base) def getImpactGraphs(self, invertColors=False, grid=False): colors = invertColors and IMPACT_COLORS_REVERSED or IMPACT_COLORS_DIRECT rv = dict() for outname in self.outputNames: base = self.baseValues.base[outname] if not isinstance(base, Real): continue exts = self.extremums[outname] exts = exts - base fig, ax = plt.subplots(1, 1) #type: plt.Figure, plt.Axes assert isinstance(fig, plt.Figure) exts.plot.barh(stacked=True, left=base, color=colors, ax=ax) ax.set(title=outname) ax.axvline(base, c="black", label="Base") ax.grid(grid) ax.legend() ax.xaxis.grid(True, ls="dotted") ax.set_axisbelow(True) rv[outname] = fig return rv def getIndivisualResponseGraphs(self, figScale=(5, 5), hideExtraXLabels=False, hideExtraYLabels=False, grid=True): nrow = len(self.outputNames) ncol = len(self.variableNames) fig, axs = plt.subplots(nrow, ncol, figsize=(figScale[0] * ncol, figScale[1] * nrow)) numericRows = [] for r in range(nrow): oname = self.outputNames[r] isNumeric = np.issubdtype(self.individualResponses[self.variableNames[0]][oname].dtype, np.number) levels = set() if isNumeric: numericRows.append(r) else: for c in range(ncol): vname = self.variableNames[c] levels = levels.union(self.individualResponses[vname][oname]) levels = sorted(levels) for c in range(ncol): vname = self.variableNames[c] ax = axs[r][c] values = self.individualResponses[vname] # type: pd.DataFrame if isNumeric: values.plot.scatter(ax=ax, x=vname, y=oname) else: sb.stripplot(vname, oname, data=values, ax=ax, jitter=False, hue_order=levels, order=levels) ax.grid(grid) if r == nrow - 1: ax.set(xlabel=vname) elif hideExtraXLabels: ax.set(xlabel="", xticklabels=[]) if c == 0: ax.set(ylabel=oname) elif hideExtraYLabels: ax.set(ylabel="", yticklabels=[]) for r in numericRows: rowmin = min(ax.get_ylim()[0] for ax in axs[r]) rowmax = max(ax.get_ylim()[1] for ax in axs[r]) for c in range(ncol): axs[r][c].set(ylim=((rowmin, rowmax))) for c in range(ncol): colmin = min(axs[r][c].get_xlim()[0] for r in range(nrow)) colmax = max(axs[r][c].get_xlim()[1] for r in range(nrow)) for r in range(nrow): axs[r][c].set(xlim=((colmin, colmax))) return fig, axs from . import Solver, Evaluator PK!nfpynetrees/solver.pyfrom collections import namedtuple from numbers import Real import pandas as pd from .valueactualizer import ValueActualizer, SCALAR_ACTUALIZER SCALAR_ZERO = 0 SERIES_ZERO = pd.Series() class Solver: def __init__(self, root: "Node", strategy: "Strategy", cashflowToValue: ValueActualizer = SCALAR_ACTUALIZER, cashflowLimits=None, valueLimits=None, strategicValueLimits=None, cashflowDistributionLimits=None, valueDistributionLimits=None, ) -> None: self.root = root # type: Node self.strategy = strategy # type: Strategy self.cashflowToValue = cashflowToValue def castLimits(limits): if limits is None: return [] return [Limit(*l) for l in limits] self.cashflowLimits = castLimits(cashflowLimits) self.valueLimits = castLimits(valueLimits) self.strategicValueLimits = castLimits(strategicValueLimits) self.cashflowDistributionLimits = castLimits(cashflowDistributionLimits) self.valueDistributionLimits = castLimits(valueDistributionLimits) self.addPayouts = None self.hasCFSeries = None def solve(self): self.reset() root = self.root self.setCashflowType() root.createPlaceholders() if self.hasCFSeries: zero = SERIES_ZERO else: zero = SCALAR_ZERO root.propagateCashflows(self, zero) root.computePossibilities(self) root.propagateEndgameDistributions(1) def reset(self): def resetNode(node: Node): node.results = NodeHolder() for t in node.transitions: t.results = TransitionHolder() if isinstance(t.target, EndGame) and t.target.placeholder: t.target = None elif t.target is not None: resetNode(t.target) resetNode(self.root) self.addPayouts = None self.hasCFSeries = None def payoutDistribution(self, node: "Node" = None): if node is None: node = self.root df = pd.DataFrame(node.results.valueDistribution, columns=("probability", "value")) df = df.groupby("value").sum() return df def strategicValue(self): return self.root.results.strategicValue def addPayoutsScalar(self, *payouts): payouts = [p for p in payouts if p is not None] if not payouts: return None return sum(payouts) def addPayoutsSeries(self, *payouts): payouts = [p for p in payouts if p is not None] if not payouts: return None return combineCashflows(payouts) def setCashflowType(self): for n in self.root.getNodesFlat(): for t in n.transitions: if t.payout is not None: if isinstance(t.payout, Real): self.addPayouts = self.addPayoutsScalar self.hasCFSeries = False else: self.addPayouts = self.addPayoutsSeries self.hasCFSeries = True return def getFailingLimits(self, node: "Node"): rv = [] for name, limit in self.cashflowLimits: # type: Limit for prob, cf in node.results.cashflowDistribution: fail, margin = limit(cf) if fail: rv.append((name, margin)) for name, limit in self.valueLimits: # type: Limit for prob, v in node.results.valueDistribution: fail, margin = limit(v) if fail: rv.append((name, margin)) for name, limit in self.valueDistributionLimits: # type: Limit fail, margin = limit(node.results.valueDistribution) if fail: rv.append((name, margin)) for name, limit in self.cashflowDistributionLimits: # type: Limit fail, margin = limit(node.results.cashflowDistribution) if fail: rv.append((name, margin)) for name, limit in self.strategicValueLimits: # type: Limit fail, margin = limit(node.results.strategicValue) if fail: rv.append((name, margin)) return rv from .strategy import Strategy from .cashflow import combineCashflows from . import Node, EndGame, NodeHolder, TransitionHolder from .limits import Limit PK!IL\\pynetrees/strategy.pyfrom math import sqrt class Strategy: def __init__(self, computeStrategicValue, selectBestStrategicValue): self.computeStrategicValue = computeStrategicValue self.selectBestStrategicValue = selectBestStrategicValue def rpExpected(strategicValues: list((float, float))) -> float: return sum(prob * sv for prob, sv in strategicValues) def rpStandardDeviation(strategicValues: list((float, float))) -> float: expected = rpExpected(strategicValues) return sqrt(sum((prob * pow(sv - expected, 2)) for prob, sv in strategicValues)) def rpMin(strategicValues: list((float, float))) -> float: return min(sv for prob, sv in strategicValues) def rpMax(strategicValues: list((float, float))) -> float: return max(sv for prob, sv in strategicValues) def selectMaxRP(strategicValues: "list(float)") -> (int, float): bestIndex = None bestSV = float("-inf") for i, rp in enumerate(strategicValues): if rp > bestSV: bestIndex = i bestSV = rp return bestIndex, bestSV def selectMinRP(strategicValues: "list(float)") -> (int, float): bestIndex = None bestSV = float("inf") for i, rp in enumerate(strategicValues): if rp < bestSV: bestIndex = i bestSV = rp return bestIndex, bestSV def createMaxExpected(): return Strategy(rpExpected, selectMaxRP) def createMinMin(): return Strategy(rpMin, selectMinRP) def createMaxMax(): return Strategy(rpMax, selectMaxRP) def createMinMax(): return Strategy(rpMax, selectMinRP) def createMaxMin(): return Strategy(rpMin, selectMaxRP) PK!?CCpynetrees/transition.pyclass Transition: def __init__(self, name, payout=None, probability=None, target: "Node" = None) -> None: if probability is not None and probability < 0: raise ValueError("Probability cannot be negative") self.name = name self.payout = payout self.target = target self.probability = probability self.results = TransitionHolder() def __str__(self) -> str: rv = "Transition: {}".format(self.name) details = [] if self.probability is not None: details.append("p={:.4%}".format(self.probability)) if self.payout is not None: details.append("impact={}".format(self.payout)) details = ", ".join(details) if details: rv = "{} ({})".format(rv, details) return rv def createPlaceHolders(self): if self.target is None: self.target = EndGame(placeholder=True) return 1 else: return self.target.createPlaceholders() def propagateCashflows(self, current, solver): current = solver.addPayouts(current, self.payout) return self.target.propagateCashflows(solver, current) def clone(self): payout = self.payout if isinstance(payout, pd.DataFrame): payout = payout.copy(True) target = self.target if target is not None: target = target.clone() return Transition(self.name, payout, self.probability, target) import pandas as pd from . import TransitionHolder from .endgame import EndGame from .node import Node PK!JjO O pynetrees/valueactualizer.pyfrom abc import ABCMeta, abstractmethod from numbers import Real import pandas as pd class ValueActualizer(metaclass=ABCMeta): @abstractmethod def actualize(self, cashflow) -> Real: pass class ScalarActualizer(ValueActualizer): def actualize(self, cashflow): assert isinstance(cashflow, Real) return cashflow class SummingActualizer(ValueActualizer): def actualize(self, cashflow) -> Real: assert isinstance(cashflow, pd.Series) return cashflow.sum() class TimeToRecoveryActualizer(ValueActualizer): def actualize(self, cashflow: pd.Series) -> Real: assert isinstance(cashflow, pd.Series) cashflow = cashflow.expanding().sum() cfiter = iter(cashflow.iteritems()) # Still positive try: while True: period, amount = next(cfiter) if amount < 0: # switching negative firstNegativePeriod = period break except StopIteration: return 0 # Went under zero try: while True: period, amount = next(cfiter) if amount >= 0: # back to positive breakEvenPeriod = period break except StopIteration: return None # Back in black try: while True: period, amount = next(cfiter) if amount < 0: # Going under zero for a second time: cannot evaluate return None except StopIteration: return breakEvenPeriod - firstNegativePeriod # Done class IndexedNPV(ValueActualizer): def __init__(self, ratePerPeriod): self.ratePerPeriod = ratePerPeriod def actualize(self, cashflow: pd.Series) -> Real: npv = 0 r = self.ratePerPeriod for period, amount in cashflow.iteritems(): npv += amount / pow(1 + self.ratePerPeriod, period) return npv class PeriodNPV(ValueActualizer): def __init__(self, ratePerPeriod, initialPeriod): self.initialPeriod = initialPeriod self.ratePerPeriod = ratePerPeriod def actualize(self, cashflow: pd.Series) -> Real: npv = 0 r = self.ratePerPeriod t0 = self.initialPeriod for period, amount in cashflow.iteritems(): period = period - t0 npv += amount / pow(1 + self.ratePerPeriod, period) return npv SCALAR_ACTUALIZER = ScalarActualizer() SUMMING_ACTUALIZER = SummingActualizer() TTR_ACTUALIZER = TimeToRecoveryActualizer() PK!HڽTUpynetrees-1.0.1.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!HQ="pynetrees-1.0.1.dist-info/METADATAKn0: +r]}pDb*,I9Q/ԃbډ&]4@H/Ā7r^VD $-GCp,N8|ipeL!3;ܩPCI,앟bvuѨbۈэö05'KUPYX-L$g]߿HIj$<ۻ*i5#zpy&f7轺Q 2άufKe^\\ÂZr48 \6;lLPkVĶxi_C|^ʋً?`$=jS Vު08y |ƏHrx80S=j 1Q͢vm92 7ƵRD]ϝSXL8-!^/%{KǓ?PK!H !Na pynetrees-1.0.1.dist-info/RECORD}˲Hy} y C"7$ |}QUitDOϹFCY0zTB6/@# Y9Ϣb2c8[ {Ԍ %̏o,}(s_ ٛ5RVY]oWj ,Y`4khX' pynetrees/decision.pyPK!pynetrees/endgame.pyPK!KMLuupynetrees/evaluator.pyPK!61H/ / 1pynetrees/event.pyPK!f22$pynetrees/jupyter.pyPK!%pynetrees/limits.pyPK!Y*pynetrees/node.pyPK!`1pynetrees/render.pyPK!;Apynetrees/sensitivity.pyPK!nfUpynetrees/solver.pyPK!IL\\agpynetrees/strategy.pyPK!?CCmpynetrees/transition.pyPK!JjO O htpynetrees/valueactualizer.pyPK!HڽTU~pynetrees-1.0.1.dist-info/WHEELPK!HQ="pynetrees-1.0.1.dist-info/METADATAPK!H !Na ppynetrees-1.0.1.dist-info/RECORDPK