package data_manage import ( "fmt" "github.com/beego/beego/v2/client/orm" "hongze/hz_crm_api/utils" "strconv" "strings" "time" ) type CffexData struct { InputValue string `orm:"column(DATA_VALUE)" description:"日期"` DataTime string `orm:"column(DATA_DATE)" description:"值"` } type BaseFromCffexDataSimple struct { Id int `orm:"column(base_from_trade_cffex_index_id);pk"` DealCode string BuyCode string SoldCode string DataTime string DealValue string BuyValue string SoldValue string } type BaseInfoFromCffex struct { DealName string BuyName string SoldName string } type BaseFromTradeCFFEXIndex struct { BaseFromTradeCffexIndexId int `orm:"column(base_from_trade_cffex_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 GetEdbDataCffexMaxOrMinDate(edbCode string) (minDate, maxDate string, err error) { o := orm.NewOrmUsingDB("data") sql := ` SELECT MIN(data_time) AS minDate,MAX(data_time) AS maxDate FROM edb_data_cffex WHERE edb_code=? ` err = o.Raw(sql, edbCode).QueryRow(&minDate, &maxDate) return } func GetEdbDataByCffex(edbCode, suffix, startDate, endDate string) (searchItem *EdbInfoSearch, err error) { o := orm.NewOrmUsingDB("data") to, err := o.Begin() if err != nil { return } searchItem = new(EdbInfoSearch) searchItem.EdbCode = edbCode cffexBaseDataAll, err := GetBaseFromCffexDataAllByIndexCode(edbCode, suffix) if err != nil && err.Error() != utils.ErrNoRow() { return } var isAdd bool addSql := ` INSERT INTO edb_data_cffex(edb_info_id,edb_code,data_time,value,create_time,modify_time,status,data_timestamp) values ` dataList := make([]*EdbInfoSearchData, 0) existMap := make(map[string]string) for _, sv := range cffexBaseDataAll { eDate := sv.DataTime dataTime, err := time.Parse(utils.FormatDate, eDate) if err != nil { fmt.Println("time.Parse Err:" + eDate) return nil, err } timestamp := dataTime.UnixNano() / 1e6 timeStr := fmt.Sprintf("%d", timestamp) if _, ok := existMap[eDate]; !ok { if suffix == "deal" { addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.DealValue) } else if suffix == "buy" { addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.BuyValue) } else { addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.SoldValue) } isAdd = true } if suffix == "deal" { existMap[eDate] = sv.DealValue } else if suffix == "buy" { existMap[eDate] = sv.BuyValue } else { existMap[eDate] = sv.SoldValue } } if isAdd { addSql = strings.TrimRight(addSql, ",") utils.FileLog.Info("addSql:" + addSql) _, err = to.Raw(addSql).Exec() if err != nil { return searchItem, err } } if err != nil { _ = to.Rollback() } else { _ = to.Commit() } size := utils.EDB_DATA_LIMIT dataList, err = GetEdbDataAllByEdbCode(edbCode, utils.DATA_SOURCE_CFFEX, size) if err != nil { utils.FileLogData.Info("GetEdbDataCffexByCode Err:%s", err.Error()) return searchItem, err } minDate, maxDate, err := GetEdbDataCffexMaxOrMinDate(edbCode) if err != nil { return searchItem, err } searchItem.DataList = dataList searchItem.StartDate = minDate searchItem.EndDate = maxDate if searchItem.DataList == nil { searchItem.DataList = make([]*EdbInfoSearchData, 0) } return } // RefreshEdbDataByCffex 刷新中金所指标数据 func RefreshEdbDataByCffex(edbInfoId int, edbCode, startDate, endDate string) (err error) { o := orm.NewOrmUsingDB("data") to, err := o.Begin() if err != nil { return } defer func() { if err != nil { _ = to.Rollback() } else { _ = to.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 := GetCffexDataByTradeCode(condition, pars) addSql := ` INSERT INTO edb_data_cffex(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 := GetEdbDataCffexByCodeAndDate(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 = ModifyEdbDataCffex(int64(edbInfoId), v.DataTime, value) if err != nil { return err } } } existMap[v.DataTime] = value } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() if err != nil { return err } } return } // RefreshAllEdbDataByCffex 全部刷新中金所 func RefreshAllEdbDataByCffex(edbInfoId, source int, edbCode, startDate, endDate string) (err error) { o := orm.NewOrmUsingDB("data") to, err := o.Begin() if err != nil { return } defer func() { if err != nil { _ = to.Rollback() } else { _ = to.Commit() } }() 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 := GetCffexDataByTradeCode(condition, pars) //获取指标所有数据 dataList := make([]*EdbDataBase, 0) dataTableName := GetEdbDataTableName(source) sql := `SELECT * FROM %s WHERE edb_info_id=? ` sql = fmt.Sprintf(sql, dataTableName) _, err = to.Raw(sql, edbInfoId).QueryRows(&dataList) if err != nil { return err } dataMap := make(map[string]string) for _, v := range dataList { dataMap[v.DataTime] = v.Value } addSql := ` INSERT INTO edb_data_cffex(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 { 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) saveValue := sValue if existVal, ok := dataMap[eDate]; !ok { addSql += GetAddSql(edbInfoIdStr, edbCode, eDate, timeStr, saveValue) isAdd = true } else { if existVal != saveValue { sql := ` UPDATE %s SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? ` sql = fmt.Sprintf(sql, dataTableName) _, err = to.Raw(sql, sValue, edbInfoId, eDate).Exec() if err != nil { return err } } } } } existMap[v.DataTime] = v.DataTime } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() if err != nil { return err } } return } // GetBaseInfoFromCffexByIndexCode 获取指标信息 func GetBaseInfoFromCffexByIndexCode(indexCode, suffix string) (list []*BaseInfoFromCffex, err error) { o := orm.NewOrmUsingDB("data") sql := `SELECT * FROM base_from_trade_cffex_index WHERE %s_code=? ` sql = fmt.Sprintf(sql, suffix) _, err = o.Raw(sql, indexCode).QueryRows(&list) return } func GetCffexDataByTradeCode(condition string, pars []interface{}) (item []*BaseFromCffexDataSimple, err error) { sql := ` SELECT * FROM base_from_trade_cffex_index WHERE 1=1 ` o := orm.NewOrmUsingDB("data") if condition != "" { sql += condition } sql += ` ORDER BY data_time DESC ` _, err = o.Raw(sql, pars).QueryRows(&item) return } func AddEdbDataCffexBySql(sqlStr string) (err error) { o := orm.NewOrmUsingDB("data") _, err = o.Raw(sqlStr).Exec() return } func GetEdbDataCffexByCode(edbCode string) (items []*EdbInfoSearchData, err error) { o := orm.NewOrmUsingDB("data") sql := ` SELECT * FROM edb_data_cffex WHERE edb_code=? ORDER BY data_time DESC LIMIT ? ` _, err = o.Raw(sql, edbCode, utils.EDB_DATA_LIMIT).QueryRows(&items) return } func GetBaseFromCffexDataAllByIndexCode(indexCode, suffix string) (list []*BaseFromTradeCFFEXIndex, err error) { o := orm.NewOrmUsingDB("data") sql := `SELECT * FROM base_from_trade_cffex_index WHERE %s_code=? ` sql = fmt.Sprintf(sql, suffix) _, err = o.Raw(sql, indexCode).QueryRows(&list) return } func GetEdbDataCffexByCodeAndDate(edbCode string, startDate string) (count int, err error) { o := orm.NewOrmUsingDB("data") sql := ` SELECT COUNT(1) AS count FROM edb_data_cffex WHERE edb_code=? AND data_time=? ` err = o.Raw(sql, edbCode, startDate).QueryRow(&count) return } func ModifyEdbDataCffex(edbInfoId int64, dataTime, value string) (err error) { o := orm.NewOrmUsingDB("data") sql := ` UPDATE edb_data_cffex SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? ` _, err = o.Raw(sql, value, edbInfoId, dataTime).Exec() return }