base_from_pb.go 4.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169
  1. package models
  2. import (
  3. "eta/eta_index_lib/utils"
  4. "fmt"
  5. "github.com/beego/beego/v2/client/orm"
  6. "strconv"
  7. "strings"
  8. "time"
  9. )
  10. type EdbDataFromPb struct {
  11. Date map[string]int64 `json:"date"`
  12. Ticker map[string]string `json:"ticker"`
  13. Field map[string]string `json:"field"`
  14. Value map[string]float64 `json:"value"`
  15. }
  16. func AddEdbDataFromPb(edbCode string, item *EdbDataFromPb) (err error) {
  17. o := orm.NewOrm()
  18. if len(item.Date) > 0 {
  19. var isAdd bool
  20. dateMap := item.Date
  21. addSql := ` INSERT INTO edb_data_pb(edb_info_id,edb_code,data_time,value,create_time,modify_time,ticker,field,data_timestamp) values `
  22. nowStr := time.Now().Format(utils.FormatDateTime)
  23. for k, v := range dateMap {
  24. timeStr := fmt.Sprintf("%d", v)
  25. v = v / 1000
  26. t := time.Unix(v, 0)
  27. dateTime := t.Format(utils.FormatDate)
  28. // //这里的目的是为了北京时间,以防万一,还是重新转一下吧
  29. {
  30. t, err = time.ParseInLocation(utils.FormatDate, dateTime, time.Local)
  31. if err != nil {
  32. return err
  33. }
  34. timeStr = fmt.Sprintf("%d", t.UnixNano()/1e6)
  35. }
  36. val := item.Value[k]
  37. field := item.Field[k]
  38. ticker := item.Ticker[k]
  39. if field == "PX_LAST" {
  40. addSql += "("
  41. addSql += "0," + "'" + edbCode + "'" + "," + "'" + dateTime + "'" + "," + utils.SubFloatToString(val, 20) + "," + "'" + nowStr + "'" +
  42. "," + "'" + nowStr + "'" + "," + "'" + ticker + "'" + "," + "'" + field + "'" + "," + "'" + timeStr + "'"
  43. addSql += "),"
  44. isAdd = true
  45. }
  46. }
  47. if isAdd {
  48. addSql = strings.TrimRight(addSql, ",")
  49. _, err = o.Raw(addSql).Exec()
  50. if err != nil {
  51. return
  52. }
  53. }
  54. }
  55. return
  56. }
  57. // 刷新彭博指标数据
  58. func RefreshEdbDataFromPb(edbInfoId int, edbCode, startDate string, item *EdbDataFromPb) (err error) {
  59. o := orm.NewOrm()
  60. source := utils.DATA_SOURCE_PB
  61. subSource := utils.DATA_SUB_SOURCE_EDB
  62. // 真实数据的最大日期 , 插入规则配置的日期
  63. var realDataMaxDate, edbDataInsertConfigDate time.Time
  64. var edbDataInsertConfig *EdbDataInsertConfig
  65. var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值
  66. {
  67. edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId)
  68. if err != nil && err.Error() != utils.ErrNoRow() {
  69. return
  70. }
  71. if edbDataInsertConfig != nil {
  72. edbDataInsertConfigDate = edbDataInsertConfig.Date
  73. }
  74. }
  75. var condition string
  76. var pars []interface{}
  77. condition += " AND edb_info_id=? "
  78. pars = append(pars, edbInfoId)
  79. condition += " AND data_time>=? "
  80. pars = append(pars, startDate)
  81. existList, err := GetEdbDataByCondition(source, subSource, condition, pars)
  82. existMap := make(map[string]*EdbInfoSearchData)
  83. for _, v := range existList {
  84. existMap[v.DataTime] = v
  85. }
  86. edbInfoIdStr := strconv.Itoa(edbInfoId)
  87. addMap := make(map[string]string)
  88. if len(item.Date) > 0 {
  89. dateMap := item.Date
  90. var isAdd bool
  91. addSql := ` INSERT INTO edb_data_pb(edb_info_id,edb_code,data_time,value,create_time,modify_time,ticker,field,data_timestamp) values `
  92. nowStr := time.Now().Format(utils.FormatDateTime)
  93. for k, v := range dateMap {
  94. timeStr := fmt.Sprintf("%d", v)
  95. v = v / 1000
  96. t := time.Unix(v, 0)
  97. dateTime := t.Format(utils.FormatDate)
  98. // //这里的目的是为了北京时间,以防万一,还是重新转一下吧
  99. {
  100. t, err = time.ParseInLocation(utils.FormatDate, dateTime, time.Local)
  101. if err != nil {
  102. return err
  103. }
  104. timeStr = fmt.Sprintf("%d", t.UnixNano()/1e6)
  105. }
  106. val := item.Value[k]
  107. field := item.Field[k]
  108. ticker := item.Ticker[k]
  109. saveValue := utils.SubFloatToString(val, 30)
  110. if field == "PX_LAST" {
  111. if findItem, ok := existMap[dateTime]; !ok {
  112. if _, addOk := addMap[dateTime]; !addOk {
  113. addSql += "("
  114. addSql += edbInfoIdStr + "," + "'" + edbCode + "'" + "," + "'" + dateTime + "'" + "," + saveValue + "," + "'" + nowStr + "'" +
  115. "," + "'" + nowStr + "'" + "," + "'" + ticker + "'" + "," + "'" + field + "'" + "," + "'" + timeStr + "'"
  116. addSql += "),"
  117. isAdd = true
  118. addMap[dateTime] = saveValue
  119. }
  120. } else {
  121. if findItem != nil && utils.SubFloatToString(findItem.Value, 30) != saveValue {
  122. err = ModifyEdbDataById(source, subSource, findItem.EdbDataId, saveValue)
  123. if err != nil {
  124. return err
  125. }
  126. }
  127. }
  128. // 下面代码主要目的是处理掉手动插入的数据判断
  129. {
  130. if realDataMaxDate.IsZero() || t.After(realDataMaxDate) {
  131. realDataMaxDate = t
  132. }
  133. if edbDataInsertConfigDate.IsZero() || t.Equal(edbDataInsertConfigDate) {
  134. isFindConfigDateRealData = true
  135. }
  136. }
  137. }
  138. }
  139. // 处理手工数据补充的配置
  140. HandleConfigInsertEdbData(realDataMaxDate, edbDataInsertConfig, edbInfoId, source, subSource, existMap, isFindConfigDateRealData)
  141. if isAdd {
  142. addSql = strings.TrimRight(addSql, ",")
  143. _, err = o.Raw(addSql).Exec()
  144. if err != nil {
  145. fmt.Println("RefreshAllEdbDataByPb add Err", err.Error())
  146. return
  147. }
  148. }
  149. }
  150. return
  151. }