Skip to content

Commit

Permalink
CombinedBlock subclasses DAG; simplified initialization, but does not…
Browse files Browse the repository at this point in the history 
… exploit it in methods yet
  • Loading branch information
@bbardoczy
bbardoczy committed Oct 11, 2021
1 parent a4cf7e3 commit 63a4d4d
Show file tree
Hide file tree
Showing 2 changed files with 5 additions and 175 deletions.
18 changes: 5 additions & 13 deletions src/sequence_jacobian/blocks/combined_block.py
View file
Original file line number Diff line number Diff line change
@@ -1,12 +1,10 @@
"""CombinedBlock class and the combine function to generate it"""

from copy import deepcopy

from .block import Block
from .auxiliary_blocks.jacobiandict_block import JacobianDictBlock
from .support.parent import Parent
from ..classes import ImpulseDict, JacobianDict
from ..utilities.graph import block_sort, find_intermediate_inputs
from ..utilities.graph import DAG, find_intermediate_inputs


def combine(blocks, name="", model_alias=False):
Expand All @@ -18,7 +16,7 @@ def create_model(blocks, **kwargs):
return combine(blocks, model_alias=True, **kwargs)


class CombinedBlock(Block, Parent):
class CombinedBlock(Block, Parent, DAG):
"""A combined `Block` object comprised of several `Block` objects, which topologically sorts them and provides
a set of partial and general equilibrium methods for evaluating their steady state, computes impulse responses,
and calculates Jacobians along the DAG"""
Expand All @@ -29,9 +27,10 @@ def __init__(self, blocks, name="", model_alias=False, sorted_indices=None, inte
super().__init__()

blocks_unsorted = [b if isinstance(b, Block) else JacobianDictBlock(b) for b in blocks]
sorted_indices = block_sort(blocks) if sorted_indices is None else sorted_indices
DAG.__init__(self, blocks_unsorted)

# TODO: deprecate this, use DAG methods instead
self._required = find_intermediate_inputs(blocks) if intermediate_inputs is None else intermediate_inputs
self.blocks = [blocks_unsorted[i] for i in sorted_indices]

if not name:
self.name = f"{self.blocks[0].name}_to_{self.blocks[-1].name}_combined"
Expand All @@ -41,13 +40,6 @@ def __init__(self, blocks, name="", model_alias=False, sorted_indices=None, inte
# now that it has a name, do Parent initialization
Parent.__init__(self, blocks)

# Find all outputs (including those used as intermediary inputs)
self.outputs = set().union(*[block.outputs for block in self.blocks])

# Find all inputs that are *not* intermediary outputs
all_inputs = set().union(*[block.inputs for block in self.blocks])
self.inputs = all_inputs.difference(self.outputs)

# If the create_model() is used instead of combine(), we will have __repr__ show this object as a 'Model'
self._model_alias = model_alias

Expand Down
162 changes: 0 additions & 162 deletions src/sequence_jacobian/utilities/graph.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -172,168 +172,6 @@ def find_intermediate_inputs(blocks):
return required


def block_sort_w_helpers(blocks, helper_blocks=None, calibration=None, return_io=False):
"""Given list of blocks (either blocks themselves or dicts of Jacobians), find a topological sort.
Relies on blocks having 'inputs' and 'outputs' attributes (unless they are dicts of Jacobians, in which case it's
inferred) that indicate their aggregate inputs and outputs
Importantly, because including helper blocks in a blocks without additional measures
can introduce cycles within the DAG, allow the user to provide the calibration that will be used in the
steady_state computation to resolve these cycles.
e.g. Consider Krusell Smith:
Suppose one specifies a helper block based on a calibrated value for "r", which outputs "K" (among other vars).
Normally block_sort would include the "firm" block as a dependency of the helper block
because the "firm" block outputs "r", which the helper block takes as an input.
However, it would also include the helper block as a dependency of the "firm" block because the "firm" block takes
"K" as an input.
This would result in a cycle. However, if a "calibration" is provided in which "r" is included, then
"firm" could be removed as a dependency of helper block and the cycle would be resolved.
blocks: `list`
A list of the blocks (SimpleBlock, HetBlock, etc.) to sort
helper_blocks: `list`
A list of helper blocks
calibration: `dict` or `None`
An optional dict of variable/parameter names and their pre-specified values to help resolve any cycles
introduced by using helper blocks. Read above docstring for more detail
return_io: `bool`
A boolean indicating whether to return the full set of input and output arguments from `blocks`
"""
if return_io:
# step 1: map outputs to blocks for topological sort
outmap, outargs = construct_output_map_w_helpers(blocks, return_output_args=True,
helper_blocks=helper_blocks, calibration=calibration)

# step 2: dependency graph for topological sort and input list
dep, inargs = construct_dependency_graph_w_helpers(blocks, outmap, return_input_args=True, outargs=outargs,
helper_blocks=helper_blocks, calibration=calibration)

return topological_sort(map_to_list(complete_reverse_graph(dep)), map_to_list(dep)), inargs, outargs
else:
# step 1: map outputs to blocks for topological sort
outmap = construct_output_map_w_helpers(blocks, helper_blocks=helper_blocks)

# step 2: dependency graph for topological sort and input list
dep = construct_dependency_graph_w_helpers(blocks, outmap, helper_blocks=helper_blocks, calibration=calibration)

return topological_sort(map_to_list(complete_reverse_graph(dep)), map_to_list(dep))


def map_to_list(m):
return [m[i] for i in range(len(m))]


def construct_output_map_w_helpers(blocks, helper_blocks=None, calibration=None, return_output_args=False):
"""Mirroring construct_output_map functionality in utilities.graph module but augmented to support
helper blocks"""
if calibration is None:
calibration = {}
if helper_blocks is None:
helper_blocks = []

helper_inputs = set().union(*[block.inputs for block in helper_blocks])

outmap = dict()
outargs = set()
for num, block in enumerate(blocks):
# Find the relevant set of outputs corresponding to a block
if hasattr(block, "outputs"):
outputs = block.outputs
elif isinstance(block, dict):
outputs = block.keys()
else:
raise ValueError(f'{block} is not recognized as block or does not provide outputs')

for o in outputs:
# Because some of the outputs of a helper block are, by construction, outputs that also appear in the
# standard blocks that comprise a DAG, ignore the fact that an output is repeated when considering
# throwing this ValueError
if o in outmap and block not in helper_blocks:
raise ValueError(f'{o} is output twice')

# Priority sorting for standard blocks:
# Ensure that the block "outmap" maps "o" to is the actual block and not a helper block if both share
# a given output, such that the dependency graph is constructed on the standard blocks, where possible
if o not in outmap:
outmap[o] = num
if return_output_args and not (o in helper_inputs and o in calibration):
outargs.add(o)
else:
continue
if return_output_args:
return outmap, outargs
else:
return outmap


def construct_dependency_graph_w_helpers(blocks, outmap, helper_blocks=None,
calibration=None, return_input_args=False, outargs=None):
"""Mirroring construct_dependency_graph functionality in utilities.graph module but augmented to support
helper blocks"""
if calibration is None:
calibration = {}
if helper_blocks is None:
helper_blocks = []
if outargs is None:
outargs = {}

dep = {num: set() for num in range(len(blocks))}
inargs = set()
for num, block in enumerate(blocks):
if hasattr(block, 'inputs'):
inputs = block.inputs
else:
inputs = set(i for o in block for i in block[o])
for i in inputs:
# Each potential input to a given block will either be 1) output by another block,
# 2) an unknown or exogenous variable, or 3) a pre-specified variable/parameter passed into
# the steady-state computation via the `calibration' dict.
# If the block is a helper block, then we want to check the calibration to see if the potential
# input is a pre-specified variable/parameter, and if it is then we will not add the block that
# produces that input as an output as a dependency.
# e.g. Krusell Smith's firm_steady_state_solution helper block and firm block would create a cyclic
# dependency, if it were not for this resolution.
if i in outmap and not (i in calibration and block in helper_blocks):
dep[num].add(outmap[i])
if return_input_args and not (i in outargs):
inargs.add(i)
if return_input_args:
return dep, inargs
else:
return dep


def find_intermediate_inputs_w_helpers(blocks, helper_blocks=None, **kwargs):
"""Mirroring find_outputs_that_are_intermediate_inputs functionality in utilities.graph module
but augmented to support helper blocks"""
required = set()
outmap = construct_output_map_w_helpers(blocks, helper_blocks=helper_blocks, **kwargs)
for num, block in enumerate(blocks):
if hasattr(block, 'inputs'):
inputs = block.inputs
else:
inputs = set(i for o in block for i in block[o])
for i in inputs:
if i in outmap:
required.add(i)
return required


def complete_reverse_graph(gph):
"""Given directed graph represented as a dict from nodes to iterables of nodes, return representation of graph that
is complete (i.e. has each vertex pointing to some iterable, even if empty), and a complete version of reversed too.
Have returns be sets, for easy removal"""

revgph = {n: set() for n in gph}
for n, e in gph.items():
for n2 in e:
n2_edges = revgph.setdefault(n2, set())
n2_edges.add(n)

return revgph


def find_cycle(dep, onlyset=None):
"""Return list giving cycle if there is one, otherwise None"""

Expand Down

0 comments on commit 63a4d4d

Please sign in to comment.