eta_task/services/data/trade_position_analysis_classify.go

package data

import (
	"eta/eta_task/models/data_manage"
	"eta/eta_task/services/alarm_msg"
	"eta/eta_task/utils"
	"fmt"
	"sort"
	"strconv"
	"time"
)

// FixPositionTask 补全缺失的合约, 注意和原先的定时任务区分开来
func FixPositionTask() (err error) {
	exchanges := []string{"zhengzhou", "dalian", "shanghai", "cffex", "ine", "guangzhou"} //郑商所，大商所，上期所，中金所，上期能源
	//exchanges := []string{"guangzhou"} //郑商所，大商所，上期所，中金所，上期能源
	for i := 194; i > 1; i-- {
		// 定时任务避开昨日和今日的数据，以免和原先的定时任务InitPositionTask冲突
		startDate := time.Now().AddDate(0, 0, -i).Format(utils.FormatDate)
		//startDate := "2024-01-08"
		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))

			var tradePosition TradePositionInterface
			if exchange == "guangzhou" {
				tradePosition = &GuangzhouTradePosition{}
			} else {
				tradePosition = &BaseTradePosition{}
			}
			tErr, errMsg := InitTradePositionClassify(exchange, startDate, endDate, tradePosition)
			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, tradePosition TradePositionInterface) (err error, errMsg string) {
	defer func() {
		if err != nil {
			tips := fmt.Sprintf("持仓分析-补全缺失的品种数据 操作失败, Exchange: %s, Err: %s, Msg: %s", exchange, err.Error(), errMsg)
			utils.FileLog.Info(tips)
			alarm_msg.SendAlarmMsg(tips, 3)
		}
	}()

	//判断合约数与数据源是否一致
	originClassifyList, tmpErr := tradePosition.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 = tradePosition.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 := tradePosition.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, tradePosition)
	if err != nil {
		errMsg = "创建净多单，净空单数据失败，createAnalysisCleanTop() Err: "
		return
	}

	err = DealYesterdayDataClassify(exchange, startDate, classifyNames, classifyTypes, tradePosition)
	if err != nil {
		errMsg = "处理昨日数据失败，DealYesterdayData() Err: "
		return
	}
	return
}

// DealYesterdayDataClassify 更新部分合约的昨日数据
func DealYesterdayDataClassify(exchange, startDate string, classifyNames, classifyTypes []string, tradePosition TradePositionInterface) (err error) {
	// 查询最早的日期
	firstItem, err := tradePosition.GetFirstBaseFromTradeIndexByDate(exchange)
	if err != nil {
		return
	}
	if startDate == firstItem.DataTime { //如果当前是起始日，则无需统计修改前一天的数据
		return
	}

	// 前一个交易日, 前两个交易日
	dates, e := tradePosition.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, tradePosition)
		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, tradePosition TradePositionInterface) (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 := tradePosition.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
}