eta_api/models/data_manage/edb_data_cffex.go

package data_manage

import (
	"eta/eta_api/utils"
	"fmt"
	"github.com/beego/beego/v2/client/orm"
	"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
}