eta_task/models/data_manage/trade_position_analysis.go

package data_manage

import (
	sql2 "database/sql"
	"eta/eta_task/global"
	"eta/eta_task/utils"
	"fmt"
	"gorm.io/gorm"
	"time"
	"unsafe"
)

// 持仓榜单表
type TradePositionTop struct {
	Id            uint64    `orm:"column(id);pk" gorm:"primaryKey"`
	ClassifyName  string    //分类名称
	ClassifyType  string    //分类名称下的类型
	DataTime      string    //数据日期
	CreateTime    time.Time //插入时间
	ModifyTime    time.Time //修改时间
	DealShortName string    //成交量公司简称
	DealValue     int       //成交量
	DealChange    int       //成交变化量
	DealType      int       //交易类型：1多单，2空单，3净多单，4净空单
	SourceType    int       //数据来源，0是原始数据的值，1是由T+1日推算出的值，2是由T日的榜单数据推算出的值
	Rank          int       //排名
}

func (m *TradePositionTop) AfterFind(db *gorm.DB) (err error) {
	m.DataTime = utils.GormDateStrToDateStr(m.DataTime)
	return
}

func (m *TradePositionTop) ConvertTimeStr() {
	m.DataTime = utils.GormDateStrToDateStr(m.DataTime)
	return
}

type TradePositionDalianTop struct {
	TradePositionTop
}

type TradePositionZhengzhouTop struct {
	TradePositionTop
}

type TradePositionCffexTop struct {
	TradePositionTop
}

type TradePositionShanghaiTop struct {
	TradePositionTop
}

type TradePositionIneTop struct {
	TradePositionTop
}

type TradePositionGuangzhouTop struct {
	TradePositionTop
}

func InsertMultiTradePositionTop(exchange string, items []*TradePositionTop) (err error) {
	o := global.DbMap[utils.DbNameIndex]
	if exchange == "dalian" {
		list := make([]*TradePositionDalianTop, 0)
		for _, v := range items {
			tmp := (*TradePositionDalianTop)(unsafe.Pointer(v))
			list = append(list, tmp)
		}
		err = o.CreateInBatches(list, utils.MultiAddNum).Error
		return
	} else if exchange == "zhengzhou" {
		list := make([]*TradePositionZhengzhouTop, 0)
		for _, v := range items {
			tmp := (*TradePositionZhengzhouTop)(unsafe.Pointer(v))
			list = append(list, tmp)
		}
		err = o.CreateInBatches(list, utils.MultiAddNum).Error
		return
	} else if exchange == "cffex" {
		list := make([]*TradePositionCffexTop, 0)
		for _, v := range items {
			tmp := (*TradePositionCffexTop)(unsafe.Pointer(v))
			list = append(list, tmp)
		}
		err = o.CreateInBatches(list, utils.MultiAddNum).Error
		return
	} else if exchange == "shanghai" {
		list := make([]*TradePositionShanghaiTop, 0)
		for _, v := range items {
			tmp := (*TradePositionShanghaiTop)(unsafe.Pointer(v))
			list = append(list, tmp)
		}
		err = o.CreateInBatches(list, utils.MultiAddNum).Error
		return
	} else if exchange == "ine" {
		list := make([]*TradePositionIneTop, 0)
		for _, v := range items {
			tmp := (*TradePositionIneTop)(unsafe.Pointer(v))
			list = append(list, tmp)
		}
		err = o.CreateInBatches(list, utils.MultiAddNum).Error
		return
	} else if exchange == "guangzhou" {
		list := make([]*TradePositionGuangzhouTop, 0)
		for _, v := range items {
			tmp := (*TradePositionGuangzhouTop)(unsafe.Pointer(v))
			list = append(list, tmp)
		}
		err = o.CreateInBatches(list, utils.MultiAddNum).Error
		return
	}

	return
}

func GetTradePositionTopByExchangeDataTime(exchange string, startDate, endDate string) (list []*TradePositionTop, err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := "SELECT * 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"
	err = o.Raw(sql, startDate, endDate).Find(&list).Error
	return
}

func GetTradePositionTopByExchangeDataTimeByClassify(exchange string, startDate, endDate string, classifyNames, classifyTypes []string) (list []*TradePositionTop, err error) {
	o := global.DbMap[utils.DbNameIndex]
	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).Find(&list).Error
	return
}

func GetTradePositionTopCountByExchangeDataTime(exchange string, startDate, endDate string) (count int64, err error) {
	o := global.DbMap[utils.DbNameIndex]
	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"
	err = o.Raw(sql, startDate, endDate).Scan(&count).Error
	return
}

func GetTradePositionTopByExchangeSourceType(exchange string, dataTime string, sourceType int) (list []*TradePositionTop, err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := "SELECT * FROM trade_position_" + exchange + "_top where data_time= ? and source_type = ? ORDER BY classify_name, classify_type, deal_type, deal_value desc"
	err = o.Raw(sql, dataTime, sourceType).Find(&list).Error
	return
}

func GetTradePositionTopByExchangeSourceTypeClassify(exchange string, dataTime string, sourceType int, classifyNames, classifyTypes []string) (list []*TradePositionTop, err error) {
	o := global.DbMap[utils.DbNameIndex]
	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).Find(&list).Error
	return
}

type TradeTopClassify struct {
	ClassifyName string //分类名称
	ClassifyType string //分类名称下的类型
}

type TradePositionSub struct {
	ClassifyName  string //分类名称
	ClassifyType  string //分类名称下的类型
	DataTime      string //数据日期
	DealShortName string //成交量公司简称
	SubValue      int    //差值
	DealType      int
}

type TradePositionSubList []*TradePositionSub

func (v TradePositionSubList) Len() int {
	return len(v)
}

func (v TradePositionSubList) Swap(i, j int) {
	v[i], v[j] = v[j], v[i]
}

func (v TradePositionSubList) Less(i, j int) bool {
	return v[i].SubValue > v[j].SubValue
}

type UpdateDealValueChange struct {
	Id         uint64
	DealValue  int //成交量
	DealChange int
	SourceType int
	ModifyTime time.Time //修改时间
}

type UpdateChangeVal struct {
	Id         uint64
	DealChange int
	ModifyTime time.Time //修改时间
}

func MultiUpdatePositionTop(exchange string, updates []UpdateDealValueChange) (err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := "UPDATE trade_position_" + exchange + "_top SET deal_value=?, deal_change=?, source_type=?, modify_time=? WHERE id = ?"
	for _, v := range updates {
		err = o.Exec(sql, v.DealValue, v.DealChange, v.SourceType, v.ModifyTime, v.Id).Error
		if err != nil {
			return
		}
	}
	return
}

func DeletePositionTopByDataTime(exchange string, dataTime string, dealType int) (err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := "delete from trade_position_" + exchange + "_top WHERE data_time=? and deal_type=?"
	err = o.Exec(sql, dataTime, dealType).Error
	return
}

func DeletePositionTopByDataTimeClassify(exchange string, dataTime string, dealType int, classifyNames, classifyTypes []string) (err error) {
	o := global.DbMap[utils.DbNameIndex]
	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.Exec(sql, dataTime, dealType, classifyNames, classifyTypes).Error
	return
}

func GetTradePositionTopByExchangeDataTimeType(exchange string, dataTime string, dealType int) (list []*TradePositionTop, err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := "select * from trade_position_" + exchange + "_top WHERE data_time=? and deal_type=?"
	err = o.Raw(sql, dataTime, dealType).Find(&list).Error
	return
}

func GetTradePositionTopByExchangeDataTimeTypeClassify(exchange string, dataTime string, dealType int, classifyNames, classifyTypes []string) (list []*TradePositionTop, err error) {
	o := global.DbMap[utils.DbNameIndex]
	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).Find(&list).Error
	return
}

func MultiInsertTradeBaseDataToTop(exchange string, startDate, endDate string) (err error) {
	o := global.DbMap[utils.DbNameIndex]
	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 buy_short_name !=' ' and data_time between ? and ?"
	err = o.Exec(sql1, now, now, startDate, endDate).Error
	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 sold_short_name !=' ' and data_time between ? and ?"
	err = o.Exec(sql2, now, now, startDate, endDate).Error
	return
}

func MultiInsertTradeBaseDataToTopByClassify(exchange string, startDate, endDate string, classifyNames, classifyTypes []string) (err error) {
	o := global.DbMap[utils.DbNameIndex]
	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 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.Exec(sql1, now, now, startDate, endDate, classifyNames, classifyTypes).Error
	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 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.Exec(sql2, now, now, startDate, endDate, classifyNames, classifyTypes).Error
	return
}

// GetTradePositionTopOriginDataTimes 获取榜单原始数据日期-正序
func GetTradePositionTopOriginDataTimes(exchange string) (dates []string, err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := `SELECT DISTINCT data_time FROM base_from_trade_%s_index ORDER BY data_time ASC`
	sql = fmt.Sprintf(sql, exchange)
	var originDates []string
	err = o.Raw(sql).Scan(&originDates).Error
	if err != nil {
		return
	}
	for _, v := range originDates {
		if v == "" {
			continue
		}
		v = utils.GormDateStrToDateStr(v)
		dates = append(dates, v)
	}
	return
}

// BaseFromTradeClassify 交易所分类表
type BaseFromTradeClassify struct {
	Id           uint64    `orm:"column(id);pk" gorm:"primaryKey"`
	ClassifyName string    //分类名称
	ClassifyType string    //分类名称下的类型
	Exchange     string    //交易所
	LatestDate   time.Time //数据最近的日期
	CreateTime   time.Time //插入时间
	ModifyTime   time.Time //修改时间
}

// GetAllBaseFromTradeClassify 获取所有的交易所分类列表
func GetAllBaseFromTradeClassify() (list []*BaseFromTradeClassify, err error) {
	sql := `SELECT * FROM base_from_trade_classify   `
	o := global.DbMap[utils.DbNameIndex]
	err = o.Raw(sql).Find(&list).Error
	return
}

// Update 更新
func (m *BaseFromTradeClassify) Update(cols []string) (err error) {
	o := global.DbMap[utils.DbNameIndex]
	err = o.Select(cols).Updates(m).Error
	return
}

// MultiAddBaseFromTradeClassify 批量插入交易所分类
func MultiAddBaseFromTradeClassify(items []*BaseFromTradeClassify) (err error) {
	if len(items) == 0 {
		return
	}
	o := global.DbMap[utils.DbNameIndex]
	err = o.CreateInBatches(items, utils.MultiAddNum).Error
	return
}

type TradeClassifyName struct {
	ClassifyName string    //分类名称
	ClassifyType string    //分类名称下的类型
	DataTime     time.Time //数据最近的日期
	ModifyTime   time.Time //数据最近的日期
}

// GetExchangeClassify 获取交易所分类列表
func GetExchangeClassify(exchange string) (list []TradeClassifyName, err error) {
	tableName := "trade_position_" + exchange + "_top"
	orderStr := "classify_name DESC, classify_type asc"
	if exchange == "zhengzhou" {
		orderStr = "classify_name asc"
	}
	sql := `SELECT classify_name, classify_type,MAX(data_time) as data_time,MAX(modify_time) as modify_time FROM ` + tableName + ` GROUP BY classify_name, classify_type  `
	sql += ` ORDER BY ` + orderStr

	o := global.DbMap[utils.DbNameIndex]
	err = o.Raw(sql).Find(&list).Error

	return
}

// GetTradePositionTopCleanByExchangeDataTime 根据时间查询净多单和净空单的值
func GetTradePositionTopCleanByExchangeDataTime(exchange string, startDate, endDate string) (list []*TradePositionTop, err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := "SELECT * FROM trade_position_" + exchange + "_top where data_time >= ? and data_time <= ? and deal_type in (3,4) ORDER BY classify_name, classify_type, deal_type, data_time, deal_value desc"
	err = o.Raw(sql, startDate, endDate).Find(&list).Error
	return
}

// GetTradePositionTopCleanByExchangeDataTimeClassify 根据时间查询净多单和净空单的值
func GetTradePositionTopCleanByExchangeDataTimeClassify(exchange string, startDate, endDate string, classifyNames, classifyTypes []string) (list []*TradePositionTop, err error) {
	o := global.DbMap[utils.DbNameIndex]
	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).Find(&list).Error
	return
}

// MultiUpdatePositionTopChangeVal 批量更新榜单里变化量的值
func MultiUpdatePositionTopChangeVal(exchange string, updates []UpdateChangeVal) (err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := "UPDATE trade_position_" + exchange + "_top SET deal_change=?, modify_time=? WHERE id = ?"
	for _, v := range updates {
		err = o.Exec(sql, v.DealChange, v.ModifyTime, v.Id).Error
		if err != nil {
			return
		}
	}
	return
}

func GetTradePositionOriginClassifyCountByExchangeDataTime(exchange string, startDate, endDate string) (count int64, err error) {
	o := global.DbMap[utils.DbNameIndex]
	var pars []interface{}
	var sql string
	if utils.DbDriverName == utils.DbDriverByDm {
		// DM不支持COUNT(DISTINCT column1, column2)这种写法
		sql = `SELECT COUNT(*)
			FROM (
				SELECT DISTINCT classify_name, classify_type
				FROM base_from_trade_ine_index
				WHERE rank < 50 
				  AND (buy_short_name != '' OR sold_short_name != '') 
				  AND (buy_short_name != ' ' OR sold_short_name != ' ') 
				  AND data_time >= ? 
				  AND data_time <= ?
			)`
		pars = append(pars, startDate, endDate)
	} else {
		sql = "SELECT COUNT(DISTINCT classify_name, classify_type) FROM base_from_trade_" + exchange + "_index where rank <50 and (buy_short_name !='' or sold_short_name !='' ) and (buy_short_name !=' ' or sold_short_name !=' ' ) and data_time >= ? and data_time <= ?"
		pars = append(pars, startDate, endDate)
	}
	err = o.Raw(sql, pars...).Scan(&count).Error
	return
}

func GetTradePositionTopClassifyCountByExchangeDataTime(exchange string, startDate, endDate string) (count int64, err error) {
	o := global.DbMap[utils.DbNameIndex]
	var pars []interface{}
	var sql string
	if utils.DbDriverName == utils.DbDriverByDm {
		sql = `SELECT COUNT(*)
			FROM (
				SELECT DISTINCT classify_name, classify_type
				FROM trade_position_ine_top
				WHERE data_time >= ?
				  AND data_time <= ?
				  AND deal_type IN (1, 2)
			)`
		pars = append(pars, startDate, endDate)
	} else {
		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) "
		pars = append(pars, startDate, endDate)
	}
	err = o.Raw(sql, pars...).Scan(&count).Error
	return
}

type TradePositionClassifyInfo struct {
	ClassifyName string //分类名称
	ClassifyType string //分类名称下的类型
}

func GetTradePositionOriginClassifyByExchangeDataTime(exchange string, startDate, endDate string) (list []TradePositionClassifyInfo, err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := "SELECT DISTINCT classify_name, classify_type FROM base_from_trade_" + exchange + "_index where rank <50 and (buy_short_name !='' or sold_short_name !='' ) and (buy_short_name !=' ' or sold_short_name !=' ' ) and data_time >= ? and data_time <= ?"
	err = o.Raw(sql, startDate, endDate).Find(&list).Error
	return
}

func GetTradePositionTopClassifyByExchangeDataTime(exchange string, startDate, endDate string) (list []TradePositionClassifyInfo, err error) {
	o := global.DbMap[utils.DbNameIndex]
	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).Find(&list).Error
	return
}

// DeleteTradePositionTopAllByExchangeDataTime 删除计算数据
func DeleteTradePositionTopAllByExchangeDataTime(exchange string, startDate, endDate string) (err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := "DELETE FROM trade_position_" + exchange + "_top where data_time >= ? and data_time <= ? "
	err = o.Exec(sql, startDate, endDate).Error
	return
}

type GetFirstBaseFromTradeIndeDate struct {
	DataTime string
}

func GetFirstBaseFromTradeIndexByDate(exchange string) (item *GetFirstBaseFromTradeIndeDate, err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := "SELECT * FROM base_from_trade_" + exchange + "_index where rank < 50 order by data_time asc"
	err = o.Raw(sql).First(&item).Error
	if err != nil {
		return
	}
	item.DataTime = utils.GormDateStrToDateStr(item.DataTime)
	return
}

// 获取持仓分析的最新数据
func GetTradePositionTopLastedDataTime(exchange string) (dataTime time.Time, err error) {
	o := global.DbMap[utils.DbNameIndex]
	sql := "SELECT max(data_time) FROM trade_position_" + exchange + "_top"
	var timeNull sql2.NullTime
	err = o.Raw(sql).Scan(&timeNull).Error
	if err != nil {
		return
	}
	if timeNull.Valid {
		dataTime = timeNull.Time
	}
	return
}