-
Notifications
You must be signed in to change notification settings - Fork 225
/
candler.go
385 lines (347 loc) · 9.47 KB
/
candler.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
package candler
import (
"fmt"
"sort"
"time"
"github.com/alpacahq/marketstore/v4/catalog"
"github.com/alpacahq/marketstore/v4/uda"
"github.com/alpacahq/marketstore/v4/utils"
"github.com/alpacahq/marketstore/v4/utils/functions"
"github.com/alpacahq/marketstore/v4/utils/io"
)
/*
Superclass for the Candler subclasses
*/
var (
/*
This is filled in for example purposes, should be overridden in implementation
*/
requiredColumns = []io.DataShape{
{Name: "Open", Type: io.FLOAT32},
{Name: "High", Type: io.FLOAT32},
{Name: "Low", Type: io.FLOAT32},
{Name: "Close", Type: io.FLOAT32},
}
/*
For the optional inputs, we'll postpend the input names mapped to each optional
for output, for example: if we map user input "Volume" to "Sum", the output
will be "Sum_Volume"
*/
optionalColumns = []io.DataShape{
{Name: "Sum", Type: io.FLOAT32},
{Name: "Avg", Type: io.FLOAT32},
}
initArgs = []io.DataShape{
{Name: "Timeframe", Type: io.STRING},
}
)
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 (ca *Candler) GetRequiredArgs() []io.DataShape {
return requiredColumns
}
func (ca *Candler) GetOptionalArgs() []io.DataShape {
return optionalColumns
}
func (ca *Candler) GetInitArgs() []io.DataShape {
return initArgs
}
/*
OVERRIDES - these methods should be overridden in a concrete implementation of this class
*/
// Accum sends new data to the aggregate.
func (ca *Candler) Accum(_ io.TimeBucketKey, _ *functions.ArgumentMap, _ io.ColumnInterface, _ *catalog.Directory,
) (*io.ColumnSeries, error) {
return nil, fmt.Errorf("accum called from base class, must override implementation")
}
/*
OVERRIDE THIS METHOD
Creates a new candler using the arguments of the specific implementation
for inputColumns and optionalInputColumns
*/
func (ca *Candler) New(argMap *functions.ArgumentMap, args ...interface{}) (c *Candler, err error) {
if len(args) != 1 {
return nil, fmt.Errorf("init requires a *utils.CandleDuration as the argument")
}
c = &Candler{
MyCD: nil,
CMap: nil,
SumNames: nil,
AvgNames: nil,
AccumSumNames: nil,
}
err = c.init(argMap, args...)
if err != nil {
return nil, err
}
return c, nil
}
func (ca *Candler) init(argMap *functions.ArgumentMap, args ...interface{}) error {
var tfstring string
switch val := args[0].(type) {
case string:
tfstring = val
case *string:
tfstring = *val
case *[]string:
if len(*val) != 1 {
return fmt.Errorf("argument passed to Init() is not a string")
}
tfstring = (*val)[0]
case []string:
if len(val) != 1 {
return fmt.Errorf("argument passed to Init() is not a string")
}
tfstring = val[0]
}
cd, err := utils.CandleDurationFromString(tfstring)
if ca == nil {
return fmt.Errorf("init called without calling New()")
}
if err != nil {
return fmt.Errorf("no suitable timeframe provided: %w", err)
}
if unmapped := argMap.Validate(); unmapped != nil {
return fmt.Errorf("unmapped columns: %s", unmapped)
}
ca.MyCD = cd
ca.CMap = make(CandleMap)
/*
Build the cache of summed input column names
*/
ca.SumNames = make([]string, 0)
ca.AvgNames = make([]string, 0)
ca.AccumSumNames = make([]string, 0)
for _, ds := range optionalColumns {
/*
This currently includes Sum and Avg
*/
switch ds.Name {
case "Sum":
for _, dds := range argMap.GetMappedColumns("Sum") {
ca.SumNames = append(ca.SumNames, dds.Name)
}
case "Avg":
for _, dds := range argMap.GetMappedColumns("Avg") {
ca.AvgNames = append(ca.AvgNames, dds.Name)
}
}
}
consMap := make(map[string]struct{})
for _, name := range ca.SumNames {
consMap[name] = struct{}{}
}
for _, name := range ca.AvgNames {
consMap[name] = struct{}{}
}
for name := range consMap {
ca.AccumSumNames = append(ca.AccumSumNames, name)
}
return nil
}
/*
Output() returns the currently valid output of this aggregate
*/
func (ca *Candler) Output() (*io.ColumnSeries, error) {
dataShapes := []io.DataShape{
{Name: "Epoch", Type: io.INT64},
{Name: "Open", Type: io.FLOAT32},
{Name: "High", Type: io.FLOAT32},
{Name: "Low", Type: io.FLOAT32},
{Name: "Close", Type: io.FLOAT32},
}
for _, name := range ca.SumNames {
dataShapes = append(dataShapes,
io.DataShape{Name: name + "_SUM", Type: io.FLOAT64})
}
for _, name := range ca.AvgNames {
dataShapes = append(dataShapes,
io.DataShape{Name: name + "_AVG", Type: io.FLOAT64})
}
var tsa uda.OrderedTime
for tkey := range ca.CMap {
tsa = append(tsa, tkey)
}
sort.Sort(tsa)
var dataBuf []byte
for _, tkey := range tsa {
cdl := ca.CMap[tkey]
dataBuf = append(dataBuf, cdl.SerializeToRowData(ca.SumNames, ca.AvgNames)...)
}
rows := io.NewRows(dataShapes, dataBuf)
return rows.ToColumnSeries()
}
func (ca *Candler) GetCandle(t time.Time, cndl ...*Candle) *Candle {
/*
Returns a candle matching the start time "t", or creates
a new one if one does not already exist
*/
candleTime := ca.MyCD.Truncate(t)
/*
If a candle is passed in, determine if it is suitable for this time
*/
if len(cndl) != 0 {
if cndl[0] != nil {
if cndl[0].StartTime == candleTime {
return cndl[0]
}
}
}
/*
We must provide a new candle if there isn't one with this time in the map
*/
if _, ok := ca.CMap[candleTime]; !ok {
ca.CMap[candleTime] = NewCandle(candleTime, ca.MyCD, ca.SumNames, ca.AvgNames)
}
return ca.CMap[candleTime]
}
/*
*********************************** 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 *******************************************.
*/
type EOHLCStruct struct {
Epoch int64
Open, High, Low, Close float32
}
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(startTime time.Time, cd *utils.CandleDuration, sumColumns, avgColumns []string) (ca *Candle) {
st := cd.Truncate(startTime)
ep := st.Unix()
ca = &Candle{
StartTime: st,
OpenTime: time.Time{},
CloseTime: time.Time{},
Duration: cd,
EOHLC: EOHLCStruct{ep, 0, 0, 0, 0},
}
if len(sumColumns) != 0 || len(avgColumns) != 0 {
ca.SumMap = make(map[string]float64)
for _, name := range sumColumns {
ca.SumMap[name] = 0
}
for _, name := range avgColumns {
ca.SumMap[name] = 0
}
}
return ca
}
func (ca *Candle) IsWithin(ts time.Time) bool {
return ca.Duration.IsWithin(ts, ca.StartTime)
}
func (ca *Candle) AddCandle(ts time.Time, prices ...float32) bool {
/*
This routine works with both ticks and candles as input
- A tick will have a single price
- A candle will have open, high, low, close prices
*/
var open, high, low, clos float32
if len(prices) == 1 {
open, high, low, clos = prices[0], prices[0], prices[0], prices[0]
} else {
open, high, low, clos = prices[0], prices[1], prices[2], prices[3]
}
/*
The input price is used to update the candle if the time is within
The return value indicates whether the time was within or not
*/
if !ca.IsWithin(ts) {
return false
}
if ca.OpenTime.IsZero() {
ca.EOHLC.Open, ca.EOHLC.High, ca.EOHLC.Low, ca.EOHLC.Close = open, high, low, clos
ca.OpenTime, ca.CloseTime = ts, ts
}
if ts.Before(ca.OpenTime) {
ca.EOHLC.Open = open
ca.OpenTime = ts
}
if ts.After(ca.CloseTime) {
ca.EOHLC.Close = clos
ca.CloseTime = ts
}
if high > ca.EOHLC.High {
ca.EOHLC.High = high
}
if low < ca.EOHLC.Low {
ca.EOHLC.Low = low
}
return true
}
func (ca *Candle) SerializeToRowData(sumNames, avgNames []string) (rowBuf []byte) {
rowBuf, _ = io.Serialize([]byte{}, ca.EOHLC)
for _, name := range sumNames {
rowBuf, _ = io.Serialize(rowBuf, ca.SumMap[name])
}
for _, name := range avgNames {
rowBuf, _ = io.Serialize(rowBuf, ca.SumMap[name]/float64(ca.Count))
}
return rowBuf
}
// CandleMap is a Map of start times to active candles.
type CandleMap map[time.Time]*Candle
/*
Utility Functions.
*/
func GetAverageColumnFloat32(cols io.ColumnInterface, srcCols []io.DataShape) (avgCol []float32, err error) {
numberCols := len(srcCols)
if numberCols == 1 {
name := srcCols[0].Name
col, err := uda.ColumnToFloat32(cols, name)
if err != nil {
return nil, err
}
return col, nil
}
/*
Average the input columns to produce the price column
*/
colLen := cols.Len()
avgCol = make([]float32, colLen)
for _, ds := range srcCols {
col, err := uda.ColumnToFloat32(cols, ds.Name)
if err != nil {
return nil, err
}
for i := 0; i < colLen; i++ {
avgCol[i] += col[i]
}
}
for i := 0; i < colLen; i++ {
avgCol[i] /= float32(numberCols)
}
return avgCol, nil
}