DEV Community 👩‍💻👨‍💻

Cover image for Converting a While Loop into a Function using an Iterator in Python
Neal Chambers
Neal Chambers

Posted on

Converting a While Loop into a Function using an Iterator in Python

I've been working on tax software to help you calculate the taxes on all those hard-earned losses gains you made on crypto over the last year. It has actually been a very interesting learning experience since there are so many exchanges out there and every one of them seems to have their own special way of building out their API. Earlier, I covered an amazing python module that helps bring some sanity to the chaos - CCXT.

And it has helped speed things along with software development. I just finished a plugin for Binance.com that allows users to pull in most of the information they need to retrieve and process their transaction information from that exchange. However, only Binance.com is available currently, and I would like to make it easier to build out more plugins for more exchanges more quickly.
That's why we decided to abstract out the CCXT goodness into an abstract class that other users can then extend and add exchange-specific logic to. Easy enough, right?

Every exchange seems to have a different way to do things though. One issue with retrieving transaction data from a server is that there could potentially be hundreds if not thousands of transactions that need to be downloaded. That obviously takes up a lot of resources that an exchange wouldn't like to tie up. So, it breaks larger pieces of data into pieces called pages. Moving through this data is done by pagination.

An illustration of the advantages of pagination. Woman is holding a scroll of paper. The label "Bad" is underneath it. To the right, there are sheets of paper and the label "Good" is underneath it.

Well, every exchange seems to have its own way of pagination. Binance.com uses date-based pagination while Coinbase uses page-based pagination. Still other exchanges use id-based pagination. All of these pagination methods are essentially while loops that keep pulling data from the server until all the data has been retrieved. Parameters will have to be adjusted on each call of the endpoint based on the data previously retrieved.

My first idea was to build a class that resolved the condition of the while loop. It would take the previous result set, evaluate and return a bool. In that same function, it would readjust the values of the parameters to pull the new set of data.

def evaluate_loop_expression(self, current_results: Any) -> bool:

        # First time the loop is executed, results will be None
        if current_results is None:
            return True

        # Did we reach the end of this market?
        end_market: bool = False

        if len(current_results):
            # All times are inclusive
            self.__since = current_results[len(current_results) - 1][_TIMESTAMP] + 1
        else:
            end_market = True

        if end_market and self.has_more_markets():
            self.__since = self.__exchange_start_time
            self.next_market()
            return True

        return False
Enter fullscreen mode Exit fullscreen mode

Basically, it shifts self.__since, which marks the first timestamp to start pulling data from forward to the end of the current_results. Or, if we pulled everything from this market (eg. BTCUSD) we can move on to the next market (eg. ETHUSD). Then, it returns True if there is more data to pull or False if not.

This would be the condition evaluated in order to keep the while loop going:

while self.evaluate_loop_expression(results):
  # Fetch a new set of results with changed parameters.
  results = client.fetchTrades(self.parameters)
Enter fullscreen mode Exit fullscreen mode

This worked okay. However, one function was serving two purposes and that is a bit of no-no in programming. So, how do you make this more readable and keep the same function?

A lot of things are happening here. We need to change the fields being used to pull data based on previous data and then break out of the while loop if there is no more data to be retrieved.

A simple iterator doesn't seem to work since we need to edit what is being returned by the iterator depending on what the previous results were.

To make something iterable in Python, we need to define __iter__, which will initialize and return an iterator.

An iterator is a class that implements __next__, which returns the next set of objects being iterated over.

Although an iterator can be iterable (implement __iter__), it doesn't have to be. For example, str is iterable and returns an iterator that iterates over the characters in the string.

So, what we need here is an iterator that we can update on each loop.

What I ended up with was 3 classes:
1) a pagination detail set that was iterable (returns an iterator when iter() is called on it)
2) a pagination iterator returned from 1) that returns new pagination details when next() is called on it.
3) a NamedTuple of the pagination details.

This is a lot more work, but it has a much cleaner design.

Here is the DateBasedPaginationDetailSet version of 1):

class AbstractPaginationDetailSet:
    def __iter__(self) -> "AbstractPaginationDetailsIterator":
        raise NotImplementedError("Abstract method")


class DateBasedPaginationDetailSet(AbstractPaginationDetailSet):
    def __init__(
        self,
        exchange_start_time: int,
        limit: Optional[int] = None,
        markets: Optional[List[str]] = None,
        params: Optional[Dict[str, Union[int, str, None]]] = None,
        window: Optional[int] = None,
    ) -> None:

        super().__init__()
        self.__exchange_start_time: int = exchange_start_time
        self.__limit: Optional[int] = limit
        self.__markets: Optional[List[str]] = markets
        self.__params: Optional[Dict[str, Union[int, str, None]]] = params
        self.__window: Optional[int] = window

    def __iter__(self) -> "DateBasedPaginationDetailsIterator":
        return DateBasedPaginationDetailsIterator(
            self.__exchange_start_time,
            self.__limit,
            self.__markets,
            self.__params,
            self.__window,
        )
Enter fullscreen mode Exit fullscreen mode

Notice that this essentially only contains the details needed to build the iterator. This makes it so all we need to do is pass 2-3 parameters to initialize it and the plugin pulling the data will do the rest.

And the iterator that __iter__ returns:

class AbstractPaginationDetailsIterator:
    def __init__(self, limit: Optional[int], markets: Optional[List[str]] = None, params: Optional[Dict[str, Union[int, str, None]]] = None) -> None:
        self.__limit: Optional[int] = limit
        self.__markets: Optional[List[str]] = markets
        self.__market_count: int = 0
        self.__params: Optional[Dict[str, Union[int, str, None]]] = params

    def _get_market(self) -> Optional[str]:
        return self.__markets[self.__market_count] if self.__markets else None

    def _has_more_markets(self) -> bool:
        return self.__market_count <= len(self.__markets) if self.__markets else False

    def _next_market(self) -> None:
        self.__market_count += 1

    def _get_limit(self) -> Optional[int]:
        return self.__limit

    def _get_params(self) -> Optional[Dict[str, Union[int, str, None]]]:
        return self.__params

    def _get_since(self) -> Optional[int]:
        return None

    def update_fetched_elements(self, current_results: Any) -> None:
        raise NotImplementedError("Abstract method")

    def __next__(self) -> PaginationDetails:
        raise NotImplementedError("Abstract method")


class DateBasedPaginationDetailsIterator(AbstractPaginationDetailsIterator):
    def __init__(
        self,
        exchange_start_time: int,
        limit: Optional[int] = None,
        markets: Optional[List[str]] = None,
        params: Optional[Dict[str, Union[int, str, None]]] = None,
        window: Optional[int] = None,
    ) -> None:

        super().__init__(limit, markets, params)
        self.__end_of_data = False
        self.__since: int = exchange_start_time
        self.__exchange_start_time: int = exchange_start_time
        self.__now: int = int(datetime.now().timestamp()) * _MS_IN_SECOND
        self.__window: int = window if window else _DEFAULT_WINDOW

    def update_fetched_elements(self, current_results: Any) -> None:

        end_of_market: bool = False

        # Update Since if needed otherwise end_of_market
        if len(current_results):
            # All times are inclusive
            self.__since = current_results[len(current_results) - 1][_TIMESTAMP] + 1
        elif self.__window:
            self.__since += self.__window

        if self.__since > self.__now:
            end_of_market = True

        if end_of_market and self._has_more_markets():
            # we have reached the end of one market, now let's move on to the next
            self.__since = self.__exchange_start_time
            self._next_market()
        else:
            self.__end_of_data = True

    def _is_end_of_data(self) -> bool:
        return self.__end_of_data

    def _get_since(self) -> int:
        return self.__since

    def _get_end_of_window(self) -> int:
        return self.__since + self.__window

    def __next__(self) -> PaginationDetails:
        while not self._is_end_of_data():
            return PaginationDetails(
                symbol=self._get_market(),
                since=self._get_since(),
                limit=self._get_limit(),
                params=self._get_params(),
            )
        raise StopIteration(self)
Enter fullscreen mode Exit fullscreen mode

Here, I added a update_fetched_elements to shift the window of time backward if we pulled the record limit (ie. we didn't pull all the records with one call). If the number of retrieve records is under the limit, I just need to move on to the next market if one exists.

And finally the NamedTuple:

class PaginationDetails(NamedTuple):
    symbol: Optional[str]
    since: Optional[int]
    limit: Optional[int]
    params: Optional[Dict[str, Union[int, str, None]]]
Enter fullscreen mode Exit fullscreen mode

Now, we have a custom iterator to use when pulling data via the REST API of the exchange.

Top comments (0)

🌱 DEV runs on 100% open source code that we started called Forem.

You can contribute to the codebase or host your own.