package models import ( "eta/eta_index_lib/utils" "fmt" "github.com/beego/beego/v2/client/orm" "strconv" "strings" "time" ) type BaseFromMtjhMapping struct { BaseFromMtjhMappingId int `orm:"column(base_from_Mtjh_mapping_id);pk"` IndexName string `description:"持买单量指标名称"` IndexCode string `description:"持买单量指标编码"` CreateTime time.Time `description:"时间"` Area string `description:"区域"` Port string `description:"港口或码头"` Variety string `description:"品种"` Unit string `description:"单位"` Frequency string `description:"频率"` } type BaseFromMtjhIndex struct { BaseFromMtjhIndexId int `orm:"column(base_from_mtjh_index_id);pk"` IndexName string `description:"持买单量指标名称"` IndexCode string `description:"持买单量指标编码"` DealValue string `description:"成交量"` DataTime string `description:"数据日期"` Area string `description:"区域"` Port string `description:"港口或码头"` Variety string `description:"品种"` Unit string `description:"单位"` Frequency string `description:"频率"` CreateTime time.Time `description:"插入时间"` ModifyTime time.Time `description:"修改时间"` } // 查询指标 func GetBaseFromMtjhMapping() (items []*BaseFromMtjhMapping, err error) { o := orm.NewOrm() sql := `SELECT * FROM base_from_mtjh_mapping` _, err = o.Raw(sql).QueryRows(&items) return } // 查询指标 func GetBaseFromMtjhIndex() (items []*BaseFromMtjhIndex, err error) { o := orm.NewOrm() sql := `SELECT * FROM base_from_mtjh_index` _, err = o.Raw(sql).QueryRows(&items) return } // 添加数据 func AddBaseFromMtjhIndex(item *BaseFromMtjhIndex) (lastId int64, err error) { o := orm.NewOrm() lastId, err = o.Insert(item) return } func AddBaseFromMtjhIndexMuti(items []*BaseFromMtjhIndex) (lastId int64, err error) { o := orm.NewOrm() lastId, err = o.InsertMulti(500, items) return } // 添加指标 func AddBaseFromMtjhMapping(item *BaseFromMtjhMapping) (lastId int64, err error) { o := orm.NewOrm() lastId, err = o.Insert(item) return } func AddBaseFromMtjhMappingMuti(items []*BaseFromMtjhMapping) (lastId int64, err error) { o := orm.NewOrm() lastId, err = o.InsertMulti(500, items) return } func UpdateBaseFromMtjhIndex(item *BaseFromMtjhIndex) (err error) { o := orm.NewOrm() sql := `UPDATE base_from_mtjh_index SET deal_value=? WHERE index_name=? AND data_time = ?` _, err = o.Raw(sql, item.DealValue, item.IndexName, item.DataTime).Exec() return } func AddEdbDataFromMtjh(edbCode string) (err error) { o := orm.NewOrm() coalBaseDataAll, err := GetMtjhindexByCode(edbCode) if err != nil && err.Error() != utils.ErrNoRow() { return } var isAdd bool addSql := ` INSERT INTO edb_data_mtjh(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values ` existMap := make(map[string]string) for _, sv := range coalBaseDataAll { eDate := sv.DataTime var timeStr string var dataTime time.Time var sDataTime string var timestamp int64 sDataTime = eDate dataTime, err = time.ParseInLocation(utils.FormatDate, eDate, time.Local) if err != nil { fmt.Println("time.Parse Err:" + eDate) return err } timestamp = dataTime.UnixNano() / 1e6 timeStr = fmt.Sprintf("%d", timestamp) value := strings.Replace(sv.DealValue, "%", "", -1) if _, ok := existMap[sDataTime]; !ok { addSql += GetAddSql("0", edbCode, sDataTime, timeStr, value) fmt.Println("edbCode:", edbCode) fmt.Println("sDataTime:", sDataTime) fmt.Println("timeStr:", timeStr) fmt.Println("value:", value) isAdd = true } existMap[eDate] = value } if isAdd { addSql = strings.TrimRight(addSql, ",") utils.FileLog.Info("addSql:" + addSql) _, err = o.Raw(addSql).Exec() if err != nil { return err } } return } func RefreshEdbDataFromMtjh(edbInfoId int, edbCode, startDate string) (err error) { source := utils.DATA_SOURCE_MTJH subSource := utils.DATA_SUB_SOURCE_EDB o := orm.NewOrm() if err != nil { return } edbInfoIdStr := strconv.Itoa(edbInfoId) //计算数据 var condition string var pars []interface{} if edbCode != "" { condition += " AND index_code=? " pars = append(pars, edbCode) } if startDate != "" { condition += " AND data_time>=? " pars = append(pars, startDate) } glDataList, err := GetMtjhDataByTradeCode(condition, pars) if err != nil { return } // 真实数据的最大日期 , 插入规则配置的日期 var realDataMaxDate, edbDataInsertConfigDate time.Time var edbDataInsertConfig *EdbDataInsertConfig var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值 { edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId) if err != nil && err.Error() != utils.ErrNoRow() { return } if edbDataInsertConfig != nil { edbDataInsertConfigDate = edbDataInsertConfig.Date } } //获取指标所有数据 var existCondition string var existPars []interface{} existCondition += " AND edb_info_id=? " existPars = append(existPars, edbInfoId) if startDate != "" { existCondition += " AND data_time>=? " existPars = append(existPars, startDate) } existList, err := GetEdbDataByCondition(source, subSource, existCondition, existPars) if err != nil { return err } existMap := make(map[string]*EdbInfoSearchData) for _, v := range existList { existMap[v.DataTime] = v } addSql := ` INSERT INTO edb_data_mtjh(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values ` var isAdd bool for _, v := range glDataList { var value string value = strings.Replace(v.DealValue, "%", "", -1) item := v itemValue := value if _, ok := existMap[v.DataTime]; !ok { eDate := item.DataTime var timeStr string var dataTime time.Time var sDataTime string var timestamp int64 sDataTime = eDate dataTime, err = time.ParseInLocation(utils.FormatDate, eDate, time.Local) if err != nil { fmt.Println("time.Parse Err:" + eDate) return err } timestamp = dataTime.UnixNano() / 1e6 timeStr = fmt.Sprintf("%d", timestamp) sValue := itemValue if sValue != "" { saveValue := sValue if findItem, ok := existMap[eDate]; !ok { addSql += GetAddSql(edbInfoIdStr, edbCode, sDataTime, timeStr, saveValue) isAdd = true } else { if findItem != nil && utils.SubFloatToString(findItem.Value, 30) != sValue { err = ModifyEdbDataById(source, subSource, findItem.EdbDataId, sValue) if err != nil { return err } } } } // 下面代码主要目的是处理掉手动插入的数据判断 { if realDataMaxDate.IsZero() || dataTime.After(realDataMaxDate) { realDataMaxDate = dataTime } if edbDataInsertConfigDate.IsZero() || dataTime.Equal(edbDataInsertConfigDate) { isFindConfigDateRealData = true } } } } // 处理手工数据补充的配置 HandleConfigInsertEdbData(realDataMaxDate, edbDataInsertConfig, edbInfoId, source, subSource, existMap, isFindConfigDateRealData) if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = o.Raw(addSql).Exec() if err != nil { return err } } return } // GetMtjhindexByCode func GetMtjhindexByCode(indexCode string) (items []*BaseFromMtjhIndex, err error) { o := orm.NewOrm() sql := "SELECT * FROM base_from_mtjh_index WHERE index_code=? " _, err = o.Raw(sql, indexCode).QueryRows(&items) return } func GetMtjhDataByTradeCode(condition string, pars []interface{}) (item []*BaseFromMtjhIndex, err error) { sql := ` SELECT * FROM base_from_mtjh_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 }