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:

$ mkts -rootDir /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 NewCandler(inputColumns, optionalInputColumns []io.DataShape) (ca *Candler)
  • func (ca *Candler) Accum(cols io.ColumnInterface) 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) Init(args ...interface{}) error
  • func (c Candler) New() (ca *Candler, am *functions.ArgumentMap)
  • func (ca *Candler) Output() *io.ColumnSeries
  • func (ca *Candler) Reset()
• type EOHLCStruct

Functions

func GetAverageColumnFloat32

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

Utility Functions

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

Map of start times to active candles

type Candler

type Candler struct {
	uda.AggInterface

	// Input arguments mapping
	ArgMap *functions.ArgumentMap
	/*
	   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 NewCandler

func NewCandler(inputColumns, optionalInputColumns []io.DataShape) (ca *Candler)

CONCRETE - these may be suitable methods for general usage

func (*Candler) Accum

func (ca *Candler) Accum(cols io.ColumnInterface) error

OVERRIDES - these methods should be overridden in a concrete implementation of this class

OVERRIDE THIS METHOD
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) Init

func (ca *Candler) Init(args ...interface{}) error

func (Candler) New

func (c Candler) New() (ca *Candler, am *functions.ArgumentMap)

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

func (*Candler) Reset

func (ca *Candler) Reset()

Reset() puts the aggregate state back to "new"

type EOHLCStruct

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

*********************************** Candle ******************************************* - Represents quantities within an interval of time - Always has Open, High, Low, Close prices defined within - Optionally has averaged and summed quantities inside - Has a starting time representing the candle, it begins the interval *********************************** Candle *******************************************