DEV Community: Neal Chambers

Dealing with Large CSV files in Python by Chunking

Neal Chambers — Tue, 07 Mar 2023 00:59:00 +0000

I'm continuing to work on Dali-RP2, which helps collect data from crypto exchanges and then matches that with historical pricing in order to prepare one's taxes. Recently, I ran into some issues pulling prices for crypto assets. Typically, to find a price for a given crypto asset for a given time, you can just access the public REST API of an exchange and it will return a OHLCV bar, which contains the Opening price, High price, Low price, Closing price and Volume.

That is all fine and good, but some exchanges can be quite stingy with their API. This is understandable, since you don't want users spamming a server for data. However, if you happen to have 1000 or so transactions you want to price it can take a long long time to price everything.

For instance, Kraken requires around 5 seconds between each call to its API. For 1000 transactions, this would take around 7 hours to pull all the pricing I need. Luckily, they have downloadable CSV files that cover everything but the last quarter. So, you can just download the files and save them to cache, right?

Unfortunately, this creates a massive cache that requires a lot of memory, and requires even more memory when it is saved to disk. Eventually, if you work with enough crypto assets this will cause the program to crash because it will run out of memory.

To avoid this problem, we have to chunk the csv files into smaller files that we then read prices from.

Chunk It, You must Chunk It

I was able to find a good comprehensive overview on the myriad of ways one can chunk things. From there, I adapted it for my particular situation.

def __split_process(self, csv_file: str, chunk_size: int = _CHUNK_SIZE) -> Generator[Tuple[str, List[List[str]]], None, None]:
    chunk: List[List[str]] = []

    lines = reader(csv_file.splitlines())
    position = _PAIR_START
    next_timestamp: Optional[int] = None

    for line in lines:
        if next_timestamp is None:
            next_timestamp = ((int(line[self.__TIMESTAMP_INDEX]) + chunk_size) // chunk_size) * chunk_size

        if int(line[self.__TIMESTAMP_INDEX]) % chunk_size == 0 or int(line[self.__TIMESTAMP_INDEX]) > next_timestamp:
            yield position, chunk
            if position == _PAIR_START:
                position = _PAIR_MIDDLE
            chunk = []
            next_timestamp += chunk_size
        chunk.append(line)
    if chunk:
        position = _PAIR_END
        yield position, chunk

def _split_chunks_size_n(self, file_name: str, csv_file: str, chunk_size: int = _CHUNK_SIZE) -> None:

    pair, duration_in_minutes = file_name.strip(".csv").split("_", 1)
    chunk_size *= min(int(duration_in_minutes), _MAX_MULTIPLIER)
    file_timestamp: str
    pair_start: Optional[int] = None
    pair_end: int
    pair_duration: str = pair + duration_in_minutes

    for position, chunk in self.__split_process(csv_file, chunk_size):
        file_timestamp = str((int(chunk[0][self.__TIMESTAMP_INDEX])) // chunk_size * chunk_size)
        if position == _PAIR_END:
            pair_end = int(chunk[-1][self.__TIMESTAMP_INDEX])
            if pair_start is None:
                pair_start = int(chunk[0][self.__TIMESTAMP_INDEX])
        elif position == _PAIR_START:
            pair_start = int(chunk[0][self.__TIMESTAMP_INDEX])

        chunk_filename: str = f'{pair}_{file_timestamp}_{duration_in_minutes}.{"csv.gz"}'
        chunk_filepath: str = path.join(self.__CACHE_DIRECTORY, chunk_filename)

        with gopen(chunk_filepath, "wt", encoding="utf-8", newline="") as chunk_file:
            csv_writer = writer(chunk_file)
            for row in chunk:
                csv_writer.writerow(row)

    if pair_start:
        self.__cached_pairs[pair_duration] = PairStartEnd(start=pair_start, end=pair_end)

Let's take a look at that line by line. First in the __split_process function. We want to be able to tell _split_chunk_size_n the start and end times of the data to make it easier to retrieve the data later. We don't want to waste time attempting to access a time frame that doesn't exist. So, we keep track of where we are in each CSV file with the position variable.

We want to yield the chunk when the timestamp (stored in the __TIMESTAMP_INDEX column) reaches a multiplier of the chunk_size or goes above it. To do that, we need to find the next multiplier of the chunk_size.

next_timestamp = ((int(line[self.__TIMESTAMP_INDEX]) + chunk_size) // chunk_size) * chunk_size

Above, we first add the chunk_size to the current timestamp in order to get a timestamp that is in the next chunk. We then "floor" the timestamp by using integer division // to divide it by chunk_size and then multiply it by the chunk_size. This will round the number down to the nearest chunk_size.

Here is a simple example of what we are doing if we had a chunk_size of 10.

36 // 10 = 3
3 * 10 = 30

When we reach the next chunk we yield the position and chunk, reset the chunk, and set the next_timestamp to the next chunk.

if int(line[self.__TIMESTAMP_INDEX]) % chunk_size == 0 or int(line[self.__TIMESTAMP_INDEX]) > next_timestamp:
    yield position, chunk
    if position == _PAIR_START:
        position = _PAIR_MIDDLE
    chunk = []
    next_timestamp += chunk_size

Otherwise, we keep appending lines to the chunk. Note that the timestamp that is a multiple of the chunk_size or just beyond it is the first bar in the next chunk.

When we reach the end of the file we yield the last chunk and tell _split_chunk_size_n our position.

if chunk:
    position = _PAIR_END
    yield position, chunk

Standardizing File Size

That worked pretty well, but I ended up with a lot of files. I had around 460 files for each pair or asset I wanted to price. This isn't too bad, but I noticed files with longer durations were smaller. That seemed like wasted space to me, so I wanted to make the files all around the same size.

The CSV files list pricing bars (OHLCV) of different durations. There are 6 files in total - 1 minute, 5 minute, 15 minute, 60 minute, 12 hour, and 24 hour. In theory, the 5 minute files would only contain one fifth of the bars that the 1 minute files did, and so forth and so on. Based on this, I decided to add a multiplier to the chunk_size, which would increase the timespan covered by the larger durations.

pair, duration_in_minutes = file_name.strip(".csv").split("_", 1)
chunk_size *= min(int(duration_in_minutes), _MAX_MULTIPLIER)

This reduced the number of files generated to only about 140 or so per asset I want to price.

Summary

I've just started working with Python. Before that, I was working with Ruby on Rails where everything just goes into a database. This was my first time to deal with a large chunk of data and how to manage it. On the surface of it, it seems like a rather primitive way to handle it, but it ended up being very fast for my needs. I'm able to price numerous assets in a relatively short period of time, and if need be fall back on the REST API for the last quarter.

What do you think? Do you have other ways you deal with large pieces of data like this? I'd love to hear how others handle this. Let me know in the comments below.

Basic Troubleshooting with Docker and Ruby on Rails

Neal Chambers — Fri, 24 Feb 2023 00:58:55 +0000

I just started volunteering with the Open Food Network to re-familiarize myself with Ruby on Rails. They are an open source, non-profit, online marketplace for food stores. They provide software to link up independent farmers with hubs and distributors and it all runs on Ruby on Rails.

Like a lot of web apps, they have a Docker setup to make things easier. Docker helps eliminate the "it works on my computer" problem that can plague other development efforts when certain software packages are not installed, or the wrong version of them is installed.

If you are not familiar, Docker is a piece of software that 'docks' system images on your computer that run a certain version of an OS. They can be customized to install certain packages when you build them. You can them perform tests with RSpec on them or boot up a server and work with the development version of the site and see if your code has the desired effect.

I found troubleshooting a little difficult at first under this setup since you need to pass commands to Docker. You can't simply use bash or any other terminal to run them. Also, the server has to be up and running in order to run those commands. This can lead to a lot of frustration.

Connecting to the DB

My first major hurdle was to connect to the database so I can see more directly what was happening 'under the hood'. Luckily, Docker, by default will forward it's ports to the localhost, so you can access it as if it were running on your own system.

If that is not the case, and the Docker image is not being forwarded to the localhost for whatever reason, you can still find the ip of a Docker image pretty simply. First, we take a look at all the images running locally.

docker ps

This will give you a list of images that are currently running that may look a little something like this:

CONTAINER ID   IMAGE                     COMMAND                  CREATED        STATUS          PORTS                                       NAMES
697b8685d509   openfoodnetwork_web       "bash -c 'wait-for-i…"   12 hours ago   Up 10 seconds   0.0.0.0:3000->3000/tcp, :::3000->3000/tcp   openfoodnetwork_web_1
8ec13ea014c2   openfoodnetwork_worker    "bash -c 'wait-for-i…"   12 hours ago   Up 11 seconds                                               openfoodnetwork_worker_1
9171e4c62f72   postgres:10.19            "docker-entrypoint.s…"   12 hours ago   Up 31 minutes   0.0.0.0:5432->5432/tcp, :::5432->5432/tcp   openfoodnetwork_db_1
5015b1895607   openfoodnetwork_webpack   "./bin/webpack-dev-s…"   12 hours ago   Up 11 seconds   0.0.0.0:3035->3035/tcp, :::3035->3035/tcp   openfoodnetwork_webpack_1
9067a38ec683   redis                     "docker-entrypoint.s…"   12 hours ago   Up 11 seconds   6379/tcp                                    openfoodnetwork_redis_1

From there, you can look up the ip address of any image with the following command:

docker inspect -f '{{range.NetworkSettings.Networks}}{{.IPAddress}}{{end}}' _container id_

From there you can plug in your details into your favorite database viewer. My personal favorite is Dbeaver, which is regularly updated and makes it very easy to connect to databases.

Running just the RSpec you want

Another issue I was running into is the enormous test suite that comes with Open Food Network. It can easily take 20-30 minutes to run the entire suite which is simply not possible to do for every little change you want to make. It's much better to just run the specific test you want.

To do this, you can run rspec with a specific line number. The following command will run the spec in enterprise_relationship_spec.rb starting at line 258.

bundle exec rspec spec/models/enterprise_relationship_spec.rb:258

To run it inside your Docker image, you have two options.

1. With a Script

You can create a script that sets up and tears down your Docker image every time you want to run an RSpec. The Open Food Network have just a script if you want to copy it for yourself.

Using the script above you just have to prepend the previous command with docker/run

docker/run bundle exec rspec spec/models/enterprise_relationship_spec.rb:258

2. Start Interactive Terminal

You can also start a container with an interactive terminal much like the terminal on your localhost.

docker exec -it $(docker-compose -f docker-compose.yml ps -q web) /bin/bash

Once you have the interactive terminal you can run commands normally as if you were using a terminal on your localhost.

bundle exec rspec spec/models/enterprise_relationship_spec.rb:258

Using `binding.pry` on the live server

You can use binding.pry anywhere in Rails code to stop the execution of the code and start an interactive terminal session. If you are running a rails server on your local machine, you can simply interact with the terminal window that you started the server from.

However, with a Docker image things are a little different. When you use binding.pry the server will still halt execution, but you will not be able to interact with it in anyway. So you need to attach to that image in order to run commands and see what is happening.

First, we need to find the image that is running the webserver. To do that we need to list the running Docker images.

docker ps

Again, we get a list of running images.

CONTAINER ID   IMAGE                     COMMAND                  CREATED        STATUS          PORTS                                       NAMES
697b8685d509   openfoodnetwork_web       "bash -c 'wait-for-i…"   12 hours ago   Up 10 seconds   0.0.0.0:3000->3000/tcp, :::3000->3000/tcp   openfoodnetwork_web_1
8ec13ea014c2   openfoodnetwork_worker    "bash -c 'wait-for-i…"   12 hours ago   Up 11 seconds                                               openfoodnetwork_worker_1
9171e4c62f72   postgres:10.19            "docker-entrypoint.s…"   12 hours ago   Up 31 minutes   0.0.0.0:5432->5432/tcp, :::5432->5432/tcp   openfoodnetwork_db_1
5015b1895607   openfoodnetwork_webpack   "./bin/webpack-dev-s…"   12 hours ago   Up 11 seconds   0.0.0.0:3035->3035/tcp, :::3035->3035/tcp   openfoodnetwork_webpack_1
9067a38ec683   redis                     "docker-entrypoint.s…"   12 hours ago   Up 11 seconds   6379/tcp                                    openfoodnetwork_redis_1

This time we want to attach to the running webserver. In my case, that is openfoodnetwork_web

docker attach 697b8685d509

From there, you can query local variables and pull records to see what is happening and troubleshoot the issue. I should mention that if you are running specs using either of the two methods I mentioned above, the interactive terminal will start normally in the terminal window you used to run the Rspec command.

Conclusion

Those are some of the basic tips I learned while trying to troubleshoot a Rails application running in Docker. Of course, there are numerous ways to setup and troubleshoot an app like this, but these are the ways I found the most useful.

Do you have any other tips for using Docker? Let me know in the comments below.

How to Get Started with Test Driven Development with Python

Neal Chambers — Mon, 06 Feb 2023 07:26:16 +0000

Everyone starts out programming with a simple program. Typically, some variation of "Hello World!" That's where I started at least and at first you make simple systems that do simple things. For example, a program that helps keep track of your to-do list items. At this point the code is pretty simple to read and see what is going on. If there is a bug somewhere, you can easily track it down and fix it.

Over time, you build more and more features on to your app. For example, you add the ability to put tags on your to-do items or you add the ability to export them to CSV. You keep adding more and more features that you need to use. Each time you add something new the code tends to get more and more complicated.

At this point things start to break and it takes more and more time to track down the specific problem in the code since everything has become interconnected. On top of that, one of your users just emailed you to tell you that some of the basic functionality that was available in the first version of the app is now no longer working.

How can you easily add functionality without running the risk of breaking a feature that already exists? Each time you add something, are you going to have to manually test every little feature to make sure it works?

Enter Test-Driven Development (TDD)

Test-Driven Development, usually abbreviated as TDD, helps you catch and fix bugs early in the development process, improves the quality of your code, and keeps your code from regressing, (ie. another feature stops working when you add a new one). TDD can help provide a fast feedback loop that will spot bugs before you release it to new users. In mission critical systems, tests are essential to keeping your system from going down when it is needed the most.

On top of that, it can help you get a clear understanding of what you actually want your code to do with the inputs you are providing it. TDD helps you define the system specifics so that you and your team has a clear understanding of what you are trying to build.

Step by Step: How to Write Tests for Software Development

Identify the functionality to be added to your system or, if you have already written the code, the expected functionality you want to test.
Write a test case for the functionality. If it is a large multi-step process your piece of software has to go through you might want to write one test for the whole process and one for each critical step in the process so that you can quickly isolate an issue later.
Run the test. Obviously, if you haven't written any code this will fail. If you have written code and the test fails, you will want to go back and verify if you have written the test properly.
Write/Refactor the code. Write code to fulfill the functionality. This is where writing tests for each step of the process is useful, since you can write each step and test to see if it is valid before moving on.
Repeat 4 & 5 to clean up and refactor your code. If the tests pass, ask yourself if there is anything you can do to make the process more efficient or simpler.
Continue this process until all functionality is implemented.

Keep in mind that in the real world, we all have deadlines to meet and you might not be able to repeat steps 4 & 5 enough times to write good, clean code. You might have to simply settle for what works for now and move on. That's okay because you have written the test, and later, when you have time, you will be able to revisit the code and refactor it more. For now though, you know that it works.

How to Install Pytest and Get Started

Install Pytest. If you are using a virtual environment, like virtualenv. Make sure you have it activated and simply type pip install pytest in your terminal.
Create a test directory. The industry standard is \test, but it could be anything.
Write tests. Write test functions in .py files within the test directory you created earlier. All test functions should start with the test_ prefix. We will go over an example later.
Run pytest. If you run the command pytest all tests in the current directory and subdirectory will be ran. I personally run it with --tb=native to receive tracebacks in native format as well as --verbose so that each test function is explicitly listed out.

How do you Write a Test?

Tests are pretty straightforward. You have to first start by importing the module that you would like to test. Let's imagine we have a hello module that has a world class and we want to make sure it always returns the str "Hello World!" when we call the greeting() function. We can do that with the following code:

from hello import world

def test_greeting():
    new_world = world()
    result = new_world.greeting()
    assert result == "Hello World!"

This is an obviously very simple test, but hopefully you can get the idea of how to get started with pytest and make your code much more reliable. I'll be walking you through more and more complex examples over the next few weeks.

Create a Simple Github Profile in 3 Steps

Neal Chambers — Sat, 28 Jan 2023 00:22:07 +0000

Having a nice looking Github profile is a huge plus for getting freelance work and introducing yourself to the world. On top of that, it doesn't take much work to set one up because there are numerous tools that you can make use of to have a snazzy profile without much effort. Recently, I took the time to actually put together a nice looking profile and thought I would share some of the things I learned. Let's take a look at 3 basic parts of a profile.

Step 1: Make a Purty Header

First things first, it is always good to greet visitors to your page. After all, they took the time to visit your page. The least you can do is say hello. Luckily, there is an easy to use tool for creating nice looking header messages.

Capsule Render

Kyechan99 has a nice header maker with plenty of options to choose from. You can define colors, widths and add any kind of text you want. They even have a handy wavy animated banner (I used the type=waving option), which is what I personally went with for my profile page.

Now, of course if you do any kind of UI/UX work or CSS animation work that you want to show off, this is the perfect place to use something more involved in order to show perspective employers or collaborators your skills.

It is also a good idea to write a very brief description of yourself. This should typically be 2-3 sentences and to the point. This is your elevator pitch if, you will.

And give people a way to get a hold of you or find out more about you. Personally, I posted links to my Twitter, Dev.to, and LinkedIn. I feel like that gives people 3 very different ways to get a hold of me, depending on what they are looking for. Here is a useful repo that contains most of the main icons.

Step 2: Make a Bullet List of your Experience

After a quick introduction, it's important to make a quick list of the tools and languages you have experience with. I personally broke these out into some common categories so visitors can quickly scan over the page and see if I can help them with anything.

I personally made an unordered bold list to title each of the categories of tools/language I use. After that, I added some badges of the key languages and tools that I feel like I have a decent amount experience with. There is a very handy list of shield.io badges with accompanying markdown available at this repo.

Step 3: Add Some Stats

You should show off the fact that you are in fact an active developer. There are a handful of very good stat widgets that can show the number of commits and PRs you make as well as what languages you most commonly use on Github.

You can find a few different stat widgets on anuraghazra's repo. They are fairly customizable and allow you to hide stats that may not be that flattering or could be misleading.

Summary

It's important to have at least a small business card available on Github, especially if you work with a lot of other developer's on open source projects. Having a good Github profile can help ease networking and connect with other like-minded developers.

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

Neal Chambers — Wed, 05 Oct 2022 14:14:09 +0000

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.

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

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)

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,
        )

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)

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]]]

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

Using CCXT Implicit API methods

Neal Chambers — Fri, 22 Jul 2022 07:32:43 +0000

I wrote about the invaluable CCXT library before, which can be used to access the common methods of an exchange's API. CCXT supports a whole host of unified methods out-of-the-box. CCXT does all the heavy lifting in terms of giving you a standard set of data from each exchange.

However, as we all know, each exchange can have its own quirks and special features. For instance, Binance, as well as several other exchanges, issues dividends if you choose to lock up certain crypto assets. For example, if you keep USDT on Binance, you can earn up to 10% APR for the first $2000.

There isn't a unified set of methods for retrieving or dealing with dividends built into CCXT, which means we need to use the exchange-specific methods or API endpoints to pull the data. Lucky for us, CCXT has a way of doing that.

To use any of the underlying endpoints you simply string together the parts of the endpoint to form a function, leaving out the version number. For example, the endpoint for dividends is /sapi/v1/asset/assetDividend, and this is a GET request as well so the function name becomes .sapiGetAssetAssetDividend(). Note the use of CamelCase and the addition of 'Get' between the name of the API and the endpoint.

It might be useful to look at the describe() function of the class for each exchange to get an idea of how these functions are formed. You can take a look at the one for Binance here.

Once you have the implicit API function, you will just need to pass the params for the endpoint in a Dict:

from ccxt import binance

client = binance(
  {
    "apiKey": "the_api_key_from_binance"
    "secret": "the_api_secret_from_binance"
  }
)

response = client.sapiGetAssetAssetDividend({"startTime":
    1656635948000, "endTime": 1658191192000, "limit": 500})

for dividend in response["rows"]:
    print(f"You received {dividend["amount"]} of
    {dividend["asset"] from {dividend["enInfo"]}.")

Each endpoint has a different set of necessary and optional parameters that you can give it. You can check the details of what is required for the Dividend Endpoint here.

Most exchanges will have documentation for their API similar to Binance. Keep in mind that they may not be up-to-date, so you will want to sample the json response from an example call to see if it is what they say it is.

Another thing I learned is that the standard set of CCXT calls will sort returned data with the most recent record in the first position, and the oldest in the last position. However, Binance and some other exchanges returns this in the opposite order. Again, double-check your data.

Conclusion

CCXT will provide a very good set of standard calls to get and post data to exchanges, but exchanges come in all sorts of shapes and sizes. Sometimes, you will need to get under the hood and call some of the endpoints unique to the exchange. These implicit API calls help you do just that. Don't be afraid to get your hands a little dirty.

This post includes affiliate links; At no additional cost to you, I may receive compensation if you purchase products or services from the links provided in this article. Buying, selling, and trading crypto assets involves risk. Always Do Your Own Research before investing any money. This post is Not Financial Advice.

Working with CCXT Python Library

Neal Chambers — Wed, 15 Jun 2022 13:39:46 +0000

The CCXT library is a powerful Python, JS, and PHP library that can be used to pull information from all of the major crypto asset exchanges.

It has made my life a lot easier since it abstracts out the interface of each exchange. Instead of having to learn each API for each exchange or importing a separate library for each one, I just need to use one module - CCXT.

It also eliminates having to worry about keeping up with all the updates of a particular API. CCXT takes care of all the updates for me, which good because some of these APIs have missing documentation. For example, Coinbase has outdated json responses.

Getting started is easy. Just instantiate an exchange with your api key and api secret, which should have the basic permissions assigned to it. I got mine off of Binance.com.

First, navigate to the site, login and click 'API Management' under your profile:

Click on create API:

Give it a name:

And then you will be met with this screen:

1) Be sure to copy down your API key AND API secret. On Binance and a lot of other exchanges, it might only give you a copy button for the API key. And the QR Code may only contain the API key, but, generally speaking, you will need both when programming or doing anything with the API really.

On top of that, you might not be able to copy the API secret once you save the new API, so be sure to copy it and save it for later.

2) Optionally, you can 'edit restrictions' and limit what can be done with your API keys. I would start with reading permissions when you are first programming. You do not want to give your program the ability to make orders until you are sure you can lock it down.

3) You can also limit access to certain IP addresses, which is really handy if you are accessing the exchange from a static IP. If you are accessing it from home, chances are you will not have a static IP though, so it is better not to check this.

After that you are all set. To initialize the exchange in CCXT, you simply have to use the constructor for it:

from ccxt import binance

client = binance(
  {
    "apiKey": "the_api_key_from_binance"
    "secret": "the_api_secret_from_binance"
  }
)

I like to import only the exchange I'm using to keep my code clean and less confusing.

Once you have instantiated the exchange, you can reuse it many times to make calls to the private or public API. For example, you can retrieve historical pricing data in order to run backtests:

results = binance.fetchOHLCV("BTC/USDT", "1m", 1655298673, 1)

print(results)
[
     [
         165529867000, // UTC timestamp in milliseconds, integer
         24235.4,        // (O)pen price, float
         24240.6,        // (H)ighest price, float
         24230.0,        // (L)owest price, float
         24230.7,        // (C)losing price, float
         37.72941911    // (V)olume (in terms of the base currency), float
     ],
     ...
 ]

There are numerous functions you can call on an exchange like this. For a full list, be sure to check the CCXT documentation.

Conclusion

If you are anything like me, you use more than a few different exchanges when you trade crypto assets. In this world, of ever-changing policy and sudden crashes, you never know when you will have to jump off an exchange and move to another, so it is good to have more than one account.

That's where learning a library like CCXT once comes in handy. You don't have to learn multiple libraries or have multiple dependencies for your project. It can also be used in javascript or PHP. Give it a try and let me know if it works for you.

DEV Community: Neal Chambers

Dealing with Large CSV files in Python by Chunking

Chunk It, You must Chunk It

Standardizing File Size

Summary

Basic Troubleshooting with Docker and Ruby on Rails

Connecting to the DB

Running just the RSpec you want

1. With a Script

2. Start Interactive Terminal

Using binding.pry on the live server

Conclusion

How to Get Started with Test Driven Development with Python

Enter Test-Driven Development (TDD)

Step by Step: How to Write Tests for Software Development

How to Install Pytest and Get Started

How do you Write a Test?

Create a Simple Github Profile in 3 Steps

Step 1: Make a Purty Header

Capsule Render

Step 2: Make a Bullet List of your Experience

Step 3: Add Some Stats

Summary

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

Using CCXT Implicit API methods

Conclusion

Working with CCXT Python Library

Conclusion

Using `binding.pry` on the live server