base_from_mtjh.go 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357
  1. package models
  2. import (
  3. "eta/eta_index_lib/global"
  4. "eta/eta_index_lib/utils"
  5. "fmt"
  6. "gorm.io/gorm"
  7. "strconv"
  8. "strings"
  9. "time"
  10. )
  11. type BaseFromMtjhMapping struct {
  12. BaseFromMtjhMappingId int `gorm:"column:base_from_Mtjh_mapping_id;primaryKey"`
  13. //BaseFromMtjhMappingId int `orm:"column(base_from_Mtjh_mapping_id);pk"`
  14. IndexName string `description:"持买单量指标名称"`
  15. IndexCode string `description:"持买单量指标编码"`
  16. CreateTime time.Time `description:"时间"`
  17. Area string `description:"区域"`
  18. Port string `description:"港口或码头"`
  19. Variety string `description:"品种"`
  20. Unit string `description:"单位"`
  21. Frequency string `description:"频率"`
  22. }
  23. type BaseFromMtjhIndex struct {
  24. BaseFromMtjhIndexId int `gorm:"column:base_from_mtjh_index_id;primaryKey"`
  25. //BaseFromMtjhIndexId int `orm:"column(base_from_mtjh_index_id);pk"`
  26. IndexName string `description:"持买单量指标名称"`
  27. IndexCode string `description:"持买单量指标编码"`
  28. DealValue string `description:"成交量"`
  29. DataTime string `description:"数据日期"`
  30. Area string `description:"区域"`
  31. Port string `description:"港口或码头"`
  32. Variety string `description:"品种"`
  33. Unit string `description:"单位"`
  34. Frequency string `description:"频率"`
  35. CreateTime time.Time `description:"插入时间"`
  36. ModifyTime time.Time `description:"修改时间"`
  37. }
  38. func (e *BaseFromMtjhIndex) AfterFind(db *gorm.DB) (err error) {
  39. e.DataTime = utils.GormDateStrToDateStr(e.DataTime)
  40. return
  41. }
  42. // 查询指标
  43. func GetBaseFromMtjhMapping() (items []*BaseFromMtjhMapping, err error) {
  44. //o := orm.NewOrm()
  45. sql := `SELECT * FROM base_from_mtjh_mapping`
  46. //_, err = o.Raw(sql).QueryRows(&items)
  47. err = global.DEFAULT_DB.Raw(sql).Find(&items).Error
  48. return
  49. }
  50. // 查询指标
  51. func GetBaseFromMtjhIndex() (items []*BaseFromMtjhIndex, err error) {
  52. //o := orm.NewOrm()
  53. sql := `SELECT * FROM base_from_mtjh_index`
  54. //_, err = o.Raw(sql).QueryRows(&items)
  55. err = global.DEFAULT_DB.Raw(sql).Find(&items).Error
  56. return
  57. }
  58. // 添加数据
  59. func AddBaseFromMtjhIndex(item *BaseFromMtjhIndex) (lastId int64, err error) {
  60. //o := orm.NewOrm()
  61. //lastId, err = o.Insert(item)
  62. err = global.DEFAULT_DB.Create(&item).Error
  63. if err != nil {
  64. return
  65. }
  66. lastId = int64(item.BaseFromMtjhIndexId)
  67. return
  68. }
  69. func AddBaseFromMtjhIndexMuti(items []*BaseFromMtjhIndex) (err error) {
  70. err = global.DEFAULT_DB.CreateInBatches(items, utils.MultiAddNum).Error
  71. return
  72. }
  73. // 添加指标
  74. func AddBaseFromMtjhMapping(item *BaseFromMtjhMapping) (lastId int64, err error) {
  75. //o := orm.NewOrm()
  76. //lastId, err = o.Insert(item)
  77. err = global.DEFAULT_DB.Create(&item).Error
  78. if err != nil {
  79. return
  80. }
  81. lastId = int64(item.BaseFromMtjhMappingId)
  82. return
  83. }
  84. func AddBaseFromMtjhMappingMuti(items []*BaseFromMtjhMapping) (lastId int64, err error) {
  85. //o := orm.NewOrm()
  86. //lastId, err = o.InsertMulti(500, items)
  87. err = global.DEFAULT_DB.CreateInBatches(items, utils.MultiAddNum).Error
  88. return
  89. }
  90. func UpdateBaseFromMtjhIndex(item *BaseFromMtjhIndex) (err error) {
  91. //o := orm.NewOrm()
  92. sql := `UPDATE base_from_mtjh_index SET deal_value=? WHERE index_name=? AND data_time = ?`
  93. //_, err = o.Raw(sql, item.DealValue, item.IndexName, item.DataTime).Exec()
  94. err = global.DEFAULT_DB.Exec(sql, item.DealValue, item.IndexName, item.DataTime).Error
  95. return
  96. }
  97. func AddEdbDataFromMtjh(edbCode string) (err error) {
  98. //o := orm.NewOrm()
  99. coalBaseDataAll, err := GetMtjhindexByCode(edbCode)
  100. if err != nil && !utils.IsErrNoRow(err) {
  101. return
  102. }
  103. var isAdd bool
  104. addSql := ` INSERT INTO edb_data_mtjh(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
  105. addSql = utils.ReplaceDriverKeywords("", addSql)
  106. existMap := make(map[string]string)
  107. for _, sv := range coalBaseDataAll {
  108. eDate := sv.DataTime
  109. var timeStr string
  110. var dataTime time.Time
  111. var sDataTime string
  112. var timestamp int64
  113. sDataTime = eDate
  114. dataTime, err = time.ParseInLocation(utils.FormatDate, eDate, time.Local)
  115. if err != nil {
  116. fmt.Println("time.Parse Err:" + eDate)
  117. return err
  118. }
  119. timestamp = dataTime.UnixNano() / 1e6
  120. timeStr = fmt.Sprintf("%d", timestamp)
  121. value := strings.Replace(sv.DealValue, "%", "", -1)
  122. if _, ok := existMap[sDataTime]; !ok {
  123. addSql += GetAddSql("0", edbCode, sDataTime, timeStr, value)
  124. fmt.Println("edbCode:", edbCode)
  125. fmt.Println("sDataTime:", sDataTime)
  126. fmt.Println("timeStr:", timeStr)
  127. fmt.Println("value:", value)
  128. isAdd = true
  129. }
  130. existMap[eDate] = value
  131. }
  132. if isAdd {
  133. addSql = strings.TrimRight(addSql, ",")
  134. utils.FileLog.Info("addSql:" + addSql)
  135. //_, err = o.Raw(addSql).Exec()
  136. err = global.DEFAULT_DB.Exec(addSql).Error
  137. if err != nil {
  138. return err
  139. }
  140. }
  141. return
  142. }
  143. func RefreshEdbDataFromMtjh(edbInfoId int, edbCode, startDate string) (err error) {
  144. source := utils.DATA_SOURCE_MTJH
  145. subSource := utils.DATA_SUB_SOURCE_EDB
  146. //o := orm.NewOrm()
  147. if err != nil {
  148. return
  149. }
  150. edbInfoIdStr := strconv.Itoa(edbInfoId)
  151. //计算数据
  152. var condition string
  153. var pars []interface{}
  154. if edbCode != "" {
  155. condition += " AND index_code=? "
  156. pars = append(pars, edbCode)
  157. }
  158. if startDate != "" {
  159. condition += " AND data_time>=? "
  160. pars = append(pars, startDate)
  161. }
  162. glDataList, err := GetMtjhDataByTradeCode(condition, pars)
  163. if err != nil {
  164. return
  165. }
  166. // 真实数据的最大日期 , 插入规则配置的日期
  167. var realDataMaxDate, edbDataInsertConfigDate time.Time
  168. var edbDataInsertConfig *EdbDataInsertConfig
  169. var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值
  170. {
  171. edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId)
  172. if err != nil && !utils.IsErrNoRow(err) {
  173. return
  174. }
  175. if edbDataInsertConfig != nil {
  176. edbDataInsertConfigDate = edbDataInsertConfig.Date
  177. }
  178. }
  179. //获取指标所有数据
  180. var existCondition string
  181. var existPars []interface{}
  182. existCondition += " AND edb_info_id=? "
  183. existPars = append(existPars, edbInfoId)
  184. if startDate != "" {
  185. existCondition += " AND data_time>=? "
  186. existPars = append(existPars, startDate)
  187. }
  188. existList, err := GetEdbDataByCondition(source, subSource, existCondition, existPars)
  189. if err != nil {
  190. return err
  191. }
  192. existMap := make(map[string]*EdbInfoSearchData)
  193. for _, v := range existList {
  194. existMap[v.DataTime] = v
  195. }
  196. addSql := ` INSERT INTO edb_data_mtjh(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
  197. addSql = utils.ReplaceDriverKeywords("", addSql)
  198. var isAdd bool
  199. for _, v := range glDataList {
  200. var value string
  201. value = strings.Replace(v.DealValue, "%", "", -1)
  202. item := v
  203. itemValue := value
  204. if _, ok := existMap[v.DataTime]; !ok {
  205. eDate := item.DataTime
  206. var timeStr string
  207. var dataTime time.Time
  208. var sDataTime string
  209. var timestamp int64
  210. sDataTime = eDate
  211. dataTime, err = time.ParseInLocation(utils.FormatDate, eDate, time.Local)
  212. if err != nil {
  213. fmt.Println("time.Parse Err:" + eDate)
  214. return err
  215. }
  216. timestamp = dataTime.UnixNano() / 1e6
  217. timeStr = fmt.Sprintf("%d", timestamp)
  218. sValue := itemValue
  219. if sValue != "" {
  220. saveValue := sValue
  221. if findItem, ok := existMap[eDate]; !ok {
  222. addSql += GetAddSql(edbInfoIdStr, edbCode, sDataTime, timeStr, saveValue)
  223. isAdd = true
  224. } else {
  225. if findItem != nil && utils.SubFloatToString(findItem.Value, 30) != sValue {
  226. err = ModifyEdbDataById(source, subSource, findItem.EdbDataId, sValue)
  227. if err != nil {
  228. return err
  229. }
  230. }
  231. }
  232. }
  233. // 下面代码主要目的是处理掉手动插入的数据判断
  234. {
  235. if realDataMaxDate.IsZero() || dataTime.After(realDataMaxDate) {
  236. realDataMaxDate = dataTime
  237. }
  238. if edbDataInsertConfigDate.IsZero() || dataTime.Equal(edbDataInsertConfigDate) {
  239. isFindConfigDateRealData = true
  240. }
  241. }
  242. }
  243. }
  244. // 处理手工数据补充的配置
  245. HandleConfigInsertEdbData(realDataMaxDate, edbDataInsertConfig, edbInfoId, source, subSource, existMap, isFindConfigDateRealData)
  246. if isAdd {
  247. addSql = strings.TrimRight(addSql, ",")
  248. //_, err = o.Raw(addSql).Exec()
  249. err = global.DEFAULT_DB.Exec(addSql).Error
  250. if err != nil {
  251. return err
  252. }
  253. }
  254. return
  255. }
  256. // GetMtjhindexByCode
  257. func GetMtjhindexByCode(indexCode string) (items []*BaseFromMtjhIndex, err error) {
  258. //o := orm.NewOrm()
  259. sql := "SELECT * FROM base_from_mtjh_index WHERE index_code=? "
  260. //_, err = o.Raw(sql, indexCode).QueryRows(&items)
  261. err = global.DEFAULT_DB.Raw(sql, indexCode).Find(&items).Error
  262. return
  263. }
  264. func GetMtjhDataByTradeCode(condition string, pars []interface{}) (item []*BaseFromMtjhIndex, err error) {
  265. sql := ` SELECT * FROM base_from_mtjh_index WHERE 1=1 `
  266. //o := orm.NewOrm()
  267. if condition != "" {
  268. sql += condition
  269. }
  270. sql += ` ORDER BY data_time DESC `
  271. //_, err = o.Raw(sql, pars).QueryRows(&item)
  272. err = global.DEFAULT_DB.Raw(sql, pars...).Find(&item).Error
  273. return
  274. }
  275. func (m *BaseFromMtjhMapping) GetMaxAndMinDateByIndexCode(indexCode string) (item *EdbInfoMaxAndMinInfo, err error) {
  276. var suffix string
  277. if strings.Contains(indexCode, "jsm") {
  278. suffix = "jsm_index"
  279. } else if strings.Contains(indexCode, "company") {
  280. suffix = "company_index"
  281. } else if strings.Contains(indexCode, "firm") {
  282. suffix = "firm_index"
  283. } else if strings.Contains(indexCode, "coastal") {
  284. suffix = "coastal_index"
  285. } else if strings.Contains(indexCode, "inland") {
  286. suffix = "inland_index"
  287. }
  288. tableName := fmt.Sprintf("base_from_coalmine_%s", suffix)
  289. //o := orm.NewOrm()
  290. sql := fmt.Sprintf(` SELECT MIN(data_time) AS min_date,MAX(data_time) AS max_date,MIN(deal_value) AS min_value,MAX(deal_value) AS max_value FROM %s WHERE index_code=? `, tableName)
  291. //err = o.Raw(sql, indexCode).QueryRow(&item)
  292. err = global.DEFAULT_DB.Raw(sql, indexCode).First(&item).Error
  293. if err != nil {
  294. return
  295. }
  296. // 获取最新值
  297. var latest_value float64
  298. sql = fmt.Sprintf(` SELECT value AS deal_value FROM %s WHERE index_code=? ORDER BY data_time DESC LIMIT 1 `, tableName)
  299. //err = o.Raw(sql, indexCode).QueryRow(&latest_value)
  300. err = global.DEFAULT_DB.Raw(sql, indexCode).Scan(&latest_value).Error
  301. if err != nil {
  302. return
  303. }
  304. item.LatestValue = latest_value
  305. return
  306. }
  307. func (m *BaseFromMtjhMapping) ModifyIndexMaxAndMinDate(indexCode string, item *EdbInfoMaxAndMinInfo) (err error) {
  308. //o := orm.NewOrm()
  309. sql := ` UPDATE base_from_mtjh_mapping SET start_date=?,end_date=?,latest_value=?,modify_time=NOW() WHERE index_code=? `
  310. //_, err = o.Raw(sql, item.MinDate, item.MaxDate, item.LatestValue, indexCode).Exec()
  311. err = global.DEFAULT_DB.Exec(sql, item.MinDate, item.MaxDate, item.LatestValue, indexCode).Error
  312. return
  313. }