同花顺高频-mongo

controllers/data_manage/base_from_ths_hf.go

@@ -5,11 +5,13 @@ import (
 	"eta/eta_api/controllers"
 	"eta/eta_api/models"
 	"eta/eta_api/models/data_manage"
+	"eta/eta_api/models/mgo"
 	"eta/eta_api/services/data"
 	"eta/eta_api/utils"
 	"fmt"
 	"github.com/rdlucklib/rdluck_tools/paging"
 	"github.com/shopspring/decimal"
+	"go.mongodb.org/mongo-driver/bson"
 	"sort"
 	"strconv"
 	"strings"
@@ -824,22 +826,15 @@ func (this *BaseFromThsHfController) Detail() {
 	if dataDate == "" {
 		dataDate = item.EndDate.Format(utils.FormatDate)
 	}
+	startTime := fmt.Sprintf("%s 00:00:00", dataDate)
+	endTime := fmt.Sprintf("%s 23:59:59", dataDate)
 
-	dataList := make([]*data_manage.BaseFromThsHfDataItem, 0)
-	{
-		dataOb := new(data_manage.BaseFromThsHfData)
-		cond := fmt.Sprintf(" AND %s = ? AND (%s BETWEEN ? AND ?)", dataOb.Cols().IndexCode, dataOb.Cols().DataTime)
-		pars := make([]interface{}, 0)
-		pars = append(pars, indexItem.IndexCode, fmt.Sprintf("%s 00:00:00", dataDate), fmt.Sprintf("%s 23:59:59", dataDate))
-		list, e := dataOb.GetItemsByCondition(cond, pars, []string{}, fmt.Sprintf("%s DESC", dataOb.Cols().DataTime))
-		if e != nil {
-			br.Msg = "获取失败"
-			br.ErrMsg = fmt.Sprintf("获取指标数据失败, %v", e)
-			return
-		}
-		for _, v := range list {
-			dataList = append(dataList, v.Format2Item())
-		}
+	// 获取数据
+	dataList, e := data.GetThsHfBaseIndexData(indexItem.IndexCode, startTime, endTime)
+	if e != nil {
+		br.Msg = "获取失败"
+		br.ErrMsg = fmt.Sprintf("获取指标数据失败, %v", e)
+		return
 	}
 
 	type DetailResp struct {
@@ -984,21 +979,29 @@ func (this *BaseFromThsHfController) Remove() {
 		}
 	}
 
-	// TODO:同步Mongo
 	if e = item.Remove(); e != nil {
 		br.Msg = "操作失败"
 		br.ErrMsg = fmt.Sprintf("删除指标失败, %v", e)
 		return
 	}
 
-	dataOb := new(data_manage.BaseFromThsHfData)
-	cond := fmt.Sprintf(" %s = ?", dataOb.Cols().BaseFromThsHfIndexId)
-	pars := make([]interface{}, 0)
-	pars = append(pars, params.IndexId)
-	if e = dataOb.RemoveByCondition(cond, pars); e != nil {
-		br.Msg = "操作失败"
-		br.ErrMsg = fmt.Sprintf("删除指标数据失败, %v", e)
-		return
+	if utils.UseMongo {
+		mogDataObj := mgo.BaseFromThsHfData{}
+		if e = mogDataObj.RemoveMany(bson.M{"index_code": item.IndexCode}); e != nil {
+			br.Msg = "操作失败"
+			br.ErrMsg = fmt.Sprintf("删除指标数据失败-Mgo, %v", e)
+			return
+		}
+	} else {
+		dataOb := new(data_manage.BaseFromThsHfData)
+		cond := fmt.Sprintf(" %s = ?", dataOb.Cols().BaseFromThsHfIndexId)
+		pars := make([]interface{}, 0)
+		pars = append(pars, params.IndexId)
+		if e = dataOb.RemoveByCondition(cond, pars); e != nil {
+			br.Msg = "操作失败"
+			br.ErrMsg = fmt.Sprintf("删除指标数据失败, %v", e)
+			return
+		}
 	}
 
 	br.Ret = 200
@@ -1317,14 +1320,26 @@ func (this *BaseFromThsHfController) MultiOpt() {
 				br.ErrMsg = fmt.Sprintf("批量删除指标失败, %v", e)
 				return
 			}
-			dataOb := new(data_manage.BaseFromThsHfData)
-			cond := fmt.Sprintf(" %s IN (%s)", dataOb.Cols().BaseFromThsHfIndexId, utils.GetOrmInReplace(len(removeIndexIds)))
-			pars := make([]interface{}, 0)
-			pars = append(pars, removeIndexIds)
-			if e := dataOb.RemoveByCondition(cond, pars); e != nil {
-				br.Msg = "删除失败"
-				br.ErrMsg = fmt.Sprintf("批量删除指标数据失败, %v", e)
-				return
+
+			if utils.UseMongo {
+				mogDataObj := mgo.BaseFromThsHfData{}
+				for _, v := range removeIndexIds {
+					if e := mogDataObj.RemoveMany(bson.M{"base_from_ths_hf_index_id": v}); e != nil {
+						br.Msg = "操作失败"
+						br.ErrMsg = fmt.Sprintf("批量删除指标数据失败-Mgo, %v", e)
+						return
+					}
+				}
+			} else {
+				dataOb := new(data_manage.BaseFromThsHfData)
+				cond := fmt.Sprintf(" %s IN (%s)", dataOb.Cols().BaseFromThsHfIndexId, utils.GetOrmInReplace(len(removeIndexIds)))
+				pars := make([]interface{}, 0)
+				pars = append(pars, removeIndexIds)
+				if e := dataOb.RemoveByCondition(cond, pars); e != nil {
+					br.Msg = "删除失败"
+					br.ErrMsg = fmt.Sprintf("批量删除指标数据失败, %v", e)
+					return
+				}
 			}
 
 			// 标记删除成功的

models/data_manage/chart_info.go

@@ -398,6 +398,9 @@ func GetEdbDataList(source, subSource, edbInfoId int, startDate, endDate string)
 	if source == utils.DATA_SOURCE_BUSINESS && utils.UseMongo {
 		return getEdbDataListByMongo(source, subSource, edbInfoId, startDate, endDate)
 	}
+	if source == utils.DATA_SOURCE_THS && subSource == utils.DATA_SUB_SOURCE_HIGH_FREQUENCY && utils.UseMongo {
+		return getThsHfEdbDataListByMongo(source, subSource, edbInfoId, startDate, endDate)
+	}
 
 	return getEdbDataListByMysql(source, subSource, edbInfoId, startDate, endDate)
 
@@ -521,6 +524,9 @@ func GetEdbDataListMinAndMax(source, subSource, edbInfoId int, startDate, endDat
 	if source == utils.DATA_SOURCE_BUSINESS && utils.UseMongo {
 		return getEdbDataListMinAndMaxByMongo(source, subSource, edbInfoId, startDate, endDate)
 	}
+	if source == utils.DATA_SOURCE_THS && subSource == utils.DATA_SUB_SOURCE_HIGH_FREQUENCY && utils.UseMongo {
+		return getThsHfEdbDataListMinAndMaxByMongo(source, subSource, edbInfoId, startDate, endDate)
+	}
 
 	return getEdbDataListMinAndMaxByMysql(source, subSource, edbInfoId, startDate, endDate)
 }
@@ -2409,3 +2415,72 @@ func GetChartInfoBySourceAndParentId(source, parentId, adminId int) (items []*Ch
 	_, err = o.Raw(sql, pars).QueryRows(&items)
 	return
 }
+
+func getThsHfEdbDataListByMongo(source, subSource, edbInfoId int, startDate, endDate string) (list []*EdbDataList, err error) {
+	list = make([]*EdbDataList, 0)
+
+	mogDataObj := mgo.EdbDataThsHf{}
+	// 构建查询条件
+	queryConditions := bson.M{
+		"edb_info_id": edbInfoId,
+	}
+
+	// 数据日期
+	dateCondition, err := mgo.BuildDateCondition(startDate, endDate)
+	if err != nil {
+		return
+	}
+	if len(dateCondition) > 0 {
+		queryConditions["data_time"] = dateCondition
+	}
+
+	// 获取列表数据
+	tmpDataList, tmpErr := mogDataObj.GetAllDataList(queryConditions, []string{"data_time"})
+	if tmpErr != nil {
+		err = tmpErr
+		return
+	}
+	for k, v := range tmpDataList {
+		list = append(list, &EdbDataList{
+			EdbDataId:     k + 1,
+			EdbInfoId:     v.EdbInfoId,
+			DataTime:      v.DataTime.Format(utils.FormatDate),
+			DataTimestamp: v.DataTimestamp,
+			Value:         v.Value,
+		})
+	}
+	return
+}
+
+func getThsHfEdbDataListMinAndMaxByMongo(source, subSource, edbInfoId int, startDate, endDate string) (minData, maxData float64, err error) {
+	mogDataObj := mgo.EdbDataThsHf{}
+	// 构建查询条件
+	queryConditions := bson.M{
+		"edb_info_id": edbInfoId,
+	}
+	// 日期
+	dateCondition, err := mgo.BuildDateCondition(startDate, endDate)
+	if err != nil {
+		return
+	}
+	if len(dateCondition) > 0 {
+		queryConditions["data_time"] = dateCondition
+	}
+
+	pipeline := []bson.M{
+		{"$match": queryConditions},
+		{"$group": bson.M{
+			"_id":       nil,
+			"min_value": bson.M{"$min": "$value"},
+			"max_value": bson.M{"$max": "$value"},
+		}},
+		{"$project": bson.M{"_id": 0}}, // 可选，如果不需要_id字段
+	}
+	result, err := mogDataObj.GetEdbInfoMaxAndMinInfo(pipeline)
+	if err != nil {
+		return
+	}
+	minData = result.MinValue
+	maxData = result.MaxValue
+	return
+}

models/data_manage/edb_data_base.go

@@ -227,6 +227,9 @@ func GetEdbDataAllByEdbCode(edbCode string, source, subSource, limit int) (items
 	if source == utils.DATA_SOURCE_BUSINESS && utils.UseMongo {
 		return GetEdbDataAllByEdbCodeByMongo(edbCode, source, subSource, limit)
 	}
+	if source == utils.DATA_SOURCE_THS && subSource == utils.DATA_SUB_SOURCE_HIGH_FREQUENCY && utils.UseMongo {
+		return GetThsHfEdbDataAllByEdbCodeByMongo(edbCode, source, subSource, limit)
+	}
 
 	return GetEdbDataAllByEdbCodeByMysql(edbCode, source, subSource, limit)
 }
@@ -405,3 +408,60 @@ func GetAddSql(edbInfoId, edbCode, dataTime, timestampStr string, value string)
 	addSql += "),"
 	return
 }
+
+func GetThsHfEdbDataAllByEdbCodeByMongo(edbCode string, source, subSource, limit int) (list []*EdbInfoSearchData, err error) {
+	list = make([]*EdbInfoSearchData, 0)
+
+	mogDataObj := mgo.EdbDataThsHf{}
+	// 构建查询条件
+	queryConditions := bson.M{
+		"edb_code": edbCode,
+	}
+
+	var tmpDataList []*mgo.EdbDataThsHf
+	// 获取列表数据
+	if limit > 0 {
+		tmpDataList, err = mogDataObj.GetLimitDataList(queryConditions, int64(limit), []string{"-data_time"})
+	} else {
+		tmpDataList, err = mogDataObj.GetLimitDataList(queryConditions, int64(limit), []string{"-data_time"})
+	}
+	if err != nil {
+		return
+	}
+	for _, v := range tmpDataList {
+		list = append(list, &EdbInfoSearchData{
+			DataTime: v.DataTime.Format(utils.FormatDate),
+			Value:    v.Value,
+		})
+	}
+	return
+}
+
+func GetThsHfEdbDataBaseMongoByEdbInfoId(edbInfoId int, source, subSource int) (list []*EdbDataBase, err error) {
+	list = make([]*EdbDataBase, 0)
+
+	mogDataObj := mgo.EdbDataThsHf{}
+	// 构建查询条件
+	queryConditions := bson.M{
+		"edb_info_id": edbInfoId,
+	}
+	// 获取列表数据
+	tmpDataList, tmpErr := mogDataObj.GetAllDataList(queryConditions, []string{"data_time"})
+	if tmpErr != nil {
+		err = tmpErr
+		return
+	}
+	for k, v := range tmpDataList {
+
+		list = append(list, &EdbDataBase{
+			EdbDataId:     k + 1,
+			EdbInfoId:     v.EdbInfoId,
+			EdbCode:       v.EdbCode,
+			DataTime:      v.DataTime.Format(utils.FormatDate),
+			DataTimestamp: v.DataTimestamp,
+			Value:         strconv.FormatFloat(v.Value, 'f', -1, 64),
+		})
+	}
+
+	return
+}

models/data_manage/edb_data_insert_config.go

@@ -135,6 +135,8 @@ func CreateEdbDataInsertConfigAndData(edbInfo *EdbInfo, date time.Time, value st
 	// 指标明细数据更新
 	if edbInfo.Source == utils.DATA_SOURCE_BUSINESS && utils.UseMongo {
 		dateStr, err = updateInsertConfigValueByMongo(to, edbInfo, oldConfigDate, date, value)
+	} else if edbInfo.Source == utils.DATA_SOURCE_THS && edbInfo.SubSource == utils.DATA_SUB_SOURCE_HIGH_FREQUENCY && utils.UseMongo {
+		dateStr, err = updateThsHfInsertConfigValueByMongo(to, edbInfo, oldConfigDate, date, value)
 	} else {
 		dateStr, err = updateInsertConfigValueByMysql(to, edbInfo, oldConfigDate, date, value)
 	}
@@ -336,3 +338,102 @@ func updateInsertConfigValueByMongo(to orm.TxOrmer, edbInfo *EdbInfo, oldConfigD
 
 	return
 }
+
+func updateThsHfInsertConfigValueByMongo(to orm.TxOrmer, edbInfo *EdbInfo, oldConfigDate, newDate time.Time, value string) (dateStr string, err error) {
+	tableName := GetEdbDataTableName(edbInfo.Source, edbInfo.SubSource)
+	if tableName == `` {
+		err = errors.New("找不到该指标的数据表")
+		return
+	}
+
+	dateStr = newDate.Format(utils.FormatDate)
+	timestamp := newDate.UnixNano() / 1e6
+	var floatValue float64
+	if value != "" {
+		floatValue, err = strconv.ParseFloat(value, 64)
+		if err != nil {
+			fmt.Println("转换失败:", err.Error())
+			return
+		}
+	}
+
+	mogDataObj := mgo.EdbDataThsHf{}
+	coll := mogDataObj.GetCollection()
+
+	var edbDateData *mgo.EdbDataThsHf
+	if !oldConfigDate.IsZero() {
+		// 构建查询条件
+		queryConditions := bson.M{
+			"edb_info_id": edbInfo.EdbInfoId,
+			"data_time":   oldConfigDate,
+		}
+
+		edbDateData, err = mogDataObj.GetItem(queryConditions)
+		//if tmpErr != nil && tmpErr == mongo.ErrNoDocuments {
+		//	err = tmpErr
+		//	return
+		//}
+		if err != nil && err != mongo.ErrNoDocuments {
+			return
+		}
+		err = nil
+
+	}
+
+	// 如果是没有历史数据，那么就需要增加数据
+	if edbDateData == nil {
+		addDataItem := mgo.EdbDataThsHf{
+			//ID:            primitive.ObjectID{},
+			EdbInfoId:     edbInfo.EdbInfoId,
+			EdbCode:       edbInfo.EdbCode,
+			DataTime:      newDate,
+			Value:         floatValue,
+			CreateTime:    time.Now(),
+			ModifyTime:    time.Now(),
+			DataTimestamp: timestamp,
+		}
+		err = mogDataObj.InsertDataByColl(coll, addDataItem)
+		if err != nil {
+			fmt.Println("mogDataObj.BatchInsertData() Err:" + err.Error())
+			return
+		}
+		return
+	}
+
+	// 数据清空
+	if value == "" {
+		queryConditions := bson.M{
+			"edb_info_id": edbInfo.EdbInfoId,
+		}
+
+		// 获取最新的两条数据
+		tmpDataList, tmpErr := mogDataObj.GetLimitDataList(queryConditions, 2, []string{"-data_time"})
+		if tmpErr != nil {
+			fmt.Println("mogDataObj.GetLimitDataList() Err:" + tmpErr.Error())
+			return
+		}
+		// 如果并没有两条数据，那么就返回
+		if len(tmpDataList) < 2 {
+			return
+		}
+
+		// 实际应该是倒数第二条数据的日期
+		dateStr = tmpDataList[1].DataTime.Format(utils.FormatDate)
+
+		// 删除插入的数据
+		err = mogDataObj.RemoveManyByColl(coll, bson.M{"_id": tmpDataList[0].ID})
+
+		return
+	}
+
+	// 更新配置的数据
+	updateData := bson.M{"$set": bson.M{
+		"value":          floatValue,
+		"modify_time":    time.Now(),
+		"data_time":      newDate,
+		"data_timestamp": timestamp,
+	}}
+	err = mogDataObj.UpdateDataByColl(coll, bson.M{"_id": edbDateData.ID}, updateData)
+
+	return
+}

models/data_manage/edb_info.go

@@ -231,6 +231,9 @@ func DeleteEdbInfoAndData(edbInfoId, source, subSource int) (err error) {
 	// 删除指标的明细数据
 	if source == utils.DATA_SOURCE_BUSINESS && utils.UseMongo {
 		err = deleteAllEdbDataByMongo(to, edbInfoId, source, subSource)
+	} else if source == utils.DATA_SOURCE_THS && subSource == utils.DATA_SUB_SOURCE_HIGH_FREQUENCY && utils.UseMongo {
+		mogDataObj := mgo.EdbDataThsHf{}
+		err = mogDataObj.RemoveMany(bson.M{"edb_info_id": edbInfoId})
 	} else {
 		err = deleteAllEdbDataByMysql(to, edbInfoId, source, subSource)
 	}
@@ -597,6 +600,9 @@ func GetEdbInfoMaxAndMinInfo(source, subSource int, edbCode string) (item *EdbIn
 	if source == utils.DATA_SOURCE_BUSINESS && utils.UseMongo {
 		return GetEdbInfoMaxAndMinInfoByMongo(source, subSource, edbCode)
 	}
+	if source == utils.DATA_SOURCE_THS && subSource == utils.DATA_SUB_SOURCE_HIGH_FREQUENCY && utils.UseMongo {
+		return GetThsHfEdbInfoMaxAndMinInfoByMongo(source, subSource, edbCode)
+	}
 
 	// 默认走mysql
 	return GetEdbInfoMaxAndMinInfoByMysql(source, subSource, edbCode)
@@ -1724,6 +1730,9 @@ func GetAllEdbDataListData(edbInfoId, source, subSource int, startDataTime strin
 	if source == utils.DATA_SOURCE_BUSINESS && utils.UseMongo {
 		return getAllDataByMongo(edbInfoId, source, subSource, startDataTime)
 	}
+	if source == utils.DATA_SOURCE_THS && subSource == utils.DATA_SUB_SOURCE_HIGH_FREQUENCY && utils.UseMongo {
+		return getThsHfAllDataByMongo(edbInfoId, source, subSource, startDataTime)
+	}
 
 	// 默认走mysql
 	return getAllDataByMysql(edbInfoId, source, subSource, startDataTime)
@@ -1858,3 +1867,81 @@ func GetEdbInfoListByCond(condition string, pars []interface{}) (list []*EdbInfo
 	_, err = o.Raw(sql, pars).QueryRows(&list)
 	return
 }
+
+func GetThsHfEdbInfoMaxAndMinInfoByMongo(source, subSource int, edbCode string) (item *EdbInfoMaxAndMinInfo, err error) {
+	mogDataObj := new(mgo.EdbDataThsHf)
+	pipeline := []bson.M{
+		{"$match": bson.M{"edb_code": edbCode}},
+		{"$group": bson.M{
+			"_id":       nil,
+			"min_date":  bson.M{"$min": "$data_time"},
+			"max_date":  bson.M{"$max": "$data_time"},
+			"min_value": bson.M{"$min": "$value"},
+			"max_value": bson.M{"$max": "$value"},
+		}},
+		{"$project": bson.M{"_id": 0}}, // 可选，如果不需要_id字段
+	}
+	result, err := mogDataObj.GetEdbInfoMaxAndMinInfo(pipeline)
+	if err != nil {
+		fmt.Println("EdbDataThsHf getEdbDataThsHfList Err:" + err.Error())
+		return
+	}
+
+	if !result.MaxDate.IsZero() {
+		whereQuery := bson.M{"edb_code": edbCode, "data_time": result.MaxDate}
+		selectParam := bson.D{{"value", 1}, {"_id", 0}}
+		latestValue, tmpErr := mogDataObj.GetLatestValue(whereQuery, selectParam)
+		if tmpErr != nil {
+			err = tmpErr
+			return
+		}
+		result.LatestValue = latestValue.Value
+		result.EndValue = latestValue.Value
+	}
+
+	item = &EdbInfoMaxAndMinInfo{
+		MinDate:     result.MinDate.Format(utils.FormatDate),
+		MaxDate:     result.MaxDate.Format(utils.FormatDate),
+		MinValue:    result.MinValue,
+		MaxValue:    result.MaxValue,
+		LatestValue: result.LatestValue,
+	}
+
+	return
+}
+
+func getThsHfAllDataByMongo(edbInfoId, source, subSource int, startDataTime string) (dataList []*EdbData, err error) {
+	dataList = make([]*EdbData, 0)
+
+	mogDataObj := mgo.EdbDataThsHf{}
+	// 构建查询条件
+	queryConditions := bson.M{
+		"edb_info_id": edbInfoId,
+	}
+
+	// 开始日期
+	dateCondition, err := mgo.BuildDateCondition(startDataTime, "")
+	if err != nil {
+		return
+	}
+	if len(dateCondition) > 0 {
+		queryConditions["data_time"] = dateCondition
+	}
+
+	// 获取列表数据
+	tmpDataList, tmpErr := mogDataObj.GetAllDataList(queryConditions, []string{"-data_time"})
+	if tmpErr != nil {
+		err = tmpErr
+		return
+	}
+	for k, v := range tmpDataList {
+		dataList = append(dataList, &EdbData{
+			EdbDataId: k + 1,
+			EdbInfoId: v.EdbInfoId,
+			DataTime:  v.DataTime.Format(utils.FormatDate),
+			Value:     v.Value,
+		})
+	}
+
+	return
+}

models/mgo/base_from_ths_hf_data.go

@@ -0,0 +1,423 @@
+package mgo
+
+import (
+	"context"
+	"errors"
+	"eta/eta_api/utils"
+	"fmt"
+	"github.com/qiniu/qmgo"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"time"
+)
+
+// BaseFromThsHfData
+// @Description: 同花顺高频集合
+type BaseFromThsHfData struct {
+	ID                   primitive.ObjectID `json:"_id" bson:"_id,omitempty"`                                   // 文档id
+	BaseFromThsHfDataId  int64              `json:"base_from_ths_hf_data_id" bson:"base_from_ths_hf_data_id"`   // 指标数据ID
+	BaseFromThsHfIndexId int64              `json:"base_from_ths_hf_index_id" bson:"base_from_ths_hf_index_id"` // 指标ID
+	IndexCode            string             `json:"index_code" bson:"index_code"`                               // 指标编码
+	DataTime             time.Time          `json:"data_time" bson:"data_time"`                                 // 数据日期
+	Value                float64            `json:"value" bson:"value"`                                         // 数据值
+	UniqueCode           string             `json:"unique_code" bson:"unique_code"`                             // 唯一编码
+	CreateTime           time.Time          `json:"create_time" bson:"create_time"`                             // 创建时间
+	ModifyTime           time.Time          `json:"modify_time" bson:"modify_time"`                             // 修改时间
+	DataTimestamp        int64              `json:"data_timestamp" bson:"data_timestamp"`                       // 数据日期时间戳
+}
+
+// CollectionName
+// @Description:  获取集合名称
+func (m *BaseFromThsHfData) CollectionName() string {
+	return "base_from_ths_hf_data"
+}
+
+// DataBaseName
+// @Description: 获取数据库名称
+func (m *BaseFromThsHfData) DataBaseName() string {
+	return utils.MgoDataDbName
+}
+
+// GetCollection
+// @Description: 获取mongodb集合的句柄
+func (m *BaseFromThsHfData) GetCollection() *qmgo.Collection {
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	return db.Collection(m.CollectionName())
+}
+
+// GetAllDataList
+// @Description: 根据条件获取所有数据
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 13:42:19
+// @param sort []string
+// @param whereParams interface{}
+// @return result []BaseFromThsHfData
+// @return err error
+func (m *BaseFromThsHfData) GetAllDataList(whereParams interface{}, sort []string) (result []*BaseFromThsHfData, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+	err = coll.Find(ctx, whereParams).Sort(sort...).All(&result)
+	if err != nil {
+		return
+	}
+
+	for _, v := range result {
+		v.DataTime = v.DataTime.In(time.Local)
+		v.CreateTime = v.CreateTime.In(time.Local)
+		v.ModifyTime = v.ModifyTime.In(time.Local)
+	}
+	return
+}
+
+// GetLimitDataList
+// @Description: 根据条件获取指定数量数据列表
+// @author: Roc
+// @receiver m
+// @datetime 2024-05-06 17:08:32
+// @param whereParams interface{}
+// @param size int64
+// @return result []*BaseFromThsHfData
+// @return err error
+func (m *BaseFromThsHfData) GetLimitDataList(whereParams interface{}, size int64, sort []string) (result []*BaseFromThsHfData, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+	err = coll.Find(ctx, whereParams).Sort(sort...).Limit(size).All(&result)
+	if err != nil {
+		return
+	}
+
+	for _, v := range result {
+		v.DataTime = v.DataTime.In(time.Local)
+		v.CreateTime = v.CreateTime.In(time.Local)
+		v.ModifyTime = v.ModifyTime.In(time.Local)
+	}
+
+	return
+}
+
+// GetPageDataList
+// @Description: 根据条件获取分页数据列表
+// @author: Roc
+// @receiver m
+// @datetime 2024-05-07 10:21:07
+// @param whereParams interface{}
+// @param startSize int64
+// @param size int64
+// @param sort []string
+// @return result []*BaseFromThsHfData
+// @return err error
+func (m *BaseFromThsHfData) GetPageDataList(whereParams interface{}, startSize, size int64, sort []string) (result []*BaseFromThsHfData, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+	err = coll.Find(ctx, whereParams).Sort(sort...).Skip(startSize).Limit(size).All(&result)
+	if err != nil {
+		return
+	}
+
+	for _, v := range result {
+		v.DataTime = v.DataTime.In(time.Local)
+		v.CreateTime = v.CreateTime.In(time.Local)
+		v.ModifyTime = v.ModifyTime.In(time.Local)
+	}
+
+	return
+}
+
+// GetCountDataList
+// @Description:  根据条件获取数据列表总数
+// @author: Roc
+// @receiver m
+// @datetime 2024-05-07 10:29:00
+// @param whereParams interface{}
+// @return count int64
+// @return err error
+func (m *BaseFromThsHfData) GetCountDataList(whereParams interface{}) (count int64, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+	count, err = coll.Find(ctx, whereParams).Count()
+
+	return
+}
+
+// InsertDataByColl
+// @Description: 写入单条数据（外部传入集合）
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 14:22:18
+// @param addData interface{}
+// @return err error
+func (m *BaseFromThsHfData) InsertDataByColl(coll *qmgo.Collection, addData interface{}) (err error) {
+	ctx := context.TODO()
+	_, err = coll.InsertOne(ctx, addData)
+	if err != nil {
+		fmt.Println("InsertDataByColl:Err:" + err.Error())
+		return
+	}
+
+	return
+}
+
+// BatchInsertData
+// @Description: 批量写入数据
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 14:22:18
+// @param bulk int 每次请求保存的数据量
+// @param dataList []interface{}
+// @return err error
+func (m *BaseFromThsHfData) BatchInsertData(bulk int, dataList []interface{}) (err error) {
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+
+	return m.BatchInsertDataByColl(coll, bulk, dataList)
+}
+
+// BatchInsertDataByColl
+// @Description: 批量写入数据（外部传入集合）
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 14:22:18
+// @param coll *qmgo.Collection
+// @param bulk int 每次请求保存的数据量
+// @param dataList []interface{}
+// @return err error
+func (m *BaseFromThsHfData) BatchInsertDataByColl(coll *qmgo.Collection, bulk int, dataList []interface{}) (err error) {
+	ctx := context.TODO()
+	dataNum := len(dataList)
+	if dataNum <= 0 {
+		return
+	}
+
+	// 不设置每次保存切片数量大小，或者实际数据量小于设置的切片数量大小，那么就直接保存吧
+	if bulk <= 0 || dataNum <= bulk {
+		_, err = coll.InsertMany(ctx, dataList)
+		if err != nil {
+			fmt.Println("BatchInsertData:Err:" + err.Error())
+			return
+		}
+		return
+	}
+
+	// 分批保存
+	i := 0
+	tmpAddDataList := make([]interface{}, 0)
+	for _, v := range dataList {
+		tmpAddDataList = append(tmpAddDataList, v)
+		i++
+		if i >= bulk {
+			_, err = coll.InsertMany(ctx, tmpAddDataList)
+			if err != nil {
+				fmt.Println("BatchInsertData:Err:" + err.Error())
+				return
+			}
+			i = 0
+			tmpAddDataList = make([]interface{}, 0)
+		}
+	}
+
+	if len(tmpAddDataList) > 0 {
+		_, err = coll.InsertMany(ctx, tmpAddDataList)
+		if err != nil {
+			fmt.Println("BatchInsertData:Err:" + err.Error())
+			return
+		}
+	}
+
+	return
+}
+
+// UpdateDataByColl
+// @Description: 单条数据修改
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 15:01:51
+// @param whereParams interface{}
+// @param updateParams interface{}
+// @return err error
+func (m *BaseFromThsHfData) UpdateDataByColl(coll *qmgo.Collection, whereParams, updateParams interface{}) (err error) {
+	ctx := context.TODO()
+	err = coll.UpdateOne(ctx, whereParams, updateParams)
+	if err != nil {
+		fmt.Println("UpdateDataByColl:Err:" + err.Error())
+		return
+	}
+	return
+}
+
+// UpdateData
+// @Description: 单条数据修改
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 15:01:51
+// @param whereParams interface{}
+// @param updateParams interface{}
+// @return err error
+func (m *BaseFromThsHfData) UpdateData(whereParams, updateParams interface{}) (err error) {
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	err = coll.UpdateOne(ctx, whereParams, updateParams)
+	if err != nil {
+		fmt.Println("UpdateData:Err:" + err.Error())
+		return
+	}
+	return
+}
+
+// HandleData
+// @Description: 事务处理数据
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-30 10:40:20
+// @param addDataList []BaseAddFromBusinessData
+// @param updateDataList []BaseFromThsHfData
+// @return result interface{}
+// @return err error
+func (m *BaseFromThsHfData) HandleData(addDataList, updateDataList []BaseFromThsHfData) (result interface{}, err error) {
+
+	ctx := context.TODO()
+
+	callback := func(sessCtx context.Context) (interface{}, error) {
+		// 重要：确保事务中的每一个操作，都使用传入的sessCtx参数
+
+		db := utils.MgoDataCli.Database(m.DataBaseName())
+		coll := db.Collection(m.CollectionName())
+
+		// 插入数据
+		if len(addDataList) > 0 {
+			_, err = coll.InsertMany(sessCtx, addDataList)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		// 修改
+
+		if len(updateDataList) > 0 {
+			for _, v := range updateDataList {
+				err = coll.UpdateOne(ctx, bson.M{"_id": v.ID}, bson.M{"$set": bson.M{"value": v.Value, "modify_time": v.ModifyTime}})
+				if err != nil {
+					fmt.Println("BatchInsertData:Err:" + err.Error())
+					return nil, err
+				}
+			}
+		}
+
+		return nil, nil
+	}
+	result, err = utils.MgoDataCli.DoTransaction(ctx, callback)
+
+	return
+}
+
+// GetEdbInfoMaxAndMinInfo
+// @Description: 获取当前指标的最大最小值
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-30 17:15:39
+// @param whereParams interface{}
+// @return result EdbInfoMaxAndMinInfo
+// @return err error
+func (m *BaseFromThsHfData) GetEdbInfoMaxAndMinInfo(whereParams interface{}) (result EdbInfoMaxAndMinInfo, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+	err = coll.Aggregate(ctx, whereParams).One(&result)
+	if err != nil {
+		return
+	}
+	result.MinDate = result.MinDate.In(time.Local)
+	result.MaxDate = result.MaxDate.In(time.Local)
+	result.LatestDate = result.LatestDate.In(time.Local)
+
+	return
+}
+
+// GetLatestValue
+// @Description: 获取当前指标的最新数据记录
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-30 17:16:15
+// @param whereParams interface{}
+// @param selectParam interface{}
+// @return latestValue LatestValue
+// @return err error
+func (m *BaseFromThsHfData) GetLatestValue(whereParams, selectParam interface{}) (latestValue LatestValue, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+
+	//var result interface{}
+	//err = coll.Find(ctx, whereParams).Select(selectParam).One(&result)
+	err = coll.Find(ctx, whereParams).Select(selectParam).One(&latestValue)
+	return
+}
+
+func (m *BaseFromThsHfData) RemoveMany(whereParams interface{}) (err error) {
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	return m.RemoveManyByColl(coll, whereParams)
+}
+
+func (m *BaseFromThsHfData) RemoveManyByColl(coll *qmgo.Collection, whereParams interface{}) (err error) {
+	ctx := context.TODO()
+	_, err = coll.RemoveAll(ctx, whereParams)
+	if err != nil {
+		fmt.Println("RemoveManyByColl:Err:" + err.Error())
+		return
+	}
+	return
+}

models/mgo/edb_data_ths_hf.go

@@ -0,0 +1,510 @@
+package mgo
+
+import (
+	"context"
+	"errors"
+	"eta/eta_api/utils"
+	"fmt"
+	"github.com/qiniu/qmgo"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"time"
+)
+
+// EdbDataThsHf
+// @Description: 同花顺高频集合（指标库）
+type EdbDataThsHf struct {
+	ID            primitive.ObjectID `json:"_id" bson:"_id,omitempty" `            // 文档id
+	EdbInfoId     int                `json:"edb_info_id" bson:"edb_info_id"`       // 指标ID
+	EdbCode       string             `json:"edb_code" bson:"edb_code"`             // 指标编码
+	DataTime      time.Time          `json:"data_time" bson:"data_time"`           // 数据日期
+	Value         float64            `json:"value" bson:"value"`                   // 数据值
+	CreateTime    time.Time          `json:"create_time" bson:"create_time"`       // 创建时间
+	ModifyTime    time.Time          `json:"modify_time" bson:"modify_time"`       // 修改时间
+	DataTimestamp int64              `json:"data_timestamp" bson:"data_timestamp"` // 数据日期时间戳
+}
+
+// CollectionName
+// @Description:  获取集合名称
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 13:41:36
+// @return string
+func (m *EdbDataThsHf) CollectionName() string {
+	return "edb_data_business"
+}
+
+// DataBaseName
+// @Description: 获取数据库名称
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 13:41:33
+// @return string
+func (m *EdbDataThsHf) DataBaseName() string {
+	return utils.MgoDataDbName
+}
+
+// GetCollection
+// @Description: 获取mongodb集合的句柄
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 13:41:33
+// @return string
+func (m *EdbDataThsHf) GetCollection() *qmgo.Collection {
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	return db.Collection(m.CollectionName())
+}
+
+// GetItem
+// @Description: 根据条件获取单条数据
+// @author: Roc
+// @receiver m
+// @datetime 2024-05-09 10:00:49
+// @param whereParams interface{}
+// @return item *EdbDataThsHf
+// @return err error
+func (m *EdbDataThsHf) GetItem(whereParams interface{}) (item *EdbDataThsHf, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+
+	return m.GetItemByColl(coll, whereParams)
+}
+
+// GetItemByColl
+// @Description: 根据条件获取单条数据
+// @author: Roc
+// @receiver m
+// @datetime 2024-05-09 13:22:06
+// @param coll *qmgo.Collection
+// @param whereParams interface{}
+// @return item *EdbDataThsHf
+// @return err error
+func (m *EdbDataThsHf) GetItemByColl(coll *qmgo.Collection, whereParams interface{}) (item *EdbDataThsHf, err error) {
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+	err = coll.Find(ctx, whereParams).One(&item)
+	if err != nil {
+		return
+	}
+
+	item.DataTime = item.DataTime.In(time.Local)
+	item.CreateTime = item.CreateTime.In(time.Local)
+	item.ModifyTime = item.ModifyTime.In(time.Local)
+
+	return
+}
+
+// GetAllDataList
+// @Description: 根据条件获取所有数据
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 13:42:19
+// @param whereParams interface{}
+// @param sort []string
+// @return result []EdbDataThsHf
+// @return err error
+func (m *EdbDataThsHf) GetAllDataList(whereParams interface{}, sort []string) (result []*EdbDataThsHf, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+	err = coll.Find(ctx, whereParams).Sort(sort...).All(&result)
+	if err != nil {
+		return
+	}
+
+	for _, v := range result {
+		v.DataTime = v.DataTime.In(time.Local)
+		v.CreateTime = v.CreateTime.In(time.Local)
+		v.ModifyTime = v.ModifyTime.In(time.Local)
+	}
+
+	return
+}
+
+// GetLimitDataList
+// @Description: 根据条件获取指定数量数据列表
+// @author: Roc
+// @receiver m
+// @datetime 2024-05-06 17:08:32
+// @param whereParams interface{}
+// @param size int64
+// @param sort []string
+// @return result []*BaseFromBusinessData
+// @return err error
+func (m *EdbDataThsHf) GetLimitDataList(whereParams interface{}, size int64, sort []string) (result []*EdbDataThsHf, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+	err = coll.Find(ctx, whereParams).Sort(sort...).Limit(size).All(&result)
+	if err != nil {
+		return
+	}
+
+	for _, v := range result {
+		v.DataTime = v.DataTime.In(time.Local)
+		v.CreateTime = v.CreateTime.In(time.Local)
+		v.ModifyTime = v.ModifyTime.In(time.Local)
+	}
+
+	return
+}
+
+// GetPageDataList
+// @Description: 根据条件获取分页数据列表
+// @author: Roc
+// @receiver m
+// @datetime 2024-05-07 10:21:07
+// @param whereParams interface{}
+// @param startSize int64
+// @param size int64
+// @param sort []string
+// @return result []*EdbDataThsHf
+// @return err error
+func (m *EdbDataThsHf) GetPageDataList(whereParams interface{}, startSize, size int64, sort []string) (result []*EdbDataThsHf, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+	err = coll.Find(ctx, whereParams).Sort(sort...).Skip(startSize).Limit(size).All(&result)
+	if err != nil {
+		return
+	}
+
+	for _, v := range result {
+		v.DataTime = v.DataTime.In(time.Local)
+		v.CreateTime = v.CreateTime.In(time.Local)
+		v.ModifyTime = v.ModifyTime.In(time.Local)
+	}
+
+	return
+}
+
+// GetCountDataList
+// @Description:  根据条件获取数据列表总数
+// @author: Roc
+// @receiver m
+// @datetime 2024-05-07 10:29:00
+// @param whereParams interface{}
+// @return count int64
+// @return err error
+func (m *EdbDataThsHf) GetCountDataList(whereParams interface{}) (count int64, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+	count, err = coll.Find(ctx, whereParams).Count()
+
+	return
+}
+
+// InsertDataByColl
+// @Description: 写入单条数据（外部传入集合）
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 14:22:18
+// @param addData interface{}
+// @return err error
+func (m *EdbDataThsHf) InsertDataByColl(coll *qmgo.Collection, addData interface{}) (err error) {
+	ctx := context.TODO()
+	_, err = coll.InsertOne(ctx, addData)
+	if err != nil {
+		fmt.Println("InsertDataByColl:Err:" + err.Error())
+		return
+	}
+
+	return
+}
+
+// BatchInsertData
+// @Description: 批量写入数据
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 14:22:18
+// @param bulk int 每次请求保存的数据量
+// @param dataList []interface{}
+// @return err error
+func (m *EdbDataThsHf) BatchInsertData(bulk int, dataList []interface{}) (err error) {
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+
+	return m.BatchInsertDataByColl(coll, bulk, dataList)
+}
+
+// BatchInsertDataByColl
+// @Description: 批量写入数据（外部传入集合）
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 14:22:18
+// @param coll *qmgo.Collection
+// @param bulk int 每次请求保存的数据量
+// @param dataList []interface{}
+// @return err error
+func (m *EdbDataThsHf) BatchInsertDataByColl(coll *qmgo.Collection, bulk int, dataList []interface{}) (err error) {
+	ctx := context.TODO()
+	dataNum := len(dataList)
+	if dataNum <= 0 {
+		return
+	}
+
+	// 不设置每次保存切片数量大小，或者实际数据量小于设置的切片数量大小，那么就直接保存吧
+	if bulk <= 0 || dataNum <= bulk {
+		_, err = coll.InsertMany(ctx, dataList)
+		if err != nil {
+			fmt.Println("BatchInsertData:Err:" + err.Error())
+			return
+		}
+		return
+	}
+
+	// 分批保存
+	i := 0
+	tmpAddDataList := make([]interface{}, 0)
+	for _, v := range dataList {
+		tmpAddDataList = append(tmpAddDataList, v)
+		i++
+		if i >= bulk {
+			_, err = coll.InsertMany(ctx, tmpAddDataList)
+			if err != nil {
+				fmt.Println("BatchInsertData:Err:" + err.Error())
+				return
+			}
+			i = 0
+			tmpAddDataList = make([]interface{}, 0)
+		}
+	}
+
+	if len(tmpAddDataList) > 0 {
+		_, err = coll.InsertMany(ctx, tmpAddDataList)
+		if err != nil {
+			fmt.Println("BatchInsertData:Err:" + err.Error())
+			return
+		}
+	}
+
+	return
+}
+
+// UpdateData
+// @Description: 单条数据修改
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 15:01:51
+// @param whereParams interface{}
+// @param updateParams interface{}
+// @return err error
+func (m *EdbDataThsHf) UpdateData(whereParams, updateParams interface{}) (err error) {
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+
+	return m.UpdateDataByColl(coll, whereParams, updateParams)
+}
+
+// UpdateDataByColl
+// @Description: 单条数据修改（外部传入集合）
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-26 15:01:51
+// @param whereParams interface{}
+// @param updateParams interface{}
+// @return err error
+func (m *EdbDataThsHf) UpdateDataByColl(coll *qmgo.Collection, whereParams, updateParams interface{}) (err error) {
+	ctx := context.TODO()
+	err = coll.UpdateOne(ctx, whereParams, updateParams)
+	if err != nil {
+		fmt.Println("UpdateDataByColl:Err:" + err.Error())
+		return
+	}
+
+	return
+}
+
+// RemoveMany
+// @Description: 根据条件删除多条数据
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-30 13:17:02
+// @param whereParams interface{}
+// @return err error
+func (m *EdbDataThsHf) RemoveMany(whereParams interface{}) (err error) {
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+
+	return m.RemoveManyByColl(coll, whereParams)
+}
+
+// RemoveManyByColl
+// @Description: 根据条件删除多条数据（外部传入集合）
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-30 13:18:42
+// @param coll *qmgo.Collection
+// @param whereParams interface{}
+// @return err error
+func (m *EdbDataThsHf) RemoveManyByColl(coll *qmgo.Collection, whereParams interface{}) (err error) {
+	ctx := context.TODO()
+	_, err = coll.RemoveAll(ctx, whereParams)
+	if err != nil {
+		fmt.Println("RemoveManyByColl:Err:" + err.Error())
+		return
+	}
+
+	return
+}
+
+// HandleData
+// @Description: 事务处理数据
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-30 10:39:01
+// @param addDataList []AddEdbDataThsHf
+// @param updateDataList []EdbDataThsHf
+// @return result interface{}
+// @return err error
+func (m *EdbDataThsHf) HandleData(addDataList, updateDataList []EdbDataThsHf) (result interface{}, err error) {
+
+	ctx := context.TODO()
+
+	callback := func(sessCtx context.Context) (interface{}, error) {
+		// 重要：确保事务中的每一个操作，都使用传入的sessCtx参数
+
+		db := utils.MgoDataCli.Database(m.DataBaseName())
+		coll := db.Collection(m.CollectionName())
+
+		// 插入数据
+		if len(addDataList) > 0 {
+			_, err = coll.InsertMany(sessCtx, addDataList)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		// 修改
+
+		if len(updateDataList) > 0 {
+			for _, v := range updateDataList {
+				err = coll.UpdateOne(ctx, bson.M{"_id": v.ID}, bson.M{"$set": bson.M{"value": v.Value, "modify_time": v.ModifyTime}})
+				if err != nil {
+					fmt.Println("BatchInsertData:Err:" + err.Error())
+					return nil, err
+				}
+			}
+		}
+
+		return nil, nil
+	}
+	result, err = utils.MgoDataCli.DoTransaction(ctx, callback)
+
+	return
+}
+
+// EdbInfoMaxAndMinInfo 指标最新数据记录结构体
+//type EdbInfoMaxAndMinInfo struct {
+//	MinDate     time.Time `description:"最小日期" bson:"min_date"`
+//	MaxDate     time.Time `description:"最大日期" bson:"max_date"`
+//	MinValue    float64   `description:"最小值" bson:"min_value"`
+//	MaxValue    float64   `description:"最大值" bson:"max_value"`
+//	LatestValue float64   `description:"最新值" bson:"latest_value"`
+//	LatestDate  time.Time `description:"实际数据最新日期" bson:"latest_date"`
+//	EndValue    float64   `description:"最新值" bson:"end_value"`
+//}
+
+// GetEdbInfoMaxAndMinInfo
+// @Description: 获取当前指标的最大最小值
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-30 17:15:39
+// @param whereParams interface{}
+// @return result EdbInfoMaxAndMinInfo
+// @return err error
+func (m *EdbDataThsHf) GetEdbInfoMaxAndMinInfo(whereParams interface{}) (result EdbInfoMaxAndMinInfo, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+	err = coll.Aggregate(ctx, whereParams).One(&result)
+	if err != nil {
+		return
+	}
+	result.MinDate = result.MinDate.In(time.Local)
+	result.MaxDate = result.MaxDate.In(time.Local)
+	result.LatestDate = result.LatestDate.In(time.Local)
+
+	return
+}
+
+// LatestValue 指标最新数据记录结构体
+//type LatestValue struct {
+//	Value float64 `description:"值" bson:"value"`
+//}
+
+// GetLatestValue
+// @Description: 获取当前指标的最新数据记录
+// @author: Roc
+// @receiver m
+// @datetime 2024-04-30 17:16:15
+// @param whereParams interface{}
+// @param selectParam interface{}
+// @return latestValue LatestValue
+// @return err error
+func (m *EdbDataThsHf) GetLatestValue(whereParams, selectParam interface{}) (latestValue LatestValue, err error) {
+	if utils.MgoDataCli == nil {
+		err = errors.New("mongodb连接失败")
+		return
+	}
+	db := utils.MgoDataCli.Database(m.DataBaseName())
+	coll := db.Collection(m.CollectionName())
+	ctx := context.TODO()
+	if err != nil {
+		fmt.Println("MgoGetColl Err:", err.Error())
+		return
+	}
+
+	//var result interface{}
+	//err = coll.Find(ctx, whereParams).Select(selectParam).One(&result)
+	err = coll.Find(ctx, whereParams).Select(selectParam).One(&latestValue)
+	return
+}

services/data/base_from_ths_hf.go

@@ -4,9 +4,12 @@ import (
 	"encoding/json"
 	"eta/eta_api/models"
 	"eta/eta_api/models/data_manage"
+	"eta/eta_api/models/mgo"
 	"eta/eta_api/services/alarm_msg"
 	"eta/eta_api/utils"
 	"fmt"
+	"go.mongodb.org/mongo-driver/bson"
+	"time"
 )
 
 // CheckExistThsHfEdb 校验已存在的同花顺高频指标
@@ -196,3 +199,104 @@ func RefreshBaseThsHfIndex(indexIds []int, refreshType int) (isAsync bool, err e
 	}
 	return
 }
+
+// GetThsHfBaseIndexData 获取指标数据
+func GetThsHfBaseIndexData(indexCode, startTime, endTime string) (dataList []*data_manage.BaseFromThsHfDataItem, err error) {
+	if utils.UseMongo {
+		list, e := getThsHfBaseIndexDataByMongo(indexCode, startTime, endTime)
+		if e != nil {
+			err = fmt.Errorf("获取指标数据失败-Mongo, %v", e)
+			return
+		}
+		dataList = list
+		return
+	}
+
+	// MySQL
+	var (
+		cond string
+		pars []interface{}
+	)
+	dataOb := new(data_manage.BaseFromThsHfData)
+	if startTime != "" && endTime != "" {
+		cond += fmt.Sprintf(" AND %s = ? AND (%s BETWEEN ? AND ?)", dataOb.Cols().IndexCode, dataOb.Cols().DataTime)
+		pars = append(pars, indexCode, startTime, endTime)
+	}
+	if startTime != "" && endTime == "" {
+		cond += fmt.Sprintf(" AND %s = ? AND %s > ?)", dataOb.Cols().IndexCode, dataOb.Cols().DataTime)
+		pars = append(pars, indexCode, startTime)
+	}
+	list, e := dataOb.GetItemsByCondition(cond, pars, []string{}, fmt.Sprintf("%s DESC", dataOb.Cols().DataTime))
+	if e != nil {
+		err = fmt.Errorf("获取指标数据失败-MySQL, %v", e)
+		return
+	}
+	for _, v := range list {
+		dataList = append(dataList, v.Format2Item())
+	}
+	return
+}
+
+// getThsHfBaseIndexDataByMongo
+func getThsHfBaseIndexDataByMongo(indexCode, startTime, endTime string) (dataList []*data_manage.BaseFromThsHfDataItem, err error) {
+	dataList = make([]*data_manage.BaseFromThsHfDataItem, 0)
+
+	mogDataObj := mgo.BaseFromThsHfData{}
+	// 构建查询条件
+	queryConditions := bson.M{
+		"index_code": indexCode,
+	}
+
+	// 时间区间
+	if startTime != "" && endTime != "" {
+		st, e := time.ParseInLocation(utils.FormatDateTime, startTime, time.Local)
+		if e != nil {
+			err = fmt.Errorf("start time parse err: %v", e)
+			return
+		}
+		ed, e := time.ParseInLocation(utils.FormatDateTime, endTime, time.Local)
+		if e != nil {
+			err = fmt.Errorf("end time parse err: %v", e)
+			return
+		}
+		queryConditions["data_time"] = bson.M{
+			"$gte": st,
+			"$lte": ed,
+		}
+	}
+	if startTime != "" && endTime == "" {
+		st, e := time.ParseInLocation(utils.FormatDateTime, startTime, time.Local)
+		if e != nil {
+			err = fmt.Errorf("start time parse err: %v", e)
+			return
+		}
+		queryConditions["data_time"] = bson.M{
+			"$gte": st,
+		}
+	}
+
+	// 获取列表数据
+	list, e := mogDataObj.GetAllDataList(queryConditions, []string{"-data_time"})
+	if e != nil {
+		err = fmt.Errorf("GetAllDataList err: %v", e)
+		return
+	}
+	for _, v := range list {
+		dataList = append(dataList, formatMgoBaseThsHfData2Item(v))
+	}
+	return
+}
+
+func formatMgoBaseThsHfData2Item(origin *mgo.BaseFromThsHfData) (item *data_manage.BaseFromThsHfDataItem) {
+	if origin == nil {
+		return
+	}
+	item = new(data_manage.BaseFromThsHfDataItem)
+	item.DataId = int(origin.BaseFromThsHfDataId)
+	item.IndexId = int(origin.BaseFromThsHfIndexId)
+	item.IndexCode = origin.IndexCode
+	item.DataTime = utils.TimeTransferString(utils.FormatDateTime, origin.DataTime)
+	item.Value = origin.Value
+	item.UniqueCode = origin.UniqueCode
+	return
+}

services/data/edb_data.go

@@ -412,6 +412,9 @@ func GetPageData(edbInfoId, source, subSource int, endDataTime string, startSize
 	if source == utils.DATA_SOURCE_BUSINESS && utils.UseMongo {
 		return getPageDataByMongo(edbInfoId, source, subSource, endDataTime, startSize, pageSize)
 	}
+	if source == utils.DATA_SOURCE_THS && subSource == utils.DATA_SUB_SOURCE_HIGH_FREQUENCY && utils.UseMongo {
+		return getThsHfPageDataByMongo(edbInfoId, source, subSource, endDataTime, startSize, pageSize)
+	}
 
 	// 默认走mysql
 	return getPageDataByMysql(edbInfoId, source, subSource, endDataTime, startSize, pageSize)
@@ -515,3 +518,47 @@ func getPageDataByMongo(edbInfoId, source, subSource int, endDataTime string, st
 
 	return
 }
+
+func getThsHfPageDataByMongo(edbInfoId, source, subSource int, endDataTime string, startSize, pageSize int) (dataCount int, dataList []*data_manage.EdbData, err error) {
+	dataList = make([]*data_manage.EdbData, 0)
+
+	mogDataObj := mgo.EdbDataThsHf{}
+	// 构建查询条件
+	queryConditions := bson.M{
+		"edb_info_id": edbInfoId,
+	}
+
+	// 结束日期
+	dateCondition, err := mgo.BuildDateCondition("", endDataTime)
+	if err != nil {
+		return
+	}
+	if len(dateCondition) > 0 {
+		queryConditions["data_time"] = dateCondition
+	}
+
+	// 获取数据总量
+	tmpCount, tmpErr := mogDataObj.GetCountDataList(queryConditions)
+	if tmpErr != nil && tmpErr.Error() != utils.ErrNoRow() {
+		err = tmpErr
+		return
+	}
+	dataCount = int(tmpCount)
+
+	// 获取列表数据
+	tmpDataList, tmpErr := mogDataObj.GetPageDataList(queryConditions, int64(startSize), int64(pageSize), []string{"-data_time"})
+	if tmpErr != nil {
+		err = tmpErr
+		return
+	}
+	for k, v := range tmpDataList {
+		dataList = append(dataList, &data_manage.EdbData{
+			EdbDataId: k + 1,
+			EdbInfoId: v.EdbInfoId,
+			DataTime:  v.DataTime.Format(utils.FormatDate),
+			Value:     v.Value,
+		})
+	}
+
+	return
+}

services/eta_forum/eta_forum_hub.go

@@ -269,6 +269,8 @@ func GetEdbListByEdbInfoId(edbInfoIds []int) (edbInfoList []*data_manage.EdbInfo
 		var dataList []*data_manage.EdbDataBase
 		if v.Source == utils.DATA_SOURCE_BUSINESS && utils.UseMongo {
 			dataList, err = data_manage.GetEdbDataBaseMongoByEdbInfoId(v.EdbInfoId, v.Source, v.SubSource)
+		} else if v.Source == utils.DATA_SOURCE_THS && v.SubSource == utils.DATA_SUB_SOURCE_HIGH_FREQUENCY && utils.UseMongo {
+			dataList, err = data_manage.GetThsHfEdbDataBaseMongoByEdbInfoId(v.EdbInfoId, v.Source, v.SubSource)
 		} else {
 			dataList, err = data_manage.GetEdbDataBaseByEdbInfoId(v.EdbInfoId, v.Source, v.SubSource)
 		}