Query historic_agg either minute or day in parallel
        for multiple symbols, and return in dict.

        symbols: list[str]
        size:    str ('day', 'minute')
        _from:   str or pd.Timestamp
        to:      str or pd.Timestamp
        limit:   str or int

        return: dict[str -> pd.DataFrame]

def _symbol_bars(
            self,
            symbols,
            size,
            _from=None,
            to=None,
            limit=None):
        '''
        Query historic_agg either minute or day in parallel
        for multiple symbols, and return in dict.

        symbols: list[str]
        size:    str ('day', 'minute')
        _from:   str or pd.Timestamp
        to:      str or pd.Timestamp
        limit:   str or int

        return: dict[str -> pd.DataFrame]
        '''
        assert size in ('day', 'minute')

        # temp workaround for less bars after masking by
        # market hours
        query_limit = limit
        if query_limit is not None:
            query_limit *= 2

        @skip_http_error((404, 504))
        def fetch(symbol):
            df = self._api.polygon.historic_agg(
                size, symbol, _from, to, query_limit).df

            # zipline -> right label
            # API result -> left label (beginning of bucket)
            if size == 'minute':
                df.index += pd.Timedelta('1min')

                # mask out bars outside market hours
                mask = self._cal.minutes_in_range(
                    df.index[0], df.index[-1],
                ).tz_convert(NY)
                df = df.reindex(mask)

            if limit is not None:
                df = df.iloc[-limit:]
            return df

        return parallelize(fetch)(symbols)