DEV Community: Robin Muhia

Unlocking Profit Potential: Building an Arbitrage Betting Client with Hexagonal Architecture in Golang

Robin Muhia — Sat, 04 May 2024 09:24:06 +0000

Introduction

What if i told you 🫵 that money can grow on trees? Sounds like a scam, right?

However, mathematics is on our side when it comes to betting.

In the fast-paced world of sports betting, the pursuit of profit often hinges on the ability to capitalize on fleeting opportunities. Arbitrage betting, a strategy that exploits pricing discrepancies in the market, stands as a beacon for those seeking consistent gains in this dynamic landscape. However, harnessing the potential of arbitrage demands more than just keen intuition—it requires robust technological solutions that can swiftly identify and exploit these fleeting differentials.

In this article, we delve into the realm of arbitrage betting and explore how to construct a powerful betting client using the principles of hexagonal architecture in the Go programming language (Golang). By employing this modular and scalable approach, we aim to empower both novice and seasoned bettors with the tools necessary to navigate the complexities of the betting market and unlock untapped profit potential.

Join us on this journey as we dissect the intricacies of arbitrage betting, unveil the inner workings of the hexagonal architecture, and demonstrate how their harmonious integration in Golang can revolutionize your approach to sports betting.

If you're only here for the repository, this is the Github link.

If you want to use the library in your application, then get the library as shown below;

go get github.com/robinmuhia/arbitrageClient@latest

What is Arbitrage Betting?

Arbitrage betting is a strategy where a bettor takes advantage of differences in odds offered by different bookmakers to guarantee a profit. By placing bets on all possible outcomes of a sports event across different bookmakers, the bettor ensures that they will make a profit regardless of the outcome. This is possible when the odds offered by different bookmakers imply probabilities that add up to less than 100%. The bettor calculates the optimal bet sizes to ensure a profit regardless of the outcome. Arbitrage opportunities are usually short-lived and require quick action to exploit.

Simply put, say a game has three possible outcomes, we can bet on all 3 possible outcomes i.e. the home team wins, the away team wins or a draw happens and in each case, we are GUARANTEED a profit. We will make money regardless of the outcome.

To learn more about arbitrage betting, read this article.

What is the hexagonal architecture?

The hexagonal architecture, also known as Ports and Adapters architecture or the onion architecture, is a software design pattern that promotes modular and loosely coupled systems. It was first introduced by Alistair Cockburn in 2005 as a way to address some of the shortcomings of traditional layered architectures.

At its core, the hexagonal architecture revolves around the idea of organizing the components of a system into concentric layers, with the business logic or core functionality residing at the center. These layers are typically represented as hexagons, hence the name.

The key principles of the hexagonal architecture include:

Separation of Concerns: The architecture emphasizes dividing the system into distinct layers, each responsible for a specific concern. This separation facilitates easier maintenance, testing, and evolution of the system.
Ports and Adapters: In hexagonal architecture, components communicate through well-defined interfaces known as ports. These ports abstract away external dependencies and allow the core business logic to remain decoupled from implementation details. Adapters are then used to connect these ports to external systems or frameworks.
Domain-Driven Design (DDD): Hexagonal architecture encourages a domain-centric approach, where the core business logic is modeled based on the problem domain rather than technical considerations. This helps ensure that the system closely aligns with the real-world problem it aims to solve.
Testability: By isolating the core business logic from external dependencies, such as databases or external services, hexagonal architecture promotes easier testing. Components can be tested in isolation using mock implementations of dependencies, leading to more robust and reliable software.

Overall, the hexagonal architecture promotes flexibility, maintainability, and testability, making it well-suited for complex and evolving software systems. It provides a solid foundation for building scalable and adaptable applications across various domains, including web development, microservices, and domain-specific applications.

Read more about the hexagonal architecture here.

Prerequisites

We will be build our Arbitrage Client around the Odds API so go to that link and grab an API key. They have a free tier which is limited to 500 requests per month. We will need around 75 requests per session so it is enough for testing the functionality and logic.

Additionally, we hope you have some basic understanding of programming so as to follow the article. Some Golang knowlegde is a plus as well.

Project Structure

You can follow along or clone the repository in a folder

git clone git@github.com:robinmuhia/arbitrageClient.git .

Make sure the project is structured as shown below;

Lets explore the folders;

Application - this holds application specific code which in our case is enums that we will use in the project.
Domain - this holds domain specific models such as how an odd, sport, bookmaker etc should be represented. The domain is only an interface for the models in the project.
Infrastructure - this is where we put our external actors that we depend on. Here we can put external services that we require such as a database or external services such as messaging client like Twilio.
Usecases - this is where we put our business logic . It is independent of other components. We can swap out our services and the business logic should work as is.

This structure allows us to swap out components as is. We can change the odds API and use another API. Our usecases should remain as is. This reduces the coupling of our code and allows us to use dependency injection efficiently.

The Arbitrage client

Application folder

We have our enums here which will be shared across the project.

endpoints.go

package enums

type endpoint string

const (
    Sport endpoint = "sports"
    Odds  endpoint = "odds"
)

func (e endpoint) String() string {
    return string(e)
}

This is where we put the endpoints we will call.

envs.go

package enums

type env string

const (
    BaseURL   env = "ODDS_API_BASE_URL"
    ApiKeyEnv env = "ODDS_API_KEY" //nolint: gosec
)

func (e env) String() string {
    return string(e)
}

This is where we will house how our environment variables are called.

params.go

package enums

type params string

const (
    ApiKey     params = "apiKey"
    Region     params = "regions"
    Markets    params = "markets"
    OddsFormat params = "oddsFormat"
    DateFormat params = "dateFormat"
)

func (e params) String() string {
    return string(e)
}

These are the parameters that we will we will use to embed in our calls to the API

Domain folder

arbs.go

package domain

// ThreeOddsArb represents the structure of how a match with three possible outcomes
// i.e a win, draw or loss will be represented in the response
type ThreeOddsArb struct {
    Title     string
    Home      string
    HomeOdds  float64
    HomeStake float64
    Draw      string
    DrawOdds  float64
    DrawStake float64
    Away      string
    AwayStake float64
    AwayOdds  float64
    GameType  string
    League    string
    Profit    float64
    GameTime  string
}

// TwoOddsArb represents the structure of how a match with three possible outcomes
// i.e a win or loss will be represented in the response
type TwoOddsArb struct {
    Title     string
    Home      string
    HomeOdds  float64
    HomeStake float64
    Away      string
    AwayStake float64
    AwayOdds  float64
    GameType  string
    League    string
    Profit    float64
    GameTime  string
}

These are how the arbs will be represented.

odds.go

package domain

// Outcome the name and price of an individual outcome of a bet eg. Bayern 1.26
type Outcome struct {
    Name  string  `json:"name"`
    Price float64 `json:"price"`
}

// Market represents the bookmarkers' odds for a game
type Market struct {
    Key        string    `json:"key"`
    LastUpdate string    `json:"last_update"`
    Outcomes   []Outcome `json:"outcomes"`
}

// Bookmaker describes the bookmarker such as bet365
type Bookmaker struct {
    Key        string   `json:"key"`
    Title      string   `json:"title"`
    LastUpdate string   `json:"last_update"`
    Markets    []Market `json:"markets"`
}

// Odds represent the odds structure of the games's odds
type Odds struct {
    ID           string      `json:"id"`
    SportKey     string      `json:"sport_key"`
    SportTitle   string      `json:"sport_title"`
    CommenceTime string      `json:"commence_time"`
    HomeTeam     string      `json:"home_team"`
    AwayTeam     string      `json:"away_team"`
    Bookmakers   []Bookmaker `json:"bookmakers"`
}

These is how odds from the API will be represented.

sport.go

package domain

// Sport represents a sport response from OddsAPI
type Sport struct {
    Key          string `json:"key"`
    Group        string `json:"group"`
    Title        string `json:"title"`
    Description  string `json:"description"`
    Active       bool   `json:"active"`
    HasOutrights bool   `json:"has_outrights"`
}

These is how sports from the API will be represented.

Infrastructure

client.go

package services

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "net/url"
    "os"
    "sync"
    "time"

    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/application/enums"
    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/domain"
)

var (
    baseURL = os.Getenv(enums.BaseURL.String())
    apiKey  = os.Getenv(enums.ApiKeyEnv.String())
)

// oddsoddsAPIHTTPClient instantiates a client to call the odds API url
type oddsAPIHTTPClient struct {
    client  *http.Client
    baseURL string
    apiKey  string
}

// ArbClient implements methods intended to be exposed by the oddsHTTPClient
type ArbClient interface {
    GetAllOdds(ctx context.Context, oddsParams OddsParams) ([]domain.Odds, error)
}

// OddsParams represent the parameters required to query for specific odds
type OddsParams struct {
    Region     string
    Markets    string
    OddsFormat string
    DateFormat string
}

These are the structs, interfaces and variables we will use to be able to call the API.

Our ArbClient interface has the GetAllOdds method. This method is the only method we will expose to other external services/folders. Thus, in future, if we were to change the api we call, we can refactor the code but the method should always be provided for this service.

// NewServiceOddsAPI returns a new instance of an OddsAPI service
func NewServiceOddsAPI() (*oddsAPIHTTPClient, error) {
    if baseURL == "" {
        return nil, fmt.Errorf("empty env variables, %s", enums.BaseURL.String())
    }

    if apiKey == "" {
        return nil, fmt.Errorf("empty env variables, %s", enums.ApiKeyEnv.String())
    }

    return &oddsAPIHTTPClient{
        client: &http.Client{
            Timeout: time.Second * 10,
        },
        baseURL: baseURL,
        apiKey:  apiKey,
    }, nil

This function is simple, it returns a HTTP client that should have the above configuration.

As seen above, the apikey and baseUrl should be provided from our environment variables.

// makeRequest calls the Odds API endpoint
func (s *oddsAPIHTTPClient) makeRequest(ctx context.Context, method, urlPath string, queryParams url.Values, _ interface{}) (*http.Response, error) {
    var request *http.Request

    switch method {
    case http.MethodGet:
        req, err := http.NewRequestWithContext(ctx, method, urlPath, nil)
        if err != nil {
            return nil, err
        }

        request = req

    default:
        return nil, fmt.Errorf("unsupported http method: %s", method)
    }

    request.Header.Set("Accept", "application/json")
    request.Header.Set("Content-Type", "application/json")

    if queryParams != nil {
        request.URL.RawQuery = queryParams.Encode()
    }

    return s.client.Do(request)
}

The makerequest function is used to make requests as explained by the docstring. Currently, it only allows GET method, constructs the requests and does the request.

// getSports returns a list of sports and an error
func (s oddsAPIHTTPClient) getSports(ctx context.Context) ([]domain.Sport, error) {
    urlPath := fmt.Sprintf("%s/%s", s.baseURL, enums.Sport.String())

    queryParams := url.Values{}
    queryParams.Add(enums.ApiKey.String(), s.apiKey)

    resp, err := s.makeRequest(ctx, http.MethodGet, urlPath, queryParams, nil)
    if err != nil {
        return nil, err
    }

    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return nil, fmt.Errorf("failed to get sports data: %s", resp.Status)
    }

    var sports []domain.Sport
    if err := json.NewDecoder(resp.Body).Decode(&sports); err != nil {
        return nil, fmt.Errorf("failed to get sports: %w", err)
    }

    return sports, nil
}

This function retrieves all the sports from the API and serializes them to our Sport domain.

A sport in this case is for example in football/soccer, EPL, La liga and Champions League are all a type of sport.

// getOdd returns all odds from one sport
func (s oddsAPIHTTPClient) getOdd(ctx context.Context, oddParams OddsParams, sport domain.Sport, wg *sync.WaitGroup) ([]domain.Odds, error) {
    defer wg.Done()

    urlPath := fmt.Sprintf("%s/%s/%s/%s", s.baseURL, enums.Sport.String(), sport.Key, enums.Odds.String())

    queryParams := url.Values{}
    queryParams.Add(enums.ApiKey.String(), s.apiKey)
    queryParams.Add(enums.Region.String(), oddParams.Region)
    queryParams.Add(enums.Markets.String(), oddParams.Markets)
    queryParams.Add(enums.OddsFormat.String(), oddParams.OddsFormat)
    queryParams.Add(enums.DateFormat.String(), oddParams.DateFormat)

    resp, err := s.makeRequest(ctx, http.MethodGet, urlPath, queryParams, nil)
    if err != nil {
        return nil, err
    }

    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return nil, fmt.Errorf("failed to get sports data: %s", resp.Status)
    }

    var odds []domain.Odds
    if err := json.NewDecoder(resp.Body).Decode(&odds); err != nil {
        return nil, fmt.Errorf("failed to decode odds data for %s: %w", sport.Title, err)
    }

    return odds, nil
}

This functions gets all the odds from one sport. It serializes the response into an odd.

// GetOdds returns a list of all available odds given various parameters across all sports
func (s oddsAPIHTTPClient) GetAllOdds(ctx context.Context, oddsParams OddsParams) ([]domain.Odds, error) {
    sports, err := s.getSports(ctx)
    if err != nil {
        return nil, err
    }

    ticker := time.NewTicker(1 * time.Second)

    var wg sync.WaitGroup

    var mu sync.Mutex

    var allOdds []domain.Odds

    for _, sport := range sports {
        if sport.Active {
            wg.Add(1)

            go func() {
                odds, err := s.getOdd(ctx, oddsParams, sport, &wg)
                if err != nil {
                    log.Print(err.Error())
                } else {
                mu.Lock()
                allOdds = append(allOdds, odds...)
                mu.Unlock()
                }
            }()
        }

        <-ticker.C // waits a second to send next goroutine, intended to prevent ddosing and rate limiting
    }

    wg.Wait()

    ticker.Stop()

    return allOdds, nil
}

This is where the magic happens, we want to get all the odds from all the sports as quick as possible. A normal for loop was getting rate limited as we are making requests for each odd.

The optimal way i found was to first get all the sports and then loop through each sport and checking if it is active. If it is active, we then get odd from each sport in a goroutine.

Since making requests is an I/O bound task, the response can take an unforeseen amount of time to execute hence the use of goroutines. Each call will execute in its own thread. We synchronize these goroutines using the sync waitgroup library. Each time a goroutine is span up, we add one to the waitgroup then in the sport function, we defer the decrement of the waitgroup.

The reason for this is to ensure all goroutines finish executing before we proceed, hence the wg.Wait(). The code will not proceed till the waitgroup reaches a value of 0.

Once we receive the odds, we append the odds to the allOdds slice. We use a mutex to make sure only one goroutine can append the odds to the slice at a time. This is done to avoid race conditions in case some goroutines complete at the same time. Thus we lock the mutex and unlock it after the odds have been appended.

client_test.go

package services

import (
    "context"
    "fmt"
    "net/http"
    "net/url"
    "sync"
    "testing"

    "github.com/jarcoal/httpmock"
    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/application/enums"
    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/domain"
)

func Test_oddsAPIHTTPClient_makeRequest(t *testing.T) {
    type args struct {
        ctx         context.Context
        method      string
        urlPath     string
        queryParams url.Values
        in4         interface{}
    }

    queryParams := url.Values{}
    queryParams.Add("foo", "bar")

    tests := []struct {
        name    string
        args    args
        wantErr bool
    }{
        {
            name: "happy case: successful request",
            args: args{
                ctx:         context.Background(),
                method:      http.MethodGet,
                urlPath:     "https://www.foo.com",
                queryParams: queryParams,
            },
            wantErr: false,
        },
        {
            name: "sad case: invalid http method",
            args: args{
                ctx:         context.Background(),
                method:      http.MethodPost,
                urlPath:     "https://www.foo.com",
                queryParams: queryParams,
            },
            wantErr: true,
        },
    }

    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            if tt.name == "happy case: successful request" {
                httpmock.RegisterResponder(http.MethodGet, tt.args.urlPath, func(r *http.Request) (*http.Response, error) { //nolint:all
                    return httpmock.NewJsonResponse(http.StatusOK, nil)
                })
            }

            httpmock.Activate()
            defer httpmock.DeactivateAndReset()

            s, _ := NewServiceOddsAPI()
            resp, err := s.makeRequest(tt.args.ctx, tt.args.method, tt.args.urlPath, tt.args.queryParams, tt.args.in4)

            if err == nil {
                defer resp.Body.Close()
            }

            if (err != nil) != tt.wantErr {
                t.Errorf("oddsAPIHTTPClient.makeRequest() error = %v, wantErr %v", err, tt.wantErr)
                return
            }
        })
    }
}

func Test_oddsAPIHTTPClient_getSports(t *testing.T) {
    type args struct {
        ctx context.Context
    }

    tests := []struct {
        name    string
        args    args
        wantErr bool
    }{
        {
            name: "happy case: successful retrieval of sport",
            args: args{
                ctx: context.Background(),
            },
            wantErr: false,
        },
        {
            name: "sad case: unable to decode sport",
            args: args{
                ctx: context.Background(),
            },
            wantErr: true,
        },
        {
            name: "sad case: unsuccessful retrieval of sport",
            args: args{
                ctx: context.Background(),
            },
            wantErr: true,
        },
    }
    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            s, _ := NewServiceOddsAPI()
            urlPath := fmt.Sprintf("%s/%s", s.baseURL, enums.Sport.String())

            if tt.name == "happy case: successful retrieval of sport" {
                httpmock.RegisterResponder(http.MethodGet, urlPath, func(r *http.Request) (*http.Response, error) {
                    resp := []domain.Sport{
                        {
                            Key:   "foo",
                            Title: "bar",
                        },
                    }

                    return httpmock.NewJsonResponse(http.StatusOK, resp)
                })
            }

            if tt.name == "sad case: unsuccessful retrieval of sport" {
                httpmock.RegisterResponder(http.MethodGet, urlPath, func(r *http.Request) (*http.Response, error) {
                    return httpmock.NewJsonResponse(http.StatusUnauthorized, nil)
                })
            }

            if tt.name == "sad case: unable to decode sport" {
                httpmock.RegisterResponder(http.MethodGet, urlPath, func(r *http.Request) (*http.Response, error) {
                    return httpmock.NewJsonResponse(http.StatusOK, "nana")
                })
            }

            httpmock.Activate()
            defer httpmock.DeactivateAndReset()

            _, err := s.getSports(tt.args.ctx)

            if (err != nil) != tt.wantErr {
                t.Errorf("oddsAPIHTTPClient.makeRequest() error = %v, wantErr %v", err, tt.wantErr)
                return
            }
        })
    }
}

func Test_oddsAPIHTTPClient_getOdd(t *testing.T) {
    type args struct {
        ctx       context.Context
        oddParams OddsParams
        sport     domain.Sport
        wg        *sync.WaitGroup
    }

    tests := []struct {
        name    string
        args    args
        wantErr bool
    }{
        {
            name: "happy case: successful retrieval of odd",
            args: args{
                ctx: context.Background(),
                oddParams: OddsParams{
                    Region:     "foo",
                    Markets:    "bar",
                    OddsFormat: "foo",
                    DateFormat: "bar",
                },
                sport: domain.Sport{
                    Key: "foo",
                },
                wg: &sync.WaitGroup{},
            },
            wantErr: false,
        },
        {
            name: "sad case: unable to decode odd",
            args: args{
                ctx: context.Background(),
                oddParams: OddsParams{
                    Region:     "foo",
                    Markets:    "bar",
                    OddsFormat: "foo",
                    DateFormat: "bar",
                },
                sport: domain.Sport{
                    Key: "foo",
                },
                wg: &sync.WaitGroup{},
            },
            wantErr: true,
        },
        {
            name: "sad case: unsuccessful retrieval of odd",
            args: args{
                ctx: context.Background(),
                oddParams: OddsParams{
                    Region:     "foo",
                    Markets:    "bar",
                    OddsFormat: "foo",
                    DateFormat: "bar",
                },
                sport: domain.Sport{
                    Key: "foo",
                },
                wg: &sync.WaitGroup{},
            },
            wantErr: true,
        },
    }

    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            s, _ := NewServiceOddsAPI()

            urlPath := fmt.Sprintf("%s/%s/%s/%s", s.baseURL, enums.Sport.String(), tt.args.sport.Key, enums.Odds.String())

            tt.args.wg.Add(1)

            if tt.name == "happy case: successful retrieval of odd" {
                httpmock.RegisterResponder(http.MethodGet, urlPath, func(r *http.Request) (*http.Response, error) {
                    resp := []domain.Odds{{
                        ID:       "foo",
                        SportKey: "bar",
                    },
                    }

                    return httpmock.NewJsonResponse(http.StatusOK, resp)
                })
            }

            if tt.name == "sad case: unsuccessful retrieval odd" {
                httpmock.RegisterResponder(http.MethodGet, urlPath, func(r *http.Request) (*http.Response, error) {
                    return httpmock.NewJsonResponse(http.StatusUnauthorized, nil)
                })
            }

            if tt.name == "sad case: unable to decode odd" {
                httpmock.RegisterResponder(http.MethodGet, urlPath, func(r *http.Request) (*http.Response, error) {
                    return httpmock.NewJsonResponse(http.StatusOK, "nana")
                })
            }

            httpmock.Activate()
            defer httpmock.DeactivateAndReset()

            _, err := s.getOdd(tt.args.ctx, tt.args.oddParams, tt.args.sport, tt.args.wg)
            if (err != nil) != tt.wantErr {
                t.Errorf("oddsAPIHTTPClient.getOdd() error = %v, wantErr %v", err, tt.wantErr)
                return
            }
        })
    }
}

func Test_oddsAPIHTTPClient_GetAllOdds(t *testing.T) {
    type args struct {
        ctx        context.Context
        oddsParams OddsParams
    }

    tests := []struct {
        name    string
        args    args
        wantErr bool
    }{
        {
            name: "happy case: successfully get all odds",
            args: args{
                ctx: context.Background(),
                oddsParams: OddsParams{
                    Region:     "foo",
                    Markets:    "bar",
                    OddsFormat: "foo",
                    DateFormat: "bar",
                },
            },
            wantErr: false,
        },
        {
            name: "sad case: unsuccessfully get all odds",
            args: args{
                ctx: context.Background(),
                oddsParams: OddsParams{
                    Region:     "foo",
                    Markets:    "bar",
                    OddsFormat: "foo",
                    DateFormat: "bar",
                },
            },
            wantErr: true,
        },
    }
    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            s, _ := NewServiceOddsAPI()

            sportPath := fmt.Sprintf("%s/%s", s.baseURL, enums.Sport.String())

            if tt.name == "happy case: successfully get all odds" {
                resp := []domain.Sport{
                    {
                        Key:    "foo",
                        Title:  "bar",
                        Active: true,
                    },
                    {
                        Key:    "bar",
                        Title:  "foo",
                        Active: true,
                    },
                }

                httpmock.RegisterResponder(http.MethodGet, sportPath, func(r *http.Request) (*http.Response, error) {
                    return httpmock.NewJsonResponse(http.StatusOK, resp)
                })

                oddPath1 := fmt.Sprintf("%s/%s/%s/%s", s.baseURL, enums.Sport.String(), resp[0].Key, enums.Odds.String())
                httpmock.RegisterResponder(http.MethodGet, oddPath1, func(r *http.Request) (*http.Response, error) {
                    oddResp1 := []domain.Odds{
                        {
                            ID:       "foo",
                            SportKey: "bar",
                        },
                    }

                    return httpmock.NewJsonResponse(http.StatusOK, oddResp1)
                })

                oddPath2 := fmt.Sprintf("%s/%s/%s/%s", s.baseURL, enums.Sport.String(), resp[1].Key, enums.Odds.String())
                httpmock.RegisterResponder(http.MethodGet, oddPath2, func(r *http.Request) (*http.Response, error) {
                    return httpmock.NewJsonResponse(http.StatusNotFound, nil)
                })
            }

            if tt.name == "sad case: unsuccessfully get all odds" {
                httpmock.RegisterResponder(http.MethodGet, sportPath, func(r *http.Request) (*http.Response, error) {
                    return httpmock.NewJsonResponse(http.StatusUnauthorized, nil)
                })
            }

            httpmock.Activate()
            defer httpmock.DeactivateAndReset()

            _, err := s.GetAllOdds(tt.args.ctx, tt.args.oddsParams)
            if (err != nil) != tt.wantErr {
                t.Errorf("oddsAPIHTTPClient.GetAllOdds() error = %v, wantErr %v", err, tt.wantErr)
                return
            }
        })
    }
}

func TestNewServiceOddsAPI(t *testing.T) {
    tests := []struct {
        name    string
        wantErr bool
    }{
        {
            name:    "happy case: successful instantiation",
            wantErr: false,
        },
        {
            name:    "sad case: no api key provided",
            wantErr: true,
        },
        {
            name:    "sad case: no base URL provided",
            wantErr: true,
        },
    }

    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            if tt.name == "sad case: no base URL provided" {
                baseURL = ""
            }

            if tt.name == "sad case: no api key provided" {
                apiKey = ""
            }

            _, err := NewServiceOddsAPI()
            if (err != nil) != tt.wantErr {
                t.Errorf("NewServiceOddsAPI() error = %v, wantErr %v", err, tt.wantErr)
                return
            }
        })
    }
}

These are unit tests that are written to ensure that the code works as expected.

Usecases

arbs.go

package arbs

import (
    "context"
    "fmt"
    "sync"

    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/domain"
    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/infrastructure/services"
)

type UseCasesArbsImpl struct {
    OddsApiClient services.ArbClient
}

We make a struct that will implement the Arbitrage functionality.

We need the APIclient to access the GetAllOdds method.

// composeTwoArbsBet process an Odd to check if an arbitrage oppurtunity exists
func (us *UseCasesArbsImpl) composeTwoArbsBet(odd domain.Odds, i int, j int) (domain.TwoOddsArb, bool) {
    homeOdd := odd.Bookmakers[i].Markets[0].Outcomes[0].Price
    awayOdd := odd.Bookmakers[j].Markets[0].Outcomes[1].Price
    arb := (1 / homeOdd) + (1 / awayOdd)

    if arb < 1.0 {
        profit := (1 - arb) * 100
        twowayArb := domain.TwoOddsArb{
            Title:     fmt.Sprintf("%s - %s", odd.HomeTeam, odd.AwayTeam),
            Home:      odd.Bookmakers[i].Title,
            HomeOdds:  homeOdd,
            HomeStake: 1 / homeOdd,
            Away:      odd.Bookmakers[j].Title,
            AwayOdds:  awayOdd,
            AwayStake: 1 / awayOdd,
            GameType:  odd.SportKey,
            League:    odd.SportTitle,
            Profit:    profit,
            GameTime:  odd.CommenceTime,
        }

        return twowayArb, true
    }

    return domain.TwoOddsArb{}, false
}

// composeThreeArbsBet processes an Odd to check if an arbitrage oppurtunity exists
func (us *UseCasesArbsImpl) composeThreeArbsBet(odd domain.Odds, i int, j int, k int) (domain.ThreeOddsArb, bool) {
    homeOdd := odd.Bookmakers[i].Markets[0].Outcomes[0].Price
    awayOdd := odd.Bookmakers[j].Markets[0].Outcomes[1].Price
    drawOdd := odd.Bookmakers[k].Markets[0].Outcomes[2].Price
    arb := (1 / homeOdd) + (1 / awayOdd) + (1 / drawOdd)

    if arb < 1.0 {
        profit := (1 - arb) * 100
        threewayArb := domain.ThreeOddsArb{
            Title:     fmt.Sprintf("%s - %s", odd.HomeTeam, odd.AwayTeam),
            Home:      odd.Bookmakers[i].Title,
            HomeOdds:  homeOdd,
            HomeStake: 1 / homeOdd,
            Away:      odd.Bookmakers[j].Title,
            AwayOdds:  awayOdd,
            AwayStake: 1 / awayOdd,
            Draw:      odd.Bookmakers[k].Title,
            DrawOdds:  drawOdd,
            DrawStake: 1 / drawOdd,
            GameType:  odd.SportKey,
            League:    odd.SportTitle,
            Profit:    profit,
            GameTime:  odd.CommenceTime,
        }

        return threewayArb, true
    }

    return domain.ThreeOddsArb{}, false
}

These two functions check if an arbitrage opportunity exists given some conditions about odds. One checks for matches that have two possible outcomes that is a home win or away win while the other checks for matches that have three possible outcomes that is home win, away win or draw.

// checkIfMarketHasEnoughGames checks whether a market has enough games to analyze an arbitrage oppurtunity
func (us *UseCasesArbsImpl) checkIfMarketHasEnoughGames(bookmarker domain.Bookmaker) bool {
    return len(bookmarker.Markets) >= 1
}

These function checks whether a markets in the bookmarker object has more than one market as there is no need to check for an arbitrage opportunity if only one market or 0 markets is available as we would never get one. This is because a bookmarker usually negates arbitrage opportunity on their own odds as they would make losses if arbs were readily available on their own platforms.

// findPossibleArbOppurtunity finds possible arbs and sends them over a channel
func (us *UseCasesArbsImpl) findPossibleArbOpportunity(odd domain.Odds,
    threeOddsCh chan<- domain.ThreeOddsArb,
    twoOddsCh chan<- domain.TwoOddsArb,
    wg *sync.WaitGroup) {
    defer wg.Done()

    if len(odd.Bookmakers) < 2 {
        return // Skip if there are not enough bookmakers for comparison
    }

    for i := 0; i < len(odd.Bookmakers); i++ {
        if !us.checkIfMarketHasEnoughGames(odd.Bookmakers[i]) {
            return
        }

        for j := 0; j < len(odd.Bookmakers); j++ {
            if !us.checkIfMarketHasEnoughGames(odd.Bookmakers[j]) {
                return
            }

            switch {
            case len(odd.Bookmakers[i].Markets[0].Outcomes) == 2 && len(odd.Bookmakers[j].Markets[0].Outcomes) == 2:
                twoWayArb, isArb := us.composeTwoArbsBet(odd, i, j)
                if isArb {
                    twoOddsCh <- twoWayArb
                }

            case len(odd.Bookmakers[i].Markets[0].Outcomes) == 3 && len(odd.Bookmakers[j].Markets[0].Outcomes) == 3:
                for k := 0; k < len(odd.Bookmakers); k++ {
                    if !us.checkIfMarketHasEnoughGames(odd.Bookmakers[k]) {
                        return
                    }

                    if len(odd.Bookmakers[k].Markets[0].Outcomes) == 3 {
                        threeWayArb, isArb := us.composeThreeArbsBet(odd, i, j, k)
                        if isArb {
                            threeOddsCh <- threeWayArb
                        }
                    }
                }
            }
        }
    }
}

This function is simple. It checks whether a game is of two outcomes or three outcomes then calls the compose method that will return an arb and true if an arb is found or an empty struct and false if an arb is not found.

if an arb is found, it sends it over a channel.

The waitgroup is then decremented by the Done method by one.

// GetArbs gets all possible arbitrage oppurtunities.
func (us *UseCasesArbsImpl) GetArbs(ctx context.Context, oddsParams services.OddsParams) ([]domain.ThreeOddsArb, []domain.TwoOddsArb, error) {
    odds, err := us.OddsApiClient.GetAllOdds(ctx, oddsParams)
    if err != nil {
        return nil, nil, err
    }

    var ThreeOddsArbs []domain.ThreeOddsArb

    var TwoOddsArbs []domain.TwoOddsArb

    // Create channels to receive arbitrage results
    threeOddsCh := make(chan domain.ThreeOddsArb)
    twoOddsCh := make(chan domain.TwoOddsArb)

    // Create a wait group to ensure all goroutines finish before returning
    var wg sync.WaitGroup

    var once sync.Once

    for _, odd := range odds {
        wg.Add(1)

        go us.findPossibleArbOpportunity(odd, threeOddsCh, twoOddsCh, &wg)
    }

    // Close the channels once all goroutines finish processing
    go func() {
        wg.Wait()
        once.Do(func() {
            close(threeOddsCh)
            close(twoOddsCh)
        })
    }()

    // Collect the results from channels
    for {
        select {
        case arb, ok := <-threeOddsCh:
            if !ok {
                threeOddsCh = nil // Set to nil to exit the loop when both channels are closed
            } else {
                ThreeOddsArbs = append(ThreeOddsArbs, arb)
            }
        case arb, ok := <-twoOddsCh:
            if !ok {
                twoOddsCh = nil // Set to nil to exit the loop when both channels are closed
            } else {
                TwoOddsArbs = append(TwoOddsArbs, arb)
            }
        }
        // Exit the loop when both channels are closed
        if threeOddsCh == nil && twoOddsCh == nil {
            break
        }
    }

    return ThreeOddsArbs, TwoOddsArbs, nil
}

The function starts by calling GetAllOdds on us.OddsApiClient to fetch all the odds using the provided oddsParams.

It then initializes two channels (threeOddsCh and twoOddsCh) to receive results for three-odds and two-odds arbitrage opportunities, respectively.
A wait group (wg) is created to ensure that all spawned goroutines finish execution before closing the channels.
For each odd fetched, a goroutine is launched to find possible arbitrage opportunities.

Each goroutine increments the wait group to signal its completion when finished.

Another goroutine is spawned to wait for all goroutines to finish (wg.Wait()), and then closes the channels (threeOddsCh and twoOddsCh) using a sync.Once to ensure they're closed only once.

The function continuously listens on both channels (threeOddsCh and twoOddsCh) using a select statement.
When a result is received on either channel, it is appended to the corresponding array (ThreeOddsArbs or TwoOddsArbs).
The function exits the loop when both channels are closed.

Once both channels are closed and all results are collected, the function returns the arrays containing arbitrage opportunities and a nil error.

arbs_test.go

package arbs

import (
    "context"
    "fmt"
    "log"
    "net/http"
    "os"
    "reflect"
    "testing"

    "github.com/jarcoal/httpmock"
    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/application/enums"
    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/domain"
    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/infrastructure/services"
)

func setUpUsecase() *UseCasesArbsImpl {
    var client services.ArbClient

    svc, err := services.NewServiceOddsAPI()
    if err != nil {
        log.Panic("error: %w", err)
    }

    client = svc

    return &UseCasesArbsImpl{
        OddsApiClient: client,
    }
}

var (
    twoWayArbodd = domain.Odds{
        HomeTeam:     "Vuvu",
        AwayTeam:     "Zela",
        SportKey:     "baseball_np",
        SportTitle:   "Baseball",
        CommenceTime: "2024-01-01T08:29:59Z",
        Bookmakers: []domain.Bookmaker{
            {
                Title: "Betway",
                Markets: []domain.Market{
                    {
                        Outcomes: []domain.Outcome{
                            {Price: 1.44},
                            {Price: 8.5},
                        }}}},
            {
                Title: "Betway",
                Markets: []domain.Market{
                    {
                        Outcomes: []domain.Outcome{
                            {Price: 0},
                            {Price: 0},
                        }}}},
        },
    }
    twoWayNotArbodd = domain.Odds{
        HomeTeam:     "Vuvu",
        AwayTeam:     "Zela",
        SportKey:     "baseball_np",
        SportTitle:   "Baseball",
        CommenceTime: "2024-01-01T08:29:59Z",
        Bookmakers: []domain.Bookmaker{
            {
                Title: "Betway",
                Markets: []domain.Market{
                    {
                        Outcomes: []domain.Outcome{
                            {Price: 1.44},
                            {Price: 1.44},
                        }}}}},
    }
    threeWayArbodd = domain.Odds{
        HomeTeam:     "Vuvu",
        AwayTeam:     "Zela",
        SportKey:     "baseball_np",
        SportTitle:   "Baseball",
        CommenceTime: "2024-01-01T08:29:59Z",
        Bookmakers: []domain.Bookmaker{
            {
                Title: "Betway",
                Markets: []domain.Market{
                    {
                        Outcomes: []domain.Outcome{
                            {Price: 4.9},
                            {Price: 17},
                            {Price: 1.57},
                        }}}},
            {
                Title: "Betway",
                Markets: []domain.Market{
                    {
                        Outcomes: []domain.Outcome{
                            {Price: 0.0},
                            {Price: 0.0},
                            {Price: 0.0},
                        }}}},
        },
    }
    threeWayNotArbodd = domain.Odds{
        HomeTeam:     "Vuvu",
        AwayTeam:     "Zela",
        SportKey:     "baseball_np",
        SportTitle:   "Baseball",
        CommenceTime: "2024-01-01T08:29:59Z",
        Bookmakers: []domain.Bookmaker{
            {
                Title: "Betway",
                Markets: []domain.Market{
                    {
                        Outcomes: []domain.Outcome{
                            {Price: 4.9},
                            {Price: 1.87},
                            {Price: 1.57},
                        }}}}},
    }
)

func TestUseCasesArbsImpl_composeTwoArbsBet(t *testing.T) {
    type args struct {
        odd domain.Odds
        i   int
        j   int
    }

    tests := []struct {
        name string
        args args
        want bool
    }{
        {
            name: "happy case: arb is found",
            args: args{
                odd: twoWayArbodd,
                i:   0,
                j:   0,
            },
            want: true,
        },
        {
            name: "sad case: arb is not found",
            args: args{
                odd: twoWayNotArbodd,
                i:   0,
                j:   0,
            },
            want: false,
        },
    }
    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            us := setUpUsecase()
            _, got := us.composeTwoArbsBet(tt.args.odd, tt.args.i, tt.args.j)

            if got != tt.want {
                t.Errorf("UseCasesArbsImpl.composeTwoArbsBet() got1 = %v, want %v", got, tt.want)
            }
        })
    }
}

func TestUseCasesArbsImpl_composeThreeArbsBet(t *testing.T) {
    type args struct {
        odd domain.Odds
        i   int
        j   int
        k   int
    }

    tests := []struct {
        name string
        args args
        want bool
    }{
        {
            name: "happy case: arb is found",
            args: args{
                odd: threeWayArbodd,
                i:   0,
                j:   0,
                k:   0,
            },
            want: true,
        },
        {
            name: "sad case: arb is not found",
            args: args{
                odd: threeWayNotArbodd,
                i:   0,
                j:   0,
                k:   0,
            },
            want: false,
        },
    }
    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            us := setUpUsecase()
            _, got := us.composeThreeArbsBet(tt.args.odd, tt.args.i, tt.args.j, tt.args.k)

            if got != tt.want {
                t.Errorf("UseCasesArbsImpl.composeThreeArbsBet() got1 = %v, want %v", got, tt.want)
            }
        })
    }
}

func TestUseCasesArbsImpl_GetArbs(t *testing.T) {
    type args struct {
        ctx        context.Context
        oddsParams services.OddsParams
    }

    tests := []struct {
        name    string
        args    args
        want    int
        want1   int
        wantErr bool
    }{
        {
            name: "happy case: get arbs",
            args: args{
                ctx:        context.Background(),
                oddsParams: services.OddsParams{},
            },
            want:    1,
            want1:   1,
            wantErr: false,
        },
    }
    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            us := setUpUsecase()

            baseUrl := os.Getenv(enums.BaseURL.String())

            sportPath := fmt.Sprintf("%s/%s", baseUrl, enums.Sport.String())

            resp := []domain.Sport{
                {
                    Key:    "foo",
                    Title:  "bar",
                    Active: true,
                },
            }

            httpmock.RegisterResponder(http.MethodGet, sportPath, func(r *http.Request) (*http.Response, error) {
                return httpmock.NewJsonResponse(http.StatusOK, resp)
            })

            oddPath := fmt.Sprintf("%s/%s/%s/%s", baseUrl, enums.Sport.String(), resp[0].Key, enums.Odds.String())
            httpmock.RegisterResponder(http.MethodGet, oddPath, func(r *http.Request) (*http.Response, error) {
                oddResp := []domain.Odds{}
                oddResp = append(oddResp, twoWayArbodd)
                oddResp = append(oddResp, twoWayNotArbodd)
                oddResp = append(oddResp, threeWayArbodd)
                oddResp = append(oddResp, threeWayNotArbodd)

                return httpmock.NewJsonResponse(http.StatusOK, oddResp)
            })

            httpmock.Activate()
            defer httpmock.DeactivateAndReset()

            got, got1, err := us.GetArbs(tt.args.ctx, tt.args.oddsParams)

            if (err != nil) != tt.wantErr {
                t.Errorf("UseCasesArbsImpl.GetArbs() error = %v, wantErr %v", err, tt.wantErr)
                return
            }

            if !reflect.DeepEqual(len(got), tt.want) {
                t.Errorf("UseCasesArbsImpl.GetArbs() got = %v, want %v", len(got), tt.want)
            }

            if !reflect.DeepEqual(len(got1), tt.want1) {
                t.Errorf("UseCasesArbsImpl.GetArbs() got1 = %v, want %v", len(got1), tt.want1)
            }
        })
    }
}

Here are the tests to validate the logic runs as expected.

Arbs.go

package arbs

import (
    "context"
    "log"

    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/domain"
    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/infrastructure/services"
    "github.com/robinmuhia/arbitrageClient/pkg/oddsWrapper/usecases/arbs"
)

var client services.ArbClient

type ArbsParams struct {
    Region     string
    Markets    string
    OddsFormat string
    DateFormat string
}

func init() {
    svc, err := services.NewServiceOddsAPI()
    if err != nil {
        log.Panic("error: %w", err)
    }

    client = svc
}

// GetAllArbs returns all possible arbs from the odds API.
func GetAllArbs(ctx context.Context, arbParams ArbsParams) ([]domain.ThreeOddsArb, []domain.TwoOddsArb, error) {
    us := arbs.UseCasesArbsImpl{
        OddsApiClient: client,
    }

    params := services.OddsParams{
        Region:     arbParams.Region,
        Markets:    arbParams.Markets,
        OddsFormat: arbParams.OddsFormat,
        DateFormat: arbParams.DateFormat,
    }

    return us.GetArbs(ctx, params)
}

Lastly, these are the functions we use to initialize our client and export only the GetAllArbs function to be used by others.

We have hidden all this complexity and only export one function for our users to use.

This library can thus be used as shown below;

package main

import (
  "context"

    "github.com/robinmuhia/arbitrageClient/arbs"
)

func main() {
    // Create a context
  ctx := context.Background()

  // We need to pass the params to get odds from specific formats
  // We currently only support decimal format for oddsFormat
  arbParams := arbs.ArbsParams{
              Region : "uk",
              Markets: "h2h",
              OddsFormat: "decimal",
              DateFormat: "iso",
             }
  threeArbs, twoArbs, err := arbs.GetAllArbs(ctx, arbParams)

  if err != nil {
    // handle err
  }

  fmt.Println(threeArbs, twoArbs)
}

Results

AOB

You can check the .github folder to learn how to configure a CI/CD for Golang services, a dependabot for Golang and an automatic workflow to mark issues as stale.

Implementing integration tests for a Django application with Elastic Search.

Robin Muhia — Sun, 17 Mar 2024 08:08:31 +0000

Introduction

In the first article, we delved into how elastic search works under the hood.

In our second article, we implemented elastic search in a Django application.

In this article, we will implement a CI/CD pipeline that can incorporate elastic search tests.

We will use the repository we cloned in our second article so catch up with those articles before proceeding with this blog.

Structure

Ensure you have a tests folder that matches the one below;

Test set up

We will be using pytest to test thus we need a pytest.ini file to configure our test settings

[pytest]
DJANGO_SETTINGS_MODULE = tests.test_settings
addopts = --ds=tests.test_settings -v  --durations=10 --cache-clear --create-db
python_files = tests.py test_*.py

Our test_settings.py file from above has the following configurations;

from elastic_search.config.settings import *  # noqa

ELASTICSEARCH_DSL_SIGNAL_PROCESSOR = "django_elasticsearch_dsl.signals.RealTimeSignalProcessor"
ELASTICSEARCH_DSL = {
    "default": {
        "hosts": ["http://localhost:9200"],
    },
}

We need to re-index documents with nested fields so we need the following utility function in our test file. It is located in elasticsearch_test.py file

from django.core.management import call_command


class ElasticSearchMixin:
    def create_elasticsearch_index(self):
        """Create elastic search indices."""
        call_command("search_index", "--rebuild", "-f")

In our conftest.py file, we have the following method that is used to clear data in indices between tests;

import pytest
from django.conf import settings
from elasticsearch import Elasticsearch


@pytest.fixture(autouse=True)
def clear_elasticsearch_indices():
    """
    Fixture to clear all Elasticsearch indices after each test case.
    """
    es = Elasticsearch(hosts=settings.ELASTICSEARCH_DSL["default"]["hosts"])

    indices = es.options().indices.get_alias(index="*").keys()

    for index_name in indices:
        es.options(ignore_status=[404, 400]).indices.delete(index=index_name)
    yield

Finally, our tests;
We first test to see if our documents are being properly indexed
.

import pytest
from django.test import TestCase
from model_bakery import baker

from elastic_search.books.documents import BookDocument
from elastic_search.books.models import Author, Book, Country, Genre
from tests.utils.elasticsearch_test import ElasticSearchMixin

pytestmark = pytest.mark.django_db


@pytest.mark.django_db
class BookDocumentTest(TestCase, ElasticSearchMixin):
    def setUp(self):
        """Set up test suite."""
        self.create_elasticsearch_index()
        self.genre = baker.make(
            Genre,
            name="action",
        )
        self.author = baker.make(Author, name="Dave Muhia")
        self.country = baker.make(Country, name="Kenya")
        self.book = baker.make(
            Book,
            title="Edge of Tommorrow",
            description="Best movie",
            year=2014,
            rating=4.2,
            author=self.author,
            genre=self.genre,
            country=self.country,
        )

    def test_document_indexing(self):
        """Make sure document is properly indexed."""
        book_hits = BookDocument().search().execute()
        assert book_hits is not None
        book_hit = book_hits.hits[0]
        assert book_hit.genre.name == "action"
        assert book_hit.country.name == "Kenya"
        assert book_hit.author.name == "Dave Muhia"
        assert book_hit.title == "Edge of Tommorrow"
        assert book_hit.description == "Best movie"
        assert book_hit.year == 2014
        assert book_hit.rating == 4.2

    def test_get_queryset(self):
        queryset = BookDocument().get_queryset()
        assert len(queryset) == 1

    def test_get_instances_from_related(self):
        """Test retrieval of related models."""
        book_document = BookDocument()

        genre_instance = book_document.get_instances_from_related(self.genre)
        assert genre_instance is not None
        assert len(genre_instance) == 1
        assert genre_instance[0].title == self.book.title

        author_instance = book_document.get_instances_from_related(self.author)
        assert author_instance is not None
        assert len(author_instance) == 1
        assert author_instance[0].title == self.book.title

        country_instance = book_document.get_instances_from_related(self.country)
        assert country_instance is not None
        assert len(country_instance) == 1
        assert country_instance[0].title == self.book.title

Then we test our views;

import pytest
from django.test import TestCase
from model_bakery import baker
from rest_framework.test import APITestCase

from elastic_search.books.models import Author, Book, Country, Genre
from tests.utils.elasticsearch_test import ElasticSearchMixin

pytestmark = pytest.mark.django_db


@pytest.mark.django_db
class BookDocumentTest(APITestCase, TestCase, ElasticSearchMixin):
    def setUp(self):
        """Set up test suite."""
        self.create_elasticsearch_index()
        self.genre = baker.make(
            Genre,
            name="action",
        )
        self.author = baker.make(Author, name="Dave Muhia")
        self.country = baker.make(Country, name="Kenya")
        self.book = baker.make(
            Book,
            title="Edge of Tommorrow",
            description="Best movie",
            year=2014,
            rating=4.2,
            author=self.author,
            genre=self.genre,
            country=self.country,
        )
        self.genre2 = baker.make(
            Genre,
            name="action",
        )
        self.author2 = baker.make(Author, name="John Connor")
        self.country2 = baker.make(Country, name="Kenya")
        self.book2 = baker.make(
            Book,
            title="Demon Slayer",
            description="Tengen Uzui is the GOAT. All your hashiras were killed.",
            year=2023,
            rating=5,
            author=self.author2,
            genre=self.genre2,
            country=self.country2,
        )

    def test_search_endpoint(self):
        """Test the search endpoint."""
        base_url = "/api/books/search/"
        # no search params
        response = self.client.get(base_url)
        assert response.status_code == 200
        assert response.data["count"] == 2
        assert response.data["results"][0]["title"] == self.book.title
        assert response.data["results"][1]["title"] == self.book2.title

        # test search
        response = self.client.get(base_url + "?search=Tommorrow")
        assert response.status_code == 200
        assert response.data["count"] == 1
        assert response.data["results"][0]["title"] == self.book.title

        # test fuzzy search on description
        response = self.client.get(base_url + "?search=kilted")
        assert response.status_code == 200
        assert response.data["count"] == 1
        assert response.data["results"][0]["title"] == self.book2.title

        # test filters
        response = self.client.get(base_url + "?search=a&year=2023")
        assert response.status_code == 200
        assert response.data["count"] == 1
        assert response.data["results"][0]["title"] == self.book2.title

CI/CD

Instead of mocking, we want to be able to test our application against an actual elastic search instance. Hence the following CI pipeline.

name: build and deploy

on:
  push:
    branches:
      - main
  pull_request:
    branches:
      - main

jobs:
  build:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:latest
        env:
          POSTGRES_DB: elastic
          POSTGRES_PASSWORD: elastic
          POSTGRES_USER: elastic
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
      elasticsearch_test:
        image: docker.elastic.co/elasticsearch/elasticsearch:8.10.2
        env:
          cluster.name: docker-cluster
          discovery.type: single-node
          bootstrap.memory_lock: "true"
          xpack.security.enabled: "false"
          ES_JAVA_OPTS: "-Xms256m -Xmx256m"
        ports:
          - 9200:9200
        options: --health-cmd="curl -f http://localhost:9200/_cluster/health || exit 1" --health-interval=10s --health-timeout=5s --health-retries=3

    steps:
      - name: Checkout code
        uses: actions/checkout@v2

      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: "3.11"

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt

      - name: Run tests
        run: |
          pytest

This configuration allows us to do so.

Viola

Conclusion

We have understood elastic search, implemented it and even added some tests to it ensuring that the workflow we intended is achieved.

Implementation of Elastic Search in Django

Robin Muhia — Sun, 17 Mar 2024 06:46:56 +0000

Introduction

In the first article, we delved into how elastic search works under the hood.

In this article, we will implement elastic search in a Django application.

This article is intended for someone familiar with Django, we will not be explaining setup deeply or functionality such as models and views.

Setup

Clone this repository into a folder of your choosing.

git clone git@github.com:robinmuhia/elasticSearchPOC.git .

Get the repo from this Github link

We need three specific libraries that we will use as they abstract a lot of what we need to implement elastic search.

django-elasticsearch-dsl==8.0
elasticsearch==8.0.0
elasticsearch-dsl==8.12.0

Create a virtual environment, activate it and install the dependencies in the requirements.txt file

python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

Your project structure should look like the image below;

Now we're ready to go.

Understanding the project

Settings file

The project is a simple Django application. It has your usual setup structure.

In the config folder, we have our settings.py file.
For the purpose of this project, our elastic search settings are simple as shown below;

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "whitenoise.runserver_nostatic",
    "django.contrib.staticfiles",
    "django_extensions",
    "django_elasticsearch_dsl",
    "rest_framework",
    "elastic_search.books",
]

ELASTICSEARCH_DSL = {
    "default": {
        "hosts": [os.getenv("ELASTICSEARCH_URL", "http://localhost:9200")],
    },
}
ELASTICSEARCH_DSL_SIGNAL_PROCESSOR = "django_elasticsearch_dsl.signals.RealTimeSignalProcessor"
ELASTICSEARCH_DSL_INDEX_SETTINGS = {}
ELASTICSEARCH_DSL_AUTOSYNC = True
ELASTICSEARCH_DSL_AUTO_REFRESH = True
ELASTICSEARCH_DSL_PARALLEL = False

In a production ready application, i would recommend using the CelerySignalProcessor. The RealTimeSignalProcessor re-indexes documents immediately any changes are made to a model. CelerySignalProcessor would handle the re-indexing asynchronously to ensure that our users would not have to experience added latency when they modify any of our models. You would have to set up Celery though.

Read more about the nuances of settings here.

Models

from django.db import models


class GenericMixin(models.Model):
    """Generic mixin to be inherited by all models."""

    id = models.AutoField(primary_key=True, editable=False, unique=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        abstract = True
        ordering = ("-updated_at", "-created_at")
class Country(GenericMixin):
    name = models.CharField(max_length=200)

    def __str__(self):
        return self.name


class Genre(GenericMixin):
    name = models.CharField(max_length=100)

    def __str__(self):
        return self.name


class Author(GenericMixin):
    name = models.CharField(max_length=200)

    def __str__(self):
        return self.name


class Book(GenericMixin):
    title = models.CharField(max_length=100)
    description = models.TextField()
    genre = models.ForeignKey(Genre, on_delete=models.CASCADE, related_name="genres")
    country = models.ForeignKey(Country, on_delete=models.CASCADE, related_name="countries")
    author = models.ForeignKey(Author, on_delete=models.CASCADE, related_name="authors")
    year = models.IntegerField()
    rating = models.FloatField()

    def __str__(self):
        return self.title

The Generic Mixin has fields that should be inherited by all models. For a production application, i would recommend using a UUID as a primary key but we will use a normal incrementing integer field as it is easier for this project.

The models are pretty self-explanatory but we will be indexing and querying the book model. Our goals are to be able to search for a book using its title and description, while also being able to filter year and rating.

Documents file

We have documents.py file in the books folder.
This folder is important and should be named as such. Our documents will be written here. For our book model, the code is shown below;

from django_elasticsearch_dsl import Document, fields
from django_elasticsearch_dsl.registries import registry

from elastic_search.books.models import Author, Book, Country, Genre


@registry.register_document
class BookDocument(Document):
    genre = fields.ObjectField(
        properties={
            "name": fields.TextField(),
        }
    )
    country = fields.NestedField(
        properties={
            "name": fields.TextField(),
        }
    )
    author = fields.NestedField(
        properties={
            "name": fields.TextField(),
        }
    )

    class Index:
        name = "books"

    class Django:
        model = Book
        fields = [
            "title",
            "description",
            "year",
            "rating",
        ]

        related_models = [Genre, Country, Author]

    def get_queryset(self):
        return super().get_queryset().select_related("genre", "author", "country")

    def get_instances_from_related(self, related_instance):
        if isinstance(related_instance, Genre):
            return related_instance.genres.all()
        elif isinstance(related_instance, Country):
            return related_instance.countries.all()
        elif isinstance(related_instance, Author):
            return related_instance.authors.all()
        else:
            return []

Import Statements:

We import necessary modules and classes from django_elasticsearch_dsl and our Django models.

Document Definition:

We define a BookDocument class which inherits from Document, provided by django_elasticsearch_dsl.

Registry Registration:

We register the BookDocument class with the registry using the @registry.register_document decorator. This tells the Elasticsearch DSL library to manage this document.

Index Configuration:

We specify the name of the Elasticsearch index for this document as "books". This index name should be unique within the Elasticsearch cluster.

Django Model Configuration:

Under the Django class nested within BookDocument, we link the document to the Django model (Book) and specify which fields of the model should be indexed.

Fields Mapping:

Inside the BookDocument class, we define fields for the Elasticsearch document. These fields map to the fields in the Django model. Some fields, such as genre, country, and author, are nested objects.

Related Models Handling:

We specify related models (Genre, Country, Author) that should be indexed along with the Book model. For each related model, we define how to retrieve instances related to the main model. This involves specifying which fields to index from related models.

Queryset Configuration:

We override the get_queryset method to specify how the queryset should be retrieved. In this case, we use select_related to fetch related objects efficiently.

Instances from Related:

We define the get_instances_from_related method to handle instances from related models. This method is used to retrieve instances related to the main model for indexing purposes.

Views

import copy
from abc import abstractmethod

from elasticsearch_dsl import Document, Q
from rest_framework.decorators import action
from rest_framework.pagination import LimitOffsetPagination
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.viewsets import ModelViewSet

from elastic_search.books.documents import BookDocument
from elastic_search.books.models import Book
from elastic_search.books.serializers import BookSerializer


class PaginatedElasticSearchAPIView(ModelViewSet, LimitOffsetPagination):
    document_class: Document = None

    @abstractmethod
    def generate_search_query(self, search_terms_list, param_filters):
        """This method should be overridden
        and return a Q() expression."""

    @action(methods=["GET"], detail=False)
    def search(self, request: Request):
        try:
            params = copy.deepcopy(request.query_params)
            search_terms = params.pop("search", None)
            query = self.generate_search_query(
                search_terms_list=search_terms, param_filters=params
            )

            search = self.document_class.search().query(query)
            response = search.to_queryset()

            results = self.paginate_queryset(response)
            serializer = self.serializer_class(results, many=True)

            return self.get_paginated_response(serializer.data)
        except Exception as e:
            return Response(e, status=500)


class BookViewSet(PaginatedElasticSearchAPIView):
    serializer_class = BookSerializer
    queryset = Book.objects.all()
    document_class = BookDocument

    def generate_search_query(self, search_terms_list: list[str], param_filters: dict):
        if search_terms_list is None:
            return Q("match_all")
        search_terms = search_terms_list[0].replace("\x00", "")
        search_terms.replace(",", " ")
        search_fields = ["title", "description"]
        filter_fields = ["year", "rating"]
        query = Q("multi_match", query=search_terms, fields=search_fields, fuzziness="auto")

        wildcard_query = Q(
            "bool",
            should=[
                Q("wildcard", **{field: f"*{search_terms.lower()}*"}) for field in search_fields
            ],
        )
        query = query | wildcard_query

        if len(param_filters) > 0:
            filters = []
            for field in filter_fields:
                if field in param_filters:
                    filters.append(Q("term", **{field: param_filters[field]}))
            filter_query = Q("bool", should=[query], filter=filters)
            query = query & filter_query

        return query

Structure

The PaginatedElasticSearchAPIView class has two important methods. The generate search query method has an abstractmethod decorator which means that any class that inherits it has to implement said method.

The other search method adds an endpoint search that accepts a get request and handles the search functionality . It copies the parameters from the URL and then passes the parameters to the generate search query function. The function should return an Elasticsearch Query which will be searched from and then converted to a queryset. The queryset will be paginated over and returned to the user.

In a production app, i would recommend handling the exception by logging the error and defaulting to use Django Rest Framework's search so at the least our search will always work.

Implementation

In the BookViewSet, we provide the document that we will execute the search on.

We also implement the abstract method. Let us explain the query one by one.

Input Parameters:

search_terms_list: These are the words or phrases a user types into the search bar when looking for a book.

param_filters: These are additional filters or conditions a user might want to apply to narrow down their search, like searching only for books published in a certain year or have a certain rating.

Understanding the Search Process:

If the user doesn't provide any search terms, it means they want to see all the books available. So, we create a "match-all" query to fetch all books.

If the user provides search terms, we want to look for those terms in specific fields of our books, like title or description. We also want to be flexible with our search, allowing for slight misspellings or variations in the search terms. That's where the "fuzziness" parameter comes into play. It helps us find similar words even if the user misspells something.

Additionally, we might want to support wildcard searches, where a user can use placeholders like '' to match any characters. For example, searching for 'hist' would match 'history', 'historic', etc.

If there are any filter parameters provided, we want to apply those filters to our search results. For example, if a user wants to see only books published in the year 2022, we want to include that condition in our search.

Constructing the Query:

We use the Elasticsearch DSL (Domain-Specific Language) to construct our search query. This query is like a set of instructions written in a language Elasticsearch understands.
We build our query step by step, considering all the different scenarios mentioned above.
We use the Q class from Elasticsearch DSL to create different parts of our query, such as match queries, wildcard queries, and filter queries.
Finally, we combine all these parts to form a comprehensive search query that captures both the user's search terms and any additional filters they might have applied.

Output:

The method returns the constructed search query, ready to be executed against our Elasticsearch index.
This query will fetch the relevant books based on the user's search terms and filters, providing them with accurate and tailored search results.

URLS

We now setup this up in our urls.py file;

from rest_framework.routers import SimpleRouter

from elastic_search.books import views

router = SimpleRouter()

router.register("books", views.BookViewSet)

urlpatterns = router.urls

Data

We need data to search against and thus there is a factories.py file that will populate the data for us in a db.

First lets create a database;
Set up postgres and run the following commands;

sudo -u postgres psql
DROP USER IF EXISTS elastic;

CREATE USER elastic WITH CREATEDB CREATEROLE SUPERUSER LOGIN PASSWORD 'elastic';

DROP DATABASE IF EXISTS elastic;

CREATE DATABASE elastic WITH OWNER postgres;

GRANT ALL ON DATABASE elastic TO elastic;
\q

Populate the data in the db;

python manage.py generate_test_data 1000

This will create a large dataset for us to run our queries against

Set up elastic search

Run the following to start a local elasticsearch instance with docker

docker run --rm --name elasticsearch_container -p 9200:9200 -e "discovery.type=single-node" -e "xpack.security.enabled=false" elasticsearch:8.10.2

Populate index

Now we can populate our index to test out the application

python manage.py search_index --rebuild

Query Time!!!

Start the server

python manage.py runserver

Head to postman or any API testing platform of your choice

Our base query will be this

http://localhost:8000/api/books/search/

A get request is shown below;

lets make a query for a movie with consumer;

Lets misspell consumer, we get same result

lets test the filter;

Conclusion

We have implemented elastic search and tested it live. We have got expected results. There exists other queries such as nested queries that can be added to include author and country into the search and filters but they are out of the scope of this tutorial. In a future article, i may add them. However, in our next article, we will add a CI/CD pipeline that can be used to test our application.

Elastic search: Why should you use it and how does it work?

Robin Muhia — Sat, 10 Feb 2024 09:24:08 +0000

Introduction

Recently, in the health-tech company where I work, I was tasked with introducing and implementing Elasticsearch to our CRM repository. Our Chief Product Officer gave me three weeks to complete this project, and while the requirements initially seemed straightforward, the process turned out to be more complex than anticipated.

The objectives were clear:

Implement Elasticsearch search functionality.
Ensure the code is testable.
Document the implementation thoroughly.

In this series of articles, I will walk through the process of integrating Elasticsearch into a simple Django repository. The first article will provide a theoretical overview, followed by a second article detailing the implementation and programming aspects. The final article will focus on testing and running these tests within a GitLab/GitHub CI environment.

Prerequisites

Before diving into the implementation, I highly recommend familiarizing yourself with the following resources:

Additionally, I want to acknowledge the invaluable insights gained from the following articles:

What is Elasticsearch?

Elasticsearch is a specialized search engine designed for efficiently querying and retrieving data. It operates on the Java platform and leverages a sophisticated indexing mechanism known as the Generalized Inverted Index.

Elasticsearch is mainly used for searching as the name suggests. A lot of users read requests usually entail some sort of search, we use Elasticsearch to spread out these read requests from our main database to Elasticsearch. Additionally, Elasticsearch allows more intuitive search by allowing fuzzy-searching that caters for misspellings, auto-completion and other powerful searching capabilities.

Intricacies of Elasticsearch

Cluster

A cluster in Elasticsearch is a collection of one or more nodes (servers) that collectively store data and provide indexing and search capabilities. Clusters are used for scalability, fault tolerance, and load distribution. They are similarly to kubernetes clusters.

Node

A node is a single server that is part of a cluster. It stores data, participates in the cluster’s indexing and search capabilities, and can be configured to perform specific roles such as master-eligible or data node. They are similarly to kubernetes nodes. Think of them as a virtual machine or a laptop in your case.

Shard

A shard is a basic unit of data in Elasticsearch. It is a subset of an index, containing a portion of the index's data. Elasticsearch distributes shards across nodes in the cluster to enable parallel processing and scalability.

Replica

A replica is a copy of a shard. Replicas provide fault tolerance and high availability by allowing data to be replicated across multiple nodes. Elasticsearch automatically manages the distribution of replicas to ensure data resilience.

Type (Deprecated in Elasticsearch 7.x)

In older versions of Elasticsearch, a type was a way to logically partition data within an index (An index is like a table in a relational database). However, starting from Elasticsearch 7.x, types are deprecated, and indices can only contain a single mapping type.

Document

A document is a basic unit of information stored in Elasticsearch. It is a JSON object that contains data and its associated metadata. Documents are indexed and stored in shards based on their index and type. Think of it as a row in a relational database.

Field

A field is a key-value pair within a document that represents a specific attribute or property of the data. Fields can be indexed for search, aggregated for analysis, and retrieved in query results. Think of this as a column in a relational database.

Strategies used by Elastic Search to store data

A Generalized Inverted Index (GIN) is a data structure used in databases to efficiently support complex queries. It stores lists of indexed items along with their corresponding keys, enabling fast retrieval based on search terms. GIN indexes are particularly effective for full-text search and support various operations like AND, OR, and phrase searches

Traditional storage engines often utilize indices by mapping unique keys to corresponding values. For instance, PostgreSQL uses B-trees, where a key is associated with a file offset pointing to the stored data. In contrast, Elasticsearch adopts a different approach: it maps textual data to unique keys, which are then linked to lists of postings representing the documents containing that text. This strategy facilitates rapid searching and retrieval.To enable this, text needs to undergo stemming and lemmatisation as well the removal of stop words.

Stop words are words that are common to a language. For example in English, words like "the" , "and" , "or" are common words that do not need to be stored.

Lemmatisation and stemming are techniques used in natural language processing to reduce words to their base or root form. Lemmatisation aims to accurately identify a word's lemma, or dictionary form, considering its context and grammatical features. Stemming, on the other hand, applies simpler rules to remove suffixes and prefixes from words, often resulting in the root form but sometimes leading to inaccuracies. While lemmatisation produces linguistically valid roots, stemming is faster and more suitable for tasks where linguistic accuracy is less critical, such as information retrieval.

Lemma: The lemma of "running" is "run". Lemmatisation would convert "running" to "run" by considering its context and grammatical features.

Stem: The stem of "running" using a stemming algorithm might be "run". Stemming applies simple rules to remove suffixes, so it may not always produce a valid root or lemma. In this case, "run" is a valid stem, but it's not necessarily the correct lemma.

Lets do an example for the famous sentence with all letters in the alphabet;

The quick brown fox jumped over the lazy dog.

Remove stop words: The stop words in the given sentence are typically common words like "the", "of", "and", etc. After removing them, we get:

"quick brown fox jumped lazy dog."

Lemmatization: Lemmatization involves reducing words to their base or dictionary form. Here, we can use lemmatization to convert words like "jumped" to "jump",

Stemming: Stemming involves removing affixes from words to obtain their root forms. For example, "jumped" can be stemmed to "jump", "brown" remains "brown", etc.

We thus get

quick brown fox jump lazy dog.

Example

The pen will be used to write in a red book by the students.
The red book is hated by most of our students.
The students will write to the red book.

Remove stop words:

"pen used write red book students"
"red book hated students."
"pen used write book."

Lemmatization:

"pen use write red book student."
"red book hate student."
"student use red book."

Stemming:

"pen use write red book student."
"red book hate student."
"student write red book."

Now we index these words:

Texts are mapped to unique keys which are mapped to the postings.

Text	Unique Keys	List of Postings
pen	1	[1]
use	2	[1,3]
write	3	[1,3]
red	4	[1,3]
book	5	[1,3]
student	6	[1,2,3]
hate	7	[2]

How Querying works under the hood?

Scoring in Elasticsearch refers to the process of ranking search results based on their relevance to the query. Elasticsearch employs various factors to calculate scores, including the frequency of search terms within documents and their proximity to each other.

Boosting allows users to assign greater importance to certain fields or documents when calculating relevance scores. This feature enables fine-tuning of search results to emphasize specific criteria.

Must and should are terms used in Elasticsearch's Boolean query syntax to define mandatory and optional criteria, respectively. Must clauses must match for a document to be considered a relevant result, while should clauses enhance the score of documents if they match but are not required for a successful search.

Fuzziness in Elasticsearch refers to the capability to find approximate matches for a given search term. This feature is particularly useful for handling typographical errors, misspellings, and variations in word forms. Elasticsearch employs algorithms like the Levenshtein distance algorithm to determine the similarity between terms and their potential matches.

Now you can query as appropriately as you want. It is up to you as the relevance engineer to choose what fields are appropriate to boost so as to make the query more relevant.

From a business perspective, we want our search to be rich and relevant so that our users can find it useful. The search also has to make business sense by for example recommending expiring products and products that have higher margins. Such factors should be considered when effectively building a query.

In summary, Elasticsearch offers a powerful and flexible solution for searching and retrieving data, leveraging advanced indexing techniques and query capabilities. Understanding its features, such as scoring, boosting, Boolean queries, and fuzzy matching, allows developers to build efficient and accurate search applications tailored to their specific needs.

Conclusion

In conclusion, Elasticsearch is a powerful tool for building scalable, real-time search applications. Understanding its architecture and intricacies, such as clusters, nodes, shards, and documents, is essential for effectively implementing and managing Elasticsearch in a production environment. By leveraging its capabilities, developers can create robust search solutions that meet the demands of modern applications.

The next two articles will delve into coding so stay tuned!!