liurui/PriceForecast
Template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
9abd7475ba
PriceForecast/test/ETA获取数据.py

757 lines
32 KiB
Python
Raw Normal View History

整理eta获取数据脚本
2024-12-06 10:15:28 +08:00
import pandas as pd
import datetime
import hashlib
import hmac
import base64
import random
import time
import string
import logging
import logging.handlers
import os
import re
import requests
class EtaReader():
def __init__(self,signature,classifylisturl,classifyidlisturl,edbcodedataurl,edbcodelist,edbdatapushurl,edbdeleteurl,edbbusinessurl):
'''
初始化 EtaReader 类的实例。
参数:
signature (str): 用于 API 请求的签名。
classifylisturl (str): 分类列表的 URL。
classifyidlisturl (str): 分类 ID 列表的 URL。
edbcodedataurl (str): EDB 代码数据的 URL。
edbdatapushurl (str): EDB 数据推送的 URL。
edbcodelist (str): EDB 代码列表的 URL。
edbdeleteurl (str): EDB 数据删除的 URL。
edbbusinessurl (str): EDB 业务数据的 URL。
返回:
None
'''
self.signature = signature
self.classifylisturl = classifylisturl
self.classifyidlisturl = classifyidlisturl
self.edbcodedataurl = edbcodedataurl
self.edbdatapushurl = edbdatapushurl
self.edbcodelist = edbcodelist
self.edbdeleteurl = edbdeleteurl
self.edbbusinessurl = edbbusinessurl
def filter_yuanyou_data(self,ClassifyName,data):
'''
指标名称保留规则
'''
# 包含 关键词 去除， 返回flase
if any(keyword in data for keyword in ['运费','检修','波动率','地缘政治','股价',
'同比','环比','环差','裂差','4WMA','变频','道琼斯','标普500','纳斯达克',
'四周均值','名占比','残差','DMA',
'连7-连9','4周平均','4周均值','滚动相关性','日本']):
return False
# 检查需要的特征
# 去掉 分析 分类下的数据
if ClassifyName == '分析':
return False
# 保留 库存中特殊关键词
if ClassifyName == '库存':
if any(keyword in data for keyword in ['原油' , '美国' ,'全球' ,'中国' ,'富查伊拉','ARA' ]):
return True
else:
pass
else:
pass
# 去掉 持仓中不是基金的数据
if ClassifyName == '持仓':
if '基金' not in data:
return False
else:
pass
else:
pass
# 去掉 航班中不是中国、美国 的数据
if ClassifyName == '需求':
if '航班' in data :
if '中国' in data or '美国' in data :
return True
else:
return False
else:
pass
else:
pass
# 分类为 期货市场，同质性数据取第一个
if ClassifyName == '期货市场':
# 去掉c1-9 以后的
if 'c1-c' in data:
try:
c = int(data.split('c1-c')[1])
except:
return False
if c > 9 :
return False
else:
pass
else:
pass
# 判断 同质性数据， 字符串开头
strstartdict = {'ICE Brent c':"ICE Brent c14",
'NYMWX WTI c':"NYMWX WTI c5",
'INE SC c':"INE SC c1",
'EFS c':"EFS c",
'Dubai Swap c':"Dubai Swap c1",
'Oman Swap c':"Oman Swap c1",
'DME Oman c':"DME Oman c1",
'Murban Futures c':"Murban Futures c1",
'Dubai连合约价格':'Dubai连1合约价格',
'美国RBOB期货月份合约价格':'美国RBOB期货2309月份合约价格',
'Brent连合约价格':'Brent连1合约价格',
'WTI连合约价格':'WTI连1合约价格',
'布伦特连合约价格':'Brent连1合约价格',
'Brent 连合约价格':'Brent连1合约价格',
'Dubai连合约价格':'Dubai连1合约价格',
'Brent连':'Brent连1合约价格',
'brent连':'Brent连1合约价格',
}
# 判断名称字符串开头是否在 strstartdict.keys中
match = re.match(r'([a-zA-Z\s]+)(\d+)', data)
if match:
part1 = match.group(1)
part2 = match.group(2)
if part1 in [i for i in strstartdict.keys()]:
if data == strstartdict[part1]:
return True
else:
return False
# data = 'Brent 连7合约价格'
# 判断名称字符串去掉数字后是否在 strstartdict.keys中
match = re.findall(r'\D+', data)
if match :
if len(match) == 2:
part1 = match[0]
part2 = match[1]
if part1+part2 in [i for i in strstartdict.keys()]:
if data == strstartdict[part1+part2]:
return True
else:
return False
else:
pass
elif len(match) == 1:
match = re.findall(r'\D+', data)
part1 = match[0]
if part1 in [i for i in strstartdict.keys()]:
if data == strstartdict[part1]:
return True
else:
return False
else:
pass
else:
pass
# 去掉kpler数据源
if 'Kpler' in ClassifyName or 'kpler' in ClassifyName:
return False
return True
def filter_pp_data(self,ClassifyName,data):
'''
指标名称保留规则
'''
# 包含 关键词 去除， 返回flase
# if any(keyword in data for keyword in ['运费','检修','波动率','地缘政治','股价',
# '同比','环比','环差','裂差','4WMA','变频','道琼斯','标普500','纳斯达克',
# '四周均值','名占比','残差','DMA',
# '连7-连9','4周平均','4周均值','滚动相关性','日本']):
# return False
# 包含 关键词 保留， 返回True
if any(keyword in data for keyword in ['拉丝']):
return True
# 检查需要的特征
# 去掉 期货市场 分类下的数据
if ClassifyName == '期货市场':
return False
else:
pass
# 保留 库存 下所有指标
if ClassifyName == '库存':
return True
else:
pass
# 保留 进出口 下所有指标
if ClassifyName == '进出口':
return True
else:
pass
# 保留 价差 下所有指标
if ClassifyName == '价差':
return True
else:
pass
# 保留 供应 下所有指标
if ClassifyName == '供应':
return True
else:
pass
# 保留 需求 下所有指标
if ClassifyName == '需求':
return True
else:
pass
return True
# 通过edbcode 获取指标数据
def edbcodegetdata(self,df,EdbCode,EdbName):
# 根据指标id，获取指标数据
url = self.edbcodedataurl+str(EdbCode)
# 发送GET请求
response = requests.get(url, headers=self.headers)
# 检查响应状态码
if response.status_code == 200:
data = response.json() # 假设接口返回的是JSON数据
all_data_items = data.get('Data')
# 列表转换为DataFrame
df3 = pd.DataFrame(all_data_items, columns=['DataTime', 'Value', 'UpdateTime'])
# df3 = pd.read_json(all_data_items, orient='records')
# 去掉UpdateTime 列
df3 = df3.drop(columns=['UpdateTime'])
# df3.set_index('DataTime')
df3.rename(columns={'Value': EdbName}, inplace=True)
# 将数据存储df1
df = pd.merge(df, df3, how='outer',on='DataTime',suffixes= ('', '_y'))
# 按时间排序
df = df.sort_values(by='DataTime', ascending=True)
return df
else:
# 请求失败，打印错误信息
logger.info(f'Error: {response.status_code}, {response.text}')
# 主动抛出异常
raise Exception(f'Error: {response.status_code}, {response.text}')
def get_eta_api_yuanyou_data(self,data_set,dataset=''):
'''
从ETA API获取原油数据
参数:
data_set (str): 数据集名称
dataset (str): 数据集ID，默认为空
返回:
None
'''
today = datetime.date.today().strftime("%Y-%m-%d")
# 定义你的headers，这里可以包含多个参数
self.headers = {
'nonce': self.signature.nonce, # 例如，一个认证令牌
'timestamp': str(self.signature.timestamp), # 自定义的header参数
'appid': self.signature.APPID, # 另一个自定义的header参数
'signature': self.signature.signature
}
# 从列表数据中获取指标名称，判断指标名称频度是否为日 ，如果是，则获取UniqueCode，然后获取指标数据，保存到xlat文件中的sheet表。
'''
df = sheetname 指标列表，存储 指标分类-指标名称-指标id-频度
df1 = sheetname 指标数据 ,存储 时间-指标名称1-指标名称2...
'''
# 构建新的DataFrame df df1
df = pd.DataFrame(columns=['指标分类', '指标名称', '指标id', '频度','指标来源','来源id','最后更新时间','更新周期','预警日期','停更周期'])
df1 = pd.DataFrame(columns=['DataTime'])
# 外网环境无法访问，请确认是否为内网环境
try:
# 发送GET请求 获取指标分类列表
response = requests.get(self.classifylisturl, headers=self.headers)
except requests.exceptions.RequestException as e:
raise Exception(f"请求失败,请确认是否为内网环境: {e}","\033[0m")
# 检查响应状态码
if response.status_code == 200:
# 获取成功， 处理响应内容
data = response.json() # 假设接口返回的是JSON数据
# 请求成功，处理响应内容
# logger.info(data.get('Data'))
# 定义你想要保留的固定值
fixed_value = 1214
# 遍历列表，只保留那些'category' key的值为固定值的数据项
filtered_data = [item for item in data.get('Data') if item.get('ParentId') == fixed_value]
#然后循环filtered_data去获取list数据，才能获取到想要获取的ClassifyId
n = 0
for item in filtered_data:
n+= 1
# if n>50:
# break
ClassifyId = item["ClassifyId"] #分类id，分类下的指标列表接口的请求参数
ClassifyName = item["ClassifyName"] #分类名称，要保存到df的指标分类列
# 根据分类id，获取指标列表
url = self.classifyidlisturl+str(ClassifyId)
response = requests.get(url, headers=self.headers)
if response.status_code == 200:
# logger.info(response.text)
data2 = response.json()
Data = data2.get('Data')
for i in Data:
# s+= 1
EdbCode = i.get('EdbCode')
EdbName = i.get('EdbName') # 指标名称，要保存到df2的指标名称列,df的指标名称列
Frequency = i.get('Frequency') # 频度，要保存到df的频度列
SourceName = i.get('SourceName') # 来源名称，要保存到df的频度列
Source = i.get('Source') # 来源ID，要保存到df的频度列
# 频度不是 日 或者 周的 跳过
if Frequency not in ['日度','周度','日','周']:
continue
# 只保留手工数据中，名称带有 海运出口 海运进口
if Source == 9 and not ('海运出口' in EdbName or '海运进口' in EdbName):
continue
# 不要wind数据
if Source == 2:
continue
# 判断名称是否需要保存
isSave = self.filter_yuanyou_data(ClassifyName,EdbName)
if isSave:
# 保存到df
df1 = self.edbcodegetdata(df1,EdbCode,EdbName)
# 取df1所有行最后一列
edbname_df = df1[['DataTime',f'{EdbName}']]
edbname_df = edbname_df.dropna()
if len(edbname_df) == 0:
logger.info(f'指标名称：{EdbName} 没有数据')
continue
try:
time_sequence = edbname_df['DataTime'].values.tolist()[-10:]
except IndexError:
time_sequence = edbname_df['DataTime'].values.tolist()
# 使用Counter来统计每个星期几的出现次数
from collections import Counter
weekday_counter = Counter(datetime.datetime.strptime(time_str, "%Y-%m-%d").strftime('%A') for time_str in time_sequence)
# 打印出现次数最多的星期几
try:
most_common_weekday = weekday_counter.most_common(1)[0][0]
# 计算两周后的日期
warning_date = (datetime.datetime.strptime(time_sequence[-1], "%Y-%m-%d") + datetime.timedelta(weeks=2)).strftime("%Y-%m-%d")
stop_update_period = (datetime.datetime.strptime(today, "%Y-%m-%d") - datetime.datetime.strptime(time_sequence[-1], "%Y-%m-%d")).days // 7
except IndexError:
most_common_weekday = '其他'
stop_update_period = 0
if '日' in Frequency:
most_common_weekday = '每天'
warning_date = (datetime.datetime.strptime(time_sequence[-1], "%Y-%m-%d") + datetime.timedelta(days=3)).strftime("%Y-%m-%d")
stop_update_period = (datetime.datetime.strptime(today, "%Y-%m-%d") - datetime.datetime.strptime(time_sequence[-1], "%Y-%m-%d")).days
# 保存频度 指标名称 分类 指标id 到 df
df2 = pd.DataFrame({'指标分类': ClassifyName, '指标名称': EdbName, '指标id': EdbCode, '频度': Frequency,'指标来源':SourceName,'来源id':Source,'最后更新时间':edbname_df['DataTime'].values[-1],'更新周期':most_common_weekday,'预警日期':warning_date,'停更周期':stop_update_period},index=[0])
# df = pd.merge(df, df2, how='outer')
df = pd.concat([df, df2])
else:
logger.info(f'跳过指标 {EdbName}')
# 找到列表中不在指标列中的指标id，保存成新的list
new_list = [item for item in self.edbcodelist if item not in df['指标id'].tolist()]
logger.info(new_list)
# 遍历new_list，获取指标数据，保存到df1
for item in new_list:
logger.info(item)
# 将item 加入到 df['指标id']中
try:
itemname = edbcodenamedict[item]
except:
itemname = item
df1 = self.edbcodegetdata(df1,item,itemname)
df = pd.concat([df, pd.DataFrame({'指标分类': '其他', '指标名称': itemname, '指标id': item, '频度': '其他','指标来源':'其他','来源id':'其他'},index=[0])])
# 按时间排序
df1.sort_values('DataTime',inplace=True,ascending=False)
df1.rename(columns={'DataTime': 'date'},inplace=True)
# df1.dropna(inplace=True)
# 去掉大于今天日期的行
df1 = df1[df1['date'] <= datetime.datetime.now().strftime('%Y-%m-%d')]
logger.info(df1.head())
# logger.info(f'{df1.head()}')
# 保存到xlsx文件的sheet表
with pd.ExcelWriter(os.path.join(dataset,data_set)) as file:
df1.to_excel(file, sheet_name='指标数据', index=False)
df.to_excel(file, sheet_name='指标列表', index=False)
df_zhibiaoshuju = df1.copy()
df_zhibiaoliebiao = df.copy()
return df_zhibiaoshuju,df_zhibiaoliebiao
def get_eta_api_pp_data(self,data_set,dataset=''):
global ClassifyId
today = datetime.date.today().strftime("%Y-%m-%d")
# 定义你的headers，这里可以包含多个参数
self.headers = {
'nonce': self.signature.nonce, # 例如，一个认证令牌
'timestamp': str(self.signature.timestamp), # 自定义的header参数
'appid': self.signature.APPID, # 另一个自定义的header参数
'signature': self.signature.signature
}
# 从列表数据中获取指标名称，判断指标名称频度是否为日 ，如果是，则获取UniqueCode，然后获取指标数据，保存到xlat文件中的sheet表。
'''
df = sheetname 指标列表，存储 指标分类-指标名称-指标id-频度
df1 = sheetname 指标数据 ,存储 时间-指标名称1-指标名称2...
'''
# 构建新的DataFrame df df1
df = pd.DataFrame(columns=['指标分类', '指标名称', '指标id', '频度'])
df1 = pd.DataFrame(columns=['DataTime'])
# 外网环境无法访问，请确认是否为内网环境
try:
# 发送GET请求 获取指标分类列表
response = requests.get(self.classifylisturl, headers=self.headers)
except requests.exceptions.RequestException as e:
raise Exception(f"请求失败,请确认是否为内网环境: {e}","\033[0m")
# 检查响应状态码
if response.status_code == 200:
# 获取成功， 处理响应内容
data = response.json() # 假设接口返回的是JSON数据
# 请求成功，处理响应内容
# logger.info(data.get('Data'))
# 定义你想要保留的固定值
fixed_value = ClassifyId
# 遍历列表，只保留那些'category' key的值为固定值的数据项
filtered_data = [item for item in data.get('Data') if item.get('ParentId') == fixed_value]
#然后循环filtered_data去获取list数据，才能获取到想要获取的ClassifyId
n = 0
for item in filtered_data:
n+= 1
# if n>50:
# break
ClassifyId = item["ClassifyId"] #分类id，分类下的指标列表接口的请求参数
ClassifyName = item["ClassifyName"] #分类名称，要保存到df的指标分类列
# 根据分类id，获取指标列表
url = self.classifyidlisturl+str(ClassifyId)
response = requests.get(url, headers=self.headers)
if response.status_code == 200:
# logger.info(response.text)
data2 = response.json()
Data = data2.get('Data')
for i in Data:
# s+= 1
EdbCode = i.get('EdbCode')
EdbName = i.get('EdbName') # 指标名称，要保存到df2的指标名称列,df的指标名称列
Frequency = i.get('Frequency') # 频度，要保存到df的频度列
# 频度不是 日 或者 周的 跳过
if Frequency not in ['日度','周度','日','周']:
continue
# 判断名称是否需要保存
isSave = self.filter_pp_data(ClassifyName,EdbName)
if isSave:
# 保存到df
# 保存频度 指标名称 分类 指标id 到 df
df2 = pd.DataFrame({'指标分类': ClassifyName, '指标名称': EdbName, '指标id': EdbCode, '频度': Frequency},index=[0])
# df = pd.merge(df, df2, how='outer')
df = pd.concat([df, df2])
df1 = self.edbcodegetdata(df1,EdbCode,EdbName)
else:
logger.info(f'跳过指标 {EdbName}')
# 找到列表中不在指标列中的指标id，保存成新的list
new_list = [item for item in self.edbcodelist if item not in df['指标id'].tolist()]
logger.info(new_list)
# 遍历new_list，获取指标数据，保存到df1
for item in new_list:
logger.info(item)
# 将item 加入到 df['指标id']中
try:
itemname = edbcodenamedict[item]
except:
itemname = item
df1 = self.edbcodegetdata(df1,item,itemname)
df = pd.concat([df, pd.DataFrame({'指标分类': '其他', '指标名称': itemname, '指标id': item, '频度': '其他'},index=[0])])
# 按时间排序
df1.sort_values('DataTime',inplace=True,ascending=False)
df1.rename(columns={'DataTime': 'date'},inplace=True)
# df1.dropna(inplace=True)
# 去掉大于今天日期的行
df1 = df1[df1['date'] <= datetime.datetime.now().strftime('%Y-%m-%d')]
logger.info(df1.head())
# logger.info(f'{df1.head()}')
# 保存到xlsx文件的sheet表
with pd.ExcelWriter(os.path.join(dataset,data_set)) as file:
df1.to_excel(file, sheet_name='指标数据', index=False)
df.to_excel(file, sheet_name='指标列表', index=False)
df_zhibiaoshuju = df1.copy()
df_zhibiaoliebiao = df.copy()
return df_zhibiaoshuju,df_zhibiaoliebiao
def push_data(self,data):
today = datetime.date.today().strftime("%Y-%m-%d")
# 定义你的headers，这里可以包含多个参数
self.headers = {
'nonce': self.signature.nonce, # 例如，一个认证令牌
'timestamp': str(self.signature.timestamp), # 自定义的header参数
'appid': self.signature.APPID, # 另一个自定义的header参数
'signature': self.signature.signature
}
# 发送post请求 上传数据
logger.info('请求参数：',data)
response = requests.post(self.edbdatapushurl, headers=self.headers,data=json.dumps(data))
# 检查响应状态码
if response.status_code == 200:
data = response.json() # 假设接口返回的是JSON数据
logger.info('上传成功，响应为：', data)
else:
# 请求失败，打印错误信息
logger.info(f'Error: {response.status_code}, {response.text}')
# 主动抛出异常
raise Exception(f'Error: {response.status_code}, {response.text}')
def del_zhibiao(self,IndexCodeList):
today = datetime.date.today().strftime("%Y-%m-%d")
# 定义你的headers，这里可以包含多个参数
self.headers = {
'nonce': self.signature.nonce, # 例如，一个认证令牌
'timestamp': str(self.signature.timestamp), # 自定义的header参数
'appid': self.signature.APPID, # 另一个自定义的header参数
'signature': self.signature.signature
}
data = {
"IndexCodeList": IndexCodeList #指标编码列表
}
# 发送post请求 上传数据
response = requests.post(self.edbdeleteurl, headers=self.headers,data=json.dumps(data))
# 检查响应状态码
if response.status_code == 200:
data = response.json() # 假设接口返回的是JSON数据
logger.info('删除成功，响应为：', data)
else:
# 请求失败，打印错误信息
logger.info(f'Error: {response.status_code}, {response.text}')
# 主动抛出异常
raise Exception(f'Error: {response.status_code}, {response.text}')
def del_business(self,data):
''''
接口地址
https://console-docs.apipost.cn/preview/fce869601d0be1d9/9a637c2f9ed0c589?target_id=d3cafcbf-a68c-42b3-b105-7bbd0e95a9cd
请求体 body
{
"IndexCode": "W001067", //指标编码
"StartDate": "2020-04-20", //指标需要删除的开始日期（>=），如果开始日期和结束日期相等，那么就是删除该日期
"EndDate": "2024-05-28" //指标需要删除的结束日期（<=），如果开始日期和结束日期相等，那么就是删除该日期
}
'''
today = datetime.date.today().strftime("%Y-%m-%d")
# 定义你的headers，这里可以包含多个参数
self.headers = {
'nonce': self.signature.nonce, # 例如，一个认证令牌
'timestamp': str(self.signature.timestamp), # 自定义的header参数
'appid': self.signature.APPID, # 另一个自定义的header参数
'signature': self.signature.signature
}
# 发送post请求 上传数据
response = requests.post(self.edbbusinessurl, headers=self.headers,data=json.dumps(data))
# 检查响应状态码
if response.status_code == 200:
data = response.json() # 假设接口返回的是JSON数据
logger.info('删除成功，响应为：', data)
else:
# 请求失败，打印错误信息
logger.info(f'Error: {response.status_code}, {response.text}')
# 主动抛出异常
raise Exception(f'Error: {response.status_code}, {response.text}')
class BinanceAPI:
'''
获取 Binance API 请求头签名
'''
def __init__(self, APPID, SECRET):
self.APPID = APPID
self.SECRET = SECRET
self.get_signature()
# 生成随机字符串作为 nonce
def generate_nonce(self, length=32):
self.nonce = ''.join(random.choices(string.ascii_letters + string.digits, k=length))
return self.nonce
# 获取当前时间戳（秒）
def get_timestamp(self):
return int(time.time())
# 构建待签名字符串
def build_sign_str(self):
return f'appid={self.APPID}&nonce={self.nonce}&timestamp={self.timestamp}'
# 使用 HMAC SHA-256 计算签名
def calculate_signature(self, secret, message):
return base64.urlsafe_b64encode(hmac.new(secret.encode('utf-8'), message.encode('utf-8'), hashlib.sha256).digest()).decode('utf-8')
def get_signature(self):
# 调用上述方法生成签名
self.nonce = self.generate_nonce()
self.timestamp = self.get_timestamp()
self.sign_str = self.build_sign_str()
self.signature = self.calculate_signature(self.SECRET, self.sign_str)
# return self.signature
### 日志配置
# 创建日志目录（如果不存在）
log_dir = 'logs'
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# 配置日志记录器
logger = logging.getLogger('pricepredict')
logger.setLevel(logging.INFO)
# 配置文件处理器，将日志记录到文件
file_handler = logging.handlers.RotatingFileHandler(os.path.join(log_dir, 'pricepredict.log'), maxBytes=1024 * 1024, backupCount=5)
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
# 配置控制台处理器，将日志打印到控制台
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter('%(message)s'))
# 将处理器添加到日志记录器
logger.addHandler(file_handler)
logger.addHandler(console_handler)
# eta 接口url
sourcelisturl = 'http://10.189.2.78:8108/v1/edb/source/list'
classifylisturl = 'http://10.189.2.78:8108/v1/edb/classify/list?ClassifyType='
uniquecodedataurl = 'http://10.189.2.78:8108/v1/edb/data?UniqueCode=4991c37becba464609b409909fe4d992&StartDate=2024-02-01'
classifyidlisturl = 'http://10.189.2.78:8108/v1/edb/list?ClassifyId='
edbcodedataurl = 'http://10.189.2.78:8108/v1/edb/data?EdbCode='
edbdatapushurl = 'http://10.189.2.78:8108/v1/edb/push'
edbdeleteurl = 'http://10.189.2.78:8108/v1/edb/business/edb/del'
edbbusinessurl = 'http://10.189.2.78:8108/v1/edb/business/data/del'
edbcodelist = ['CO1 Comdty', 'ovx index', 'C2404194834', 'C2404199738', 'dxy curncy', 'C2403128043', 'C2403150124',
'DOESCRUD Index', 'WTRBM1 EEGC Index', 'FVHCM1 INDEX', 'doedtprd index', 'CFFDQMMN INDEX',
'C2403083739', 'C2404167878', 'C2403250571', 'lmcads03 lme comdty', 'GC1 COMB Comdty',
'C2404171822','C2404167855',
# 'W000825','W000826','G.IPE', # 美国汽柴油
# 'S5131019','ID00135604','FSGAM1 Index','S5120408','ID00136724', # 新加坡汽柴油
]
预生产数据库配置
2024-12-11 14:52:45 +08:00
# eta自有数据指标编码
modelsindex = {
'NHITS': 'SELF0000001',
'Informer':'SELF0000057',
'LSTM':'SELF0000058',
'iTransformer':'SELF0000059',
'TSMixer':'SELF0000060',
'TSMixerx':'SELF0000061',
'PatchTST':'SELF0000062',
'RNN':'SELF0000063',
'GRU':'SELF0000064',
'TCN':'SELF0000065',
'BiTCN':'SELF0000066',
'DilatedRNN':'SELF0000067',
'MLP':'SELF0000068',
'DLinear':'SELF0000069',
'NLinear':'SELF0000070',
'TFT':'SELF0000071',
'FEDformer':'SELF0000072',
'StemGNN':'SELF0000073',
'MLPMultivariate':'SELF0000074',
'TiDE':'SELF0000075',
'DeepNPTS':'SELF0000076'
}
edbcodelist = edbcodelist+list(modelsindex.values())
整理eta获取数据脚本
2024-12-06 10:15:28 +08:00
### 文件
预生产数据库配置
2024-12-11 14:52:45 +08:00
data_set = '历史预测结果.xlsx' # 数据集文件
整理eta获取数据脚本
2024-12-06 10:15:28 +08:00
# data_set = 'INE_OIL(1).csv'
### 文件夹
dataset = 'yuanyoudataset' # 数据集文件夹
# eta 接口token
APPID = "XNLDvxZHHugj7wJ7"
SECRET = "iSeU4s6cKKBVbt94htVY1p0sqUMqb2xa"
signature = BinanceAPI(APPID, SECRET)
etadata = EtaReader(signature=signature,
classifylisturl=classifylisturl,
classifyidlisturl=classifyidlisturl,
edbcodedataurl=edbcodedataurl,
edbcodelist=edbcodelist,
edbdatapushurl=edbdatapushurl,
edbdeleteurl=edbdeleteurl,
edbbusinessurl=edbbusinessurl,
)
df_zhibiaoshuju, df_zhibiaoliebiao = etadata.get_eta_api_yuanyou_data(data_set=data_set, dataset=dataset) # 原始数据，未处理
Reference in New Issue Copy Permalink