base_predict_from_calculate.go 9.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310
  1. package models
  2. import (
  3. "errors"
  4. "eta/eta_index_lib/utils"
  5. "fmt"
  6. "github.com/beego/beego/v2/client/orm"
  7. "github.com/shopspring/decimal"
  8. "github.com/yidane/formula"
  9. "strconv"
  10. "strings"
  11. "time"
  12. )
  13. // PredictEdbInfoCalculateSaveReq 预测计算(运算)指标请求参数
  14. type PredictEdbInfoCalculateSaveReq struct {
  15. AdminId int `description:"添加人id"`
  16. AdminName string `description:"添加人名称"`
  17. EdbName string `description:"指标名称"`
  18. Frequency string `description:"频率"`
  19. Unit string `description:"单位"`
  20. ClassifyId int `description:"分类id"`
  21. CalculateFormula string `description:"计算公式"`
  22. EdbInfoIdArr []struct {
  23. EdbInfoId int `description:"指标id"`
  24. FromTag string `description:"指标对应标签"`
  25. }
  26. }
  27. // PredictCalculateItems 预测计算(运算)指标信息
  28. type PredictCalculateItems struct {
  29. EdbInfoId int
  30. DataMap map[string]float64
  31. }
  32. // EditPredictCalculate 新增预测计算(运算)指标
  33. func EditPredictCalculate(edbInfo *EdbInfo, calculateMappingList []*EdbInfoCalculateMapping) (err error) {
  34. o := orm.NewOrm()
  35. to, err := o.Begin()
  36. if err != nil {
  37. return
  38. }
  39. defer func() {
  40. if err != nil {
  41. tmpErr := to.Rollback()
  42. errMsg := fmt.Sprintf("AddPredictCalculate Err:%s" + err.Error())
  43. if tmpErr != nil {
  44. errMsg += `;rollback失败:` + tmpErr.Error()
  45. }
  46. utils.FileLog.Info(errMsg)
  47. return
  48. }
  49. tmpErr := to.Commit()
  50. if tmpErr != nil {
  51. errMsg := fmt.Sprintf("AddPredictCalculate commit Err:%s" + tmpErr.Error())
  52. utils.FileLog.Info(errMsg)
  53. }
  54. }()
  55. // 修改指标信息
  56. _, err = to.Update(edbInfo, "EdbName", "Frequency", "Unit", "ClassifyId", "CalculateFormula")
  57. if err != nil {
  58. return
  59. }
  60. //删除,计算指标关联的,基础指标的关联关系
  61. sql := ` DELETE FROM edb_info_calculate_mapping WHERE edb_info_id = ? `
  62. _, err = to.Raw(sql, edbInfo.EdbInfoId).Exec()
  63. if err != nil {
  64. err = errors.New("删除计算指标关联关系失败,Err:" + err.Error())
  65. return
  66. }
  67. // 插入新的关系表
  68. if len(calculateMappingList) > 0 {
  69. _, err = to.InsertMulti(len(calculateMappingList), calculateMappingList)
  70. }
  71. //数据表
  72. tableName := GetEdbDataTableName(edbInfo.Source)
  73. //清空原有数据
  74. sql = ` DELETE FROM ` + tableName + ` WHERE edb_info_id = ? `
  75. _, err = o.Raw(sql, edbInfo.EdbInfoId).Exec()
  76. if err != nil {
  77. return
  78. }
  79. return
  80. }
  81. // AddPredictCalculateData 新增预测计算(运算)指标的数据
  82. func AddPredictCalculateData(edbInfoIdList []*EdbInfo, edbInfo *EdbInfo, edbCode, formulaStr string, edbInfoIdBytes []string) (latestDateStr string, latestValue float64, err error) {
  83. o := orm.NewOrm()
  84. to, err := o.Begin()
  85. if err != nil {
  86. return
  87. }
  88. defer func() {
  89. if err != nil {
  90. fmt.Println("AddPredictCalculateData Calculate,Err:" + err.Error())
  91. _ = to.Rollback()
  92. } else {
  93. _ = to.Commit()
  94. }
  95. }()
  96. latestDateStr, latestValue, err = refreshAllPredictCalculate(to, edbInfoIdList, edbInfo.EdbInfoId, edbInfo.Source, edbCode, formulaStr, "", edbInfoIdBytes)
  97. return
  98. }
  99. // RefreshAllPredictCalculate 刷新预测计算指标的全部数据
  100. func RefreshAllPredictCalculate(edbInfoIdList []*EdbInfo, edbInfoId, source int, edbCode, formulaStr, startDate string, edbInfoIdBytes []string) (latestDateStr string, latestValue float64, err error) {
  101. o := orm.NewOrm()
  102. to, err := o.Begin()
  103. if err != nil {
  104. return
  105. }
  106. defer func() {
  107. if err != nil {
  108. fmt.Println("RefreshAllCalculate,Err:" + err.Error())
  109. _ = to.Rollback()
  110. } else {
  111. _ = to.Commit()
  112. }
  113. }()
  114. latestDateStr, latestValue, err = refreshAllPredictCalculate(to, edbInfoIdList, edbInfoId, source, edbCode, formulaStr, startDate, edbInfoIdBytes)
  115. return
  116. }
  117. // refreshAllPredictCalculate 刷新预测计算指标的全部数据
  118. func refreshAllPredictCalculate(to orm.TxOrmer, edbInfoIdList []*EdbInfo, edbInfoId, source int, edbCode, formulaStr, startDate string, edbInfoIdBytes []string) (latestDateStr string, latestValue float64, err error) {
  119. fmt.Println("startDate:", startDate)
  120. // 最小的结束日期 , 最晚的数据开始日期
  121. var minLatestDate, maxStartDate time.Time
  122. dateList := make([]string, 0) // 第一个指标的日期数据
  123. realSaveDataMap := make(map[string]map[int]float64)
  124. saveDataMap := make(map[string]map[int]float64)
  125. for edbInfoIndex, v := range edbInfoIdList {
  126. tmpLatestDate, tmpErr := time.ParseInLocation(utils.FormatDate, v.EndDate, time.Local)
  127. if tmpErr != nil {
  128. err = errors.New(`最近的日期格式化失败;日期:` + v.EndDate + `;err:` + tmpErr.Error())
  129. return
  130. }
  131. if minLatestDate.IsZero() || tmpLatestDate.Before(minLatestDate) {
  132. minLatestDate = tmpLatestDate
  133. }
  134. tmpStartDate, tmpErr := time.ParseInLocation(utils.FormatDate, v.StartDate, time.Local)
  135. if tmpErr != nil {
  136. err = errors.New(`最近的日期格式化失败;日期:` + v.StartDate + `;err:` + tmpErr.Error())
  137. return
  138. }
  139. if maxStartDate.IsZero() || maxStartDate.Before(tmpStartDate) {
  140. maxStartDate = tmpStartDate
  141. }
  142. // 获取关联指标数据
  143. dataList, tmpErr := GetPredictEdbDataListAllByStartDate(v, 1, startDate)
  144. if tmpErr != nil {
  145. err = tmpErr
  146. return
  147. }
  148. for _, dv := range dataList { // 现有实际数据
  149. if val, ok := realSaveDataMap[dv.DataTime]; ok {
  150. if _, ok := val[v.EdbInfoId]; !ok {
  151. val[v.EdbInfoId] = dv.Value
  152. }
  153. } else {
  154. temp := make(map[int]float64)
  155. temp[v.EdbInfoId] = dv.Value
  156. realSaveDataMap[dv.DataTime] = temp
  157. }
  158. // 待处理的数据
  159. if val, ok := saveDataMap[dv.DataTime]; ok {
  160. if _, ok := val[v.EdbInfoId]; !ok {
  161. val[v.EdbInfoId] = dv.Value
  162. }
  163. } else {
  164. temp := make(map[int]float64)
  165. temp[v.EdbInfoId] = dv.Value
  166. saveDataMap[dv.DataTime] = temp
  167. }
  168. if edbInfoIndex == 0 {
  169. dateList = append(dateList, dv.DataTime)
  170. }
  171. }
  172. }
  173. //数据处理,将日期内不全的数据做填补
  174. handleDateSaveDataMap(dateList, maxStartDate, minLatestDate, realSaveDataMap, saveDataMap, edbInfoIdList)
  175. formulaMap := utils.CheckFormula(formulaStr)
  176. addSql := ` INSERT INTO edb_data_predict_calculate (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
  177. var isAdd bool
  178. if !minLatestDate.IsZero() {
  179. latestDateStr = minLatestDate.Format(utils.FormatDate)
  180. }
  181. //获取指标所有数据
  182. dataList := make([]*EdbData, 0)
  183. dataTableName := GetEdbDataTableName(source)
  184. sql := `SELECT * FROM %s WHERE edb_info_id=? `
  185. sql = fmt.Sprintf(sql, dataTableName)
  186. _, err = to.Raw(sql, edbInfoId).QueryRows(&dataList)
  187. if err != nil {
  188. return
  189. }
  190. dataMap := make(map[string]string)
  191. for _, v := range dataList {
  192. dataMap[v.DataTime] = v.Value
  193. }
  194. edbInfoIdStr := strconv.Itoa(edbInfoId)
  195. existDataMap := make(map[string]string)
  196. removeDateList := make([]string, 0) //需要移除的日期
  197. for sk, sv := range saveDataMap {
  198. //fmt.Println(sk, sv)
  199. formulaStr = strings.ToUpper(formulaStr)
  200. formulaFormStr := ReplaceFormula(edbInfoIdList, sv, formulaMap, formulaStr, edbInfoIdBytes)
  201. if formulaFormStr != "" {
  202. //utils.FileLog.Info(fmt.Sprintf("formulaFormStr:%s", formulaFormStr))
  203. expression := formula.NewExpression(formulaFormStr)
  204. calResult, tmpErr := expression.Evaluate()
  205. if tmpErr != nil {
  206. // 分母为0的报错
  207. if strings.Contains(err.Error(), "divide by zero") {
  208. removeDateList = append(removeDateList, sk)
  209. continue
  210. }
  211. err = errors.New("计算失败:Err:" + tmpErr.Error() + ";formulaStr:" + formulaFormStr)
  212. fmt.Println(err)
  213. return
  214. }
  215. calVal, tmpErr := calResult.Float64()
  216. if tmpErr != nil {
  217. err = errors.New("计算失败:获取计算值失败 Err:" + tmpErr.Error() + ";formulaStr:" + formulaFormStr)
  218. fmt.Println(err)
  219. return
  220. }
  221. // 最近实际日期的实际数据
  222. if latestDateStr == sk {
  223. latestValue, _ = decimal.NewFromFloat(calVal).Truncate(4).Float64() //保留4位小数
  224. }
  225. saveValue := decimal.NewFromFloat(calVal).RoundCeil(4).String() //utils.SubFloatToString(calVal, 4)
  226. if existVal, ok := dataMap[sk]; !ok {
  227. dataTime, _ := time.ParseInLocation(utils.FormatDate, sk, time.Local)
  228. timestamp := dataTime.UnixNano() / 1e6
  229. timeStr := fmt.Sprintf("%d", timestamp)
  230. if _, existOk := existDataMap[sk]; !existOk {
  231. addSql += GetAddSql(edbInfoIdStr, edbCode, sk, timeStr, saveValue)
  232. isAdd = true
  233. }
  234. existDataMap[sk] = sk
  235. } else {
  236. existValDecimal, tmpErr := decimal.NewFromString(existVal)
  237. if tmpErr != nil {
  238. err = tmpErr
  239. return
  240. }
  241. existStr := existValDecimal.String()
  242. if existStr != saveValue {
  243. sql := ` UPDATE %s SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? `
  244. sql = fmt.Sprintf(sql, dataTableName)
  245. _, err = to.Raw(sql, saveValue, edbInfoId, sk).Exec()
  246. if err != nil {
  247. return
  248. }
  249. }
  250. }
  251. } else {
  252. //计算公式异常,那么就移除该指标
  253. removeDateList = append(removeDateList, sk)
  254. continue
  255. }
  256. }
  257. if isAdd {
  258. addSql = strings.TrimRight(addSql, ",")
  259. _, err = to.Raw(addSql).Exec()
  260. if err != nil {
  261. fmt.Println("RefreshAllCalculate add Err", err.Error())
  262. return
  263. }
  264. }
  265. if len(removeDateList) > 0 {
  266. removeDateStr := strings.Join(removeDateList, `","`)
  267. removeDateStr = `"` + removeDateStr + `"`
  268. //如果拼接指标变更了,那么需要删除所有的指标数据
  269. tableName := GetEdbDataTableName(source)
  270. sql := fmt.Sprintf(` DELETE FROM %s WHERE edb_info_id = ? and data_time in (%s) `, tableName, removeDateStr)
  271. _, err = to.Raw(sql, edbInfoId).Exec()
  272. if err != nil {
  273. err = fmt.Errorf("删除计算失败的计算指标数据失败,Err:" + err.Error())
  274. return
  275. }
  276. }
  277. return
  278. }