package rpc_fea import ( "github.com/banbox/banbot/biz" "github.com/banbox/banbot/btime" "github.com/banbox/banbot/core" "github.com/banbox/banbot/orm" "github.com/banbox/banexg" "github.com/banbox/banexg/errs" "github.com/banbox/banexg/log" utils2 "github.com/banbox/banexg/utils" "github.com/banbox/banta" "go.uber.org/zap" "gonum.org/v1/gonum/mat" "math" ) type AIFeaState struct { cg *banta.CGraph cgBig *banta.CGraph matBig [3]*mat.Dense feasBig *mat.Dense rateBig float64 avgCostBigs []float64 //大周期的成本价 } func pubAiFea(exsList []*orm.ExSymbol, req *biz.SubReq, rsp biz.FeaFeeder_SubFeaturesServer) error { var sample = int(req.Sample) if sample <= 0 { sample = 10 } startDt := btime.ToDateStr(req.Start, core.DefaultDateFmt) log.Info("start ai fea2", zap.Int("pair", len(exsList)), zap.String("start", startDt), zap.Int("sample", sample)) var seqNum = 150 // 特征序列长度 var penNum = 120 // 笔矩阵长度 var depth = 100 // 笔矩阵深度:100个深度 tfSml, tfBig := "5m", "1h" tfMSecs := int64(utils2.TFToSecs(tfSml) * 1000) tfMSecsBig := int64(utils2.TFToSecs(tfBig) * 1000) states := make(map[string]*AIFeaState) for _, exs := range exsList { states[exs.Symbol] = &AIFeaState{ cg: &banta.CGraph{ParseLevel: banta.ParseLevelPen, MaxPen: penNum * 3 / 2}, cgBig: &banta.CGraph{ParseLevel: banta.ParseLevelPen, MaxPen: penNum * 3 / 2}, } } task := &GenFeaTask{ TfSml: tfSml, TfBig: tfBig, WarmSml: 3000, WarmBig: 3000, TfMSecs: tfMSecs, TfMSecsBig: tfMSecsBig, NextObsNum: 30, } batch := NewBatchMap(len(exsList)) task.OnBar = func(e *banta.BarEnv, b *banexg.Kline, futs []*banexg.Kline) *errs.Error { //if !core.IsWarmUp { // log.Info("OnBar", zap.Int64("time", e.TimeStop)) //} state := states[e.Symbol] state.cg.AddBar(e) state.cg.Parse() feas := onAiFeatures(e, seqNum) if e.BarNum%sample > 0 || len(state.cg.Pens) < penNum || state.rateBig == 0 || len(futs) == 0 { // 预热时futs为nil return nil } // 收集特征,发送到监听端 minChgVal := banta.ATR(e.High, e.Low, e.Close, 30).Get(0) * 5 futProbs := genFutProbsV1(futs, minChgVal, 0.4) mats, rate := GenFeaMats(state.cg, penNum, depth, false) avgCosts := avgCostN(e, 10, 20, 40, 80, 160, 320) info := make([]float64, 0, 25) info = append(info, float64(b.Time/1000), b.Open, b.High, b.Low, b.Close, b.Volume, math.Log1p(rate), math.Log1p(state.rateBig)) info = append(info, avgCosts...) info = append(info, state.avgCostBigs...) info = append(info, futProbs[0], futProbs[1], futProbs[2]) return batch.Add(rsp, e, b.Time, map[string]*biz.NumArr{ "time": matToNumArr(mats[0]), "buy": matToNumArr(mats[1]), "sell": matToNumArr(mats[2]), "feas": matToNumArr(feas), "time2": matToNumArr(state.matBig[0]), "buy2": matToNumArr(state.matBig[1]), "sell2": matToNumArr(state.matBig[2]), "feas2": matToNumArr(state.feasBig), "info": {Data: info, Shape: []int32{int32(len(info))}}, }) } task.OnInfoBar = func(e *banta.BarEnv, b *banexg.Kline) *errs.Error { //if !core.IsWarmUp { // log.Info("OnInfoBar", zap.Int64("time", e.TimeStop)) //} state := states[e.Symbol] state.cgBig.AddBar(e) state.cgBig.Parse() state.feasBig = onAiFeatures(e, seqNum) state.avgCostBigs = avgCostN(e, 10, 20, 40, 80, 160, 320) if len(state.cgBig.Pens) >= penNum { state.matBig, state.rateBig = GenFeaMats(state.cgBig, penNum, depth, false) } return nil } task.OnEnvEnd = func(bar *banexg.PairTFKline, adj *orm.AdjInfo) { if bar == nil { return } state := states[bar.Symbol] state.cg = &banta.CGraph{ParseLevel: banta.ParseLevelPen, MaxPen: penNum * 3 / 2} state.cgBig = &banta.CGraph{ParseLevel: banta.ParseLevelPen, MaxPen: penNum * 3 / 2} state.rateBig = 0 } return pubFeaBase(exsList, req, task) }