<commit_before>"""
Compute Engine for FFC API
"""
from abc import (
    ABCMeta,
    abstractmethod,
)
from operator import and_
from six import (
    iteritems,
    with_metaclass,
)
from six.moves import (
    reduce,
    zip,
    zip_longest,
)

from networkx import (
    DiGraph,
    get_node_attributes,
    topological_sort,
)
from numpy import (
    add,
    empty_like,
)
from pandas import (
    DataFrame,
    date_range,
    MultiIndex,
)

from zipline.lib.adjusted_array import ensure_ndarray
from zipline.errors import NoFurtherDataError
from zipline.modelling.factor import Factor
from zipline.modelling.filter import Filter


# TODO: Move this somewhere else.
class CyclicDependency(Exception):
    pass


def build_dependency_graph(terms):
    """
    Build a dependency graph containing the given terms and their dependencies.

    Parameters
    ----------
    terms : iterable
        An iterable of zipline.modelling.term.Term.

    Returns
    -------
    dependencies : networkx.DiGraph
        A directed graph representing the dependencies of the desired inputs.

        Each node in the graph has an `extra_rows` attribute, indicating how
        many, if any, extra rows we should compute for the node.  Extra rows
        are most often needed when a term is an input to a rolling window
        computation.  For example, if we compute a 30 day moving average of
        price from day X to day Y, we need to load price data for the range
        from day (X - 29) to day Y.
    """
    dependencies = DiGraph()
    parents = set()
    for term in terms:
        _add_to_graph(
            term,
            dependencies,
            parents,
            extra_rows=0,
        )
        # No parents should be left between top-level terms.
        assert not parents
    return dependencies


def _add_to_graph(term,
                  dependencies,
                  parents,
                  extra_rows):
    """
    Add the term and all its inputs to dependencies.
    """
    # If we've seen this node already as a parent of the current traversal,
    # it means we have an unsatisifiable dependency.  This should only be
    # possible if the term's inputs are mutated after construction.
    if term in parents:
        raise CyclicDependency(term)
    parents.add(term)

    try:
        existing = dependencies.node[term]
    except KeyError:
        # We're not yet in the graph: add the term with the specified number of
        # extra rows.
        dependencies.add_node(term, extra_rows=extra_rows)
    else:
        # We're already in the graph because we've been traversed by
        # another parent.  Ensure that we have enough extra rows to satisfy
        # all of our parents.
        existing['extra_rows'] = max(extra_rows, existing['extra_rows'])

    for subterm in term.inputs:
        _add_to_graph(
            subterm,
            dependencies,
            parents,
            extra_rows=extra_rows + term.extra_input_rows,
        )
        dependencies.add_edge(subterm, term)

    parents.remove(term)


class FFCEngine(with_metaclass(ABCMeta)):

    @abstractmethod
    def factor_matrix(self, terms, start_date, end_date):
        """
        Compute values for `terms` between `start_date` and `end_date`.

        Returns a DataFrame with a MultiIndex of (date, asset) pairs on the
        index.  On each date, we return a row for each asset that passed all
        instances of `Filter` in `terms, and the columns of the returned frame
        will be the keys in `terms` whose values are instances of `Factor`.

        Parameters
        ----------
        terms : dict
            Map from str -> zipline.modelling.term.Term.
        start_date : datetime
            The first date of the matrix.
        end_date : datetime
            The last date of the matrix.

        Returns
        -------
        matrix : pd.DataFrame
            A matrix of factors
        """
        raise NotImplementedError("factor_matrix")


class NoOpFFCEngine(FFCEngine):
    """
    FFCEngine that doesn't do anything.
    """

    def factor_matrix(self, terms, start, end):
        return DataFrame(
            index=MultiIndex.from_product(
                [date_range(start=start, end=end, freq='D'), ()],
            ),
            columns=sorted(terms.keys())
        )


class SimpleFFCEngine(object):
    """
    FFC Engine class that computes each term independently.

    Parameters
    ----------
    loader : FFCLoader
        A loader to use to retrieve raw data for atomic terms.
    calendar : DatetimeIndex
        Array of dates to consider as trading days when computing a range
        between a fixed start and end.
    asset_finder : zipline.assets.AssetFinder<commit_msg>STY: Use same names in subclass.<commit_after>"""
Compute Engine for FFC API
"""
from abc import (
    ABCMeta,
    abstractmethod,
)
from operator import and_
from six import (
    iteritems,
    with_metaclass,
)
from six.moves import (
    reduce,
    zip,
    zip_longest,
)

from networkx import (
    DiGraph,
    get_node_attributes,
    topological_sort,
)
from numpy import (
    add,
    empty_like,
)
from pandas import (
    DataFrame,
    date_range,
    MultiIndex,
)

from zipline.lib.adjusted_array import ensure_ndarray
from zipline.errors import NoFurtherDataError
from zipline.modelling.factor import Factor
from zipline.modelling.filter import Filter


# TODO: Move this somewhere else.
class CyclicDependency(Exception):
    pass


def build_dependency_graph(terms):
    """
    Build a dependency graph containing the given terms and their dependencies.

    Parameters
    ----------
    terms : iterable
        An iterable of zipline.modelling.term.Term.

    Returns
    -------
    dependencies : networkx.DiGraph
        A directed graph representing the dependencies of the desired inputs.

        Each node in the graph has an `extra_rows` attribute, indicating how
        many, if any, extra rows we should compute for the node.  Extra rows
        are most often needed when a term is an input to a rolling window
        computation.  For example, if we compute a 30 day moving average of
        price from day X to day Y, we need to load price data for the range
        from day (X - 29) to day Y.
    """
    dependencies = DiGraph()
    parents = set()
    for term in terms:
        _add_to_graph(
            term,
            dependencies,
            parents,
            extra_rows=0,
        )
        # No parents should be left between top-level terms.
        assert not parents
    return dependencies


def _add_to_graph(term,
                  dependencies,
                  parents,
                  extra_rows):
    """
    Add the term and all its inputs to dependencies.
    """
    # If we've seen this node already as a parent of the current traversal,
    # it means we have an unsatisifiable dependency.  This should only be
    # possible if the term's inputs are mutated after construction.
    if term in parents:
        raise CyclicDependency(term)
    parents.add(term)

    try:
        existing = dependencies.node[term]
    except KeyError:
        # We're not yet in the graph: add the term with the specified number of
        # extra rows.
        dependencies.add_node(term, extra_rows=extra_rows)
    else:
        # We're already in the graph because we've been traversed by
        # another parent.  Ensure that we have enough extra rows to satisfy
        # all of our parents.
        existing['extra_rows'] = max(extra_rows, existing['extra_rows'])

    for subterm in term.inputs:
        _add_to_graph(
            subterm,
            dependencies,
            parents,
            extra_rows=extra_rows + term.extra_input_rows,
        )
        dependencies.add_edge(subterm, term)

    parents.remove(term)


class FFCEngine(with_metaclass(ABCMeta)):

    @abstractmethod
    def factor_matrix(self, terms, start_date, end_date):
        """
        Compute values for `terms` between `start_date` and `end_date`.

        Returns a DataFrame with a MultiIndex of (date, asset) pairs on the
        index.  On each date, we return a row for each asset that passed all
        instances of `Filter` in `terms, and the columns of the returned frame
        will be the keys in `terms` whose values are instances of `Factor`.

        Parameters
        ----------
        terms : dict
            Map from str -> zipline.modelling.term.Term.
        start_date : datetime
            The first date of the matrix.
        end_date : datetime
            The last date of the matrix.

        Returns
        -------
        matrix : pd.DataFrame
            A matrix of factors
        """
        raise NotImplementedError("factor_matrix")


class NoOpFFCEngine(FFCEngine):
    """
    FFCEngine that doesn't do anything.
    """

    def factor_matrix(self, terms, start_date, end_date):
        return DataFrame(
            index=MultiIndex.from_product(
                [date_range(start=start_date, end=end_date, freq='D'), ()],
            ),
            columns=sorted(terms.keys())
        )


class SimpleFFCEngine(object):
    """
    FFC Engine class that computes each term independently.

    Parameters
    ----------
    loader : FFCLoader
        A loader to use to retrieve raw data for atomic terms.
    calendar : DatetimeIndex
        Array of dates to consider as trading days when computing a range
        between a fixed start and end.
    asset_finder : zipline.assets.AssetFinder
{'fraction_non_alphanumeric': 0.0710319539025668, 'fraction_numerical': 0.0010476689366160294, 'mean_word_length': 2.1809396867710764, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4864135', 'n_tokens_mistral': 2631, 'n_tokens_neox': 2523, 'n_words': 966}