base_from_pcsg.go 7.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266
  1. package services
  2. import (
  3. "encoding/json"
  4. "eta/eta_index_lib/logic"
  5. "eta/eta_index_lib/models"
  6. "eta/eta_index_lib/services/alarm_msg"
  7. "eta/eta_index_lib/utils"
  8. "fmt"
  9. "io/ioutil"
  10. "net/http"
  11. "strings"
  12. "time"
  13. )
  14. var (
  15. PCSGBloombergGeneralIndexDataUrl = "/api/pcsg/bloomberg/index_data/general" // 通用指标API
  16. )
  17. type PCSGBloombergApiReq struct {
  18. TaskKey string `description:"任务key"`
  19. Frequency string `description:"指标频度"`
  20. }
  21. type PCSGBloombergTask struct {
  22. TaskKey string `json:"TaskKey"`
  23. Frequency string `json:"Frequency"`
  24. VCode bool `json:"VCode"`
  25. ExtraLetter string `json:"ExtraLetter"`
  26. IndexNamePrefix string `json:"IndexNamePrefix" description:"指标名称前缀"`
  27. IndexCodeSuffix string `json:"IndexCodeSuffix" description:"指标编码后缀"`
  28. }
  29. // LoadPCSGBloombergTask 加载配置
  30. func LoadPCSGBloombergTask() (tasks []*PCSGBloombergTask, err error) {
  31. filePath := "./static/pcsg_task.json"
  32. b, e := ioutil.ReadFile(filePath)
  33. if e != nil {
  34. err = fmt.Errorf("读取配置失败, err: %v", e)
  35. return
  36. }
  37. if e = json.Unmarshal(b, &tasks); e != nil {
  38. err = fmt.Errorf("解析配置失败, err: %v", e)
  39. return
  40. }
  41. return
  42. }
  43. // GetPCSGBloombergGeneralIndexFromBridge 获取通用数据类型指标
  44. func GetPCSGBloombergGeneralIndexFromBridge(params PCSGBloombergApiReq) (indexes []models.BaseFromBloombergApiIndexAndData, err error) {
  45. defer func() {
  46. if err != nil {
  47. tips := fmt.Sprintf("GetPCSGBloombergGeneralIndexFromBridge-获取指标数据失败, err: %s", err.Error())
  48. utils.FileLog.Info(tips)
  49. go alarm_msg.SendAlarmMsg(tips, 3)
  50. }
  51. }()
  52. p, e := json.Marshal(params)
  53. if e != nil {
  54. err = fmt.Errorf("params json marshal err: %v", e)
  55. return
  56. }
  57. url := fmt.Sprint(utils.EtaBridgeUrl, PCSGBloombergGeneralIndexDataUrl)
  58. body := ioutil.NopCloser(strings.NewReader(string(p)))
  59. client := &http.Client{}
  60. req, e := http.NewRequest("POST", url, body)
  61. if e != nil {
  62. err = fmt.Errorf("http create request err: %s", e.Error())
  63. return
  64. }
  65. checkToken := utils.MD5(utils.EtaBridgeAppNameEn + utils.EtaBridgeMd5Key)
  66. contentType := "application/json;charset=utf-8"
  67. req.Header.Set("Content-Type", contentType)
  68. req.Header.Set("Authorization", checkToken)
  69. resp, e := client.Do(req)
  70. if e != nil {
  71. err = fmt.Errorf("http client do err: %s", e.Error())
  72. return
  73. }
  74. defer func() {
  75. _ = resp.Body.Close()
  76. }()
  77. b, e := ioutil.ReadAll(resp.Body)
  78. if e != nil {
  79. err = fmt.Errorf("resp body read err: %s", e.Error())
  80. return
  81. }
  82. if len(b) == 0 {
  83. err = fmt.Errorf("resp body is empty")
  84. return
  85. }
  86. // 生产环境解密
  87. if utils.RunMode == "release" {
  88. str := string(b)
  89. str = strings.Trim(str, `"`)
  90. b = utils.DesBase64Decrypt([]byte(str), utils.EtaBridgeDesKey)
  91. }
  92. result := new(models.BridgePCSGBloombergResultData)
  93. if e = json.Unmarshal(b, &result); e != nil {
  94. err = fmt.Errorf("result unmarshal err: %s\nresult: %s", e.Error(), string(b))
  95. return
  96. }
  97. if result.Code != 200 {
  98. err = fmt.Errorf("result: %s", string(b))
  99. return
  100. }
  101. indexes = result.Data
  102. return
  103. }
  104. // PCSGWrite2BaseBloomberg 写入彭博数据源
  105. func PCSGWrite2BaseBloomberg(indexes []models.BaseFromBloombergApiIndexAndData, isVCode bool, extraLetter, namePrefix string) (err error) {
  106. defer func() {
  107. if err != nil {
  108. tips := fmt.Sprintf("PCSGWrite2BaseBloomberg-写入彭博数据源失败, err: %s", err.Error())
  109. utils.FileLog.Info(tips)
  110. go alarm_msg.SendAlarmMsg(tips, 3)
  111. }
  112. }()
  113. // 这里挡一下...万一没限制加进库了不好删...
  114. if isVCode && extraLetter == "" {
  115. err = fmt.Errorf("中间字母有误")
  116. return
  117. }
  118. for _, v := range indexes {
  119. if v.IndexCode == "" {
  120. continue
  121. }
  122. // 无数据的情况不处理
  123. if len(v.Data) == 0 {
  124. continue
  125. }
  126. if isVCode {
  127. v.IndexCode = utils.InsertStr2StrIdx(v.IndexCode, " ", 1, extraLetter)
  128. }
  129. // 指标是否存在
  130. index, e := models.GetBaseFromBloombergIndexByCode(v.IndexCode)
  131. if e != nil && e.Error() != utils.ErrNoRow() {
  132. err = fmt.Errorf("获取Bloomberg原始指标失败, err: %s", e.Error())
  133. return
  134. }
  135. // 指标名称+前缀
  136. indexName := v.IndexName
  137. if indexName != "" && namePrefix != "" {
  138. indexName = fmt.Sprint(namePrefix, indexName)
  139. }
  140. // 新增指标
  141. if index == nil {
  142. newIndex := new(models.BaseFromBloombergIndex)
  143. newIndex.IndexCode = v.IndexCode
  144. newIndex.IndexName = indexName
  145. newIndex.Unit = v.Unit
  146. newIndex.Source = utils.DATA_SOURCE_BLOOMBERG
  147. newIndex.Frequency = v.Frequency
  148. newIndex.CreateTime = time.Now().Local()
  149. newIndex.ModifyTime = time.Now().Local()
  150. if e = newIndex.Create(); e != nil {
  151. err = fmt.Errorf("新增Bloomberg原始指标失败, err: %s", e.Error())
  152. return
  153. }
  154. index = newIndex
  155. } else {
  156. // 无指标名称的情况下更新指标基础信息
  157. if index.IndexName == "" {
  158. index.IndexName = indexName
  159. index.Unit = v.Unit
  160. index.Frequency = v.Frequency
  161. index.ModifyTime = time.Now().Local()
  162. if e = index.Update([]string{"IndexName", "Unit", "Frequency", "ModifyTime"}); e != nil {
  163. err = fmt.Errorf("更新Bloomberg原始指标失败, err: %s", e.Error())
  164. return
  165. }
  166. }
  167. }
  168. // 更新指标数据
  169. var cond string
  170. var pars []interface{}
  171. cond += ` AND index_code = ? `
  172. pars = append(pars, v.IndexCode)
  173. indexData, e := models.GetBaseFromBloombergDataByCondition(cond, pars)
  174. if e != nil {
  175. err = fmt.Errorf("获取Bloomberg历史数据失败, err: %s", e.Error())
  176. return
  177. }
  178. dateExist := make(map[string]*models.BaseFromBloombergData)
  179. newValExist := make(map[string]bool)
  180. if len(indexData) > 0 {
  181. for _, d := range indexData {
  182. strDate := d.DataTime.Format(utils.FormatDate)
  183. dateExist[strDate] = d
  184. }
  185. }
  186. // 筛选新增/更新数据
  187. updateData := make([]*models.BaseFromBloombergData, 0)
  188. insertData := make([]*models.BaseFromBloombergData, 0)
  189. for _, d := range v.Data {
  190. strDate := d.DataTime.Format(utils.FormatDate)
  191. originData := dateExist[strDate]
  192. if originData != nil {
  193. if utils.FloatAlmostEqual(originData.Value, d.Value) {
  194. continue
  195. }
  196. originData.Value = d.Value
  197. originData.ModifyTime = time.Now().Local()
  198. updateData = append(updateData, originData)
  199. } else {
  200. // 新增的数据去重
  201. if newValExist[strDate] {
  202. continue
  203. }
  204. newValExist[strDate] = true
  205. newData := new(models.BaseFromBloombergData)
  206. newData.BaseFromBloombergIndexId = index.BaseFromBloombergIndexId
  207. newData.IndexCode = index.IndexCode
  208. newData.DataTime = d.DataTime
  209. newData.Value = d.Value
  210. newData.CreateTime = time.Now()
  211. newData.ModifyTime = time.Now()
  212. timestamp := d.DataTime.UnixNano() / 1e6
  213. newData.DataTimestamp = int(timestamp)
  214. insertData = append(insertData, newData)
  215. }
  216. }
  217. if e = models.MultiInsertOrUpdateBaseFromBloombergData(insertData, updateData); e != nil {
  218. err = fmt.Errorf("新增/更新Bloomberg指标数据失败, err: %s", e.Error())
  219. return
  220. }
  221. // 更新指标开始结束时间
  222. minMax, e := models.GetBaseFromBloombergIndexMinMax(index.IndexCode)
  223. if e == nil && minMax != nil {
  224. e = models.ModifyBaseFromBloombergIndexMinMax(index.IndexCode, minMax)
  225. if e != nil {
  226. err = fmt.Errorf("更新Bloomberg开始结束时间失败, err: %s", e.Error())
  227. return
  228. }
  229. }
  230. // 同步刷新指标库
  231. go func() {
  232. edb, e := models.GetEdbInfoByEdbCode(utils.DATA_SOURCE_BLOOMBERG, index.IndexCode)
  233. if e != nil && e.Error() != utils.ErrNoRow() {
  234. utils.FileLog.Info("获取Bloomberg指标库信息失败, err: " + e.Error())
  235. return
  236. }
  237. if edb != nil {
  238. _, _, e = logic.RefreshBaseEdbInfo(edb, ``)
  239. if e != nil {
  240. utils.FileLog.Info(fmt.Sprintf("Bloomberg RefreshBaseEdbInfo, edbCode: %s, err: %v", index.IndexCode, e))
  241. return
  242. }
  243. }
  244. }()
  245. }
  246. return
  247. }