tickserver/framework/base/candle_calc.go

package base

//一个更加简单的计算K线的接口
//提供
//1. Peek
//2. Align
//3. Next
import "tickserver/markinfo"
import "encoding/binary"
import "io"
import "fmt"
import "unsafe"
import "errors"

type result struct {
	data  *OhlcGo
	isnew bool
	err   error
}

var ErrNoData = errors.New("NoData")
var ErrAlignTimeTooBig = fmt.Errorf("time > peek Time align time error.")
var tickSize = int(unsafe.Sizeof(TickGo{}))

type CandleCalc struct {
	candle    *Candle
	reader    *AsynReader
	tick      TickGo
	retTick   TickGo
	nextTick  TickGo
	num       int
	peek      int
	ret       result
	last      result
	symbolId  int
	period    int
	maxComp   int
	minPeriod int
}

func NewCandleCalc(symbolId int, period int, trans *TimezoneTrans, reader io.Reader) (*CandleCalc, error) {
	name, err := markinfo.SymbolName(symbolId)
	if err != nil {
		return nil, err
	}
	point, err := markinfo.SymbolPoint(name)
	if err != nil {
		return nil, err
	}
	candle, err := NewCandle(period, point, nil, 0)
	if err != nil {
		return nil, err
	}
	candle.SetTimezoneTrans(trans)
	calc := CandleCalc{}
	calc.candle = candle
	calc.reader = NewAsynReader(reader, tickSize)
	calc.symbolId = symbolId
	calc.period = period
	calc.maxComp = 1
	calc.minPeriod = 60
	return &calc, nil
}

func (calc *CandleCalc) SetConf(key int, conf interface{}) (int, error) {
	if key == CANDLE_AUTOCOMPLETE_MAX {
		calc.maxComp = conf.(int)
	}
	return calc.candle.Set(key, conf)
}

func (calc *CandleCalc) SetMinPeriod(period int) {
	calc.minPeriod = period
}

func (calc *CandleCalc) PeriodSecond() int {
	return calc.period
}

func (calc *CandleCalc) NodataAlignTime() int {
	return calc.maxComp
}

func (calc *CandleCalc) readTick(asyn bool) error {
	calc.retTick = calc.tick
	if calc.tick.Time == 0 { //first time
		err := calc.readNextTick(asyn)
		if err != nil {
			return err
		}
		calc.tick = calc.nextTick
		return nil
	} else {
		if calc.nextTick == calc.tick { //nextTick 已经被读取了
			err := calc.readNextTick(asyn)
			if err != nil {
				return err
			}
		}
		t1 := calc.minPeriod * (int(calc.nextTick.Time) / calc.minPeriod)
		t2 := calc.minPeriod * (int(calc.tick.Time) / calc.minPeriod)
		if t1-t2 > calc.minPeriod && t1-t2 < calc.maxComp {//需要补全,不需要读取next
			calc.tick.Time = int32(t2 + calc.minPeriod)
			return nil
		} else {
			calc.tick = calc.nextTick
			return nil
		}
	}
	return nil
}

//异步的读取tick
func (calc *CandleCalc) readNextTick(asyn bool) error {
	//保存上一个tick
	for {
		if asyn {
			ok, err := calc.reader.CanRead(tickSize)
			if err != nil {
				return err
			}
			if !ok {
				return ErrNoData
			}
		}
		err := binary.Read(calc.reader, binary.LittleEndian, &calc.nextTick)
		if err != nil {
			return err
		}
		//检查tick是否往回走了，避免这样的tick
		if calc.retTick.Time > calc.nextTick.Time || (calc.retTick.Time == calc.nextTick.Time && calc.retTick.Ms > calc.nextTick.Ms) {
			//直接忽略这样的tick
			fmt.Println("error tick.", calc.retTick, calc.nextTick)
			continue
		}
		break
	}
	return nil
}

func (calc *CandleCalc) Next() (*OhlcGo, bool, error) {
	if calc.peek > 0 {
		calc.peek = 0
		return calc.ret.data, calc.ret.isnew, calc.ret.err
	}
	if calc.num > 0 {
		calc.num--
		calc.ret.isnew = true
		return calc.readNext()
	}
	err := calc.readTick(false)
	if err != nil {
		return nil, false, err
	}
	calc.num = calc.candle.UpdateTick((*Tick)(unsafe.Pointer(&calc.tick)))
	if calc.num == 0 {
		calc.ret.isnew = false
		return calc.readNext()
	} else {
		calc.num--
		calc.ret.isnew = true
		return calc.readNext()
	}
	return nil, false, io.EOF
}

func (calc *CandleCalc) AsynNext() (*OhlcGo, bool, error) {
	if calc.peek > 0 {
		calc.peek = 0
		return calc.ret.data, calc.ret.isnew, calc.ret.err
	}
	if calc.num > 0 {
		calc.num--
		calc.ret.isnew = true
		return calc.readNext()
	}
	err := calc.readTick(true)
	if err != nil {
		if err == ErrNoData {
			return calc.ret.data, calc.ret.isnew, ErrNoData
		}
		return nil, false, err
	}
	calc.num = calc.candle.UpdateTick((*Tick)(unsafe.Pointer(&calc.tick)))
	if calc.num == 0 {
		calc.ret.isnew = false
		return calc.readNext()
	}
    calc.num--
    calc.ret.isnew = true
    return calc.readNext()
}

func (calc *CandleCalc) readNext() (*OhlcGo, bool, error) {
	ohlc := Ohlc{}
	calc.candle.Next(&ohlc)
	if calc.last.data != nil && calc.ret.data != nil && calc.last.data.Time == calc.ret.data.Time {
		calc.last.isnew = false
	}
	calc.last = calc.ret
	//赋值
	tmp := ohlc.ToGOStruct()
	calc.ret.data = &tmp
	calc.ret.err = nil
	//fmt.Println(tmp)
	return calc.ret.data, calc.ret.isnew, calc.ret.err
}

//只能向前查看一个数据
func (calc *CandleCalc) Peek() (*OhlcGo, bool, error) {
	if calc.peek == 0 || (calc.ret.err != nil && calc.ret.err.Error() == "ErrNoData") {
        calc.peek = 0
		calc.ret.data, calc.ret.isnew, calc.ret.err = calc.Next()
	}
	calc.peek++
	return calc.ret.data, calc.ret.isnew, calc.ret.err
}

func (calc *CandleCalc) AsynPeek() (*OhlcGo, bool, error) {
	if calc.peek == 0 || (calc.ret.err != nil && calc.ret.err.Error() == "ErrNoData") {
        calc.peek = 0
		calc.ret.data, calc.ret.isnew, calc.ret.err = calc.AsynNext()
	}
	calc.peek++
	return calc.ret.data, calc.ret.isnew, calc.ret.err
}

func (calc *CandleCalc) AsynAlign(time int32) (*OhlcGo, bool, error) {
	data, isnew, err := calc.Align(time)
	if err == ErrAlignTimeTooBig {
		ohlc := *calc.ret.data
		ohlc.Time = time
		//上次返回的数据
		calc.last = calc.ret
		calc.last.data = &ohlc
		return &ohlc, true, nil
	}
	return data, isnew, err
}

func (calc *CandleCalc) Align(time int32) (*OhlcGo, bool, error) {
	//补全前先往前查看一下
	calc.Peek()
	if calc.ret.data == nil && calc.last.data == nil { //保证有数据
		return nil, false, fmt.Errorf("no current or last data, align failed")
	}
	//第一个点的补全比较特殊，可以前后补全
	if calc.last.data == nil {
		ohlc := *calc.ret.data
		ohlc.Time = time
		//上次返回的数据
		calc.last = calc.ret
		calc.last.data = &ohlc
		return &ohlc, calc.ret.isnew, calc.ret.err
	}
	if time == 0 {
		return calc.last.data, false, calc.last.err
	}
	current := *calc.last.data
	if calc.ret.data == nil {
		goto align
	}
	//补全不能大于ret的时间,否则时间会往回走, 所以一般补全前要peek一下
	if time > calc.ret.data.Time {
		if calc.ret.data.TickVolumn == 0 { //补全的数据，忽略到time
			goto align
		}
		return nil, true, ErrAlignTimeTooBig
	}

	if time < current.Time {
		return nil, true, fmt.Errorf("time < current Time align time error.")
	}
align:
	//时间没有变，那么就用current
	if time == current.Time {
		calc.last.isnew = false
		return &current, false, calc.last.err
	}
	//往前补一个
	current.Time = time
	current.Open = current.Close
	current.High = current.Close
	current.Low = current.Close
	current.TickVolumn = 0
	current.RealVolumn = 0
	//更新last
	calc.last.data = &current
	calc.last.isnew = true
	calc.last.err = nil
	if calc.ret.data != nil && calc.ret.data.Time == current.Time {
		calc.ret.isnew = false
	}
	return &current, true, nil
}

func (calc *CandleCalc) LastTick() TickGo {
	return calc.tick
}

func (calc *CandleCalc) Last() TickGo {
	return calc.tick
}

func (calc *CandleCalc) RetTime() int32 {
	return calc.ret.data.Time
}

func (calc *CandleCalc) LastTime() int32 {
	return calc.last.data.Time
}

func (calc *CandleCalc) TickTime() int64 {
	return int64(calc.tick.Time)*1000 + int64(calc.tick.Ms)
}

func (calc *CandleCalc) AlignTick() TickGo {
	return calc.retTick
}

func (calc *CandleCalc) AlignTickTime(t int64) TickGo {
	tick := calc.retTick
	tick.Time = int32(t / 1000)
	tick.Ms = int16(t % 1000)
	return tick
}