package services import ( "context" "eta/eta_task/services/data" "eta/eta_task/services/data/future_good" "eta/eta_task/utils" "fmt" "github.com/beego/beego/v2/task" "sync" "time" ) func Task() { fmt.Println("task start") //如果是生产环境,才需要走这些任务 if utils.RunMode == "release" { releaseTask() } task.StartTask() fmt.Println("task end") } // 生产环境需要走的任务 func releaseTask() { //同步指标 if utils.BusinessCode != "E2023080900" { syncHzDataIndex := task.NewTask("syncHzDataIndex", "0 10,20,40,50 16,18 * * *", SyncHzDataIndex) task.AddTask("syncHzDataIndex", syncHzDataIndex) } // 定时统计交易所的持仓分析数据 initPositionTask := task.NewTask("initPositionTask", "0 20,40 16-18 * * *", data.InitPositionTask) task.AddTask("initPositionTask", initPositionTask) //刷新指标数据 refreshData := task.NewTask("refreshData", "0 30 0,19 * * *", RefreshData) task.AddTask("refreshData", refreshData) //同步弘则数据库中来自,钢联,隆众,有色,人工等基础数据--每隔五分钟,同步一次最新数据 syncBaseData := task.NewTask("syncBaseData", "0 */5 * * * * ", SyncBaseData) task.AddTask("syncBaseData", syncBaseData) syncBaseDataExt := task.NewTask("syncBaseDataExt", "0 */30 * * * * ", SyncBaseDataExt) task.AddTask("syncBaseDataExt", syncBaseDataExt) //初始化指标更新状态 resetEdbInfoIsUpdate := task.NewTask("resetEdbInfoIsUpdate", "0 0 0 * * *", data.ResetEdbInfoIsUpdate) task.AddTask("resetEdbInfoIsUpdate", resetEdbInfoIsUpdate) // 刷新商品数据 refreshFutureGoodData := task.NewTask("refreshFutureGoodData", "0 30 0,19 * * *", RefreshFutureGoodData) task.AddTask("refreshFutureGoodData", refreshFutureGoodData) //刷新交易所指标数据 refreshTradeData := task.NewTask("refreshData", "0 1 4 * * *", RefreshTradeData) task.AddTask("refreshTradeData", refreshTradeData) //刷新欧洲天然气指标数据 refreshEicData := task.NewTask("refreshData", "0 1 3,7 * * *", RefreshEicData) task.AddTask("refreshEicData", refreshEicData) //检测数据服务器 checkDataServer := task.NewTask("checkDataServer", "0 */2 * * * * ", checkDataServer) task.AddTask("checkDataServer", checkDataServer) //删除日志 report_save_log,ppt_v2_save_log,保留一个月的 deleteLog := task.NewTask("syncSubStatus", "0 0 2 2 * *", DeleteLog) task.AddTask("deleteLog", deleteLog) // 定时发布研报 publishReport := task.NewTask("publishReport", "0 */1 * * * *", PublishReport) task.AddTask("定时发布研报", publishReport) // 定时发布英文研报 publishEnglishReport := task.NewTask("publishEnglishReport", "0 */1 * * * *", PublishEnglishReport) task.AddTask("定时发布英文研报", publishEnglishReport) // 每天清理两周前的报告保存日志 clearReportSaveLog := task.NewTask("clearReportSaveLog", "0 15 23 * * *", ClearReportSaveLog) task.AddTask("定时清理报告保存日志", clearReportSaveLog) } func RefreshData(cont context.Context) (err error) { wg := sync.WaitGroup{} wg.Add(16) //hour := time.Now().Hour() //if hour != 0 { //} go data.RefreshDataFromWind(&wg) //同花顺 go data.RefreshDataFromThs(&wg) //彭博 go data.RefreshDataFromPb(&wg) //彭博财务 go data.RefreshDataFromPbFinance(&wg) //手工数据 go data.RefreshDataFromManual(&wg) //隆众数据 //go data.RefreshDataFromLz(&wg) //有色 go data.RefreshDataFromYs(&wg) //钢联 go data.RefreshDataFromGl(&wg) //路透 go data.RefreshDataFromLt(&wg) //煤炭 go data.RefreshDataFromCoal(&wg) //谷歌出行数据 go data.RefreshDataFromGoogleTravel(&wg) //钢联化工 go data.RefreshDataFromMysteelChemical(&wg) //eia steo报告指标 go data.RefreshDataFromEiaSteo(&wg) //UN报告指标 go data.RefreshDataFromComTrade(&wg) //卓创报告指标 go data.RefreshDataFromSci(&wg) //国家统计局指标 go data.RefreshDataFromNationalStatistics(&wg) wg.Wait() ////计算指标 data.RefreshDataFromCalculateAll() // 预测计算指标 data.RefreshPredictDataFromCalculateAll() time.Sleep(5 * time.Second) data.RefreshNotice() fmt.Println("Refresh End") return } // SyncBaseData 刷新基础数据 func SyncBaseData(cont context.Context) (err error) { now := time.Now() if now.Hour() == 0 || now.Hour() == 19 { return nil } //同步钢联基础数据 go data.SyncGlDataBase() return } // SyncBaseDataExt 刷新基础数据 func SyncBaseDataExt(cont context.Context) (err error) { now := time.Now() if now.Hour() == 0 || now.Hour() == 19 { return nil } //同步手工数据 go data.SyncManualDataBase() //刷新图表中,指标的最新日期 go data.SetChartEdbEndDate() return } // RefreshFutureGoodData 刷新商品数据 func RefreshFutureGoodData(cont context.Context) (err error) { future_good.RefreshFutureGoodDataFromThs() fmt.Println("Refresh End") return } // RefreshTradeData 刷新交易所数据 func RefreshTradeData(cont context.Context) (err error) { wg := sync.WaitGroup{} wg.Add(5) //郑商所 go data.RefreshDataFromZz(&wg) //上期所 go data.RefreshDataFromSh(&wg) //上期能源 go data.RefreshDataFromShfe(&wg) //中金所 go data.RefreshDataFromCffex(&wg) //大商所 go data.RefreshDataFromDl(&wg) wg.Wait() //计算指标 data.RefreshDataFromCalculateAll() // 计算预测指标 data.RefreshPredictDataFromCalculateAll() return } // RefreshEicData 刷新欧洲天然气数据 func RefreshEicData(cont context.Context) (err error) { wg := sync.WaitGroup{} wg.Add(1) //欧洲天然气 go data.RefreshDataFromEic(&wg) wg.Wait() //计算指标 data.RefreshDataFromCalculateAll() // 计算预测指标 data.RefreshPredictDataFromCalculateAll() return } // 检测数据服务 func checkDataServer(cont context.Context) (err error) { //检测wind新服务器 go data.CheckWindDataInterface(cont) //检测同花顺数据服务器 //go data.CheckThsDataInterface(cont) //检测路透数据服务器 go data.CheckLtDataInterface(cont) //检测彭博 go data.CheckPbDataInterface(cont) return }