def compute_chunk(self, graph, dates, sids, initial_workspace):
        """
        Compute the Pipeline terms in the graph for the requested start and end
        dates.

        This is where we do the actual work of running a pipeline.

        Parameters
        ----------
        graph : zipline.pipeline.graph.ExecutionPlan
            Dependency graph of the terms to be executed.
        dates : pd.DatetimeIndex
            Row labels for our root mask.
        assets : pd.Int64Index
            Column labels for our root mask.
        initial_workspace : dict
            Map from term -> output.
            Must contain at least entry for `self._root_mask_term` whose shape
            is `(len(dates), len(assets))`, but may contain additional
            pre-computed terms for testing or optimization purposes.

        Returns
        -------
        results : dict
            Dictionary mapping requested results to outputs.
        """
        self._validate_compute_chunk_params(
            graph, dates, sids, initial_workspace,
        )
        get_loader = self._get_loader

        # Copy the supplied initial workspace so we don't mutate it in place.
        workspace = initial_workspace.copy()
        refcounts = graph.initial_refcounts(workspace)
        execution_order = graph.execution_order(refcounts)
        domain = graph.domain

        # Many loaders can fetch data more efficiently if we ask them to
        # retrieve all their inputs at once. For example, a loader backed by a
        # SQL database can fetch multiple columns from the database in a single
        # query.
        #
        # To enable these loaders to fetch their data efficiently, we group
        # together requests for LoadableTerms if they are provided by the same
        # loader and they require the same number of extra rows.
        #
        # The extra rows condition is a simplification: we don't currently have
        # a mechanism for asking a loader to fetch different windows of data
        # for different terms, so we only batch requests together when they're
        # going to produce data for the same set of dates. That may change in
        # the future if we find a loader that can still benefit significantly
        # from batching unequal-length requests.
        def loader_group_key(term):
            loader = get_loader(term)
            extra_rows = graph.extra_rows[term]
            return loader, extra_rows

        # Only produce loader groups for the terms we expect to load.  This
        # ensures that we can run pipelines for graphs where we don't have a
        # loader registered for an atomic term if all the dependencies of that
        # term were supplied in the initial workspace.
        will_be_loaded = graph.loadable_terms - viewkeys(workspace)
        loader_groups = groupby(
            loader_group_key,
            (t for t in execution_order if t in will_be_loaded),
        )

        for term in graph.execution_order(refcounts):
            # `term` may have been supplied in `initial_workspace`, and in the
            # future we may pre-compute loadable terms coming from the same
            # dataset.  In either case, we will already have an entry for this
            # term, which we shouldn't re-compute.
            if term in workspace:
                continue

            # Asset labels are always the same, but date labels vary by how
            # many extra rows are needed.
            mask, mask_dates = graph.mask_and_dates_for_term(
                term,
                self._root_mask_term,
                workspace,
                dates,
            )

            if isinstance(term, LoadableTerm):
                loader = get_loader(term)
                to_load = sorted(
                    loader_groups[loader_group_key(term)],
                    key=lambda t: t.dataset
                )
                loaded = loader.load_adjusted_array(
                    domain, to_load, mask_dates, sids, mask,
                )
                assert set(loaded) == set(to_load), (
                    'loader did not return an AdjustedArray for each column\n'
                    'expected: %r\n'
                    'got:      %r' % (sorted(to_load), sorted(loaded))
                )
                workspace.update(loaded)
            else:
                workspace[term] = term._compute(
                    self._inputs_for_term(term, workspace, graph, domain),
                    mask_dates,
                    sids,
                    mask,
                )
                if term.ndim == 2:
                    assert workspace[term].shape == mask.shape
                else:
                    assert workspace[term].shape == (mask.shape[0], 1)

                # Decref dependencies of ``term``, and clear any terms whose
                # refcounts hit 0.
                for garbage_term in graph.decref_dependencies(term, refcounts):
                    del workspace[garbage_term]

        # At this point, all the output terms are in the workspace.
        out = {}
        graph_extra_rows = graph.extra_rows
        for name, term in iteritems(graph.outputs):
            # Truncate off extra rows from outputs.
            out[name] = workspace[term][graph_extra_rows[term]:]
        return out
def compute_chunk ( self , graph , dates , sids , initial_workspace ) : self . _validate_compute_chunk_params ( graph , dates , sids , initial_workspace , ) get_loader = self . _get_loader workspace = initial_workspace . copy ( ) refcounts = graph . initial_refcounts ( workspace ) execution_order = graph . execution_order ( refcounts ) domain = graph . domain def loader_group_key ( term ) : loader = get_loader ( term ) extra_rows = graph . extra_rows [ term ] return loader , extra_rows will_be_loaded = graph . loadable_terms - viewkeys ( workspace ) loader_groups = groupby ( loader_group_key , ( t for t in execution_order if t in will_be_loaded ) , ) for term in graph . execution_order ( refcounts ) : if term in workspace : continue mask , mask_dates = graph . mask_and_dates_for_term ( term , self . _root_mask_term , workspace , dates , ) if isinstance ( term , LoadableTerm ) : loader = get_loader ( term ) to_load = sorted ( loader_groups [ loader_group_key ( term ) ] , key = lambda t : t . dataset ) loaded = loader . load_adjusted_array ( domain , to_load , mask_dates , sids , mask , ) assert set ( loaded ) == set ( to_load ) , ( 'loader did not return an AdjustedArray for each column\n' 'expected: %r\n' 'got:      %r' % ( sorted ( to_load ) , sorted ( loaded ) ) ) workspace . update ( loaded ) else : workspace [ term ] = term . _compute ( self . _inputs_for_term ( term , workspace , graph , domain ) , mask_dates , sids , mask , ) if term . ndim == 2 : assert workspace [ term ] . shape == mask . shape else : assert workspace [ term ] . shape == ( mask . shape [ 0 ] , 1 ) for garbage_term in graph . decref_dependencies ( term , refcounts ) : del workspace [ garbage_term ] out = { } graph_extra_rows = graph . extra_rows for name , term in iteritems ( graph . outputs ) : out [ name ] = workspace [ term ] [ graph_extra_rows [ term ] : ] return out
Compute the Pipeline terms in the graph for the requested start and end
        dates.

        This is where we do the actual work of running a pipeline.

        Parameters
        ----------
        graph : zipline.pipeline.graph.ExecutionPlan
            Dependency graph of the terms to be executed.
        dates : pd.DatetimeIndex
            Row labels for our root mask.
        assets : pd.Int64Index
            Column labels for our root mask.
        initial_workspace : dict
            Map from term -> output.
            Must contain at least entry for `self._root_mask_term` whose shape
            is `(len(dates), len(assets))`, but may contain additional
            pre-computed terms for testing or optimization purposes.

        Returns
        -------
        results : dict
            Dictionary mapping requested results to outputs.