eta_index_lib/models/predict_edb.go

package models

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/beego/beego/v2/client/orm"
	"github.com/shopspring/decimal"
	"github.com/yidane/formula"
	"hongze/hongze_edb_lib/services"
	"hongze/hongze_edb_lib/services/alarm_msg"
	"hongze/hongze_edb_lib/utils"
	"strings"
	"time"
)

// CalculateRule 预测指标 规则 计算
type CalculateRule struct {
	EdbInfoId                  int `description:"指标id"`
	ConfigId                   int `description:"配置id"`
	TrendsCalculateMappingList []*PredictEdbConfCalculateMapping
	EdbInfoList                []*EdbInfo
	EdbInfoIdBytes             []string
	Formula                    string
	RuleType                   int              `description:"预测规则，1：最新，2：固定值，3：同比，4：同差，5：环比，6：环差，7：N期移动均值，8：N期段线性外推值，9：动态环差"`
	EndDate                    string           `description:"截止日期"`
	EdbInfoIdArr               []EdbInfoFromTag `description:"指标信息"`
}

// RefreshCalculateByRuleBy9 刷新计算
func RefreshCalculateByRuleBy9(rule CalculateRule) (resultDataList []*EdbInfoSearchData, err error) {
	o := orm.NewOrm()
	to, err := o.Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			tmpErr := to.Rollback()
			if tmpErr != nil {
				go alarm_msg.SendAlarmMsg("RefreshCalculateByRuleBy9 事务回滚失败，Err："+tmpErr.Error(), 3)
			}
		} else {
			err = to.Commit()
		}
	}()

	resultDataList, err = CalculateByRuleBy9(to, rule)

	return
}

// CalculateByRuleBy9 动态环差规则计算入库
func CalculateByRuleBy9(to orm.TxOrmer, rule CalculateRule) (resultDataList []*EdbInfoSearchData, err error) {
	realSaveDataMap := make(map[string]map[int]float64)
	saveDataMap := make(map[string]map[int]float64)
	dateList := make([]string, 0)

	formulaStr := strings.ToUpper(rule.Formula)
	// 获取关联指标数据
	for edbInfoIndex, v := range rule.EdbInfoList {
		dataList, tmpErr := GetPredictEdbDataListAll(v, 1)
		if tmpErr != nil {
			err = tmpErr
			return
		}
		dataMap := make(map[string]float64)
		for _, dv := range dataList {
			// 现有实际数据
			if val, ok := realSaveDataMap[dv.DataTime]; ok {
				if _, ok := val[v.EdbInfoId]; !ok {
					val[v.EdbInfoId] = dv.Value
				}
			} else {
				temp := make(map[int]float64)
				temp[v.EdbInfoId] = dv.Value
				realSaveDataMap[dv.DataTime] = temp
			}

			// 待处理的数据
			if val, ok := saveDataMap[dv.DataTime]; ok {
				if _, ok := val[v.EdbInfoId]; !ok {
					val[v.EdbInfoId] = dv.Value
				}
			} else {
				temp := make(map[int]float64)
				temp[v.EdbInfoId] = dv.Value
				saveDataMap[dv.DataTime] = temp
			}

			// 以第一个指标的日期作为基准日期
			if edbInfoIndex == 0 {
				dateList = append(dateList, dv.DataTime)
			}
		}
		item := new(CalculateItems)
		item.EdbInfoId = v.EdbInfoId
		item.DataMap = dataMap
	}

	//数据处理，将日期内不全的数据做补全
	handleDateSaveDataMap(dateList, realSaveDataMap, saveDataMap, rule.EdbInfoList)

	// 添加数据
	addDataList := make([]*PredictEdbRuleData, 0)

	// 计算规则
	formulaMap := services.CheckFormula(formulaStr)

	//获取指标所有数据
	dataList := make([]*PredictEdbRuleData, 0)
	sql := `SELECT * FROM predict_edb_rule_data WHERE config_id = ?`
	_, err = to.Raw(sql, rule.ConfigId).QueryRows(&dataList)
	if err != nil {
		return
	}
	dataMap := make(map[string]*PredictEdbRuleData)
	for _, v := range dataList {
		dataMap[v.DataTime] = v
	}
	existDataMap := make(map[string]string)

	removeDateList := make([]string, 0) //需要移除的日期
	for sk, sv := range saveDataMap {
		//fmt.Println(sk, sv)
		formulaFormStr := ReplaceFormula(rule.EdbInfoList, sv, formulaMap, formulaStr, rule.EdbInfoIdBytes)
		//计算公式异常，那么就移除该指标
		if formulaFormStr == "" {
			removeDateList = append(removeDateList, sk)
			continue
		}

		//utils.FileLog.Info(fmt.Sprintf("formulaFormStr:%s", formulaFormStr))
		expression := formula.NewExpression(formulaFormStr)
		calResult, tmpErr := expression.Evaluate()
		if tmpErr != nil {
			// 分母为0的报错
			if strings.Contains(tmpErr.Error(), "divide by zero") {
				removeDateList = append(removeDateList, sk)
				continue
			}
			err = errors.New("计算失败：Err:" + tmpErr.Error() + ";formulaStr:" + formulaFormStr)
			//fmt.Println(err)
			return
		}
		calVal, tmpErr := calResult.Float64()
		if tmpErr != nil {
			err = errors.New("计算失败：获取计算值失败 Err:" + tmpErr.Error() + ";formulaStr:" + formulaFormStr)
			//fmt.Println(err)
			return
		}

		saveValue := decimal.NewFromFloat(calVal).RoundCeil(4).String() //utils.SubFloatToString(calVal, 4)
		existPredictEdbRuleData, ok := dataMap[sk]
		if !ok {
			dataTime, _ := time.ParseInLocation(utils.FormatDate, sk, time.Local)
			timestamp := dataTime.UnixNano() / 1e6

			if _, existOk := existDataMap[sk]; !existOk {
				tmpPredictEdbRuleData := &PredictEdbRuleData{
					//PredictEdbRuleDataId: 0,
					EdbInfoId:     rule.EdbInfoId,
					ConfigId:      rule.ConfigId,
					DataTime:      sk,
					Value:         saveValue,
					CreateTime:    time.Now(),
					ModifyTime:    time.Now(),
					DataTimestamp: timestamp,
				}
				addDataList = append(addDataList, tmpPredictEdbRuleData)
			}
			existDataMap[sk] = sk
		} else {
			existValDecimal, tmpErr := decimal.NewFromString(existPredictEdbRuleData.Value)
			if tmpErr != nil {
				err = tmpErr
				return
			}
			existStr := existValDecimal.String()
			if existStr != saveValue {
				existPredictEdbRuleData.Value = saveValue
				existPredictEdbRuleData.ModifyTime = time.Now()
				_, err = to.Update(existPredictEdbRuleData, "Value", "ModifyTime")
				if err != nil {
					return
				}
			}
		}

		// 计算出来的结果集
		resultDataList = append(resultDataList, &EdbInfoSearchData{
			//EdbDataId: 0,
			DataTime: sk,
			Value:    calVal,
		})
	}

	// 添加计算出来的值入库
	lenAddDataList := len(addDataList)
	if lenAddDataList > 0 {
		_, err = to.InsertMulti(lenAddDataList, addDataList)
		if err != nil {
			return
		}
	}

	//删除多余的值
	lenRemoveDateList := len(removeDateList)
	if lenRemoveDateList > 0 {
		//如果拼接指标变更了，那么需要删除所有的指标数据
		sql := ` DELETE FROM predict_edb_rule_data WHERE config_id = ? and data_time in (` + utils.GetOrmInReplace(lenRemoveDateList) + `) `

		_, err = to.Raw(sql, rule.ConfigId, removeDateList).Exec()
		if err != nil {
			err = fmt.Errorf("删除计算失败的预测规则计算指标数据失败,Err:" + err.Error())
			return
		}
	}

	return
}

// RefreshCalculateByRuleByLineNh 刷新动态结果计算（线性拟合）
func RefreshCalculateByRuleByLineNh(predictEdbInfo EdbInfo, predictEdbConfAndDataList []*PredictEdbConfAndData, rule PredictEdbConf) (err error, errMsg string) {
	o := orm.NewOrm()
	to, err := o.Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			tmpErr := to.Rollback()
			if tmpErr != nil {
				go alarm_msg.SendAlarmMsg("RefreshCalculateByRuleBy9 事务回滚失败，Err："+tmpErr.Error(), 3)
			}
		} else {
			err = to.Commit()
		}
	}()
	err, errMsg = CalculateByRuleByRuleLineNh(to, predictEdbInfo, predictEdbConfAndDataList, rule)
	return
}

// CalculateByRuleByRuleLineNh 一元线性拟合规则计算入库
func CalculateByRuleByRuleLineNh(to orm.TxOrmer, predictEdbInfo EdbInfo, predictEdbConfAndDataList []*PredictEdbConfAndData, rule PredictEdbConf) (err error, errMsg string) {
	var secondDataList []*EdbInfoSearchData
	predictEdbInfoId := predictEdbInfo.EdbInfoId // 预测指标id

	// 规则
	var ruleConf RuleLineNhConf
	tmpErr := json.Unmarshal([]byte(rule.Value), &ruleConf)
	if tmpErr != nil {
		errMsg = `季节性配置信息异常`
		err = errors.New("季节性配置信息异常:" + tmpErr.Error())
		return
	}

	// 获取自身指标的数据
	{
		// 来源指标
		var sourceEdbInfoItem *EdbInfo
		sql := ` SELECT * FROM edb_info WHERE edb_info_id=? `
		err = to.Raw(sql, rule.SourceEdbInfoId).QueryRow(&sourceEdbInfoItem)
		if err != nil {
			return
		}

		predictEdbInfo.EdbInfoId = 0
		secondDataList, err, _ = GetPredictDataListByPredictEdbConfList(&predictEdbInfo, sourceEdbInfoItem, predictEdbConfAndDataList, 1, ``)
		if err != nil {
			return
		}

	}
	lenSecondData := len(secondDataList)
	if lenSecondData <= 0 {
		return
	}

	newNhccDataMap, err, errMsg := getCalculateNhccData(secondDataList, ruleConf)
	if err != nil {
		return
	}

	//将最后计算出来的结果数据处理（新增入库、编辑日期的值、删除日期）
	{
		// 获取需要预测的日期
		startDateStr := secondDataList[lenSecondData-1].DataTime
		startDate, _ := time.ParseInLocation(utils.FormatDate, startDateStr, time.Local)
		//endDate, _ := time.ParseInLocation(utils.FormatDate, ruleConf.EndDate, time.Local)
		endDate := rule.EndDate
		dayList := getPredictEdbDayList(startDate, endDate, predictEdbInfo.Frequency)
		if len(dayList) <= 0 { // 如果未来没有日期的话，那么就退出当前循环，进入下一个循环
			return
		}

		//获取该配置的所有数据
		dataList := make([]*PredictEdbRuleData, 0)
		sql := `SELECT * FROM predict_edb_rule_data WHERE config_id = ?`
		_, err = to.Raw(sql, rule.ConfigId).QueryRows(&dataList)
		if err != nil {
			return
		}
		dataMap := make(map[string]*PredictEdbRuleData)
		for _, v := range dataList {
			dataMap[v.DataTime] = v
		}

		//需要移除的日期
		removeDateList := make([]string, 0)
		// 已经操作过的日期
		existDataMap := make(map[string]string)
		// 添加数据
		addDataList := make([]*PredictEdbRuleData, 0)

		for _, currentDate := range dayList {
			// 动态拟合残差值数据
			currentDateStr := currentDate.Format(utils.FormatDate)
			val, ok := newNhccDataMap[currentDateStr]
			// 找不到数据，那么就移除该日期的数据
			if !ok {
				removeDateList = append(removeDateList, currentDateStr)
				continue
			}

			saveValue := decimal.NewFromFloat(val).RoundCeil(4).String() //utils.SubFloatToString(calVal, 4)
			existPredictEdbRuleData, ok := dataMap[currentDateStr]
			if !ok {
				timestamp := currentDate.UnixNano() / 1e6

				if _, existOk := existDataMap[currentDateStr]; !existOk {
					tmpPredictEdbRuleData := &PredictEdbRuleData{
						//PredictEdbRuleDataId: 0,
						EdbInfoId:     predictEdbInfoId,
						ConfigId:      rule.ConfigId,
						DataTime:      currentDateStr,
						Value:         saveValue,
						CreateTime:    time.Now(),
						ModifyTime:    time.Now(),
						DataTimestamp: timestamp,
					}
					addDataList = append(addDataList, tmpPredictEdbRuleData)
				}
				existDataMap[currentDateStr] = currentDateStr
			} else {
				existValDecimal, tmpErr := decimal.NewFromString(existPredictEdbRuleData.Value)
				if tmpErr != nil {
					err = tmpErr
					return
				}
				existStr := existValDecimal.String()
				if existStr != saveValue {
					existPredictEdbRuleData.Value = saveValue
					existPredictEdbRuleData.ModifyTime = time.Now()
					_, err = to.Update(existPredictEdbRuleData, "Value", "ModifyTime")
					if err != nil {
						return
					}
				}
			}
		}

		// 添加计算出来的值入库
		lenAddDataList := len(addDataList)
		if lenAddDataList > 0 {
			_, err = to.InsertMulti(lenAddDataList, addDataList)
			if err != nil {
				return
			}
		}

		//删除多余的值
		lenRemoveDateList := len(removeDateList)
		if lenRemoveDateList > 0 {
			//如果拼接指标变更了，那么需要删除所有的指标数据
			sql := ` DELETE FROM predict_edb_rule_data WHERE config_id = ? and data_time in (` + utils.GetOrmInReplace(lenRemoveDateList) + `) `

			_, err = to.Raw(sql, rule.ConfigId, removeDateList).Exec()
			if err != nil {
				err = fmt.Errorf("删除计算失败的预测规则计算指标数据失败,Err:" + err.Error())
				return
			}
		}
	}

	return
}