package services import ( "encoding/json" "errors" "eta/eta_index_lib/models" "eta/eta_index_lib/models/future_good" "eta/eta_index_lib/services/alarm_msg" "eta/eta_index_lib/utils" "fmt" "github.com/shopspring/decimal" "io" netHttp "net/http" "reflect" "strings" "time" ) var tokenRedisKey = `THS_SERVER_ACCESS_TOKEN` // getEdbDataFromThsHttp 通过url获取同花顺的普通数据 func getEdbDataFromThsHttp(edbCode, startDate, endDate, thsRefreshToken, token string) (item models.EdbDataFromThs, err error) { thsUrl := `https://quantapi.51ifind.com/api/v1/edb_service` //indicators 是 半角逗号分隔的所有指标,宏观指标过多,推荐使用Windows超级命令生成。 "indicators":"M001620326,M002822183" //functionpara 否 key-value格式,省略时不进行更新时间筛选。两个时间控件更新起始时间(startrtime)和更新结束时间(endrtime),不勾选时省略见下方代码块 //startdate 是 开始日期,支持”YYYYMMDD"”YYYY-MM-DD"”YYYY/MM/DD"三种时间格式 "startdate":"2018-01-01" //enddate 是 结束日期,支持”YYYYMMDD"”YYYY-MM-DD"”YYYY/MM/DD"三种日期格式 "enddate":"2018-01-01 //发送创建请求 dataMap := map[string]interface{}{ "indicators": edbCode, "startdate": startDate, "enddate": endDate, } body, err, _ := postCurl(thsUrl, dataMap, 0, thsRefreshToken, token) if err != nil { return } tmpItems := new(EdbDataFromThsInterface) err = json.Unmarshal(body, &tmpItems) if err != nil { err = errors.New("GetEdbDataFromThs json.Unmarshal Err:" + err.Error()) return } if tmpItems.Errorcode != 0 { err = errors.New(tmpItems.Errmsg) return } // 因为table里面的value有的时候返回的是string,有的是float64,所以需要用interface来反射取值 tablesList := make([]models.Tables, 0) for _, table := range tmpItems.Tables { tableIdList := make([]string, 0) tableTimeList := make([]string, 0) tableValueList := make([]float64, 0) for _, tableId := range table.ID { tableIdList = append(tableIdList, tableId) } for _, tableTime := range table.Time { tableTimeList = append(tableTimeList, tableTime) } //指标数据 for _, tmpValue := range table.Value { var tableValue float64 if reflect.TypeOf(tmpValue).Kind() == reflect.Float64 { tableValue = reflect.ValueOf(tmpValue).Float() } else if reflect.TypeOf(tmpValue).Kind() == reflect.String { tmpTableValue, tmpErr := decimal.NewFromString(reflect.ValueOf(tmpValue).String()) if tmpErr != nil { err = tmpErr return } tableValue, _ = tmpTableValue.Truncate(4).Float64() } else { err = errors.New("错误的数据类型" + reflect.TypeOf(tmpValue).String()) return } tableValueList = append(tableValueList, tableValue) } tmpTable := models.Tables{ ID: tableIdList, Time: tableTimeList, Value: tableValueList, } tablesList = append(tablesList, tmpTable) } item = models.EdbDataFromThs{ DataVol: tmpItems.DataVol, Errmsg: tmpItems.Errmsg, Errorcode: tmpItems.Errorcode, Perf: tmpItems.Perf, Tables: tablesList, } return } // getFutureGoodDataFromThsHttp 通过url获取同花顺的商品数据 func getFutureGoodDataFromThsHttp(edbCode, startDate, endDate, thsRefreshToken, token string) (item future_good.FutureGoodDataFromThs, err error) { thsUrl := `https://quantapi.51ifind.com/api/v1/cmd_history_quotation` //codes 是 半角逗号分隔的所有代码 "codes":"300033.SZ,600030.SH" //indicators 是 半角逗号分隔的所有指标 "indicators":"preClose,open" //functionpara 否 /key-value格式。所有key均取默认时,functionpara省略。 见下方说明 //startdate 是 开始日期,支持"YYYYMMDD""YYYY-MMDD""YYYY/MM/DD"三种日期格式 //"startdate":"2018-01-01" //enddate 是 结束日期,支持"YYYYMMDD""YYYY-MMDD""YYYY/MM/DD"三种日期格式 //发送创建请求 dataMap := map[string]interface{}{ "codes": edbCode, "indicators": `lastclose,open,high,low,close,avgprice,change,changeper,volume,amount,hsl,lastsettlement,settlement,zdsettlement,zdfsettlement,ccl,ccbd,zf,zjlx,zjcd`, "startdate": startDate, "enddate": endDate, } body, err, _ := postCurl(thsUrl, dataMap, 0, thsRefreshToken, token) if err != nil { return } tmpItems := new(FutureGoodDataFromThsInterface) err = json.Unmarshal(body, &tmpItems) if err != nil { err = errors.New("GetEdbDataFromThs json.Unmarshal Err:" + err.Error()) return } if tmpItems.Errorcode != 0 { err = errors.New(tmpItems.Errmsg) return } if len(tmpItems.Tables) <= 0 { return } table := tmpItems.Tables[0] item = future_good.FutureGoodDataFromThs{ DataVol: tmpItems.DataVol, Errmsg: tmpItems.Errmsg, Errorcode: tmpItems.Errorcode, Perf: tmpItems.Perf, Tables: future_good.FutureGoodDataTables{ Time: table.Time, Open: table.Table.Open, High: table.Table.High, Low: table.Table.Low, Close: table.Table.Close, Volume: table.Table.Volume, Amount: table.Table.Amount, Ccl: table.Table.Ccl, Settlement: table.Table.Settlement, }, } return } // BaseThsInterface 同花顺基础返回 type BaseThsInterface struct { ErrMsg string `json:"errmsg"` ErrorCode int64 `json:"errorcode"` Tables interface{} `json:"tables"` } // postCurl post请求同花顺接口 func postCurl(urlStr string, dataMap map[string]interface{}, num int, thsRefreshToken, token string) (body []byte, err error, errMsg string) { logMsg := `` defer func() { if err != nil { if logMsg != `` { errMsg = logMsg go alarm_msg.SendAlarmMsg("post请求同花顺接口失败,ERR:"+err.Error()+";errMsg:"+errMsg, 3) } } }() jsonStrByte, err := json.Marshal(dataMap) if err != nil { return } reqStr := string(jsonStrByte) req, _ := netHttp.NewRequest("POST", urlStr, strings.NewReader(reqStr)) req.Header.Add("Content-Type", "application/json") req.Header.Add("access_token", token) res, err := netHttp.DefaultClient.Do(req) if err != nil { return } defer res.Body.Close() //解析resp并且存入关联表 body, err = io.ReadAll(res.Body) if err != nil { logMsg = fmt.Sprint("post err; request:", reqStr, "; errMsg:", err.Error()) utils.FileLog.Info(logMsg) return } //logMsg = fmt.Sprint("post request:", reqStr, "; response:", string(body)) //utils.FileLog.Info(logMsg) logMsg = fmt.Sprint("post request url:", urlStr, ";token:", token, ";params:", reqStr, ";response:", string(body)) var response BaseThsInterface err = json.Unmarshal(body, &response) if err != nil { utils.FileLog.Info("post Err:", err.Error(), ";url:", urlStr, ";params:", reqStr, ";response:", string(body)) err = errors.New("Unmarshal Err:" + err.Error()) return } utils.FileLog.Info(fmt.Sprint("post request url:", urlStr, ";token:", token, ";params:", reqStr, ";response:", string(body))) //如果是token失效,同时只是第一次请求(没有尝试强制刷新token,那么重新请求) if utils.InArrayByInt([]int{-1010, -1302}, int(response.ErrorCode)) && num <= 0 { //token失效 token, err = refreshAccessToken(thsRefreshToken) if err != nil { return } num++ return postCurl(urlStr, dataMap, num, thsRefreshToken, token) } else if response.ErrorCode != 0 { utils.FileLog.Info(fmt.Sprint("post data err:", response.ErrMsg, ";url:", urlStr, ";params:", reqStr, ";response:", string(body))) err = errors.New(response.ErrMsg) return } return } // GetAccessToken 获取accessToken func GetAccessToken(isRefresh bool, thsRefreshToken string) (token string, err error) { defer func() { if err != nil { go alarm_msg.SendAlarmMsg("获取同花顺的token失败,ERR:"+err.Error(), 3) } }() redisKey := utils.CACHE_EDB_THS_SERVER_TOKEN + thsRefreshToken + ":" token, redisErr := utils.Rc.RedisString(redisKey) //如果从redis中accessToken 获取失败或者token为空了,再或者需要强制刷新了,那么重新获取accessToken if redisErr != nil || token == `` || isRefresh { token, _ = refreshAccessToken(thsRefreshToken) return } return } // refreshAccessToken 强制刷新获取accessToken func refreshAccessToken(thsRefreshToken string) (token string, err error) { defer func() { if err != nil { go alarm_msg.SendAlarmMsg("刷新同花顺的token失败;ERR:"+err.Error(), 3) } }() tokenInfo, tmpErr := getAccessToken(thsRefreshToken) if tmpErr != nil { err = tmpErr return } token = tokenInfo.AccessToken expireTime, err := time.ParseInLocation(utils.FormatDateTime, tokenInfo.ExpiredTime, time.Local) if err != nil { err = fmt.Errorf("获取同花顺的token失败;同花顺token截止日期转换失败,ERR:%s", err.Error()) return } //token存入redis //err = utils.Rc.Put(tokenRedisKey, token, time.Duration(expireTime.Unix()-600)*time.Second) // 本来是要设置下600s的过期时间,但因为不是强制刷新token,就不获取了 redisKey := utils.CACHE_EDB_THS_SERVER_TOKEN + thsRefreshToken + ":" err = utils.Rc.Put(redisKey, token, time.Duration(expireTime.Unix())*time.Second) if err != nil { err = fmt.Errorf("获取同花顺的token成功;同花顺token存入redis失败,ERR:%s", err.Error()) return } return } type GetTokenResp struct { ErrorCode int `json:"errorcode"` ErrMsg string `json:"errmsg"` Data TokenData `json:"data"` } type TokenData struct { AccessToken string `json:"access_token"` //ExpireIn int `json:"expire_in"` ExpiredTime string `json:"expired_time"` } // getAccessToken token内部请求接口 func getAccessToken(thsRefreshToken string) (tokenData TokenData, err error) { /*if utils.ThsRefreshToken == `` { err = errors.New("同花顺token未配置") return }*/ defer func() { if err != nil { go alarm_msg.SendAlarmMsg("更新同花顺的token失败;ERR:"+err.Error(), 3) } }() if thsRefreshToken == "" { err = errors.New("同花顺token未配置") return } //getUrl := `https://quantapi.51ifind.com/api/v1/update_access_token` // 强制刷新token;目前因为生产和测试都是使用的一个账号,所以token不能因为环境的不同而改变 getUrl := `https://quantapi.51ifind.com/api/v1/get_access_token` // 获取当前token req, _ := netHttp.NewRequest("GET", getUrl, nil) req.Header.Add("Content-Type", "application/json") req.Header.Add("refresh_token", thsRefreshToken) res, err := netHttp.DefaultClient.Do(req) if err != nil { return } defer res.Body.Close() body, err := io.ReadAll(res.Body) if err != nil { err = errors.New("NewRequest Err:" + err.Error()) return } utils.FileLog.Info("同花顺刷新token:" + string(body)) var tokenResp GetTokenResp err = json.Unmarshal(body, &tokenResp) if err != nil { err = errors.New("Unmarshal Err:" + err.Error()) return } if tokenResp.ErrorCode != 0 { err = errors.New("getAccessToken err:" + tokenResp.ErrMsg) return } tokenData = tokenResp.Data return }