tacitus/ops/aggregator.go

package ops

import (
	"github.com/kcotugno/tacitus"

	"sync"
	"time"
)

type Aggregator struct {
	Database tacitus.DatabaseClientService
	Logger   tacitus.Logger
	Products []string

	isDone bool
	doneMu sync.Mutex
}

func (a *Aggregator) Start(interval time.Duration) {
	go func() {
		a.setDone(false)

		var timer *time.Ticker

		correction := time.Until(time.Now().Truncate(interval).Add(interval))
		a.Logger.Info("aggregator: time correction")

		// We add 1.5 seconds to ensure that we do not miss one or two
		// that may still be being inserted into the databast.
		// This may be changed based off of further testing.
		time.Sleep(correction + (1500 * time.Millisecond))

		timer = time.NewTicker(interval)
		a.Logger.Info("aggregator: starting")

		for !a.done() {
			select {
			case <-timer.C:
				if a.Products == nil {
					a.Logger.Info("aggregator: products are nil")
					break
				}

				for _, p := range a.Products {
					go a.aggregate(p, interval)
				}

			}
		}
	}()
}

func (a *Aggregator) aggregate(product string, interval time.Duration) {
	a.Logger.Info(`aggregator: start product="%v" interval="%v"`, product, interval)

	end := time.Now().Truncate(interval)
	start := end.Add(-interval)

	trades, err := a.Database.TradeService().TradesInDateRange(product, start, end)
	if err != nil {
		a.Logger.Info(`aggregator: database error="%v"`, err)
		return
	}

	var agg tacitus.Aggregation
	agg.Interval = int(interval.Seconds())
	agg.Product = product
	agg.Timestamp = end

	for _, t := range trades {
		agg.Price = t.Price

		if t.Buy {
			agg.BuyVolume = agg.BuyVolume.Add(t.Size)
			agg.BuyTransactions++
		} else {
			agg.SellVolume = agg.SellVolume.Add(t.Size)
			agg.SellTransactions++
		}
	}

	_, err = a.Database.AggregationService().CreateAggregation(agg)
	if err != nil {
		a.Logger.Info(`aggregator: database error="%v"`, err)
	}

	a.Logger.Info(`aggregator: DONE product="%v" interval="%v"`, product, interval)
}

func (a *Aggregator) Stop() {
	a.setDone(true)
}

func (a *Aggregator) done() bool {
	a.doneMu.Lock()
	defer a.doneMu.Unlock()

	return a.isDone
}

func (a *Aggregator) setDone(done bool) {
	a.doneMu.Lock()
	defer a.doneMu.Unlock()

	a.isDone = done
}