Repository: quantopian/zipline
File: zipline/pipeline/engine.py (Python)

Code before:
                # remaining ``maybe_specialize`` calls floating around in this
                # file.
                #
                # In the meantime, disallowing ComputableTerms in the initial
                # workspace would break almost every test in
                # `test_filter`/`test_factor`/`test_classifier`, and fixing
                # them would require updating all those tests to compute with
                # more specialized terms. Once we have full specialization, we
                # can fix all the tests without a large volume of edits by
                # simply specializing their workspaces, so for now I'm leaving
                # this in place as a somewhat sharp edge.
                if isinstance(term, LoadableTerm):
                    raise ValueError(
                        "Loadable workspace terms must be specialized to a "
                        "domain, but got generic term {}".format(term)
                    )

            elif term.domain != graph.domain:
                raise ValueError(
                    "Initial workspace term {} has domain {}. "
                    "Does not match pipeline domain {}".format(
                        term, term.domain, graph.domain,
                    )
                )

    def adjust_date(self, pipeline, dt):
        """
        Given a date, align it to the calendar of the pipeline's domain.

        Returns
        -------
        pd.Timestamp
        """
        dt = Timestamp(dt, tz='UTC')

        domain = self._resolve_domain(pipeline)
        if domain.calendar.is_session(dt):
            return dt

        trading_days = domain.all_sessions()
        return trading_days[trading_days.searchsorted(dt)]

    def _resolve_domain(self, pipeline):
        """Resolve a concrete domain for ``pipeline``.
        """
        domain = pipeline.domain(default=self._default_domain)
        if domain is GENERIC:
            raise ValueError(
                "Unable to determine domain for Pipeline.\n"
                "Pass domain=<desired domain> to your Pipeline to set a "
                "domain."

Review comment (refactor):
Talked about this in person a bit. I think my preference here would be to split this into two piece of functionality:

1. `resolve_domain` (which already exists, but needs to be made public and tested)
2. `adjust_dates`, which can be part of the `Domain` interface.

The rationale for this split is twofold:

1. We might have consumers that have a domain and want to align dates to the domain's calendar, but don't want to construct a full-blown pipeline engine.
2. The domain is the logical "owner" of date information, so it feels like a more clean separation to have it be responsible for the rolling logic.

Code after:
                # remaining ``maybe_specialize`` calls floating around in this
                # file.
                #
                # In the meantime, disallowing ComputableTerms in the initial
                # workspace would break almost every test in
                # `test_filter`/`test_factor`/`test_classifier`, and fixing
                # them would require updating all those tests to compute with
                # more specialized terms. Once we have full specialization, we
                # can fix all the tests without a large volume of edits by
                # simply specializing their workspaces, so for now I'm leaving
                # this in place as a somewhat sharp edge.
                if isinstance(term, LoadableTerm):
                    raise ValueError(
                        "Loadable workspace terms must be specialized to a "
                        "domain, but got generic term {}".format(term)
                    )

            elif term.domain != graph.domain:
                raise ValueError(
                    "Initial workspace term {} has domain {}. "
                    "Does not match pipeline domain {}".format(
                        term, term.domain, graph.domain,
                    )
                )

    def resolve_domain(self, pipeline):
        """Resolve a concrete domain for ``pipeline``.
        """
        domain = pipeline.domain(default=self._default_domain)
        if domain is GENERIC:
            raise ValueError(
                "Unable to determine domain for Pipeline.\n"
                "Pass domain=<desired domain> to your Pipeline to set a "
                "domain."
            )
        return domain

    def _is_special_root_term(self, term):
        return (
            term is self._root_mask_term
            or term is self._root_mask_dates_term
        )