ronantakizawa/codereview-bench huggingface dataset:
Repository: quantopian/zipline
File: tests/pipeline/test_pipeline_algo.py (Python)

Code before:

    def test_pipeline_compute_before_bts(self):

        # for storing and keeping track of calls to BTS and TestFactor.compute
        history = []
        counter = [0]

        class TestFactor(CustomFactor, SingleInputMixin):

            inputs = [TestingDataSet.float_col]

            def compute(self, today, assets, out, inputs):
                history.append(
                    "CustomFactor call"
                )

        def initialize(context):
            pipeline = attach_pipeline(Pipeline(), 'my_pipeline')
            test_factor = TestFactor(window_length=30)
            pipeline.add(test_factor, 'test_factor')

        def handle_data(context, data):
            pass

        def before_trading_start(context, data):
            context.results = pipeline_output('my_pipeline')
            history.append("BTS call %s" % counter[0])
            counter[0] += 1

        self.run_algorithm(
            initialize=initialize,
            handle_data=handle_data,
            before_trading_start=before_trading_start,
        )

        self.assertTrue(counter[0] > 0)

        # All pipeline computation calls should occur before any BTS calls,
        # so there shouldn't be any compute calls in the second half of
        # this list
        self.assertFalse("CustomFactor call" in history[int(len(history)/2):])


Review comment (suggestion):
I would just call `pipeline_output` here. Storing the result on `context.results` makes me think that that's somehow relevant to the behavior under test here.

Code after:

        class TestFactor(CustomFactor):
            inputs = ()

            # window_length doesn't actually matter for this test case
            window_length = 1

            def compute(self, today, assets, out):
                trace.append("CustomFactor call")

        def initialize(context):
            pipeline = attach_pipeline(Pipeline(), 'my_pipeline')
            test_factor = TestFactor()
            pipeline.add(test_factor, 'test_factor')

        def before_trading_start(context, data):
            trace.append("BTS call")
            pipeline_output('my_pipeline')

        self.run_algorithm(
            initialize=initialize,
            before_trading_start=before_trading_start,
            sim_params=SimulationParameters(
                start_session=pd.Timestamp('2014-12-29', tz='utc'),
                end_session=pd.Timestamp('2014-12-31', tz='utc'),
                data_frequency='daily',
                emission_rate='daily',
                trading_calendar=self.trading_calendar,
            )
        )

        # All pipeline computation calls should occur before any BTS calls,
        # and the algorithm is being run for 3 days, so the first 3 calls
        # should be to the custom factor and the next 3 calls should be to BTS
        expected_result = ["CustomFactor call"] * 3 + ["BTS call"] * 3
        self.assertEqual(trace, expected_result)
Repository: quantopian/zipline
File: zipline/pipeline/visualize.py (Python)

Code before:
    """
    try:
        import IPython.display as display
    except ImportError:
        raise NoIPython("IPython is not installed.  Can't display graph.")

    if format == 'svg':
        display_cls = display.SVG
    elif format in ("jpeg", "png"):
        display_cls = partial(display.Image, format=format, embed=True)

    out = BytesIO()
    _render(g, out, format, include_asset_exists=include_asset_exists)
    return display_cls(data=out.getvalue())


def writeln(f, s):
    f.write((s + '\n').encode('utf-8'))


def fmt(obj):
    if isinstance(obj, Term):
        if hasattr(obj, 'short_repr'):
            r = obj.short_repr()
        else:
            r = type(obj).__name__
    else:
        r = obj
    return '"%s"' % r


def add_term_node(f, term):
    declare_node(f, id(term), attrs_for_node(term))


def declare_node(f, name, attributes):
    writeln(f, "{0} {1};".format(name, format_attrs(attributes)))


def add_edge(f, source, dest):
    writeln(f, "{0} -> {1};".format(source, dest))


def attrs_for_node(term, **overrides):
    attrs = {
        'shape': 'box',
        'colorscheme': 'pastel19',
        'style': 'filled',
        'label': fmt(term),
    }
    if isinstance(term, BoundColumn):

Review comment (bug):
This formatting is broken for strings. If you want to repr strings with quotes use `%r` I wouldn't block on this comment though.

Code after:
    """
    try:
        import IPython.display as display
    except ImportError:
        raise NoIPython("IPython is not installed.  Can't display graph.")

    if format == 'svg':
        display_cls = display.SVG
    elif format in ("jpeg", "png"):
        display_cls = partial(display.Image, format=format, embed=True)

    out = BytesIO()
    _render(g, out, format, include_asset_exists=include_asset_exists)
    return display_cls(data=out.getvalue())


def writeln(f, s):
    f.write((s + '\n').encode('utf-8'))


def fmt(obj):
    if isinstance(obj, Term):
        r = obj.graph_repr()
    else:
        r = obj
    return '"%s"' % r


def add_term_node(f, term):
    declare_node(f, id(term), attrs_for_node(term))


def declare_node(f, name, attributes):
    writeln(f, "{0} {1};".format(name, format_attrs(attributes)))


def add_edge(f, source, dest):
    writeln(f, "{0} -> {1};".format(source, dest))


def attrs_for_node(term, **overrides):
    attrs = {
        'shape': 'box',
        'colorscheme': 'pastel19',
        'style': 'filled',
        'label': fmt(term),
    }
    if isinstance(term, BoundColumn):
        attrs['fillcolor'] = '1'
    if isinstance(term, Factor):
        attrs['fillcolor'] = '2'
Repository: quantopian/zipline
File: zipline/algorithm.py (Python)

Code before:
        """
        Get a default pipeline domain for algorithms running on ``calendar``.

        This will be used to infer a domain for pipelines that only use generic
        datasets when running in the context of a TradingAlgorithm.
        """
        return _DEFAULT_DOMAINS.get(calendar.name, domain.GENERIC)

    ##################
    # End Pipeline API
    ##################

    @classmethod
    def all_api_methods(cls):
        """
        Return a list of all the TradingAlgorithm API methods.
        """
        return [
            fn for fn in itervalues(vars(cls))
            if getattr(fn, 'is_api_method', False)
        ]


# Map from calendar name to default domain for that calendar.
_DEFAULT_DOMAINS = {
    'XBRU': domain.BE_EQUITIES,
    'XTSE': domain.CA_EQUITIES,
    'XSWX': domain.CH_EQUITIES,
    'XFRA': domain.DE_EQUITIES,
    'XPAR': domain.FR_EQUITIES,
    'XLON': domain.GB_EQUITIES,
    'XTKS': domain.JP_EQUITIES,
    'XAMS': domain.NL_EQUITIES,
    'XLIS': domain.PT_EQUITIES,
    'XNYS': domain.US_EQUITIES,
}


Review comment (suggestion):
If we have a list of built-in domains defined in the domain module, then this can probably turn into something like:

```python
_DEFAULT_DOMAINS = {d.calendar_name: d for d in domain.BUILT_IN_DOMAINS}
```
(We should make `calendar_name` public if we do ^).

Code after:
        """
        Get a default pipeline domain for algorithms running on ``calendar``.

        This will be used to infer a domain for pipelines that only use generic
        datasets when running in the context of a TradingAlgorithm.
        """
        return _DEFAULT_DOMAINS.get(calendar.name, domain.GENERIC)

    ##################
    # End Pipeline API
    ##################

    @classmethod
    def all_api_methods(cls):
        """
        Return a list of all the TradingAlgorithm API methods.
        """
        return [
            fn for fn in itervalues(vars(cls))
            if getattr(fn, 'is_api_method', False)
        ]


# Map from calendar name to default domain for that calendar.
_DEFAULT_DOMAINS = {d.calendar_name: d for d in domain.BUILT_IN_DOMAINS}
Repository: quantopian/zipline
File: zipline/data/hdf5_daily_bars.py (Python)

Code before:
FIELDS = (OPEN, HIGH, LOW, CLOSE, VOLUME)

DAY = 'day'
SID = 'sid'

START_DATE = 'start_date'
END_DATE = 'end_date'


DEFAULT_SCALING_FACTORS = {
    # Retain 3 decimal places for prices.
    OPEN: 1000,
    HIGH: 1000,
    LOW: 1000,
    CLOSE: 1000,
    # Volume is expected to be a whole integer.
    VOLUME: 1,
}


def coerce_to_uint32(a, field, scaling_factor):
    """
    Returns a copy of the array as uint32, applying a scaling factor to
    maintain precision if supplied.
    """
    return (a * scaling_factor).round().astype('uint32')


def days_and_sids_for_frames(frames):
    """
    Returns the date index and sid columns shared by a list of dataframes,
    ensuring they all match.

    Parameters
    ----------
    frames : list[pd.DataFrame]
        A list of dataframes indexed by day, with a column per sid.

    Returns
    -------
    days : np.array[datetime64[ns]]
        The days in these dataframes.
    sids : np.array[int64]
        The sids in these dataframes.

    Raises
    ------
    ValueError
        If the dataframes passed are not all indexed by the same days
        and sids.
    """

Review comment (suggestion):
This looks good to me, though it appears that `field` is unused in this function? (I think we used to just pass field and look up the value in DEFAULT_SCALING_FACTORS).

Code after:
FIELDS = (OPEN, HIGH, LOW, CLOSE, VOLUME)

DAY = 'day'
SID = 'sid'

START_DATE = 'start_date'
END_DATE = 'end_date'


DEFAULT_SCALING_FACTORS = {
    # Retain 3 decimal places for prices.
    OPEN: 1000,
    HIGH: 1000,
    LOW: 1000,
    CLOSE: 1000,
    # Volume is expected to be a whole integer.
    VOLUME: 1,
}


def coerce_to_uint32(a, scaling_factor):
    """
    Returns a copy of the array as uint32, applying a scaling factor to
    maintain precision if supplied.
    """
    return (a * scaling_factor).round().astype('uint32')


def days_and_sids_for_frames(frames):
    """
    Returns the date index and sid columns shared by a list of dataframes,
    ensuring they all match.

    Parameters
    ----------
    frames : list[pd.DataFrame]
        A list of dataframes indexed by day, with a column per sid.

    Returns
    -------
    days : np.array[datetime64[ns]]
        The days in these dataframes.
    sids : np.array[int64]
        The sids in these dataframes.

    Raises
    ------
    ValueError
        If the dataframes passed are not all indexed by the same days
        and sids.
    """
Repository: quantopian/zipline
File: zipline/algorithm.py (Python)

Code before:
        return _DEFAULT_FETCH_CSV_COUNTRY_CODES.get(calendar.name)

    ##################
    # End Pipeline API
    ##################

    @classmethod
    def all_api_methods(cls):
        """
        Return a list of all the TradingAlgorithm API methods.
        """
        return [
            fn for fn in itervalues(vars(cls))
            if getattr(fn, 'is_api_method', False)
        ]


# Map from calendar name to default domain for that calendar.
_DEFAULT_DOMAINS = {d.calendar_name: d for d in domain.BUILT_IN_DOMAINS}
# Map from calendar name to default country code for that calendar.
_DEFAULT_FETCH_CSV_COUNTRY_CODES = {
    d.calendar_name: d.country_code for d in domain.BUILT_IN_DOMAINS
}
# Include us_futures, which doesn't have a pipeline domain.
_DEFAULT_FETCH_CSV_COUNTRY_CODES['us_futures'] = 'US'


Review comment (suggestion):
Maybe this could just be called `_DEFAULT_CALENDAR_COUNTRY_CODES` or somesuch, since it isn't necessarily tied to `fetch_csv`.

Code after:

        This will be used to disambiguate symbol lookups for fetch_csv calls if
        our asset db contains entries with the same ticker spread across
        multiple
        """
        return _DEFAULT_FETCH_CSV_COUNTRY_CODES.get(calendar.name)

    ##################
    # End Pipeline API
    ##################

    @classmethod
    def all_api_methods(cls):
        """
        Return a list of all the TradingAlgorithm API methods.
        """
        return [
            fn for fn in itervalues(vars(cls))
            if getattr(fn, 'is_api_method', False)
        ]


# Map from calendar name to default domain for that calendar.
_DEFAULT_DOMAINS = {d.calendar_name: d for d in domain.BUILT_IN_DOMAINS}
# Map from calendar name to default country code for that calendar.
_DEFAULT_FETCH_CSV_COUNTRY_CODES = {
    d.calendar_name: d.country_code for d in domain.BUILT_IN_DOMAINS
}
# Include us_futures, which doesn't have a pipeline domain.
_DEFAULT_FETCH_CSV_COUNTRY_CODES['us_futures'] = 'US'
Repository: quantopian/zipline
File: zipline/errors.py (Python)

Code before:
Possible options: {options}
    """.strip()


class MultipleSymbolsFoundForFuzzySymbol(ZiplineError):
    """
    Raised when a fuzzy symbol lookup is not resolvable without additional
    information.
    """
    msg = dedent("""\
        Multiple symbols were found fuzzy matching the name '{symbol}'. Use
        the as_of_date and/or country_code arguments to to specify the date
        and country for the symbol-lookup.

        Possible options: {options}
    """)


class SameSymbolUsedAcrossCountries(ZiplineError):
    """
    Raised when a symbol() call contains a symbol that is used in more than
    one country and is thus not resolvable without a country_code.
    """
    msg = dedent("""\
        The symbol '{symbol}' is used in more than one country. Use the
        country_code' argument to to specify the country.

        Possible options: {options}
    """)


class SymbolNotFound(ZiplineError):
    """
    Raised when a symbol() call contains a non-existant symbol.
    """
    msg = """
Symbol '{symbol}' was not found.
""".strip()


class RootSymbolNotFound(ZiplineError):
    """
    Raised when a lookup_future_chain() call contains a non-existant symbol.
    """
    msg = """
Root symbol '{root_symbol}' was not found.
""".strip()


class ValueNotFoundForField(ZiplineError):
    """

Review comment (suggestion):
I think the `'` is cruft?
```suggestion
        country_code argument to to specify the country.
```

Code after:
Possible options: {options}
    """.strip()


class MultipleSymbolsFoundForFuzzySymbol(MultipleSymbolsFound):
    """
    Raised when a fuzzy symbol lookup is not resolvable without additional
    information.
    """
    msg = dedent("""\
        Multiple symbols were found fuzzy matching the name '{symbol}'. Use
        the as_of_date and/or country_code arguments to to specify the date
        and country for the symbol-lookup.

        Possible options: {options}
    """)


class SameSymbolUsedAcrossCountries(MultipleSymbolsFound):
    """
    Raised when a symbol() call contains a symbol that is used in more than
    one country and is thus not resolvable without a country_code.
    """
    msg = dedent("""\
        The symbol '{symbol}' is used in more than one country. Use the
        country_code argument to to specify the country.

        Possible options by country: {options}
    """)


class SymbolNotFound(ZiplineError):
    """
    Raised when a symbol() call contains a non-existant symbol.
    """
    msg = """
Symbol '{symbol}' was not found.
""".strip()


class RootSymbolNotFound(ZiplineError):
    """
    Raised when a lookup_future_chain() call contains a non-existant symbol.
    """
    msg = """
Root symbol '{root_symbol}' was not found.
""".strip()


class ValueNotFoundForField(ZiplineError):
    """
Repository: quantopian/zipline
File: zipline/pipeline/domain.py (Python)

Code before:
BUILT_IN_DOMAINS = [
    AT_EQUITIES,
    AU_EQUITIES,
    BE_EQUITIES,
    BR_EQUITIES,
    CA_EQUITIES,
    CH_EQUITIES,
    CN_EQUITIES,
    DE_EQUITIES,
    DK_EQUITIES,
    ES_EQUITIES,
    FI_EQUITIES,
    FR_EQUITIES,
    GB_EQUITIES,
    HK_EQUITIES,
    IE_EQUITIES,
    IN_EQUITIES,
    IT_EQUITIES,
    JP_EQUITIES,
    NL_EQUITIES,
    NO_EQUITIES,
    NZ_EQUITIES,
    PT_EQUITIES,
    SE_EQUITIES,
    SG_EQUITIES,
    SK_EQUITIES,
    US_EQUITIES,
]


def infer_domain(terms):
    """
    Infer the domain from a collection of terms.

    The algorithm for inferring domains is as follows:

    - If all input terms have a domain of GENERIC, the result is GENERIC.

    - If there is exactly one non-generic domain in the input terms, the result
      is that domain.

    - Otherwise, an AmbiguousDomain error is raised.

    Parameters
    ----------
    terms : iterable[zipline.pipeline.term.Term]

    Returns
    -------
    inferred : Domain or NotSpecified


Review comment (suggestion):
You should update `test_domain.py` with this change as well. See https://github.com/quantopian/zipline/pull/2341

Code after:
BUILT_IN_DOMAINS = [
    AT_EQUITIES,
    AU_EQUITIES,
    BE_EQUITIES,
    BR_EQUITIES,
    CA_EQUITIES,
    CH_EQUITIES,
    CN_EQUITIES,
    DE_EQUITIES,
    DK_EQUITIES,
    ES_EQUITIES,
    FI_EQUITIES,
    FR_EQUITIES,
    GB_EQUITIES,
    HK_EQUITIES,
    IE_EQUITIES,
    IN_EQUITIES,
    IT_EQUITIES,
    JP_EQUITIES,
    KR_EQUITIES,
    NL_EQUITIES,
    NO_EQUITIES,
    NZ_EQUITIES,
    PT_EQUITIES,
    SE_EQUITIES,
    SG_EQUITIES,
    US_EQUITIES,
]


def infer_domain(terms):
    """
    Infer the domain from a collection of terms.

    The algorithm for inferring domains is as follows:

    - If all input terms have a domain of GENERIC, the result is GENERIC.

    - If there is exactly one non-generic domain in the input terms, the result
      is that domain.

    - Otherwise, an AmbiguousDomain error is raised.

    Parameters
    ----------
    terms : iterable[zipline.pipeline.term.Term]

    Returns
    -------
    inferred : Domain or NotSpecified
Repository: quantopian/zipline
File: zipline/testing/core.py (Python)

Code before:
    """
    @abstractproperty
    def _reader_cls(self):
        raise NotImplementedError('_reader')

    @abstractmethod
    def _write(self, cal, days, path, data):
        raise NotImplementedError('_write')

    def __init__(self, cal, days, data, path=None):
        super(_TmpBarReader, self).__init__(path=path)
        self._cal = cal
        self._days = days
        self._data = data

    def __enter__(self):
        tmpdir = super(_TmpBarReader, self).__enter__()
        try:
            self._write(
                self._cal,
                self._days,
                tmpdir.path,
                self._data,
            )
            return self._reader_cls(tmpdir.path)
        except BaseException:  # Gotta catch 'em all
            self.__exit__(None, None, None)
            raise


class tmp_bcolz_equity_minute_bar_reader(_TmpBarReader):
    """A temporary BcolzMinuteBarReader object.

    Parameters
    ----------
    cal : TradingCalendar
        The trading calendar for which we're writing data.
    days : pd.DatetimeIndex
        The days to write for.
    data : iterable[(int, pd.DataFrame)]
        The data to write.
    path : str, optional
        The path to the directory to write the data into. If not given, this
        will be a unique name.

    See Also
    --------
    tmp_bcolz_equity_daily_bar_reader
    """
    _reader_cls = BcolzMinuteBarReader
    _write = staticmethod(write_bcolz_minute_data)

Review comment (suggestion):
I almost don't even think this is a reasonable change. If you actually want to catch all exceptions `except:` does that and is pretty clear about it.

Code after:
    """
    @abstractproperty
    def _reader_cls(self):
        raise NotImplementedError('_reader')

    @abstractmethod
    def _write(self, cal, days, path, data):
        raise NotImplementedError('_write')

    def __init__(self, cal, days, data, path=None):
        super(_TmpBarReader, self).__init__(path=path)
        self._cal = cal
        self._days = days
        self._data = data

    def __enter__(self):
        tmpdir = super(_TmpBarReader, self).__enter__()
        try:
            self._write(
                self._cal,
                self._days,
                tmpdir.path,
                self._data,
            )
            return self._reader_cls(tmpdir.path)
        except BaseException:  # Clean up even on KeyboardInterrupt
            self.__exit__(None, None, None)
            raise


class tmp_bcolz_equity_minute_bar_reader(_TmpBarReader):
    """A temporary BcolzMinuteBarReader object.

    Parameters
    ----------
    cal : TradingCalendar
        The trading calendar for which we're writing data.
    days : pd.DatetimeIndex
        The days to write for.
    data : iterable[(int, pd.DataFrame)]
        The data to write.
    path : str, optional
        The path to the directory to write the data into. If not given, this
        will be a unique name.

    See Also
    --------
    tmp_bcolz_equity_daily_bar_reader
    """
    _reader_cls = BcolzMinuteBarReader
    _write = staticmethod(write_bcolz_minute_data)
Repository: quantopian/zipline
File: tests/pipeline/test_adjusted_array.py (Python)

Code before:
        (
            True,
            {
                1: [Float64Multiply(0, 3, 0, 0, 4.0),
                    Float64Overwrite(2, 3, 0, 0, 1.0)],
                3: [Float64Multiply(0, 3, 0, 0, 4.0)],
            },
        ),
        # Append new adjustment to front of the list if index already contains
        # adjustments.
        (
            False,
            {
                1: [Float64Overwrite(2, 3, 0, 0, 1.0),
                    Float64Multiply(0, 3, 0, 0, 4.0)],
                3: [Float64Multiply(0, 3, 0, 0, 4.0)],
            },
        )
    ])
    def test_append_adjustments(self, append_back, expected_adjustments):
        adjustments = {
            1: [Float64Multiply(0, 3, 0, 0, 4.0)],
        }
        adjustments_to_append = {
            1: [Float64Overwrite(2, 3, 0, 0, 1.0)],
            3: [Float64Multiply(0, 3, 0, 0, 4.0)],
        }

        data = arange(30, dtype=float).reshape(6, 5)
        adjusted_array = AdjustedArray(data, adjustments, float('nan'))

        adjusted_array.append_adjustments(
            adjustments_to_append, append_back=append_back
        )
        self.assertEqual(adjusted_array.adjustments, expected_adjustments)


Review comment (bug):
Is there a reason we're using the same values for the multiplies here? That makes it theoretically possible for us to write a bug where the multiplies get swapped, but we don't notice because they're the same.

Code after:
    D = Float64Overwrite(0, 3, 0, 0, 4.0)
    E = Float64Overwrite(0, 0, 1, 1, 3.7)
    F = Float64Multiply(0, 4, 3, 3, 10.0)
    G = Float64Overwrite(5, 5, 4, 4, 1.7)
    H = Float64Multiply(0, 4, 2, 2, 0.99)
    S = Float64Multiply(0, 1, 4, 4, 5.06)

    @parameterized.expand([(
        # Initial adjustments
        {
            1: [A, B],
            2: [C],
            4: [D],
        },

        # Adjustments to add
        {
            1: [E],
            2: [F, G],
            3: [H, S],
        },

        # Expected adjustments with 'append'
        {
            1: [A, B, E],
            2: [C, F, G],
            3: [H, S],
            4: [D],
        },

        # Expected adjustments with 'prepend'
        {
            1: [E, A, B],
            2: [F, G, C],
            3: [H, S],
            4: [D],
        },
    )])
    def test_update_adjustments(self,
                                initial_adjustments,
                                adjustments_to_add,
                                expected_adjustments_with_append,
                                expected_adjustments_with_prepend):
        methods = ['append', 'prepend']
        expected_outputs = [
            expected_adjustments_with_append, expected_adjustments_with_prepend
        ]

        for method, expected_output in zip(methods, expected_outputs):
            data = arange(30, dtype=float).reshape(6, 5)
            adjusted_array = AdjustedArray(