eta_index_lib/controllers/base_from_ths_hf.go

package controllers

import (
	"encoding/json"
	"eta/eta_index_lib/logic"
	"eta/eta_index_lib/models"
	"eta/eta_index_lib/services"
	"eta/eta_index_lib/utils"
	"fmt"
	"strconv"
	"strings"
	"time"
)

// ThsHfController 同花顺高频数据
type ThsHfController struct {
	BaseAuthController
}

// GetData
// @Title 同花顺高频数据-获取数据
// @Description 同花顺高频数据-获取数据
// @Success 200 {object} models.ThsHfSearchEdbReq
// @router /hf/edb_data [post]
func (this *ThsHfController) GetData() {
	br := new(models.BaseResponse).Init()
	defer func() {
		if br.ErrMsg == "" {
			br.IsSendEmail = false
		}
		this.Data["json"] = br
		this.ServeJSON()
	}()
	var params models.ThsHfSearchEdbReq
	if e := json.Unmarshal(this.Ctx.Input.RequestBody, &params); e != nil {
		br.Msg = "参数解析异常"
		br.ErrMsg = fmt.Sprintf("参数解析失败, %v", e)
		return
	}
	params.StockCode = strings.TrimSpace(params.StockCode)
	if params.StockCode == "" {
		br.Msg = "请输入证券代码"
		return
	}
	stockCodes := strings.Split(params.StockCode, ",")
	if len(stockCodes) == 0 {
		br.Msg = "请输入证券代码"
		return
	}
	if len(stockCodes) > 10 {
		br.Msg = "最多输入10个证券代码"
		return
	}
	params.EdbCode = strings.TrimSpace(params.EdbCode)
	if params.EdbCode == "" {
		br.Msg = "请输入指标代码"
		return
	}
	edbCodes := strings.Split(params.EdbCode, ",")
	if len(edbCodes) == 0 {
		br.Msg = "请输入指标代码"
		return
	}
	if len(edbCodes) > 20 {
		br.Msg = "最多选择/输入20个指标代码"
		return
	}
	if params.StartTime == "" {
		br.Msg = "请选择起始时间"
		return
	}
	_, e := time.ParseInLocation(utils.FormatDateTime, params.StartTime, time.Local)
	if e != nil {
		br.Msg = "起始时间格式有误"
		br.ErrMsg = fmt.Sprintf("起始时间格式有误, %v", e)
		return
	}
	// 结束时间选填, 不填则为当前时间
	if params.EndTime != "" {
		_, e := time.ParseInLocation(utils.FormatDateTime, params.EndTime, time.Local)
		if e != nil {
			br.Msg = "截止时间格式有误"
			br.ErrMsg = fmt.Sprintf("截止时间格式有误, %v", e)
			return
		}
	}
	if params.EndTime == "" {
		params.EndTime = time.Now().Local().Format(utils.FormatDateTime)
	}
	if !utils.InArrayByInt(models.ThsHfPeriodArr, params.Interval) {
		br.Msg = "时间周期有误"
		br.ErrMsg = fmt.Sprintf("时间周期有误, Interval: %d", params.Interval)
		return
	}
	if params.CPS != "" && !utils.InArrayByStr(models.ThsHfCPSArr, params.CPS) {
		br.Msg = "复权方式有误"
		br.ErrMsg = fmt.Sprintf("复权方式有误, CPS: %s", params.CPS)
		return
	}
	if params.BaseDate != "" {
		_, e = time.ParseInLocation(utils.FormatDate, params.BaseDate, time.Local)
		if e != nil {
			br.Msg = "复权基点格式有误"
			br.ErrMsg = fmt.Sprintf("复权基点格式有误, %v", e)
			return
		}
	}
	if params.Fill != "" && !utils.InArrayByStr(models.ThsHfFillArr, params.Fill) {
		br.Msg = "非交易间隔处理有误"
		br.ErrMsg = fmt.Sprintf("非交易间隔处理有误, Fill: %s", params.Fill)
		return
	}

	// 根据配置获取指标数据
	indexes, e := services.GetEdbDataFromThsHf(params, "")
	if e != nil {
		br.Msg = "获取失败"
		br.ErrMsg = fmt.Sprintf("获取同花顺高频指标失败, %v", e)
		return
	}

	br.Data = indexes
	br.Ret = 200
	br.Success = true
	br.Msg = "获取成功"
}

// BaseAdd
// @Title 同花顺高频数据-新增至数据源
// @Description 同花顺高频数据-新增至数据源
// @Success 200 {object} models.ThsHfBaseAddReq
// @router /hf/base/add [post]
func (this *ThsHfController) BaseAdd() {
	br := new(models.BaseResponse).Init()
	defer func() {
		if br.ErrMsg == "" {
			br.IsSendEmail = false
		}
		this.Data["json"] = br
		this.ServeJSON()
	}()
	var params models.ThsHfBaseAddReq
	if e := json.Unmarshal(this.Ctx.Input.RequestBody, &params); e != nil {
		br.Msg = "参数解析异常"
		br.ErrMsg = fmt.Sprintf("参数解析失败, %v", e)
		return
	}
	params.StockCode = strings.TrimSpace(params.StockCode)
	if params.StockCode == "" {
		br.Msg = "请输入证券代码"
		return
	}
	params.EdbCode = strings.TrimSpace(params.EdbCode)
	if params.EdbCode == "" {
		br.Msg = "请输入指标代码"
		return
	}
	if params.StartTime == "" {
		br.Msg = "请选择起始时间"
		return
	}
	startTime, e := time.ParseInLocation(utils.FormatDateTime, params.StartTime, time.Local)
	if e != nil {
		br.Msg = "起始时间格式有误"
		br.ErrMsg = fmt.Sprintf("起始时间格式有误, %v", e)
		return
	}
	var endTime time.Time
	if params.EndTime != "" {
		ed, e := time.ParseInLocation(utils.FormatDateTime, params.EndTime, time.Local)
		if e != nil {
			br.Msg = "截止时间格式有误"
			br.ErrMsg = fmt.Sprintf("截止时间格式有误, %v", e)
			return
		}
		endTime = ed
	}
	if !utils.InArrayByInt(models.ThsHfPeriodArr, params.Interval) {
		br.Msg = "时间周期有误"
		br.ErrMsg = fmt.Sprintf("时间周期有误, Interval: %d", params.Interval)
		return
	}
	if params.CPS != "" && !utils.InArrayByStr(models.ThsHfCPSArr, params.CPS) {
		br.Msg = "复权方式有误"
		br.ErrMsg = fmt.Sprintf("复权方式有误, CPS: %s", params.CPS)
		return
	}
	if params.BaseDate != "" {
		_, e = time.ParseInLocation(utils.FormatDate, params.BaseDate, time.Local)
		if e != nil {
			br.Msg = "复权基点格式有误"
			br.ErrMsg = fmt.Sprintf("复权基点格式有误, %v", e)
			return
		}
	}
	if params.Fill != "" && !utils.InArrayByStr(models.ThsHfFillArr, params.Fill) {
		br.Msg = "非交易间隔处理有误"
		br.ErrMsg = fmt.Sprintf("非交易间隔处理有误, Fill: %s", params.Fill)
		return
	}
	if params.ClassifyId <= 0 {
		br.Msg = "请选择分类"
		return
	}
	params.IndexName = strings.TrimSpace(params.IndexName)
	if params.IndexName == "" {
		br.Msg = "请输入指标名称"
		return
	}
	if params.Frequency == "" {
		br.Msg = "请输入频度"
		return
	}

	// 缓存
	source := utils.DATA_SOURCE_THS
	subSource := utils.DATA_SUB_SOURCE_HIGH_FREQUENCY
	cacheKey := fmt.Sprintf("%s_%d_%d_%s_%s", utils.CACHE_BASE_EDB_ADD, source, subSource, params.StockCode, params.EdbCode)
	defer func() {
		_ = utils.Rc.Delete(cacheKey)
	}()
	if utils.Rc.IsExist(cacheKey) {
		br.Ret = 501
		br.Success = true
		br.Msg = "系统处理中，请稍后重试"
		return
	}
	utils.Rc.SetNX(cacheKey, 1, 3*time.Minute)

	// 已添加则忽略
	indexOb := new(models.BaseFromThsHfIndex)
	{
		cond := fmt.Sprintf(" AND %s = ? AND %s = ?", indexOb.Cols().StockCode, indexOb.Cols().Indicator)
		pars := make([]interface{}, 0)
		pars = append(pars, params.StockCode, params.EdbCode)
		item, e := indexOb.GetItemByCondition(cond, pars, "")
		if e != nil && e.Error() != utils.ErrNoRow() {
			br.Msg = "操作失败"
			br.ErrMsg = fmt.Sprintf("获取原始指标失败, %v", e)
			return
		}
		if item != nil {
			br.Ret = 200
			br.Success = true
			br.Msg = "操作成功"
			return
		}
	}

	// 获取指标数据
	var apiPars models.ThsHfSearchEdbReq
	apiPars.StockCode = params.StockCode
	apiPars.EdbCode = params.EdbCode
	apiPars.StartTime = params.StartTime
	apiPars.EndTime = params.EndTime
	apiPars.Interval = params.Interval
	apiPars.Fill = params.Fill
	apiPars.CPS = params.CPS
	apiPars.BaseDate = params.BaseDate
	indexes, e := services.GetEdbDataFromThsHf(apiPars, "")
	if e != nil {
		br.Msg = "操作失败"
		br.ErrMsg = fmt.Sprintf("获取同花顺高频指标失败, %v", e)
		return
	}
	if len(indexes) == 0 {
		br.Msg = "未搜索到指标"
		br.ErrMsg = fmt.Sprintf("未搜索到指标")
		return
	}
	indexWithData := indexes[0]

	indexItem := new(models.BaseFromThsHfIndex)
	indexItem.BaseFromThsHfClassifyId = params.ClassifyId
	indexItem.IndexCode = fmt.Sprintf("%s%s%s%s", utils.ThsHf, params.StockCode, params.EdbCode, params.Frequency)
	indexItem.IndexName = params.IndexName
	indexItem.Unit = params.Unit
	indexItem.Frequency = params.Frequency
	indexItem.StartDate = startTime
	indexItem.EndDate = endTime
	indexItem.SysUserId = params.SysAdminId
	indexItem.SysUserRealName = params.SysAdminName
	if params.EndTime == "" {
		indexItem.EndDate = time.Now().Local()
	}
	// TODO:指定终端号
	terminal, e := services.GetFirstTerminal(utils.DATA_SOURCE_THS, "")
	if e != nil {
		br.Msg = "终端未配置"
		br.ErrMsg = fmt.Sprintf("终端未配置, %v", e)
		return
	}
	indexItem.TerminalCode = terminal.TerminalCode
	indexItem.StockCode = params.StockCode
	indexItem.Indicator = params.EdbCode
	b, e := json.Marshal(apiPars)
	if e != nil {
		br.Msg = "操作失败"
		br.ErrMsg = fmt.Sprintf("API入参JSON格式化失败, %v", e)
		return
	}
	indexItem.ApiPars = string(b)
	if len(indexWithData.IndexData) > 0 {
		indexItem.LatestValue = indexWithData.IndexData[0].Value
	}
	indexItem.CreateTime = time.Now().Local()
	indexItem.ModifyTime = time.Now().Local()

	// 新增至数据源和数据
	itemData := make([]*models.BaseFromThsHfData, 0)
	for _, v := range indexWithData.IndexData {
		t := new(models.BaseFromThsHfData)
		t.IndexCode = indexItem.IndexCode
		t.DataTime = v.DataTime
		// TODO:数据格式化
		t.Value = v.Value
		t.UniqueCode = utils.MD5(fmt.Sprint(indexItem.IndexCode, v.DataTime.Format("2006-01-02 15:04")))
		t.CreateTime = time.Now().Local()
		t.ModifyTime = time.Now().Local()
		t.DataTimestamp = v.DataTime.UnixNano() / 1e6
		itemData = append(itemData, t)
	}

	// 新增指标和数据
	if e = indexOb.CreateIndexAndData(indexItem, itemData); e != nil {
		br.Msg = "操作失败"
		br.ErrMsg = fmt.Sprintf("新增指标和数据失败, %v", e)
		return
	}

	br.Ret = 200
	br.Success = true
	br.Msg = "操作成功"
}

// TODO:BaseRefresh
// @Title 同花顺高频数据-数据源刷新
// @Description 同花顺高频数据-数据源刷新
// @Success 200 {object} models.RefreshEdbInfoReq
// @router /hf/base/refresh [post]
func (this *ThsHfController) BaseRefresh() {
	br := new(models.BaseResponse).Init()
	defer func() {
		if br.ErrMsg == "" {
			br.IsSendEmail = false
		}
		this.Data["json"] = br
		this.ServeJSON()
	}()

	br.Ret = 200
	br.Success = true
	br.Msg = "操作成功"
}

// EdbAdd
// @Title 同花顺高频数据-新增至指标库
// @Description 同花顺高频数据-新增至指标库
// @Success 200 {object} models.ThsHfEdbAddReq
// @router /hf/edb/add [post]
func (this *ThsHfController) EdbAdd() {
	br := new(models.BaseResponse).Init()
	defer func() {
		if br.ErrMsg == "" {
			br.IsSendEmail = false
		}
		this.Data["json"] = br
		this.ServeJSON()
	}()
	var params models.ThsHfEdbAddReq
	if e := json.Unmarshal(this.Ctx.Input.RequestBody, &params); e != nil {
		br.Msg = "参数解析异常"
		br.ErrMsg = fmt.Sprintf("参数解析失败, %v", e)
		return
	}
	if params.NewIndex == nil {
		br.Msg = "参数有误"
		br.ErrMsg = "参数有误, 指标信息有误"
		return
	}
	params.NewIndex.NewIndexName = strings.TrimSpace(params.NewIndex.NewIndexName)
	if params.NewIndex.NewIndexName == "" {
		br.Msg = "请输入指标名称"
		return
	}
	if params.NewIndex.ClassifyId <= 0 {
		br.Msg = "请选择分类"
		return
	}
	if params.NewIndex.Unit == "" {
		params.NewIndex.Unit = "无"
	}
	if params.NewIndex.NewFrequency == "" {
		br.Msg = "请输入频度"
		return
	}

	// 校验转换规则
	convertRule := params.ConvertRule
	if convertRule.ConvertType != 1 && convertRule.ConvertType != 2 {
		br.Msg = "请选择数据转换方式"
		return
	}
	if convertRule.ConvertType == 1 {
		if convertRule.ConvertFixed.FixedDay != 1 && convertRule.ConvertFixed.FixedDay != 2 {
			br.Msg = "请选择指定时间"
			return
		}
		if convertRule.ConvertFixed.FixedTime == "" {
			br.Msg = "请选择指定时间"
			return
		}
		timePrefix := time.Now().Local().Format(utils.FormatDate)
		st := fmt.Sprintf("%s %s", timePrefix, convertRule.ConvertFixed.FixedTime)
		_, e := time.Parse(utils.FormatDateTime, st)
		if e != nil {
			br.Msg = "指定时间格式有误"
			return
		}
	}
	if convertRule.ConvertType == 2 {
		if convertRule.ConvertArea.StartDay != 1 && convertRule.ConvertArea.StartDay != 2 {
			br.Msg = "请选择起始时间"
			return
		}
		if convertRule.ConvertArea.StartTime == "" {
			br.Msg = "请选择起始时间"
			return
		}
		var startTimePre string
		if convertRule.ConvertArea.StartDay == 1 {
			startTimePre = time.Now().Local().Format(utils.FormatDate)
		}
		if convertRule.ConvertArea.StartDay == 2 {
			startTimePre = time.Now().Local().AddDate(0, 0, -1).Format(utils.FormatDate)
		}
		st := fmt.Sprintf("%s %s", startTimePre, convertRule.ConvertArea.StartTime)
		startTime, e := time.Parse(utils.FormatDateTime, st)
		if e != nil {
			br.Msg = "起始时间格式有误"
			return
		}

		if convertRule.ConvertArea.EndDay != 1 && convertRule.ConvertArea.EndDay != 2 {
			br.Msg = "请选择截止时间"
			return
		}
		if convertRule.ConvertArea.EndTime == "" {
			br.Msg = "请选择截止时间"
			return
		}
		var endTimePre string
		if convertRule.ConvertArea.EndDay == 1 {
			endTimePre = time.Now().Local().Format(utils.FormatDate)
		}
		if convertRule.ConvertArea.EndDay == 2 {
			endTimePre = time.Now().Local().AddDate(0, 0, -1).Format(utils.FormatDate)
		}
		ed := fmt.Sprintf("%s %s", endTimePre, convertRule.ConvertArea.EndTime)
		endTime, e := time.Parse(utils.FormatDateTime, ed)
		if e != nil {
			br.Msg = "截止时间格式有误"
			return
		}
		if startTime.After(endTime) {
			br.Msg = "起始日期不可早于截止日期"
			return
		}
	}
	convertRuleByte, e := json.Marshal(params.ConvertRule)
	if e != nil {
		br.Msg = "操作失败"
		br.ErrMsg = fmt.Sprintf("转换规则JSON格式化失败, %v", e)
		return
	}

	// 缓存
	source := utils.DATA_SOURCE_THS
	subSource := utils.DATA_SUB_SOURCE_HIGH_FREQUENCY
	cacheKey := fmt.Sprintf("%s_%d_%d_%s_%s_%s", utils.CACHE_EDB_DATA_ADD, source, subSource, params.NewIndex.StockCode, params.NewIndex.EdbCode, params.NewIndex.NewFrequency)
	defer func() {
		_ = utils.Rc.Delete(cacheKey)
	}()
	if utils.Rc.IsExist(cacheKey) {
		br.Ret = 501
		br.Success = true
		br.Msg = "系统处理中，请稍后重试"
		return
	}
	utils.Rc.SetNX(cacheKey, 1, 3*time.Minute)

	// 校验指标/分类
	baseIndexOb := new(models.BaseFromThsHfIndex)
	baseIndex, e := baseIndexOb.GetItemById(params.NewIndex.IndexId)
	if e != nil {
		br.Msg = "原指标不存在"
		br.ErrMsg = fmt.Sprintf("原指标不存在, %v", e)
		return
	}
	_, e = models.GetEdbClassifyById(params.NewIndex.ClassifyId)
	if e != nil {
		br.Msg = "分类信息有误"
		br.ErrMsg = fmt.Sprintf("获取分类失败, %v", e)
		return
	}

	// 判断指标名称是否已存在
	{
		var cond string
		var pars []interface{}
		if this.Lang == utils.EnLangVersion {
			cond += " AND edb_name_en = ? "
		} else {
			cond += " AND edb_name = ?"
		}
		pars = append(pars, params.NewIndex.NewIndexName)
		count, e := models.GetEdbInfoCountByCondition(cond, pars)
		if e != nil {
			br.Msg = "操作失败"
			br.ErrMsg = fmt.Sprintf("获取重名指标失败, %v", e)
			return
		}
		if count > 0 {
			br.Msg = "指标名称已存在"
			return
		}
	}

	// 排序/指标编码
	sortMax, e := models.GetEdbClassifyMaxSort(params.NewIndex.ClassifyId, 0)
	if e != nil {
		br.Msg = "操作失败"
		br.ErrMsg = fmt.Sprintf("获取最大排序失败, %v", e)
		return
	}
	edbCode, e := utils.GenerateEdbCode(1, "")
	if e != nil {
		br.Msg = "操作失败"
		br.ErrMsg = fmt.Sprintf("生成指标编码失败, %v", e)
		return
	}
	timestamp := strconv.FormatInt(time.Now().UnixNano(), 10)
	uniqueCode := utils.MD5(utils.DATA_PREFIX + "_" + timestamp)

	thsOb := new(models.EdbThsHf)
	var addPars models.ThsHfAddBaseParams
	addPars.EdbCode = edbCode
	addPars.EdbName = params.NewIndex.NewIndexName
	addPars.Unit = params.NewIndex.Unit
	addPars.Frequency = params.NewIndex.NewFrequency
	addPars.Sort = sortMax + 1
	addPars.ClassifyId = params.NewIndex.ClassifyId
	addPars.SysUserId = params.NewIndex.SysAdminId
	addPars.SysUserRealName = params.NewIndex.SysAdminName
	addPars.UniqueCode = uniqueCode
	addPars.ConvertRule = string(convertRuleByte)
	edbInfo, e := thsOb.Add(addPars, baseIndex)
	if e != nil {
		br.Msg = "新增失败"
		br.ErrMsg = fmt.Sprintf("新增指标失败, %v", e)
		return
	}

	// 更新指标最值
	e, _ = models.UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
	if e != nil {
		br.Msg = "刷新指标失败"
		br.ErrMsg = fmt.Sprintf("更新指标最值失败, %v", e)
		return
	}

	// 添加到es
	go logic.UpdateEs(edbInfo.EdbInfoId)

	br.Ret = 200
	br.Success = true
	br.Msg = "操作成功"
}

// EdbRefresh
// @Title 同花顺高频数据-指标库刷新
// @Description 同花顺高频数据-指标库刷新
// @Success 200 {object} models.RefreshEdbInfoReq
// @router /hf/edb/refresh [post]
func (this *ThsHfController) EdbRefresh() {
	br := new(models.BaseResponse).Init()
	defer func() {
		if br.ErrMsg == "" {
			br.IsSendEmail = false
		}
		this.Data["json"] = br
		this.ServeJSON()
	}()
	var req models.RefreshEdbInfoReq
	err := json.Unmarshal(this.Ctx.Input.RequestBody, &req)
	if err != nil {
		br.Msg = "参数解析异常！"
		br.ErrMsg = "参数解析失败，Err:" + err.Error()
		return
	}
	if req.EdbCode == "" {
		br.Msg = "请输入指标编码！"
		br.ErrMsg = "请输入指标编码，指标编码为空"
		return
	}
	if req.EdbInfoId < 0 {
		br.Msg = "请输入指标ID！"
		br.ErrMsg = "请输入指标ID"
		return
	}
	thsOb := new(models.EdbThsHf)
	source := thsOb.GetSource()
	subSource := thsOb.GetSubSource()
	cacheKey := fmt.Sprintf("%s_%d_%d_%s", utils.CACHE_EDB_DATA_REFRESH, source, subSource, req.EdbCode)
	if utils.Rc.IsExist(cacheKey) {
		br.Ret = 501
		br.Success = true
		br.Msg = "系统处理中，请稍后重试"
		return
	}
	utils.Rc.SetNX(cacheKey, 1, 3*time.Minute)
	defer func() {
		_ = utils.Rc.Delete(cacheKey)
	}()

	// 获取指标详情
	edbInfo, e := models.GetEdbInfoByEdbCode(source, req.EdbCode)
	if e != nil {
		br.Msg = "指标不存在"
		br.ErrMsg = fmt.Sprintf("指标不存在, %v", e)
		return
	}

	// 获取指标关联信息
	baseMapping := new(models.BaseFromEdbMapping)
	{
		ob := new(models.BaseFromEdbMapping)
		cond := fmt.Sprintf(" AND %s = ? AND %s = ? AND %s = ?", ob.Cols().EdbCode, ob.Cols().Source, ob.Cols().SubSource)
		pars := make([]interface{}, 0)
		pars = append(pars, req.EdbCode, thsOb.GetSource(), thsOb.GetSubSource())
		mapping, e := ob.GetItemByCondition(cond, pars, "")
		if e != nil {
			br.Msg = "刷新失败"
			br.ErrMsg = fmt.Sprintf("指标关联信息有误, %v", e)
			return
		}
		baseMapping = mapping
	}

	// 刷新指标
	if e = thsOb.Refresh(edbInfo, baseMapping, req.StartDate); e != nil {
		br.Msg = "刷新指标失败"
		br.ErrMsg = fmt.Sprintf("刷新指标失败, %v", e)
		return
	}

	// 更新指标最值
	e, _ = models.UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
	if e != nil {
		br.Msg = "刷新指标失败"
		br.ErrMsg = fmt.Sprintf("更新指标最值失败, %v", e)
		return
	}

	// 更新ES
	go logic.UpdateEs(edbInfo.EdbInfoId)

	br.Ret = 200
	br.Success = true
	br.Msg = "操作成功"
}