siaga/modules/candle.go

package module

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/phyer/core"
	"os"
	"strconv"
	"strings"
	// "sync"
	"time"
	//
	// simple "github.com/bitly/go-simplejson"
	// "github.com/go-redis/redis"
	// "github.com/phyer/core/utils"
	logrus "github.com/sirupsen/logrus"
)

type MyCandle struct {
	core.Candle
}

func (cd *MyCandle) Process(cr *core.Core) {
	tmi := ToInt64(cd.Data[0])
	time.UnixMilli(tmi).Format("01-02 15:04")
	// 如果sardine是运行在远端，不能让candle存盘, 因为这是tunnas该干的事情，不能跟它抢
	founded := cd.Filter(cr)
	if !founded {
		return
	}
	// 把candle存下来
	go func() {
		saveCandle := os.Getenv("SIAGA_SAVECANDLE")
		if saveCandle == "true" {
			_, err := cd.SetToKey(cr)
			if err != nil {
				logrus.Warning("SetToKey err: ", err)
			}
		}
		// 对于软candle，推到elasticSearch
		if strings.HasPrefix(cd.From, "soft") {
			cd.PushToWriteLogChan(cr)
		}
	}()
	// TODO update plate and coaster
	go func() {
		makeSeries := os.Getenv("SIAGA_MAKESERIES")
		if makeSeries == "true" {
			_, err := cd.InsertIntoPlate(cr)
			if err != nil {
				logrus.Warning("SetToKey err: ", err)
			}
		}
	}()

	// 由于max可以从远端直接拿，不需要sardine自己做，所以可以sardine做也可以不做
	go func(cad *MyCandle) {
		makeMaX := os.Getenv("SIAGA_MAKEMAX")
		if makeMaX == "true" {
			// 等一会儿防止candle还没有加进CoinMap
			time.Sleep(200 * time.Millisecond)
			cr.MakeMaXsChan <- &cad.Candle
		}
	}(cd)

	go func(cad *MyCandle) {
		// 触发制作插值candle
		makeSoft := false
		// makeVolSoft := true
		// makeVolSoft := false
		if os.Getenv("SIAGA_MAKESOFTCANDLE") == "true" {
			makeSoft = true
		}
		// 根据低维度candle，模拟出"软"的高纬度candle
		if cad.Period == "1m" && makeSoft {
			fmt.Println("makeSoft:", cad.Period, cad.InstID)
			MakeSoftCandles(cr, cad)
		}
	}(cd)
	// 再次发布到redis channel， 給可能的收听者，我觉得没有这个必要了吧
	// go func(cad *MyCandle) {
	// 	time.Sleep(100 * time.Millisecond)
	// 	cad.AddToGeneralCandleChnl(cr)
	// }(cd)
}

func (cd *MyCandle) InsertIntoPlate(cr *core.Core) (*core.Sample, error) {
	cr.Mu.Lock()
	defer cr.Mu.Unlock()
	// pl, plateFounded := cr.PlateMap[cd.InstID]
	// if !plateFounded || pl == nil {
	pl, _ := LoadPlate(cr, cd.InstID)
	cr.PlateMap[cd.InstID] = pl
	// }
	po, coasterFounded := pl.CoasterMap["period"+cd.Period]
	err := errors.New("")
	if !coasterFounded {
		pl.MakeCoaster(cr, cd.Period)
	}

	if len(po.InstID) == 0 {
		// logrus.Debug("candle coaster: ", cd.Period, pl.CoasterMap["period"+cd.Period], pl.CoasterMap)
		//创建失败的原因是原始数据不够，一般发生在服务中断了，缺少部分数据的情况下, 后续需要数据补全措施
		erstr := fmt.Sprintln("coaster创建失败 candle instID: "+cd.InstID+"; period: "+cd.Period, "coasterFounded: ", coasterFounded, " ", err)
		err := errors.New(erstr)
		return nil, err
	}
	sm, err := po.RPushSample(cr, &cd.Candle, "candle")
	return sm, err
}
func (cad *MyCandle) AddToGeneralCandleChnl(cr *core.Core) {
	suffix := ""
	env := os.Getenv("GO_ENV")
	if strings.Contains(env, "demoEnv") {
		suffix = "-demoEnv"
	}
	redisCli := cr.RedisLocalCli
	ab, _ := json.Marshal(cad.Candle)
	_, err := redisCli.Publish(core.ALLCANDLES_PUBLISH+suffix, string(ab)).Result()
	// logrus.Debug("publish, res,err:", res, err, "candle:", string(ab))
	if err != nil {
		logrus.Debug("err of ma7|ma30 add to redis2:", err, cad.Candle.From)
	}
}

// TODO  用key名 BTC-USDT|3M|CandleInfo 维护唯一的一份。
// 当这个key不存在的时候，创建这个key。周期的起始时间，目前MakeSoftCandle函数已经实现了这个逻辑。
// 当存在时，检查现有key里的时间信息，和刚创建的周期其实时间一致不一致，
// 如果一致，更新这个CandleInfo的last值，如果需要的话，更新high和low值。
// 当周期起始时间和已经保存的周期起始时间不一致的时候，说明到了跨周期的时间点了。这时候更新现有的key里保存的open，high，low，close信息为当前tickerInfo的last值。
func (mcd MyCandle) GetSetCandleInfo(cr *core.Core, newPeriod string, ts int64) []interface{} {
	tss := strconv.FormatInt(ts, 10)
	cd := mcd.Candle
	keyName := cd.InstID + "|" + newPeriod + "|candleData"
	str, _ := cr.RedisLocalCli.Get(keyName).Result()
	odata := []interface{}{}
	founded := false

	if len(str) > 0 {
		err := json.Unmarshal([]byte(str), &odata)
		if err != nil {
			logrus.Panic(GetFuncName(), " str2:", str, " err:", err)
		} else {
			founded = true
		}
	}
	if !founded {
		cd.Data[0] = tss        // 时间，数据都是新的
		cd.Data[1] = cd.Data[4] //开盘价
		cd.Data[2] = cd.Data[4] //最高价
		cd.Data[3] = cd.Data[4] //最低价
		bj, _ := json.Marshal(cd.Data)
		cr.RedisLocalCli.Set(keyName, string(bj), 0)
		return cd.Data
	} else {
		// 发现原有的candleData，且还没过期

		if (odata[0]).(string) == tss {
			oHigh := ToFloat64(odata[2])
			oLow := ToFloat64(odata[3])
			cd.Data[0] = tss
			cd.Data[1] = odata[1]
			cd.Data[5] = odata[5]
			cd.Data[6] = odata[6]
			if oHigh <= ToFloat64(cd.Data[4]) {
				cd.Data[2] = cd.Data[4]
			} else {
				cd.Data[2] = odata[2]
			}
			if oLow >= ToFloat64(cd.Data[4]) {
				cd.Data[3] = cd.Data[4]
			} else {
				cd.Data[3] = odata[3]
			}
			bj, _ := json.Marshal(cd.Data)
			cr.RedisLocalCli.Set(keyName, string(bj), 0)
			return cd.Data
		} else {
			// 发现原有的candleData，但是已经过期了
			cd.Data[0] = tss        // 新开始一个cd周期，时间，数据都是新的
			cd.Data[1] = cd.Data[4] //开盘价
			cd.Data[2] = cd.Data[4] //最高价
			cd.Data[3] = cd.Data[4] //最低价
			cd.Data[5] = odata[5]
			cd.Data[6] = odata[6]
			bj, _ := json.Marshal(cd.Data)
			cr.RedisLocalCli.Set(keyName, string(bj), 0)
			return cd.Data
		}
	}
}