package models import ( "errors" "eta/eta_index_lib/utils" "fmt" "github.com/beego/beego/v2/client/orm" "github.com/shopspring/decimal" "reflect" "strconv" "strings" "time" ) // PredictStandardBase 基础预测指标 type PredictStandardBase struct { } // AddPredictEdb 添加预测指标 // edbInfo, calculateMappingList, predictEdbConfList,calculateRule9List,trendsMappingList func AddPredictEdb(item *EdbInfo, calculateMappingList []*EdbInfoCalculateMapping, predictEdbConfList []*PredictEdbConf, calculateRuleMap map[int]CalculateRule) (err error, errMsg string) { var latestDateStr string var latestValue float64 o := orm.NewOrm() tx, err := o.Begin() if err != nil { return } defer func() { if err != nil { tx.Rollback() } else { err = tx.Commit() // 更新指标最大最小值 go UnifiedModifyPredictEdbInfoMaxAndMinInfo(item, latestDateStr, latestValue) } }() // 新增预测指标 edbInfoId, err := tx.Insert(item) if err != nil { return } item.EdbInfoId = int(edbInfoId) // 新增预测指标的关联关系 lenCalculateMapping := len(calculateMappingList) if lenCalculateMapping > 0 { for _, calculateMappingItem := range calculateMappingList { calculateMappingItem.EdbInfoId = item.EdbInfoId calculateMappingItem.EdbCode = item.EdbCode } _, err = tx.InsertMulti(lenCalculateMapping, calculateMappingList) if err != nil { return } } predictEdbConfAndDataList := make([]*PredictEdbConfAndData, 0) // 新增预测指标配置 for k, v := range predictEdbConfList { v.PredictEdbInfoId = item.EdbInfoId configId, tmpErr := tx.Insert(v) if tmpErr != nil { err = tmpErr return } v.ConfigId = int(configId) // 每次规则计算的时候,产生的临时数据 resultDataList := make([]*EdbInfoSearchData, 0) switch v.RuleType { case 9: //动态环差规则 calculateRule := calculateRuleMap[k] calculateRule.ConfigId = v.ConfigId calculateRule.EdbInfoId = v.PredictEdbInfoId // 指标与规则的动态数据生成入库 resultDataList, err = CalculateByRuleBy9(tx, calculateRule) if err != nil { return } // 规则与指标的关系入库 lenTrendsCalculateMapping := len(calculateRule.TrendsCalculateMappingList) if lenTrendsCalculateMapping > 0 { for _, vv := range calculateRule.TrendsCalculateMappingList { vv.EdbInfoId = item.EdbInfoId vv.ConfigId = v.ConfigId } _, err = tx.InsertMulti(lenTrendsCalculateMapping, calculateRule.TrendsCalculateMappingList) if err != nil { return } } case 14: //14:根据 一元线性拟合 规则获取预测数据 calculateRule := calculateRuleMap[k] calculateRule.ConfigId = v.ConfigId calculateRule.EdbInfoId = v.PredictEdbInfoId // 指标与规则的动态数据(拟合数据)生成入库 err, errMsg = CalculateByRuleByRuleLineNh(tx, *item, predictEdbConfAndDataList, *v) if err != nil { return } // 规则与指标的关系入库 lenTrendsCalculateMapping := len(calculateRule.TrendsCalculateMappingList) if lenTrendsCalculateMapping > 0 { for _, vv := range calculateRule.TrendsCalculateMappingList { vv.EdbInfoId = item.EdbInfoId vv.ConfigId = v.ConfigId } _, err = tx.InsertMulti(lenTrendsCalculateMapping, calculateRule.TrendsCalculateMappingList) if err != nil { return } } } // 规则配置(含数据) tmpPredictEdbConfAndData := &PredictEdbConfAndData{ ConfigId: 0, PredictEdbInfoId: 0, SourceEdbInfoId: v.SourceEdbInfoId, RuleType: v.RuleType, FixedValue: v.FixedValue, Value: v.Value, EndDate: v.EndDate, EndNum: v.EndNum, ModifyTime: v.ModifyTime, CreateTime: v.CreateTime, DataList: resultDataList, } predictEdbConfAndDataList = append(predictEdbConfAndDataList, tmpPredictEdbConfAndData) } // 重新预测数据 obj := new(PredictStandardBase) latestDateStr, latestValue, err, errMsg = obj.refresh(tx, item, predictEdbConfAndDataList, item.EdbCode, "") return } // EditPredictEdb 修改预测指标 func EditPredictEdb(edbInfo *EdbInfo, updateEdbInfoCol []string, calculateMappingList []*EdbInfoCalculateMapping, predictEdbConfList []*PredictEdbConf, calculateRuleMap map[int]CalculateRule) (err error, errMsg string) { var latestDateStr string var latestValue float64 o := orm.NewOrm() tx, err := o.Begin() if err != nil { return } defer func() { if err != nil { tx.Rollback() } else { err = tx.Commit() // 更新指标最大最小值 go UnifiedModifyPredictEdbInfoMaxAndMinInfo(edbInfo, latestDateStr, latestValue) } }() // 修改预测指标 _, err = tx.Update(edbInfo, updateEdbInfoCol...) if err != nil { return } // 先删除原有的预测指标 与 其他指标的 关联关系 sql := ` DELETE FROM edb_info_calculate_mapping WHERE edb_info_id = ?` _, err = tx.Raw(sql, edbInfo.EdbInfoId).Exec() if err != nil { return } // 先删除原有的配置 sql = ` DELETE FROM predict_edb_conf WHERE predict_edb_info_id = ?` _, err = tx.Raw(sql, edbInfo.EdbInfoId).Exec() if err != nil { return } // 删除基础预测指标 规则配置 与 其他指标的 关联关系 sql = ` DELETE FROM predict_edb_conf_calculate_mapping WHERE edb_info_id = ?` _, err = tx.Raw(sql, edbInfo.EdbInfoId).Exec() if err != nil { return } // 删除基础预测指标 规则配置 生成的动态数据值 sql = ` DELETE FROM predict_edb_rule_data WHERE edb_info_id = ?` _, err = tx.Raw(sql, edbInfo.EdbInfoId).Exec() if err != nil { return } // 新增预测指标的关联关系 lenCalculateMapping := len(calculateMappingList) if lenCalculateMapping > 0 { for _, calculateMappingItem := range calculateMappingList { calculateMappingItem.EdbInfoId = edbInfo.EdbInfoId calculateMappingItem.EdbCode = edbInfo.EdbCode } _, err = tx.InsertMulti(lenCalculateMapping, calculateMappingList) if err != nil { return } } calculateRuleIndex := 0 // 预测计算规则下标 predictEdbConfAndDataList := make([]*PredictEdbConfAndData, 0) // 新增预测指标配置 for confIndex, v := range predictEdbConfList { configId, tmpErr := tx.Insert(v) if tmpErr != nil { err = tmpErr return } v.ConfigId = int(configId) // 每次规则计算的时候,产生的临时数据 resultDataList := make([]*EdbInfoSearchData, 0) switch v.RuleType { case 9: //动态环差规则 calculateRule := calculateRuleMap[confIndex] calculateRule.ConfigId = v.ConfigId calculateRule.EdbInfoId = v.PredictEdbInfoId // 指标与规则的动态数据生成入库 resultDataList, err = CalculateByRuleBy9(tx, calculateRule) if err != nil { return } // 规则与指标的关系入库 lenTrendsCalculateMapping := len(calculateRule.TrendsCalculateMappingList) if lenTrendsCalculateMapping > 0 { for _, vv := range calculateRule.TrendsCalculateMappingList { vv.EdbInfoId = edbInfo.EdbInfoId vv.ConfigId = v.ConfigId } _, err = tx.InsertMulti(lenTrendsCalculateMapping, calculateRule.TrendsCalculateMappingList) if err != nil { return } } case 14: //14:根据 一元线性拟合 规则获取预测数据 calculateRule := calculateRuleMap[confIndex] calculateRule.ConfigId = v.ConfigId calculateRule.EdbInfoId = v.PredictEdbInfoId // 指标与规则的动态数据(拟合数据)生成入库 err, errMsg = CalculateByRuleByRuleLineNh(tx, *edbInfo, predictEdbConfAndDataList, *v) if err != nil { return } // 规则与指标的关系入库 lenTrendsCalculateMapping := len(calculateRule.TrendsCalculateMappingList) if lenTrendsCalculateMapping > 0 { for _, vv := range calculateRule.TrendsCalculateMappingList { vv.EdbInfoId = edbInfo.EdbInfoId vv.ConfigId = v.ConfigId } _, err = tx.InsertMulti(lenTrendsCalculateMapping, calculateRule.TrendsCalculateMappingList) if err != nil { return } } } calculateRuleIndex++ // 规则配置(含数据) tmpPredictEdbConfAndData := &PredictEdbConfAndData{ ConfigId: 0, PredictEdbInfoId: 0, SourceEdbInfoId: v.SourceEdbInfoId, RuleType: v.RuleType, FixedValue: v.FixedValue, Value: v.Value, EndDate: v.EndDate, ModifyTime: v.ModifyTime, CreateTime: v.CreateTime, DataList: resultDataList, } predictEdbConfAndDataList = append(predictEdbConfAndDataList, tmpPredictEdbConfAndData) } // 重新预测数据 obj := new(PredictStandardBase) latestDateStr, latestValue, err, errMsg = obj.refresh(tx, edbInfo, predictEdbConfAndDataList, edbInfo.EdbCode, "") return } // Add 添加 func (obj PredictStandardBase) Add(params BatchSaveCalculateBatchParams) (edbInfo *EdbInfo, latestDateStr string, latestValue float64, err error, errMsg string) { err = errors.New("暂不支持该方法") return } // Edit 编辑 func (obj PredictStandardBase) Edit(params BatchSaveCalculateBatchParams) (latestDateStr string, latestValue float64, err error, errMsg string) { err = errors.New("暂不支持该方法") return } // Refresh 刷新 func (obj PredictStandardBase) Refresh(params RefreshParams) (latestDateStr string, latestValue float64, err error, errMsg string) { edbInfo := params.EdbInfo if edbInfo == nil { errMsg = `错误的指标` err = errors.New(errMsg) return } // 查找该预测指标配置 predictEdbConfList, err := GetPredictEdbConfAndDataListById(edbInfo.EdbInfoId) if err != nil && err.Error() != utils.ErrNoRow() { errMsg = "获取预测指标配置信息失败" return } if len(predictEdbConfList) == 0 { errMsg = "获取预测指标配置信息失败" err = errors.New(errMsg) return } o := orm.NewOrm() to, err := o.Begin() if err != nil { return } defer func() { if err != nil { _ = to.Rollback() fmt.Println(reflect.TypeOf(obj).Name(), ";Refresh,Err:"+err.Error()) } else { _ = to.Commit() } }() // 重新预测数据 latestDateStr, latestValue, err, errMsg = obj.refresh(to, edbInfo, predictEdbConfList, edbInfo.EdbCode, "") return } // GetSource 获取来源编码id func (obj PredictStandardBase) GetSource() int { return utils.DATA_SOURCE_PREDICT } // GetSourceName 获取来源名称 func (obj PredictStandardBase) GetSourceName() string { return utils.DATA_SOURCE_NAME_PREDICT } func (obj PredictStandardBase) refresh(to orm.TxOrmer, edbInfo *EdbInfo, predictEdbConfAndDataList []*PredictEdbConfAndData, edbCode, startDate string) (latestDateStr string, latestValue float64, err error, errMsg string) { edbInfoId := edbInfo.EdbInfoId dataTableName := GetEdbDataTableName(edbInfo.Source, edbInfo.SubSource) edbInfoIdStr := strconv.Itoa(edbInfoId) // 获取标准差图表的指标数据 fromDataList, fromEdbInfo, err, errMsg := obj.GetCalculateDataList(edbInfo, predictEdbConfAndDataList, 1, startDate) if err != nil { return } latestDateStr = fromEdbInfo.LatestDate //获取指标所有数据 existDataList := make([]*EdbData, 0) sql := `SELECT * FROM %s WHERE edb_info_id=? ` sql = fmt.Sprintf(sql, dataTableName) _, err = to.Raw(sql, edbInfoId).QueryRows(&existDataList) if err != nil { return } existDataMap := make(map[string]string) removeDataTimeMap := make(map[string]int) //需要移除的日期数据 for _, v := range existDataList { existDataMap[v.DataTime] = v.Value removeDataTimeMap[v.DataTime] = 1 } needAddDateMap := make(map[time.Time]int) addSql := ` INSERT INTO ` + dataTableName + `(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values ` var isAdd bool for _, tmpData := range fromDataList { currDateStr := tmpData.DataTime currTime, tmpErr := time.ParseInLocation(utils.FormatDate, currDateStr, time.Local) if tmpErr != nil { err = tmpErr return } // 当前的实际值 saveValue := decimal.NewFromFloat(tmpData.Value).Round(4).String() existVal, ok := existDataMap[currDateStr] // 如果库中已经存在该数据的话,那么就进行值的变更操作 if ok { //校验待删除日期数据里面是否存在该元素,如果存在的话,那么移除该日期 delete(removeDataTimeMap, currDateStr) // 格式化库中的指标数据,并保留4位小数 existValDec, tmpErr := decimal.NewFromString(existVal) if tmpErr != nil { err = tmpErr return } existVal = existValDec.Round(4).String() if existVal != saveValue { sql := ` UPDATE %s SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? ` sql = fmt.Sprintf(sql, dataTableName) _, err = to.Raw(sql, saveValue, edbInfoId, currDateStr).Exec() if err != nil { return } } continue } // 库中不存在该日期的数据 timestamp := currTime.UnixNano() / 1e6 timeStr := fmt.Sprintf("%d", timestamp) if _, existOk := needAddDateMap[currTime]; !existOk { addSql += GetAddSql(edbInfoIdStr, edbCode, currDateStr, timeStr, saveValue) isAdd = true } needAddDateMap[currTime] = 1 } //删除已经不存在的指标数据(由于该指标当日的数据删除了) { removeDateList := make([]string, 0) for dateTime := range removeDataTimeMap { removeDateList = append(removeDateList, dateTime) } removeNum := len(removeDateList) if removeNum > 0 { sql := fmt.Sprintf(` DELETE FROM %s WHERE edb_info_id = ? and data_time in (`+utils.GetOrmInReplace(removeNum)+`) `, dataTableName) _, err = to.Raw(sql, edbInfo.EdbInfoId, removeDateList).Exec() if err != nil { fmt.Println(reflect.TypeOf(obj).Name(), " add data ;delete Err", err.Error()) err = fmt.Errorf("删除不存在的指标数据失败,Err:" + err.Error()) return } } } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() if err != nil { fmt.Println(reflect.TypeOf(obj).Name(), " add data Err", err.Error()) return } } //确定实际数据的最终值 { finalLast, tmpErr := GetFinalLastByTo(to, edbInfoId, edbInfo.Source, edbInfo.SubSource, fromEdbInfo.LatestDate) if tmpErr != nil && tmpErr.Error() != utils.ErrNoRow() { return } if tmpErr == nil { latestDateStr = finalLast.DataTime latestValue = finalLast.Value } } return } // GetCalculateDataList // @Description: 获取计算后的数据 // @author: Roc // @receiver obj // @datetime 2024-04-24 10:42:51 // @param edbInfo *EdbInfo // @param predictEdbConfList []*PredictEdbConfAndData // @param order int order:1升序,其余值为降序 // @param startDate string // @return dataList []*EdbInfoSearchData // @return sourceEdbInfoItem *EdbInfo // @return err error // @return errMsg string func (obj PredictStandardBase) GetCalculateDataList(edbInfo *EdbInfo, predictEdbConfList []*PredictEdbConfAndData, order int, startDate string) (dataList []*EdbInfoSearchData, sourceEdbInfoItem *EdbInfo, err error, errMsg string) { if len(predictEdbConfList) == 0 { errMsg = "获取预测指标配置信息失败" err = errors.New(errMsg) return } predictEdbConf := predictEdbConfList[0] // 来源指标 sourceEdbInfoItem, err = GetEdbInfoById(predictEdbConf.SourceEdbInfoId) if err != nil { if err.Error() == utils.ErrNoRow() { errMsg = "找不到来源指标信息" err = errors.New(errMsg) } return } dataList, err, errMsg = GetPredictDataListByPredictEdbConfList(edbInfo, sourceEdbInfoItem, predictEdbConfList, order, startDate) return } // RefreshPredictStandardBaseByGeneralEdbInfoId // @Description: 根据普通指标id刷新预测标准基础数据 // @author: Roc // @datetime 2024-04-24 13:40:59 // @param sourceEdbInfoId int // @return err error func RefreshPredictStandardBaseByGeneralEdbInfoId(sourceEdbInfoId int) { errList := make([]string, 0) defer func() { if len(errList) > 0 { fmt.Println(errList) utils.FileLog.Info("RefreshPredictStandardBaseByGeneralEdbInfoId更新失败,来源ID: ErrMsg:" + strings.Join(errList, "\n")) } }() o := orm.NewOrm() var list []*PredictEdbConf sql := ` SELECT * FROM predict_edb_conf WHERE source_edb_info_id=? ` total, err := o.Raw(sql, sourceEdbInfoId).QueryRows(&list) if err != nil { errList = append(errList, fmt.Sprintf("根据来源换指标id获取配置项失败,来源指标ID:%d;err:%s", sourceEdbInfoId, err.Error())) return } if total > 0 { idList := make([]int, 0) for _, v := range list { idList = append(idList, v.PredictEdbInfoId) edbInfo, tmpErr := GetEdbInfoById(v.PredictEdbInfoId) if tmpErr != nil { errList = append(errList, fmt.Sprintf("获取预测指标配置信息失败,预测指标配置ID:%d;err:%s", v.PredictEdbInfoId, tmpErr.Error())) continue } refreshParams := RefreshParams{ EdbInfo: edbInfo, } obj := new(PredictStandardBase) latestDateStr, latestValue, tmpErr, errMsg := obj.Refresh(refreshParams) if tmpErr != nil { errList = append(errList, fmt.Sprintf("更新失败,预测指标配置ID:%d;err:%s;errMsg:%s", v.PredictEdbInfoId, tmpErr.Error(), errMsg)) continue } // 更新指标最大最小值 tmpErr, errMsg = UnifiedModifyPredictEdbInfoMaxAndMinInfo(edbInfo, latestDateStr, latestValue) if tmpErr != nil { errList = append(errList, fmt.Sprintf("更新指标最大最小值失败,预测指标配置ID:%d;err:%s;errMsg:%s", v.PredictEdbInfoId, tmpErr.Error(), errMsg)) continue } } } return }