eta_index_lib/models/base_from_pb.go

package models

import (
	"eta_gn/eta_index_lib/utils"
	"fmt"
	"github.com/beego/beego/v2/client/orm"
	"strconv"
	"strings"
	"time"
)

type EdbDataFromPb struct {
	Date   map[string]int64   `json:"date"`
	Ticker map[string]string  `json:"ticker"`
	Field  map[string]string  `json:"field"`
	Value  map[string]float64 `json:"value"`
}

func AddEdbDataFromPb(edbCode string, item *EdbDataFromPb) (err error) {
	o := orm.NewOrm()
	if len(item.Date) > 0 {
		var isAdd bool
		dateMap := item.Date
		addSql := ` INSERT INTO edb_data_pb(edb_info_id,edb_code,data_time,value,create_time,modify_time,ticker,field,data_timestamp) values `
		nowStr := time.Now().Format(utils.FormatDateTime)
		for k, v := range dateMap {
			timeStr := fmt.Sprintf("%d", v)
			v = v / 1000
			t := time.Unix(v, 0)
			dateTime := t.Format(utils.FormatDate)

			//  //这里的目的是为了北京时间，以防万一，还是重新转一下吧
			{
				t, err = time.ParseInLocation(utils.FormatDate, dateTime, time.Local)
				if err != nil {
					return err
				}
				timeStr = fmt.Sprintf("%d", t.UnixNano()/1e6)
			}

			val := item.Value[k]
			field := item.Field[k]
			ticker := item.Ticker[k]

			if field == "PX_LAST" {
				addSql += "("
				addSql += "0," + "'" + edbCode + "'" + "," + "'" + dateTime + "'" + "," + utils.SubFloatToString(val, 20) + "," + "'" + nowStr + "'" +
					"," + "'" + nowStr + "'" + "," + "'" + ticker + "'" + "," + "'" + field + "'" + "," + "'" + timeStr + "'"
				addSql += "),"
				isAdd = true
			}
		}
		if isAdd {
			addSql = strings.TrimRight(addSql, ",")
			_, err = o.Raw(addSql).Exec()
			if err != nil {
				return
			}
		}
	}
	return
}

// 刷新彭博指标数据
func RefreshEdbDataFromPb(edbInfoId int, edbCode, startDate string, item *EdbDataFromPb) (err error) {
	o := orm.NewOrm()
	source := utils.DATA_SOURCE_PB
	subSource := utils.DATA_SUB_SOURCE_EDB

	// 真实数据的最大日期  ， 插入规则配置的日期
	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 condition string
	var pars []interface{}

	condition += " AND edb_info_id=? "
	pars = append(pars, edbInfoId)

	condition += " AND data_time>=? "
	pars = append(pars, startDate)
	existList, err := GetEdbDataByCondition(source, subSource, condition, pars)
	existMap := make(map[string]*EdbInfoSearchData)
	for _, v := range existList {
		existMap[v.DataTime] = v
	}

	edbInfoIdStr := strconv.Itoa(edbInfoId)
	addMap := make(map[string]string)
	if len(item.Date) > 0 {
		dateMap := item.Date
		var isAdd bool
		addSql := ` INSERT INTO edb_data_pb(edb_info_id,edb_code,data_time,value,create_time,modify_time,ticker,field,data_timestamp) values `
		nowStr := time.Now().Format(utils.FormatDateTime)
		for k, v := range dateMap {
			timeStr := fmt.Sprintf("%d", v)
			v = v / 1000
			t := time.Unix(v, 0)
			dateTime := t.Format(utils.FormatDate)

			//  //这里的目的是为了北京时间，以防万一，还是重新转一下吧
			{
				t, err = time.ParseInLocation(utils.FormatDate, dateTime, time.Local)
				if err != nil {
					return err
				}
				timeStr = fmt.Sprintf("%d", t.UnixNano()/1e6)
			}

			val := item.Value[k]
			field := item.Field[k]
			ticker := item.Ticker[k]
			saveValue := utils.SubFloatToString(val, 30)
			if field == "PX_LAST" {
				if findItem, ok := existMap[dateTime]; !ok {
					if _, addOk := addMap[dateTime]; !addOk {
						addSql += "("
						addSql += edbInfoIdStr + "," + "'" + edbCode + "'" + "," + "'" + dateTime + "'" + "," + saveValue + "," + "'" + nowStr + "'" +
							"," + "'" + nowStr + "'" + "," + "'" + ticker + "'" + "," + "'" + field + "'" + "," + "'" + timeStr + "'"
						addSql += "),"
						isAdd = true
						addMap[dateTime] = saveValue
					}
				} else {
					if findItem != nil && utils.SubFloatToString(findItem.Value, 30) != saveValue {
						err = ModifyEdbDataById(source, subSource, findItem.EdbDataId, saveValue)
						if err != nil {
							return err
						}
					}
				}

				// 下面代码主要目的是处理掉手动插入的数据判断
				{
					if realDataMaxDate.IsZero() || t.After(realDataMaxDate) {
						realDataMaxDate = t
					}
					if edbDataInsertConfigDate.IsZero() || t.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 {
				fmt.Println("RefreshAllEdbDataByPb add Err", err.Error())
				return
			}
		}
	}
	return
}