base_from_pb.go 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168
  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. // 真实数据的最大日期 , 插入规则配置的日期
  62. var realDataMaxDate, edbDataInsertConfigDate time.Time
  63. var edbDataInsertConfig *EdbDataInsertConfig
  64. var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值
  65. {
  66. edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId)
  67. if err != nil && err.Error() != utils.ErrNoRow() {
  68. return
  69. }
  70. if edbDataInsertConfig != nil {
  71. edbDataInsertConfigDate = edbDataInsertConfig.Date
  72. }
  73. }
  74. var condition string
  75. var pars []interface{}
  76. condition += " AND edb_info_id=? "
  77. pars = append(pars, edbInfoId)
  78. condition += " AND data_time>=? "
  79. pars = append(pars, startDate)
  80. existList, err := GetEdbDataByCondition(source, condition, pars)
  81. existMap := make(map[string]*EdbInfoSearchData)
  82. for _, v := range existList {
  83. existMap[v.DataTime] = v
  84. }
  85. edbInfoIdStr := strconv.Itoa(edbInfoId)
  86. addMap := make(map[string]string)
  87. if len(item.Date) > 0 {
  88. dateMap := item.Date
  89. var isAdd bool
  90. addSql := ` INSERT INTO edb_data_pb(edb_info_id,edb_code,data_time,value,create_time,modify_time,ticker,field,data_timestamp) values `
  91. nowStr := time.Now().Format(utils.FormatDateTime)
  92. for k, v := range dateMap {
  93. timeStr := fmt.Sprintf("%d", v)
  94. v = v / 1000
  95. t := time.Unix(v, 0)
  96. dateTime := t.Format(utils.FormatDate)
  97. // //这里的目的是为了北京时间,以防万一,还是重新转一下吧
  98. {
  99. t, err = time.ParseInLocation(utils.FormatDate, dateTime, time.Local)
  100. if err != nil {
  101. return err
  102. }
  103. timeStr = fmt.Sprintf("%d", t.UnixNano()/1e6)
  104. }
  105. val := item.Value[k]
  106. field := item.Field[k]
  107. ticker := item.Ticker[k]
  108. saveValue := utils.SubFloatToString(val, 30)
  109. if field == "PX_LAST" {
  110. if findItem, ok := existMap[dateTime]; !ok {
  111. if _, addOk := addMap[dateTime]; !addOk {
  112. addSql += "("
  113. addSql += edbInfoIdStr + "," + "'" + edbCode + "'" + "," + "'" + dateTime + "'" + "," + saveValue + "," + "'" + nowStr + "'" +
  114. "," + "'" + nowStr + "'" + "," + "'" + ticker + "'" + "," + "'" + field + "'" + "," + "'" + timeStr + "'"
  115. addSql += "),"
  116. isAdd = true
  117. addMap[dateTime] = saveValue
  118. }
  119. } else {
  120. if findItem != nil && utils.SubFloatToString(findItem.Value, 30) != saveValue {
  121. err = ModifyEdbDataById(source, findItem.EdbDataId, saveValue)
  122. if err != nil {
  123. return err
  124. }
  125. }
  126. }
  127. // 下面代码主要目的是处理掉手动插入的数据判断
  128. {
  129. if realDataMaxDate.IsZero() || t.After(realDataMaxDate) {
  130. realDataMaxDate = t
  131. }
  132. if edbDataInsertConfigDate.IsZero() || t.Equal(edbDataInsertConfigDate) {
  133. isFindConfigDateRealData = true
  134. }
  135. }
  136. }
  137. }
  138. // 处理手工数据补充的配置
  139. HandleConfigInsertEdbData(realDataMaxDate, edbDataInsertConfig, edbInfoId, source, existMap, isFindConfigDateRealData)
  140. if isAdd {
  141. addSql = strings.TrimRight(addSql, ",")
  142. _, err = o.Raw(addSql).Exec()
  143. if err != nil {
  144. fmt.Println("RefreshAllEdbDataByPb add Err", err.Error())
  145. return
  146. }
  147. }
  148. }
  149. return
  150. }