Merge branch 'fix/lost_trade_top' of hongze/hongze_task_trial into master

models/data_manage/trade_position_analysis.go

@@ -98,6 +98,13 @@ func GetTradePositionTopByExchangeDataTime(exchange string, startDate, endDate s
 	return
 }
 
+func GetTradePositionTopByExchangeDataTimeByClassify(exchange string, startDate, endDate string, classifyNames, classifyTypes []string) (list []*TradePositionTop, err error) {
+	o := orm.NewOrm()
+	sql := `SELECT * FROM trade_position_` + exchange + `_top where data_time >= ? and data_time <= ? and deal_type in (1,2) and classify_name in (` + utils.GetOrmInReplace(len(classifyNames)) + `)  and classify_type in (` + utils.GetOrmInReplace(len(classifyTypes)) + `) ORDER BY classify_name, classify_type, deal_type, data_time, deal_value desc`
+	_, err = o.Raw(sql, startDate, endDate, classifyNames, classifyTypes).QueryRows(&list)
+	return
+}
+
 func GetTradePositionTopCountByExchangeDataTime(exchange string, startDate, endDate string) (count int64, err error) {
 	o := orm.NewOrm()
 	sql := "SELECT count(*) FROM trade_position_" + exchange + "_top where data_time >= ? and data_time <= ? and deal_type in (1,2) ORDER BY classify_name, classify_type, deal_type, data_time, deal_value desc"
@@ -112,6 +119,13 @@ func GetTradePositionTopByExchangeSourceType(exchange string, dataTime string, s
 	return
 }
 
+func GetTradePositionTopByExchangeSourceTypeClassify(exchange string, dataTime string, sourceType int, classifyNames, classifyTypes []string) (list []*TradePositionTop, err error) {
+	o := orm.NewOrm()
+	sql := `SELECT * FROM trade_position_` + exchange + `_top where data_time= ? and source_type = ? and classify_name in (` + utils.GetOrmInReplace(len(classifyNames)) + `)  and classify_type in (` + utils.GetOrmInReplace(len(classifyTypes)) + `) ORDER BY classify_name, classify_type, deal_type, deal_value desc`
+	_, err = o.Raw(sql, dataTime, sourceType, classifyNames, classifyTypes).QueryRows(&list)
+	return
+}
+
 type TradeTopClassify struct {
 	ClassifyName string //分类名称
 	ClassifyType string //分类名称下的类型
@@ -179,6 +193,13 @@ func DeletePositionTopByDataTime(exchange string, dataTime string, dealType int)
 	return
 }
 
+func DeletePositionTopByDataTimeClassify(exchange string, dataTime string, dealType int, classifyNames, classifyTypes []string) (err error) {
+	o := orm.NewOrm()
+	sql := `delete from trade_position_` + exchange + `_top WHERE data_time=? and deal_type=? and classify_name in (` + utils.GetOrmInReplace(len(classifyNames)) + `)  and classify_type in (` + utils.GetOrmInReplace(len(classifyTypes)) + `)`
+	_, err = o.Raw(sql, dataTime, dealType, classifyNames, classifyTypes).Exec()
+	return
+}
+
 func GetTradePositionTopByExchangeDataTimeType(exchange string, dataTime string, dealType int) (list []TradePositionTop, err error) {
 	o := orm.NewOrm()
 	sql := "select * from trade_position_" + exchange + "_top WHERE data_time=? and deal_type=?"
@@ -186,6 +207,13 @@ func GetTradePositionTopByExchangeDataTimeType(exchange string, dataTime string,
 	return
 }
 
+func GetTradePositionTopByExchangeDataTimeTypeClassify(exchange string, dataTime string, dealType int, classifyNames, classifyTypes []string) (list []TradePositionTop, err error) {
+	o := orm.NewOrm()
+	sql := `select * from trade_position_` + exchange + `_top WHERE data_time=? and deal_type=? and classify_name in (` + utils.GetOrmInReplace(len(classifyNames)) + `)  and classify_type in (` + utils.GetOrmInReplace(len(classifyTypes)) + `)`
+	_, err = o.Raw(sql, dataTime, dealType, classifyNames, classifyTypes).QueryRows(&list)
+	return
+}
+
 func MultiInsertTradeBaseDataToTop(exchange string, startDate, endDate string) (err error) {
 	o := orm.NewOrm()
 	now := time.Now().Format(utils.FormatDateTime)
@@ -201,6 +229,21 @@ SELECT classify_name,classify_type,sold_short_name,sold_value,sold_change,data_t
 	return
 }
 
+func MultiInsertTradeBaseDataToTopByClassify(exchange string, startDate, endDate string, classifyNames, classifyTypes []string) (err error) {
+	o := orm.NewOrm()
+	now := time.Now().Format(utils.FormatDateTime)
+	sql1 := `INSERT INTO trade_position_` + exchange + `_top(classify_name,classify_type,deal_short_name,deal_value,deal_change,data_time,deal_type,source_type,rank,create_time,modify_time)
+	SELECT classify_name,classify_type,buy_short_name,buy_value,buy_change,data_time,1,0,rank,?,? FROM base_from_trade_` + exchange + `_index where rank <50 and buy_short_name !="" and data_time between ? and ? and classify_name in (` + utils.GetOrmInReplace(len(classifyNames)) + `)  and classify_type in (` + utils.GetOrmInReplace(len(classifyTypes)) + `)`
+	_, err = o.Raw(sql1, now, now, startDate, endDate, classifyNames, classifyTypes).Exec()
+	if err != nil {
+		return
+	}
+	sql2 := `INSERT INTO trade_position_` + exchange + `_top(classify_name,classify_type,deal_short_name,deal_value,deal_change,data_time,deal_type,source_type,rank,create_time,modify_time)
+SELECT classify_name,classify_type,sold_short_name,sold_value,sold_change,data_time,2,0,rank,?,? FROM base_from_trade_` + exchange + `_index where rank <50 and sold_short_name !="" and data_time between ? and ? and classify_name in (` + utils.GetOrmInReplace(len(classifyNames)) + `)  and classify_type in (` + utils.GetOrmInReplace(len(classifyTypes)) + `)`
+	_, err = o.Raw(sql2, now, now, startDate, endDate, classifyNames, classifyTypes).Exec()
+	return
+}
+
 // GetTradePositionTopOriginDataTimes 获取榜单原始数据日期-正序
 func GetTradePositionTopOriginDataTimes(exchange string) (dates []string, err error) {
 	o := orm.NewOrm()
@@ -252,7 +295,7 @@ type TradeClassifyName struct {
 	ClassifyName string    //分类名称
 	ClassifyType string    //分类名称下的类型
 	DataTime     time.Time //数据最近的日期
-	ModifyTime   time.Time //数据最近修改的日期
+	ModifyTime   time.Time //数据最近的日期
 }
 
 // GetExchangeClassify 获取交易所分类列表
@@ -279,6 +322,14 @@ func GetTradePositionTopCleanByExchangeDataTime(exchange string, startDate, endD
 	return
 }
 
+// GetTradePositionTopCleanByExchangeDataTimeClassify 根据时间查询净多单和净空单的值
+func GetTradePositionTopCleanByExchangeDataTimeClassify(exchange string, startDate, endDate string, classifyNames, classifyTypes []string) (list []*TradePositionTop, err error) {
+	o := orm.NewOrm()
+	sql := `SELECT * FROM trade_position_` + exchange + `_top where data_time >= ? and data_time <= ? and deal_type in (3,4) and classify_name in (` + utils.GetOrmInReplace(len(classifyNames)) + `)  and classify_type in (` + utils.GetOrmInReplace(len(classifyTypes)) + `) ORDER BY classify_name, classify_type, deal_type, data_time, deal_value desc`
+	_, err = o.Raw(sql, startDate, endDate, classifyNames, classifyTypes).QueryRows(&list)
+	return
+}
+
 // MultiUpdatePositionTopChangeVal 批量更新榜单里变化量的值
 func MultiUpdatePositionTopChangeVal(exchange string, updates []UpdateChangeVal) (err error) {
 	o := orm.NewOrm()
@@ -297,3 +348,44 @@ func MultiUpdatePositionTopChangeVal(exchange string, updates []UpdateChangeVal)
 	}
 	return
 }
+
+func GetTradePositionOriginClassifyCountByExchangeDataTime(exchange string, startDate, endDate string) (count int64, err error) {
+	o := orm.NewOrm()
+	sql := `SELECT COUNT(DISTINCT classify_name, classify_type) FROM base_from_trade_` + exchange + `_index where rank <50 and (buy_short_name !="" || sold_short_name !="" ) and data_time >= ? and data_time <= ?`
+	err = o.Raw(sql, startDate, endDate).QueryRow(&count)
+	return
+}
+
+func GetTradePositionTopClassifyCountByExchangeDataTime(exchange string, startDate, endDate string) (count int64, err error) {
+	o := orm.NewOrm()
+	sql := "SELECT COUNT(DISTINCT classify_name, classify_type)  FROM trade_position_" + exchange + "_top where data_time >= ? and data_time <= ? and deal_type in (1,2) "
+	err = o.Raw(sql, startDate, endDate).QueryRow(&count)
+	return
+}
+
+type TradePositionClassifyInfo struct {
+	ClassifyName string //分类名称
+	ClassifyType string //分类名称下的类型
+}
+
+func GetTradePositionOriginClassifyByExchangeDataTime(exchange string, startDate, endDate string) (list []TradePositionClassifyInfo, err error) {
+	o := orm.NewOrm()
+	sql := `SELECT DISTINCT classify_name, classify_type FROM base_from_trade_` + exchange + `_index where rank <50 and (buy_short_name !="" || sold_short_name !="" ) and data_time >= ? and data_time <= ?`
+	_, err = o.Raw(sql, startDate, endDate).QueryRows(&list)
+	return
+}
+
+func GetTradePositionTopClassifyByExchangeDataTime(exchange string, startDate, endDate string) (list []TradePositionClassifyInfo, err error) {
+	o := orm.NewOrm()
+	sql := "SELECT DISTINCT classify_name, classify_type  FROM trade_position_" + exchange + "_top where data_time >= ? and data_time <= ? and deal_type in (1,2) "
+	_, err = o.Raw(sql, startDate, endDate).QueryRows(&list)
+	return
+}
+
+// DeleteTradePositionTopAllByExchangeDataTime 删除计算数据
+func DeleteTradePositionTopAllByExchangeDataTime(exchange string, startDate, endDate string) (err error) {
+	o := orm.NewOrm()
+	sql := "DELETE FROM trade_position_" + exchange + "_top where data_time >= ? and data_time <= ? "
+	_, err = o.Raw(sql, startDate, endDate).Exec()
+	return
+}

services/data/trade_position_analysis.go

@@ -68,6 +68,7 @@ func InitPositionTask(cont context.Context) (err error) {
 						ClassifyName: classify.ClassifyName,
 						ClassifyType: classify.ClassifyType,
 						Exchange:     v,
+						LatestDate:   classify.DataTime,
 						CreateTime:   time.Now(),
 						ModifyTime:   classify.ModifyTime,
 					})
@@ -115,7 +116,31 @@ func InitTradePosition(exchange, startDate, endDate string) (err error, errMsg s
 	}
 	if num > 0 {
 		//err = fmt.Errorf("数据已存在，无需处理")
-		return
+		//数据存在不同步的情况，有些合约会提早更新，有些合约会延迟更新
+		//判断合约数是否一致
+
+		originNum, tmpErr := data_manage.GetTradePositionOriginClassifyCountByExchangeDataTime(exchange, startDate, endDate)
+		if tmpErr != nil {
+			err = tmpErr
+			errMsg = "查询原始数据分类个数失败，GetTradePositionOriginClassifyCountByExchangeDataTime() Err: "
+			return
+		}
+		topNum, tmpErr := data_manage.GetTradePositionTopClassifyCountByExchangeDataTime(exchange, startDate, endDate)
+		if tmpErr != nil {
+			err = tmpErr
+			errMsg = "查询榜单数据分类个数失败，GetTradePositionTopClassifyCountByExchangeDataTime() Err: "
+			return
+		}
+		if originNum == topNum {
+			//err = fmt.Errorf("数据已存在，无需处理")
+			return
+		}
+		//如果合约数不一致，则删除今日数据
+		err = data_manage.DeleteTradePositionTopAllByExchangeDataTime(exchange, startDate, endDate)
+		if err != nil {
+			errMsg = "删除榜单数据失败，DeleteTradePositionTopAllByExchangeDataTime() Err: "
+			return
+		}
 	}
 	err = data_manage.MultiInsertTradeBaseDataToTop(exchange, startDate, endDate)
 	if err != nil {
@@ -238,7 +263,7 @@ func InitTradePosition(exchange, startDate, endDate string) (err error, errMsg s
 		errMsg = "创建净多单，净空单数据失败，createAnalysisCleanTop() Err: "
 		return
 	}
-	// 特殊处理起始日期前一天的数据
+
 	err = DealYesterdayData(exchange, startDate)
 	if err != nil {
 		errMsg = "处理昨日数据失败，DealYesterdayData() Err: "
@@ -619,24 +644,6 @@ func createAnalysisCleanTop(exchange, startDate, endDate string) (err error) {
 	return
 }
 
-func getYesterdayDate(today string) (yesterday string, err error) {
-	i := 1
-	tmpTime, err := time.ParseInLocation(utils.FormatDate, today, time.Local)
-	if err != nil {
-		return
-	}
-	tmpTimeDate := tmpTime.AddDate(0, 0, -i)
-	weekStr := tmpTimeDate.Weekday().String()
-	if weekStr == "Sunday" {
-		i += 2
-	} else if weekStr == "Saturday" {
-		i += 1
-	}
-	tmpTimeDate = tmpTime.AddDate(0, 0, -i)
-	yesterday = tmpTimeDate.Format(utils.FormatDate)
-	return
-}
-
 // getPrevTradeDataDate 获取指定日期上一个交易日日期
 func getPrevTradeDataDate(date string, dates []string) string {
 	pre := -1

services/data/trade_position_analysis_classify.go

@@ -0,0 +1,592 @@
+package data
+
+import (
+	"fmt"
+	"hongze/hongze_task_trial/models/data_manage"
+	"hongze/hongze_task_trial/services/alarm_msg"
+	"hongze/hongze_task_trial/utils"
+	"sort"
+	"strconv"
+	"time"
+)
+
+// FixPositionTask 补全缺失的合约, 注意和原先的定时任务区分开来
+func FixPositionTask() (err error) {
+	exchanges := []string{"zhengzhou", "dalian", "shanghai", "cffex", "ine"} //郑商所，大商所，上期所，中金所，上期能源
+	for i := 194; i > 1; i-- {
+		// 定时任务避开昨日和今日的数据，以免和原先的定时任务InitPositionTask冲突
+		startDate := time.Now().AddDate(0, 0, -i).Format(utils.FormatDate)
+		endDate := startDate
+		for _, v := range exchanges {
+			exchange := v
+			err = nil
+			fmt.Println("FixPositionTask:	启动：" + exchange)
+			utils.FileLog.Info("FixPositionTask:	启动：" + exchange)
+
+			fmt.Println("开始" + startDate + "结束" + endDate)
+			utils.FileLog.Info(fmt.Sprintf("FixPositionTask:开始:%s; 结束:%s", startDate, endDate))
+			tErr, errMsg := InitTradePositionClassify(exchange, startDate, endDate)
+			if tErr != nil {
+				err = tErr
+				fmt.Println("FixPositionTask: 操作失败：" + errMsg + tErr.Error())
+				utils.FileLog.Info(fmt.Sprintf("InitTradePosition: 操作失败：%s:%s", errMsg, tErr.Error()))
+				continue
+			}
+
+			fmt.Println("FixPositionTask:" + exchange + "已完成")
+			utils.FileLog.Info("FixPositionTask:" + exchange + "已完成")
+		}
+	}
+	return
+}
+
+// InitTradePositionClassify 持仓分析补全缺失的合约
+func InitTradePositionClassify(exchange, startDate, endDate string) (err error, errMsg string) {
+	defer func() {
+		if err != nil {
+			tips := fmt.Sprintf("持仓分析-补全缺失的品种数据 操作失败, Exchange: %s, Err: %s, Msg: %s", exchange, err.Error(), errMsg)
+			alarm_msg.SendAlarmMsg(tips, 3)
+		}
+	}()
+
+	//判断合约数与数据源是否一致
+	originClassifyList, tmpErr := data_manage.GetTradePositionOriginClassifyByExchangeDataTime(exchange, startDate, endDate)
+	if tmpErr != nil {
+		err = tmpErr
+		errMsg = "查询原始数据分类个数失败，GetTradePositionOriginClassifyCountByExchangeDataTime() Err: "
+		return
+	}
+	topClassifyList, tmpErr := data_manage.GetTradePositionTopClassifyByExchangeDataTime(exchange, startDate, endDate)
+	if tmpErr != nil {
+		err = tmpErr
+		errMsg = "查询榜单数据分类个数失败，GetTradePositionTopClassifyCountByExchangeDataTime() Err: "
+		return
+	}
+
+	if len(originClassifyList) == len(topClassifyList) {
+		//合约数目一致，无需补全
+		return
+	}
+	if len(originClassifyList) == 0 {
+		return
+	}
+	topClassifyMap := make(map[string][]data_manage.TradePositionClassifyInfo)
+	for _, v := range topClassifyList {
+		str := fmt.Sprintf("%s_%s", v.ClassifyName, v.ClassifyType)
+		topClassifyMap[str] = append(topClassifyMap[str], v)
+	}
+	classifyMap := make(map[string]struct{})
+	classifyNames := make([]string, 0)
+	classifyTypes := make([]string, 0)
+	for _, v := range originClassifyList {
+		str := fmt.Sprintf("%s_%s", v.ClassifyName, v.ClassifyType)
+		if _, ok := topClassifyMap[str]; !ok {
+			classifyTypes = append(classifyTypes, v.ClassifyType)
+			if _, ok1 := classifyMap[v.ClassifyName]; !ok1 {
+				classifyNames = append(classifyNames, v.ClassifyName)
+				classifyMap[v.ClassifyName] = struct{}{}
+			}
+		}
+	}
+
+	if len(classifyTypes) == 0 {
+		return
+	}
+
+	//批量导入缺失的合约
+	err = data_manage.MultiInsertTradeBaseDataToTopByClassify(exchange, startDate, endDate, classifyNames, classifyTypes)
+	if err != nil {
+		errMsg = "新增原始数据失败，MultiInsertTradeBaseDataToTop() Err: "
+		return
+	}
+	originList, err := data_manage.GetTradePositionTopByExchangeDataTimeByClassify(exchange, startDate, endDate, classifyNames, classifyTypes)
+	if err != nil {
+		errMsg = "查询原始数据失败， GetTradePositionTopByExchangeDataTime() Err: "
+		return
+	}
+	if len(originList) <= 0 {
+		// 忽略周末
+		w := time.Now().Weekday().String()
+		if w == "Saturday" || w == "Sunday" {
+			return
+		}
+		// 每天最后一个小时执行依旧无数据时, 才进行邮件提示
+		if time.Now().Hour() != 23 {
+			return
+		}
+		err = fmt.Errorf("原始数据没有值")
+		return
+	}
+
+	// 原始数据日期
+	dates, e := data_manage.GetTradePositionTopOriginDataTimes(exchange)
+	if e != nil {
+		err = fmt.Errorf("GetTradePositionTopOriginDataTimes err: %s", e.Error())
+		return
+	}
+
+	now := time.Now()
+	dataTimeMap := make(map[string]*data_manage.TradePositionTop)
+	onlyEmptyMap := make(map[string]bool)
+	onlyEmptyNameMap := make(map[string]*data_manage.TradePositionTop)
+	topLastMap := make(map[string]int)
+	topLastRankMap := make(map[string]int)
+	list := make([]*data_manage.TradePositionTop, 0)
+	for _, v := range originList {
+		tmp0, tmpErr := dealTradeOriginData(dataTimeMap, onlyEmptyMap, onlyEmptyNameMap, v, topLastMap, topLastRankMap, startDate, now, dates)
+		if tmpErr != nil {
+			err = tmpErr
+			errMsg = "处理原始数据失败 dealTradeOriginData() Err: "
+			return
+		}
+		if tmp0 != nil {
+			list = append(list, tmp0)
+		}
+		if len(list) >= 1000 {
+			err = data_manage.InsertMultiTradePositionTop(exchange, list)
+			if err != nil {
+				errMsg = "批量新增昨日数据失败，InsertMultiTradePositionTop() Err: "
+				return
+			}
+			list = make([]*data_manage.TradePositionTop, 0)
+		}
+	}
+	if len(list) > 0 {
+		err = data_manage.InsertMultiTradePositionTop(exchange, list)
+		if err != nil {
+			errMsg = "批量新增昨日数据失败，InsertMultiTradePositionTop() Err: "
+			return
+		}
+		list = make([]*data_manage.TradePositionTop, 0)
+	}
+	// 处理某个期货公司只有买单没有卖单，或者只有卖单没有买单的情况
+	for k, v := range onlyEmptyNameMap {
+		_, ok1 := onlyEmptyMap[k+"_1"]
+		_, ok2 := onlyEmptyMap[k+"_2"]
+		var dealType int
+		if ok1 && !ok2 {
+			dealType = 2 //只有买单没有卖单
+		} else if !ok1 && ok2 {
+			dealType = 1 //只有卖单没有买单的情况
+		} else {
+			continue
+		}
+		if dealType > 0 {
+			str := v.ClassifyName + "_" + v.ClassifyType + "_" + v.DataTime + "_" + strconv.Itoa(dealType)
+			dealValue := 0
+			if lastVal, ok := topLastMap[str]; ok {
+				dealValue = int(float64(lastVal)*0.7 + 0.5)
+			}
+
+			tmp := &data_manage.TradePositionTop{
+				ClassifyName:  v.ClassifyName,
+				ClassifyType:  v.ClassifyType,
+				DealShortName: v.DealShortName,
+				DataTime:      v.DataTime,
+				DealValue:     dealValue,
+				CreateTime:    now,
+				ModifyTime:    now,
+				DealType:      dealType,
+				SourceType:    2,
+			}
+			list = append(list, tmp)
+			if len(list) >= 1000 {
+				err = data_manage.InsertMultiTradePositionTop(exchange, list)
+				if err != nil {
+					errMsg = "批量新增前日数据失败，InsertMultiTradePositionTop() Err: "
+					return
+				}
+				list = make([]*data_manage.TradePositionTop, 0)
+			}
+		}
+	}
+
+	if len(list) > 0 {
+		err = data_manage.InsertMultiTradePositionTop(exchange, list)
+		if err != nil {
+			errMsg = "批量新增前日数据失败，InsertMultiTradePositionTop() Err: "
+			return
+		}
+	}
+
+	//生成净多单，净空单榜单
+	err = createAnalysisCleanTopClassify(exchange, startDate, endDate, classifyNames, classifyTypes)
+	if err != nil {
+		errMsg = "创建净多单，净空单数据失败，createAnalysisCleanTop() Err: "
+		return
+	}
+
+	err = DealYesterdayDataClassify(exchange, startDate, classifyNames, classifyTypes)
+	if err != nil {
+		errMsg = "处理昨日数据失败，DealYesterdayData() Err: "
+		return
+	}
+	return
+}
+
+// DealYesterdayDataClassify 更新部分合约的昨日数据
+func DealYesterdayDataClassify(exchange, startDate string, classifyNames, classifyTypes []string) (err error) {
+	// 查询最早的日期
+	firstItem, err := data_manage.GetFirstBaseFromTradeIndexByDate(exchange)
+	if err != nil {
+		return
+	}
+	if startDate == firstItem.DataTime { //如果当前是起始日，则无需统计修改前一天的数据
+		return
+	}
+
+	// 前一个交易日, 前两个交易日
+	dates, e := data_manage.GetTradePositionTopOriginDataTimes(exchange)
+	if e != nil {
+		err = fmt.Errorf("GetTradePositionTopOriginDataTimes err: %s", e.Error())
+		return
+	}
+	yesterdayStr := getPrevTradeDataDate(startDate, dates)
+	beforeYesterdayStr := getPrevTradeDataDate(yesterdayStr, dates)
+
+	// 先查出T日最原始的数据
+	originList, err := data_manage.GetTradePositionTopByExchangeDataTimeByClassify(exchange, startDate, startDate, classifyNames, classifyTypes)
+	if err != nil {
+		return
+	}
+	originBuyMap := make(map[string]*data_manage.TradePositionTop)
+	originSoldMap := make(map[string]*data_manage.TradePositionTop)
+	for _, v := range originList {
+		if v.SourceType != 0 {
+			continue
+		}
+		str := v.ClassifyName + "_" + v.ClassifyType + "_" + v.DealShortName
+		if v.DealType == 1 {
+			originBuyMap[str] = v
+		} else if v.DealType == 2 {
+			originSoldMap[str] = v
+		}
+	}
+
+	// 然后查询T-1中数据来源类型是2的数据
+	changeList, err := data_manage.GetTradePositionTopByExchangeSourceTypeClassify(exchange, yesterdayStr, 2, classifyNames, classifyTypes)
+	if err != nil {
+		return
+	}
+	if len(changeList) <= 0 {
+		//err = fmt.Errorf("前天的数据无需修改")
+		return
+	}
+	// 查询出前日的成交量
+	beforeYesterdayList, err := data_manage.GetTradePositionTopByExchangeDataTimeByClassify(exchange, beforeYesterdayStr, beforeYesterdayStr, classifyNames, classifyTypes)
+	if err != nil {
+		return
+	}
+	beforeYesterdayMap1 := make(map[string]int)
+	beforeYesterdayMap2 := make(map[string]int)
+	if len(beforeYesterdayList) > 0 {
+		for _, v := range beforeYesterdayList {
+			if v.SourceType == 2 {
+				continue
+			}
+			str := v.ClassifyName + "_" + v.ClassifyType + "_" + v.DealShortName
+			if v.DealType == 1 {
+				beforeYesterdayMap1[str] = v.DealValue
+			} else if v.DealType == 2 {
+				beforeYesterdayMap2[str] = v.DealValue
+			}
+		}
+	}
+	// 根据原始数据中的值推算出最新的值
+	now := time.Now()
+	// 批量更新到分析表中，
+	var updateAnalysisData []data_manage.UpdateDealValueChange
+	for _, v := range changeList {
+		str := v.ClassifyName + "_" + v.ClassifyType + "_" + v.DealShortName
+		dealValue := 0
+		dealChange := 0
+		if v.DealType == 1 {
+			if n, ok := originBuyMap[str]; ok {
+				dealValue = n.DealValue - n.DealChange
+				if beforeVal, ok1 := beforeYesterdayMap1[str]; ok1 {
+					dealChange = dealValue - beforeVal
+				}
+				tmp := data_manage.UpdateDealValueChange{
+					Id:         v.Id,
+					DealValue:  dealValue,
+					DealChange: dealChange,
+					SourceType: 1,
+					ModifyTime: now,
+				}
+				updateAnalysisData = append(updateAnalysisData, tmp)
+			}
+		} else if v.DealType == 2 {
+			if n, ok := originSoldMap[str]; ok {
+				dealValue = n.DealValue - n.DealChange
+				if beforeVal, ok1 := beforeYesterdayMap2[str]; ok1 {
+					dealChange = dealValue - beforeVal
+				}
+				tmp := data_manage.UpdateDealValueChange{
+					Id:         v.Id,
+					DealValue:  dealValue,
+					DealChange: dealChange,
+					SourceType: 1,
+					ModifyTime: now,
+				}
+				updateAnalysisData = append(updateAnalysisData, tmp)
+			}
+		}
+	}
+	if len(updateAnalysisData) > 0 {
+		err = data_manage.MultiUpdatePositionTop(exchange, updateAnalysisData)
+		if err != nil {
+			return
+		}
+		//删除T-1日净多单和净空单的榜单
+		err = data_manage.DeletePositionTopByDataTimeClassify(exchange, yesterdayStr, 3, classifyNames, classifyTypes)
+		if err != nil {
+			return
+		}
+
+		err = data_manage.DeletePositionTopByDataTimeClassify(exchange, yesterdayStr, 4, classifyNames, classifyTypes)
+		if err != nil {
+			return
+		}
+
+		//重新生成净多单和净空单的榜单
+		err = createAnalysisCleanTopClassify(exchange, yesterdayStr, yesterdayStr, classifyNames, classifyTypes)
+		if err != nil {
+			return
+		}
+
+		//T-1日重新生成净多单和净空单的榜单后，需要更新T日净多单和净空单榜单里的变化量
+		err = updateAnalysisCleanTopChangeValClassify(exchange, startDate, yesterdayStr, classifyNames, classifyTypes)
+		if err != nil {
+			return
+		}
+	}
+
+	return
+}
+
+// createAnalysisCleanTopClassify 生成部分合约的净多单，净空单榜单
+func createAnalysisCleanTopClassify(exchange, startDate, endDate string, classifyNames, classifyTypes []string) (err error) {
+	defer func() {
+		if err != nil {
+			fmt.Println("createAnalysisCleanTop err: " + err.Error())
+		}
+	}()
+
+	topList := make([]*data_manage.TradePositionTop, 0)
+	now := time.Now()
+	var subDataList data_manage.TradePositionSubList
+
+	subChangeMap1 := make(map[string]int) //净多单map
+	subChangeMap2 := make(map[string]int) //净空单map
+
+	// 2023-05-10 此处取前一个交易日, 不一定是昨日
+	dates, e := data_manage.GetTradePositionTopOriginDataTimes(exchange)
+	if e != nil {
+		err = fmt.Errorf("GetTradePositionTopOriginDataTimes err: %s", e.Error())
+		return
+	}
+	yesterday := getPrevTradeDataDate(startDate, dates)
+
+	// 上一个交易日的净多单
+	yesterdayTopList1, tErr := data_manage.GetTradePositionTopByExchangeDataTimeTypeClassify(exchange, yesterday, 3, classifyNames, classifyTypes)
+	if tErr != nil {
+		err = tErr
+		return
+	}
+	if len(yesterdayTopList1) > 0 {
+		for _, v := range yesterdayTopList1 {
+			nameStr := v.ClassifyName + "_" + v.ClassifyType + "_" + v.DataTime + "_" + v.DealShortName
+			subChangeMap1[nameStr] = v.DealValue
+		}
+	}
+
+	// 上一个交易日的净空单
+	yesterdayTopList2, tErr := data_manage.GetTradePositionTopByExchangeDataTimeTypeClassify(exchange, yesterday, 4, classifyNames, classifyTypes)
+	if tErr != nil {
+		err = tErr
+		return
+	}
+	if len(yesterdayTopList2) > 0 {
+		for _, v := range yesterdayTopList2 {
+			nameStr := v.ClassifyName + "_" + v.ClassifyType + "_" + v.DataTime + "_" + v.DealShortName
+			subChangeMap2[nameStr] = v.DealValue
+		}
+	}
+
+	// 根据当日多单/空单数据, 生成净多单/净空单数据
+	originDataList, err := data_manage.GetTradePositionTopByExchangeDataTimeByClassify(exchange, startDate, endDate, classifyNames, classifyTypes)
+	if err != nil {
+		return
+	}
+	buyDataMap := make(map[string]int)
+	for _, v := range originDataList {
+		str := v.ClassifyName + "_" + v.ClassifyType + "_" + v.DataTime + "_" + v.DealShortName
+		if v.DealType == 1 {
+			buyDataMap[str] = v.DealValue
+		} else if v.DealType == 2 {
+			subValue := 0
+			dealType := 0
+			if buy, ok := buyDataMap[str]; ok {
+				subValue = buy - v.DealValue
+				if subValue >= 0 {
+					dealType = 3
+				} else {
+					subValue = -subValue
+					dealType = 4
+				}
+			}
+			tmp := &data_manage.TradePositionSub{
+				ClassifyName:  v.ClassifyName,
+				ClassifyType:  v.ClassifyType,
+				DataTime:      v.DataTime,
+				DealShortName: v.DealShortName,
+				SubValue:      subValue,
+				DealType:      dealType,
+			}
+			subDataList = append(subDataList, tmp)
+		}
+	}
+	if len(subDataList) > 0 {
+		sort.Sort(subDataList)
+	}
+
+	// 根据净多单/净空单数据, 比对上一个交易日的日期计算成交变化量, 并写入
+	var dealType int
+	rankMap := make(map[string]int)
+	for _, v := range subDataList {
+		subValue := v.SubValue
+		nameStr := v.ClassifyName + "_" + v.ClassifyType + "_" + v.DataTime + "_" + v.DealShortName
+		if v.DealType == 3 {
+			subChangeMap1[nameStr] = subValue
+			dealType = 3
+			if _, ok := rankMap[v.ClassifyName+"_"+v.ClassifyType+"_"+v.DataTime+"_3"]; !ok {
+				rankMap[v.ClassifyName+"_"+v.ClassifyType+"_"+v.DataTime+"_3"] = 1
+			} else {
+				rankMap[v.ClassifyName+"_"+v.ClassifyType+"_"+v.DataTime+"_3"]++
+			}
+		} else if v.DealType == 4 {
+			subChangeMap2[nameStr] = subValue
+			dealType = 4
+			if _, ok := rankMap[v.ClassifyName+"_"+v.ClassifyType+"_"+v.DataTime+"_4"]; !ok {
+				rankMap[v.ClassifyName+"_"+v.ClassifyType+"_"+v.DataTime+"_4"] = 1
+			} else {
+				rankMap[v.ClassifyName+"_"+v.ClassifyType+"_"+v.DataTime+"_4"]++
+			}
+		}
+
+		// 2023-05-10 目前看该方法的引用startDate和endDate其实是同一天, 所以前一个交易日直接用上面的yesterday
+		tmpTimeStr := yesterday
+
+		//和T-1日比较差值
+		//var tmpTimeStr string
+		//tmpTimeStr, err = getYesterdayDate(v.DataTime)
+		//if err != nil {
+		//	return
+		//}
+		yesterdayStr := v.ClassifyName + "_" + v.ClassifyType + "_" + tmpTimeStr + "_" + v.DealShortName
+		dealChange := 0
+		if dealType == 3 {
+			if c, ok := subChangeMap1[yesterdayStr]; ok {
+				dealChange = subValue - c
+			}
+		} else if dealType == 4 {
+			if c, ok := subChangeMap2[yesterdayStr]; ok {
+				dealChange = subValue - c
+			}
+		}
+		tmp := &data_manage.TradePositionTop{
+			ClassifyName:  v.ClassifyName,
+			ClassifyType:  v.ClassifyType,
+			DataTime:      v.DataTime,
+			CreateTime:    now,
+			ModifyTime:    now,
+			DealShortName: v.DealShortName,
+			DealValue:     subValue,
+			DealChange:    dealChange,
+			DealType:      dealType,
+			Rank:          rankMap[v.ClassifyName+"_"+v.ClassifyType+"_"+v.DataTime+"_"+strconv.Itoa(dealType)],
+		}
+		topList = append(topList, tmp)
+		if len(topList) >= 1000 {
+			err = data_manage.InsertMultiTradePositionTop(exchange, topList)
+			if err != nil {
+				return
+			}
+			topList = make([]*data_manage.TradePositionTop, 0)
+		}
+	}
+
+	if len(topList) >= 0 {
+		err = data_manage.InsertMultiTradePositionTop(exchange, topList)
+		if err != nil {
+			return
+		}
+	}
+	return
+}
+
+// updateAnalysisCleanTopChangeValClassify T-1日重新生成净多单和净空单的榜单后，需要更新T日净多单和净空单榜单里的变化量
+func updateAnalysisCleanTopChangeValClassify(exchange, startDate, yesterday string, classifyNames, classifyTypes []string) (err error) {
+	defer func() {
+		if err != nil {
+			fmt.Println("updateAnalysisCleanTopChangeVal err: " + err.Error())
+		}
+	}()
+
+	//查询T日的净多单和净空单榜单列表
+	//查询T-1日的净多单和净空单列表
+	//组装数据，计算T日与T-1日的变更值
+	//更新变更值
+	topList := make([]*data_manage.TradePositionTop, 0)      //T日和T+1日列表
+	todayTopList := make([]*data_manage.TradePositionTop, 0) //T日列表
+
+	yesterdayTopListMap := make(map[string]int) //净多单净空单持仓量map
+
+	// 查询T日和T-1日的净多单和净空单列表
+	topList, err = data_manage.GetTradePositionTopCleanByExchangeDataTimeClassify(exchange, yesterday, startDate, classifyNames, classifyTypes)
+	if err != nil {
+		return
+	}
+	if len(topList) == 0 {
+		return
+	}
+	for _, v := range topList {
+		if v.DataTime == startDate {
+			todayTopList = append(todayTopList, v)
+		} else if v.DataTime == yesterday {
+			nameStr := v.ClassifyName + "_" + v.ClassifyType + "_" + v.DataTime + "_" + v.DealShortName + "_" + strconv.Itoa(v.DealType)
+			yesterdayTopListMap[nameStr] = v.DealValue
+		}
+	}
+
+	if len(todayTopList) == 0 {
+		return
+	}
+	// 根据净多单/净空单数据, 比对上一个交易日的日期计算成交变化量, 并写入
+	now := time.Now()
+	updateList := make([]data_manage.UpdateChangeVal, 0)
+	for _, v := range todayTopList {
+		//T日值-T-1日值
+		yesterdayStr := v.ClassifyName + "_" + v.ClassifyType + "_" + yesterday + "_" + v.DealShortName + "_" + strconv.Itoa(v.DealType)
+		dealChange := 0
+		if c, ok := yesterdayTopListMap[yesterdayStr]; ok {
+			dealChange = v.DealValue - c
+		}
+		if dealChange != v.DealChange {
+			tmp := data_manage.UpdateChangeVal{
+				Id:         v.Id,
+				ModifyTime: now,
+				DealChange: dealChange,
+			}
+			updateList = append(updateList, tmp)
+		}
+	}
+
+	if len(updateList) > 0 {
+		err = data_manage.MultiUpdatePositionTopChangeVal(exchange, updateList)
+		if err != nil {
+			return
+		}
+	}
+	return
+}