task.go 9.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310
  1. package services
  2. import (
  3. "context"
  4. "eta/eta_task/services/data"
  5. "eta/eta_task/services/data/future_good"
  6. "eta/eta_task/services/data_stat"
  7. "eta/eta_task/utils"
  8. "fmt"
  9. "sync"
  10. "time"
  11. "github.com/beego/beego/v2/task"
  12. )
  13. func Task() {
  14. fmt.Println("task start")
  15. //如果是生产环境,才需要走这些任务
  16. if utils.RunMode == "release" {
  17. releaseTask()
  18. } else {
  19. // 根据配置刷新指标数据
  20. configRefreshData := task.NewTask("syncBaseDataExt", "0 */30 * * * * ", ConfigRefreshData)
  21. task.AddTask("configRefreshData", configRefreshData)
  22. }
  23. // 定时发布智能研报
  24. publishSmartReport := task.NewTask("publishSmartReport", "0 */1 * * * *", PublishSmartReport)
  25. task.AddTask("定时发布智能研报", publishSmartReport)
  26. // 定时汇总数据源终端指标更新情况
  27. setEdbSourceStatTask := task.NewTask("setEdbSourceStatTask", "0 20 19,23 * * *", data_stat.SetEdbSourceStatTask)
  28. task.AddTask("数据源统计表", setEdbSourceStatTask)
  29. XyTask()
  30. task.StartTask()
  31. fmt.Println("task end")
  32. }
  33. // 生产环境需要走的任务
  34. func releaseTask() {
  35. //同步指标
  36. if utils.BusinessCode != utils.BusinessCodeRelease {
  37. syncHzDataIndex := task.NewTask("syncHzDataIndex", "0 10,20,40,50 16,18 * * *", SyncHzDataIndex)
  38. task.AddTask("syncHzDataIndex", syncHzDataIndex)
  39. syncRankingFromDalian := task.NewTask("syncRankingFromDalian", "0 30,40,50 16,18 * * *", SyncRankingFromDalian)
  40. task.AddTask("syncRankingFromDalian", syncRankingFromDalian)
  41. }
  42. // 定时统计交易所的持仓分析数据
  43. initPositionTask := task.NewTask("initPositionTask", "0 20,40 16-19 * * *", data.InitPositionTask)
  44. task.AddTask("initPositionTask", initPositionTask)
  45. //刷新指标数据
  46. refreshData := task.NewTask("refreshData", "0 30 0,19 * * *", RefreshData)
  47. task.AddTask("refreshData", refreshData)
  48. // 根据配置刷新指标数据
  49. configRefreshData := task.NewTask("syncBaseDataExt", "0 */30 * * * * ", ConfigRefreshData)
  50. task.AddTask("configRefreshData", configRefreshData)
  51. // 定时禁用钢联化工和wind指标的刷新状态
  52. disableEdbRefresh := task.NewTask("disableEdbRefresh", "0 0 10 * * *", DisableEdbRefresh)
  53. task.AddTask("disableEdbRefresh", disableEdbRefresh)
  54. //同步弘则数据库中来自,钢联,隆众,有色,人工等基础数据--每隔五分钟,同步一次最新数据
  55. syncBaseData := task.NewTask("syncBaseData", "0 */5 * * * * ", SyncBaseData)
  56. task.AddTask("syncBaseData", syncBaseData)
  57. syncBaseDataExt := task.NewTask("syncBaseDataExt", "0 */30 * * * * ", SyncBaseDataExt)
  58. task.AddTask("syncBaseDataExt", syncBaseDataExt)
  59. //初始化指标更新状态
  60. resetEdbInfoIsUpdate := task.NewTask("resetEdbInfoIsUpdate", "0 0 0 * * *", data.ResetEdbInfoIsUpdate)
  61. task.AddTask("resetEdbInfoIsUpdate", resetEdbInfoIsUpdate)
  62. // 刷新商品数据
  63. refreshFutureGoodData := task.NewTask("refreshFutureGoodData", "0 30 0,19 * * *", RefreshFutureGoodData)
  64. task.AddTask("refreshFutureGoodData", refreshFutureGoodData)
  65. //刷新交易所指标数据
  66. refreshTradeData := task.NewTask("refreshData", "0 1 4 * * *", RefreshTradeData)
  67. task.AddTask("refreshTradeData", refreshTradeData)
  68. //刷新欧洲天然气指标数据
  69. refreshEicData := task.NewTask("refreshData", "0 1 3,7 * * *", RefreshEicData)
  70. task.AddTask("refreshEicData", refreshEicData)
  71. //检测数据服务器
  72. checkDataServer := task.NewTask("checkDataServer", "0 */2 * * * * ", checkDataServer)
  73. task.AddTask("checkDataServer", checkDataServer)
  74. //删除日志 report_save_log,ppt_v2_save_log,保留一个月的
  75. deleteLog := task.NewTask("syncSubStatus", "0 0 2 2 * *", DeleteLog)
  76. task.AddTask("deleteLog", deleteLog)
  77. // 定时发布研报
  78. publishReport := task.NewTask("publishReport", "0 */1 * * * *", PublishReport)
  79. task.AddTask("定时发布研报", publishReport)
  80. // 定时发布英文研报
  81. publishEnglishReport := task.NewTask("publishEnglishReport", "0 */1 * * * *", PublishEnglishReport)
  82. task.AddTask("定时发布英文研报", publishEnglishReport)
  83. // 每天清理两周前的报告保存日志
  84. clearReportSaveLog := task.NewTask("clearReportSaveLog", "0 15 23 * * *", ClearReportSaveLog)
  85. task.AddTask("定时清理报告保存日志", clearReportSaveLog)
  86. // 每天清理三个月前的用户操作日志
  87. clearAdminOperateLog := task.NewTask("clearAdminOperateLog", "0 20 23 * * *", ClearAdminOperateLog)
  88. task.AddTask("定时清理用户操作日志", clearAdminOperateLog)
  89. // 嘉悦物产
  90. if utils.BusinessCode == utils.BusinessCodeJiaYue {
  91. // 每10分钟定时同步增量指标
  92. syncJiaYueNewIndex := task.NewTask("syncJiaYueNewIndex", "0 */10 * * * *", data.SyncJiaYueNewIndex)
  93. task.AddTask("定时同步嘉悦物产增量指标", syncJiaYueNewIndex)
  94. // 每30分钟同步一次数据宝指标数据
  95. syncJiaYueDataBaby := task.NewTask("syncJiaYueDataBaby", "0 */30 * * * * ", data.RefreshJiaYueDataFromBridge)
  96. task.AddTask("syncJiaYueDataBaby", syncJiaYueDataBaby)
  97. }
  98. if utils.BusinessCode == utils.BusinessCodeZhongJi {
  99. // 每天同步一次指标列表
  100. syncZhongJiIndexList := task.NewTask("syncZhongJiIndexList", "0 0 17 * * *", data.SyncZhongJiIndexList)
  101. task.AddTask("定时同步中基宁波SMM指标列表", syncZhongJiIndexList)
  102. }
  103. // 中石油新加坡
  104. if utils.IsPCSG == "1" {
  105. refreshPCSGBloomberg := task.NewTask("refreshPCSGBloombergDaily", "0 30 9 * * *", data.RefreshPCSGBloomberg)
  106. task.AddTask("中石油新加坡-每日Bloomberg指标刷新", refreshPCSGBloomberg)
  107. }
  108. }
  109. func RefreshData(cont context.Context) (err error) {
  110. wg := sync.WaitGroup{}
  111. wg.Add(14)
  112. //hour := time.Now().Hour()
  113. //if hour != 0 {
  114. //}
  115. //彭博
  116. go data.RefreshDataFromPb(&wg)
  117. //彭博财务
  118. go data.RefreshDataFromPbFinance(&wg)
  119. //手工数据
  120. go data.RefreshDataFromManual(&wg)
  121. //隆众数据
  122. //go data.RefreshDataFromLz(&wg)
  123. //有色
  124. go data.RefreshDataFromYs(&wg)
  125. //钢联
  126. go data.RefreshDataFromGl(&wg)
  127. //路透
  128. go data.RefreshDataFromLt(&wg)
  129. //煤炭
  130. go data.RefreshDataFromCoal(&wg)
  131. //谷歌出行数据
  132. go data.RefreshDataFromGoogleTravel(&wg)
  133. //钢联化工
  134. go data.RefreshDataFromMysteelChemical(&wg)
  135. //eia steo报告指标
  136. go data.RefreshDataFromEiaSteo(&wg)
  137. //UN报告指标
  138. go data.RefreshDataFromComTrade(&wg)
  139. //卓创报告指标
  140. go data.RefreshDataFromSci(&wg)
  141. //国家统计局指标
  142. go data.RefreshDataFromNationalStatistics(&wg)
  143. //富宝指标刷新
  144. go data.RefreshDataFromFubao(&wg)
  145. // Bloomberg
  146. go func() {
  147. wg.Add(1)
  148. _ = data.RefreshDataFromBloomberg(&wg)
  149. }()
  150. // CCF化纤信息
  151. go func() {
  152. wg.Add(1)
  153. _ = data.RefreshDataFromCCF(&wg)
  154. }()
  155. wg.Wait()
  156. ////计算指标
  157. data.RefreshDataFromCalculateAll()
  158. // 刷新所有的基础预测指标
  159. data.RefreshBasePredictDataAll()
  160. // 预测计算指标
  161. data.RefreshPredictDataFromCalculateAll()
  162. // 指标系列计算数据
  163. _ = data.RefreshFactorEdbCalculateData()
  164. // 指标系列图表计算数据
  165. _ = data.RefreshFactorEdbChartCalculateData()
  166. time.Sleep(5 * time.Second)
  167. //data.RefreshNotice()
  168. fmt.Println("Refresh End")
  169. return
  170. }
  171. // SyncBaseData 刷新基础数据
  172. func SyncBaseData(cont context.Context) (err error) {
  173. now := time.Now()
  174. if now.Hour() == 0 || now.Hour() == 19 {
  175. return nil
  176. }
  177. //同步钢联基础数据
  178. if utils.BusinessCode == utils.BusinessCodeRelease {
  179. go data.SyncGlDataBase()
  180. }
  181. return
  182. }
  183. // SyncBaseDataExt 刷新基础数据
  184. func SyncBaseDataExt(cont context.Context) (err error) {
  185. now := time.Now()
  186. if now.Hour() == 0 || now.Hour() == 19 {
  187. return nil
  188. }
  189. //同步手工数据
  190. go data.SyncManualDataBase()
  191. //刷新图表中,指标的最新日期
  192. go data.SetChartEdbEndDate()
  193. return
  194. }
  195. // RefreshFutureGoodData 刷新商品数据
  196. func RefreshFutureGoodData(cont context.Context) (err error) {
  197. future_good.RefreshFutureGoodDataFromThs()
  198. fmt.Println("Refresh End")
  199. return
  200. }
  201. // RefreshTradeData 刷新交易所数据
  202. func RefreshTradeData(cont context.Context) (err error) {
  203. wg := sync.WaitGroup{}
  204. wg.Add(6)
  205. //郑商所
  206. go data.RefreshDataFromZz(&wg)
  207. //上期所
  208. go data.RefreshDataFromSh(&wg)
  209. //上期能源
  210. go data.RefreshDataFromShfe(&wg)
  211. //中金所
  212. go data.RefreshDataFromCffex(&wg)
  213. //大商所
  214. go data.RefreshDataFromDl(&wg)
  215. //大商所
  216. go data.RefreshDataFromGz(&wg)
  217. wg.Wait()
  218. //计算指标
  219. data.RefreshDataFromCalculateAll()
  220. // 刷新所有的基础预测指标
  221. data.RefreshBasePredictDataAll()
  222. // 计算预测指标
  223. data.RefreshPredictDataFromCalculateAll()
  224. return
  225. }
  226. // RefreshEicData 刷新欧洲天然气数据
  227. func RefreshEicData(cont context.Context) (err error) {
  228. wg := sync.WaitGroup{}
  229. wg.Add(1)
  230. //欧洲天然气
  231. go data.RefreshDataFromEic(&wg)
  232. wg.Wait()
  233. //计算指标
  234. data.RefreshDataFromCalculateAll()
  235. // 刷新所有的基础预测指标
  236. data.RefreshBasePredictDataAll()
  237. // 计算预测指标
  238. data.RefreshPredictDataFromCalculateAll()
  239. return
  240. }
  241. // 检测数据服务
  242. func checkDataServer(cont context.Context) (err error) {
  243. //检测wind新服务器
  244. go data.CheckWindDataInterface(cont)
  245. //检测同花顺数据服务器
  246. //go data.CheckThsDataInterface(cont)
  247. //检测路透数据服务器
  248. go data.CheckLtDataInterface(cont)
  249. //检测彭博
  250. go data.CheckPbDataInterface(cont)
  251. return
  252. }
  253. func TaskInit() {
  254. fmt.Println("TaskInit start")
  255. //data.InitGuangzhouPositionTask()
  256. fmt.Println("TaskInit end")
  257. return
  258. }