Originally published on my blog
Recently I discovered the MOEX API, a few weeks ago to be more precise, with the recent events I thought that it may be interesting for you to know about this API. I say interesting, because you can consume the data that this API provides, basically data from the Moscow Exchange, and see how this Exchange behaves with the current crisis in Ukraine and the sanctions imposed from other countries to Russia. I'm not going to make an analysis of this data here, that's not my interest, you can make it by your own ).
A point that I want to clarify, this is a technical post, don't spec anything else from the reading of this post. If I wish to write about politics, I would write about Cuba, where I lived 26 years of my life, and where there's a lot of things to talk about. I barely know Belarus, that's where I'm leaving now. Saying that, let's start describing the requirements.
Requirements
- Docker
- An account in the MOEX API, this is optional, the API allow you to consumes the data that I will be using for free. The creation of the account is totally free, no credit card needed.
Having docker, is really not necessary to have anything else on your machine maybe. Just in case it would be easier for you to have:
- Python
- The following Python external libraries:
Introduction
The MOEX API have a lot of interesting data, at least for me, maybe for you is boring. For our use case, what I'm going to do is basically
expose the data that you can use to make some candlestick charts. For this purpose I wrote a pretty simple Python client to interact with this API, you can find it here, is called moexiss.
Let's focus on what is important for this tutorial.
- I hope, you will see how to define a "model" of the data you want to return on a given endpoint of the API.
- Define parameters to receive on GET endpoints.
- Build a Docker container with your FastAPI API, very redundant I know.
Endpoints
boards
Our API will expose the boards available in the stock engine and the index market. For this particular endpoint you won't need to provide any parameter. We are going to return the list of boards according to this structure.
[
{
"id": 0,
"board_group_id": 0,
"boardid": "string",
"title": "string",
"is_traded": true
}
]
securities
In case of the securities endpoint you must provide the boardid parameter in the GET request, and the API will return you a list of available securities, with the following data.
[
{
"secid": "string",
"boardid": "string",
"name": "string",
"decimals": 0,
"shortname": "string",
"annualhigh": 0,
"annuallow": 0,
"currencyid": "string",
"calcmode": "string"
}
]
candles
Last but not least, we are going to serve you also data that you could use to draw some candlestick charts. For this case you will need to provide the following parameters:
- board, board id that can be fetched with the boards endpoint.
- security, security id that can be fetched with the securities endpoint.
- start_date and end_date, in this case you must provide a range of date that you want to fetch data from. You must take into consideration that there are days where the index market is not working, so if you choose one of these days don't be surprised if you don't get any data.
The structure of the data to be returned would be as follow:
{
"board": "string",
"security": "string",
"start_date": "string",
"end_date": "string",
"records": [
{
"open": 0,
"close": 0,
"high": 0,
"low": 0,
"value": 0,
"volume": 0,
"begin": "string",
"end": "string"
}
]
}
Code
The tree of the final directory would be this:
.
├── app.py
├── Dockerfile
├── models
│ ├── __init__.py
│ └── model.py
├── moexiss
│ ├── client.py
│ ├── __init__.py
│ └── README.md
└── requirements.txt
The code related to moexiss could be found here, feel free to git clone.
First, let's define our models to be returned on the endpoints of the API. In order to define these models we are going to need the following imports,
importing pydantic and typing
from pydantic import BaseModel
from typing import List, Optional
Board model
File: models/model.py
class Board(BaseModel):
id: int
board_group_id: int
boardid: str
title: str
is_traded: bool
Now given this model, the API needs to return a List of this model, for that reason our endpoint that will return the boards need to be defined in the following way:
boards endpoint definition
File: app.py
@app.get("/boards", response_model=List[Board])
async def boards():
# code related to processing boards
resp = client.boards("stock", "index")
data = form_data(resp)
boards = [Board(**d) for d in data]
return boards
The most important detail here, is the response_model provided in the decorator call, in this case we are using the List from typing library, and specifying that this endpoint will return a list of the model previously defined. Pretty straight forward.
Here I make use of an utility function called form_data, the definition of this function would be:
def form_data(resp):
return [dict(zip(resp["columns"], record)) for record in resp["data"]]
Is a small transformation of the data fetched from the MOEX API.
Security model
File: models/model.py
class Security(BaseModel):
secid: str
boardid: str
name: str
decimals: int
shortname: str
annualhigh: float
annuallow: float
currencyid: str
calcmode: str = None
The endpoint for this case, would receive one parameter called boardid that would be a string, and would return a List of Security, the code related to it would be as follows:
securities endpoint definition
File: app.py
@app.get("/securities", response_model=List[Security])
async def securities(boardid: str):
resp = client.securities("stock", "index", boardid)
resp["columns"] = [col.lower() for col in resp["columns"]]
data = form_data(resp)
secs = [Security(**d) for d in data]
return secs
Notice that the parameter is defined in the definition of the endpoint handler. In case you failed to provide this parameter, FastAPI by default will return you and 422 HTTP Unprocessable Entity. In the body of this response you could find the reason why the API is returning you this error, is just awesome, that requires zero effort by your part.
In the case of the last endpoint, candles, is basically the same, only that here I will need to define another model called Record
Security model
File: models/model.py
class Record(BaseModel):
open: float
close: float
high: float
low: float
value: float
volume: float
begin: str
end: str
class Candle(BaseModel):
board: str
security: str
start_date: str
end_date: str
records: Optional[List[Record]]
Our endpoint would be similar to the previous one, only with more parameters
candles endpoint definition
File: app.py
@app.get("/candles", response_model=Candle)
async def candles(board: str, security: str, start_date: str, end_date: str):
args = {
"engine": "stock",
"market": "index",
"board": board,
"security": security,
"start_date": start_date,
"end_date": end_date,
}
resp = client.candle_data(**args)
candle = Candle(**args)
candle.records = form_data(resp)
return candle
Docker container
Now for our docker image I use the Python3.9 image as the base image, honestly I tried to make other way at first. I tried to use a distroless container, but is a pain in the ass, really annoying. This is only with Python, with Golang images works like a charm, no complaining about that, but with Python man that's another story. The main problem that you will find is with uvicorn, no more comment on this matters, if you know how let me know.
Back to business, the Dockerfile is pretty simple
FROM python:3.9
WORKDIR /app
COPY . /app
RUN pip install --no-cache-dir --upgrade -r /app/requirements.txt
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "80"]
The image built with this dockerfile is about 991MB, I don't like images so big, that's the reason on trying to use distroless container.
Note
I don't provide the whole code on purpose, is to force you to write the missing parts by yourself. There's no too much to add really, but at least you do something.
Top comments (0)