candler

README

Financial Analysis UDA: Candler - converts raw price data to Candles for technical analysis

This module aggregates time series data into candles using the marketstore UDA interface.

The interesting functions can be called from SQL

Example:

$ marketstore connect --dir /data/market-data/mktsdb/
» select candlecandler('12Min',Open,High,Low,Close,Avg::Volume,Sum::Volume) from `TSLA/1Min/OHLCV` where Epoch > '2017-01-01' limit 10;
              =============================  ==========  ==========  ==========  ==========  ==========  ==========
                                      Epoch  Open        High        Low         Close       Volume_SUM  Volume_AVG
              =============================  ==========  ==========  ==========  ==========  ==========  ==========
              2017-01-03 14:24:00 +0000 UTC  214.86      215.3       210.96      212.5378    342337      57056.168
              2017-01-03 14:36:00 +0000 UTC  212.555     216.143     211.4       215.82      501743      41811.918
              2017-01-03 14:48:00 +0000 UTC  215.82      218.39      215.39      217.775     520423      43368.582
              2017-01-03 15:00:00 +0000 UTC  217.82      220.33      217.56      220.255     667460      55621.668
              2017-01-03 15:12:00 +0000 UTC  220.29      220.29      218.77      219.65      255877      21323.084
              2017-01-03 15:24:00 +0000 UTC  219.63      220         218.9       219.59      210390      17532.5
              2017-01-03 15:36:00 +0000 UTC  219.55      219.87      219.0444    219.1       140869      11739.083
              2017-01-03 15:48:00 +0000 UTC  219.18      219.25      217.46      217.71      203272      16939.334
              2017-01-03 16:00:00 +0000 UTC  217.72      218.25      216.59      217.87      122625      10218.75
              2017-01-03 16:12:00 +0000 UTC  218.04      218.317     216.61      216.68      87799       7316.5835
              =============================  ==========  ==========  ==========  ==========  ==========  ==========
              Elapsed query time: 63.754 ms

Notice that the parameters of the candlecandler() function in some cases have a notation like "AVG:Volume" - those are dynamic tags used by the function to take a variable number of inputs which are used by the candlecandler().

/*
	Accum() sends new data to the aggregate
*/
func (ca *CandleCandler) Accum(cols io.ColumnInterface) error {
	if cols.Len() == 0 {
		return fmt.Errorf("Empty input to Accum")
	}
	/*
		Get the input column for "Price"
	*/
	openCols := ca.ArgMap.GetMappedColumns(requiredColumns[0].Name)
	highCols := ca.ArgMap.GetMappedColumns(requiredColumns[1].Name)
	lowCols := ca.ArgMap.GetMappedColumns(requiredColumns[2].Name)
	closeCols := ca.ArgMap.GetMappedColumns(requiredColumns[3].Name)
	open, err := candler.GetAverageColumnFloat32(cols, openCols)
	if err != nil {
		return err
	}
	high, err := candler.GetAverageColumnFloat32(cols, highCols)
	if err != nil {
		return err
	}

Documentation

Index

• func GetAverageColumnFloat32(cols io.ColumnInterface, srcCols []io.DataShape) (avgCol []float32, err error)
• type Candle
  • func NewCandle(startTime time.Time, cd *utils.CandleDuration, sumColumns, avgColumns []string) (ca *Candle)
  • func (ca *Candle) AddCandle(ts time.Time, prices ...float32) bool
  • func (ca *Candle) IsWithin(ts time.Time) bool
  • func (ca *Candle) SerializeToRowData(sumNames, avgNames []string) (rowBuf []byte)
• type CandleMap
• type Candler
  • func (ca *Candler) Accum(_ io.TimeBucketKey, _ *functions.ArgumentMap, _ io.ColumnInterface, ...) (*io.ColumnSeries, error)
  • func (ca *Candler) GetCandle(t time.Time, cndl ...*Candle) *Candle
  • func (ca *Candler) GetInitArgs() []io.DataShape
  • func (ca *Candler) GetOptionalArgs() []io.DataShape
  • func (ca *Candler) GetRequiredArgs() []io.DataShape
  • func (ca *Candler) New(argMap *functions.ArgumentMap, args ...interface{}) (c *Candler, err error)
  • func (ca *Candler) Output() *io.ColumnSeries
• type EOHLCStruct

Functions

func GetAverageColumnFloat32

func GetAverageColumnFloat32(cols io.ColumnInterface, srcCols []io.DataShape) (avgCol []float32, err error)

Types

type Candle

type Candle struct {
	StartTime time.Time
	Duration  *utils.CandleDuration
	/*
		Every candle has OHLC
	*/
	EOHLC               EOHLCStruct
	OpenTime, CloseTime time.Time // The time at which the Open and Close prices happened
	/*
		Some candles optionally sum quantities like "Volume" from the
		input columns
	*/
	SumMap map[string]float64 // One sum per mapped column
	Count  int64              // Counts the elements incorporated in the Sums
	/*
		Does this candle have complete data? Sometimes we can tell...
	*/
	Complete bool
}

func NewCandle

func NewCandle(startTime time.Time, cd *utils.CandleDuration, sumColumns, avgColumns []string) (ca *Candle)

func (*Candle) AddCandle

func (ca *Candle) AddCandle(ts time.Time, prices ...float32) bool

func (*Candle) IsWithin

func (ca *Candle) IsWithin(ts time.Time) bool

func (*Candle) SerializeToRowData

func (ca *Candle) SerializeToRowData(sumNames, avgNames []string) (rowBuf []byte)

type CandleMap

type CandleMap map[time.Time]*Candle

CandleMap is a Map of start times to active candles.

type Candler

type Candler struct {
	uda.AggInterface

	/*
	   Manages one timeframe, creates candles in that timeframe from
	   input and outputs them on demand.
	*/
	MyCD *utils.CandleDuration
	CMap CandleMap
	/*
	   We need to keep an ordered list of sum and avg names to ensure ordered output
	*/
	SumNames, AvgNames []string // A cache of the column names that are summed and averaged in the candles
	AccumSumNames      []string // Consolidated list of names either summed or averaged
}

func (*Candler) Accum

func (ca *Candler) Accum(_ io.TimeBucketKey, _ *functions.ArgumentMap, _ io.ColumnInterface, _ *catalog.Directory,
) (*io.ColumnSeries, error)

Accum sends new data to the aggregate.

func (*Candler) GetCandle

func (ca *Candler) GetCandle(t time.Time, cndl ...*Candle) *Candle

func (*Candler) GetInitArgs

func (ca *Candler) GetInitArgs() []io.DataShape

func (*Candler) GetOptionalArgs

func (ca *Candler) GetOptionalArgs() []io.DataShape

func (*Candler) GetRequiredArgs

func (ca *Candler) GetRequiredArgs() []io.DataShape

func (*Candler) New

func (ca *Candler) New(argMap *functions.ArgumentMap, args ...interface{}) (c *Candler, err error)

OVERRIDE THIS METHOD Creates a new candler using the arguments of the specific implementation for inputColumns and optionalInputColumns

func (*Candler) Output

func (ca *Candler) Output() *io.ColumnSeries

Output() returns the currently valid output of this aggregate

type EOHLCStruct

type EOHLCStruct struct {
	Epoch                  int64
	Open, High, Low, Close float32
}