package data_manage import ( "eta/eta_task/utils" "fmt" "github.com/beego/beego/v2/client/orm" "strconv" "strings" "time" ) type ShfeData struct { InputValue string `orm:"column(DATA_VALUE)" description:"日期"` DataTime string `orm:"column(DATA_DATE)" description:"值"` } type BaseFromShfeDataSimple struct { Id int `orm:"column(base_from_trade_ine_index_id);pk"` DealCode string BuyCode string SoldCode string DataTime string DealValue string BuyValue string SoldValue string } type BaseInfoFromShfe struct { DealName string BuyName string SoldName string } type BaseFromTradeShfeIndex struct { BaseFromTradeShfeIndexId int `orm:"column(base_from_trade_ine_index_id);pk"` Rank int DealShortName string DealName string DealCode string DealValue string DealChange int BuyShortName string BuyName string BuyCode string BuyValue string BuyChange int SoldShortName string SoldName string SoldCode string SoldValue string SoldChange int Frequency string ClassifyName string ClassifyType string CreateTime time.Time ModifyTime time.Time DataTime string } func GetEdbDataShfeMaxOrMinDate(edbCode string) (minDate, maxDate string, err error) { o := orm.NewOrm() sql := ` SELECT MIN(data_time) AS minDate,MAX(data_time) AS maxDate FROM edb_data_ine WHERE edb_code=? ` err = o.Raw(sql, edbCode).QueryRow(&minDate, &maxDate) return } // RefreshEdbDataByShfe 刷新上期能源指标数据 func RefreshEdbDataByShfe(edbInfoId int, edbCode, startDate, endDate string) (err error) { o := orm.NewOrm() tx, err := o.Begin() if err != nil { return err } defer func() { if err != nil { tx.Rollback() } else { tx.Commit() } }() if err != nil { return } var suffix string if strings.Contains(edbCode, "deal") { suffix = "deal" } else if strings.Contains(edbCode, "buy") { suffix = "buy" } else if strings.Contains(edbCode, "sold") { suffix = "sold" } edbInfoIdStr := strconv.Itoa(edbInfoId) //计算数据 var condition string var pars []interface{} if edbCode != "" { if suffix == "deal" { condition += " AND deal_code=? " } else if suffix == "buy" { condition += " AND buy_code=? " } else { condition += " AND sold_code=? " } pars = append(pars, edbCode) } if startDate != "" { condition += " AND data_time>=? " pars = append(pars, startDate) } if endDate != "" { condition += " AND data_time<=? " pars = append(pars, endDate) } glDataList, err := GetShfeDataByTradeCode(condition, pars) addSql := ` INSERT INTO edb_data_ine(edb_info_id,edb_code,data_time,value,create_time,modify_time,status,data_timestamp) values ` var isAdd bool existMap := make(map[string]string) for _, v := range glDataList { var value string if suffix == "deal" { value = v.DealValue } else if suffix == "buy" { value = v.BuyValue } else { value = v.SoldValue } item := v itemValue := value if _, ok := existMap[v.DataTime]; !ok { count, err := GetEdbDataShfeByCodeAndDate(edbCode, v.DataTime) if err != nil && err.Error() != utils.ErrNoRow() { return err } if count <= 0 { eDate := item.DataTime sValue := itemValue if sValue != "" { dataTime, err := time.Parse(utils.FormatDate, eDate) if err != nil { return err } timestamp := dataTime.UnixNano() / 1e6 timeStr := fmt.Sprintf("%d", timestamp) addSql += GetAddSql(edbInfoIdStr, edbCode, eDate, timeStr, sValue) isAdd = true } } else { err = ModifyEdbDataShfe(int64(edbInfoId), v.DataTime, value) if err != nil { return err } } } existMap[v.DataTime] = value } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = tx.Raw(addSql).Exec() if err != nil { return err } } return } // GetBaseInfoFromShfeByIndexCode 获取指标信息 func GetBaseInfoFromShfeByIndexCode(indexCode, suffix string) (list []*BaseInfoFromShfe, err error) { o := orm.NewOrm() sql := `SELECT * FROM base_from_trade_ine_index WHERE %s_code=? ` sql = fmt.Sprintf(sql, suffix) _, err = o.Raw(sql, indexCode).QueryRows(&list) return } func GetShfeDataByTradeCode(condition string, pars []interface{}) (item []*BaseFromShfeDataSimple, err error) { sql := ` SELECT * FROM base_from_trade_ine_index WHERE 1=1 ` o := orm.NewOrm() if condition != "" { sql += condition } sql += ` ORDER BY data_time DESC ` _, err = o.Raw(sql, pars).QueryRows(&item) return } func AddEdbDataShfeBySql(sqlStr string) (err error) { o := orm.NewOrm() _, err = o.Raw(sqlStr).Exec() return } func GetBaseFromShfeDataAllByIndexCode(indexCode, suffix string) (list []*BaseFromTradeShfeIndex, err error) { o := orm.NewOrm() sql := `SELECT * FROM base_from_trade_ine_index WHERE %s_code=? ` sql = fmt.Sprintf(sql, suffix) _, err = o.Raw(sql, indexCode).QueryRows(&list) return } func GetEdbDataShfeByCodeAndDate(edbCode string, startDate string) (count int, err error) { o := orm.NewOrm() sql := ` SELECT COUNT(1) AS count FROM edb_data_ine WHERE edb_code=? AND data_time=? ` err = o.Raw(sql, edbCode, startDate).QueryRow(&count) return } func ModifyEdbDataShfe(edbInfoId int64, dataTime, value string) (err error) { o := orm.NewOrm() sql := ` UPDATE edb_data_ine SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? ` _, err = o.Raw(sql, value, edbInfoId, dataTime).Exec() return }