From 765ca10b5b34efed0582f06017cc55fc3ba452a6 Mon Sep 17 00:00:00 2001 From: workpc Date: Thu, 6 Mar 2025 14:59:18 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8E=9F=E6=B2=B9=E6=9C=88=E5=BA=A6=E8=B0=83?= =?UTF-8?q?=E8=AF=95=E9=80=9A=E8=BF=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config_jingbo_yuedu.py | 273 +++++++++++++------------- lib/dataread.py | 4 +- main_yuanyou_yuedu.py | 274 +++++++++++++++++--------- main_yuanyou_zhoudu.py | 428 ++++++++++++++++++++--------------------- 4 files changed, 531 insertions(+), 448 deletions(-) diff --git a/config_jingbo_yuedu.py b/config_jingbo_yuedu.py index 04c7e92..6693283 100644 --- a/config_jingbo_yuedu.py +++ b/config_jingbo_yuedu.py @@ -2,7 +2,7 @@ import logging import os import logging.handlers import datetime -from lib.tools import MySQLDB,SQLiteHandler +from lib.tools import MySQLDB, SQLiteHandler # eta 接口token @@ -10,66 +10,65 @@ APPID = "XNLDvxZHHugj7wJ7" SECRET = "iSeU4s6cKKBVbt94htVY1p0sqUMqb2xa" # eta 接口url -sourcelisturl = 'http://10.189.2.78:8108/v1/edb/source/list' +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' +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', # 新加坡汽柴油 - ] + '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', # 新加坡汽柴油 + ] # 临时写死用指定的列,与上面的edbcode对应，后面更改 edbnamelist = [ - 'ds','y', - 'Brent c1-c6','Brent c1-c3','Brent-WTI','美国商业原油库存', - 'DFL','美国汽油裂解价差','ovx index','dxy curncy','lmcads03 lme comdty', - 'C2403128043','C2403150124','FVHCM1 INDEX','doedtprd index','CFFDQMMN INDEX', - 'C2403083739','C2404167878', - 'GC1 COMB Comdty','C2404167855', + 'ds', 'y', + 'Brent c1-c6', 'Brent c1-c3', 'Brent-WTI', '美国商业原油库存', + 'DFL', '美国汽油裂解价差', 'ovx index', 'dxy curncy', 'lmcads03 lme comdty', + 'C2403128043', 'C2403150124', 'FVHCM1 INDEX', 'doedtprd index', 'CFFDQMMN INDEX', + 'C2403083739', 'C2404167878', + 'GC1 COMB Comdty', 'C2404167855', # 'A汽油价格','W000826','ICE柴油价格', # '新加坡(含硫0.05%) 柴油现货价','柴油：10ppm：国际市场：FOB中间价：新加坡（日）','Bloomberg Commodity Fair Value Singapore Mogas 92 Swap Month 1','97#汽油FOB新加坡现货价','无铅汽油：97#：国际市场：FOB中间价：新加坡（日）' - ] - +] # 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' - } + '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' +} # eta 上传预测结果的请求体，后面发起请求的时候更改 model datalist 数据 data = { - "IndexCode": "", - "IndexName": "价格预测模型", - "Unit": "无", + "IndexCode": "", + "IndexName": "价格预测模型", + "Unit": "无", "Frequency": "日度", "SourceName": f"价格预测", "Remark": 'ddd', @@ -79,19 +78,18 @@ data = { "Value": 333444 } ] - } +} # eta 分类 # level：3才可以获取到数据，所以需要人工把能源化工下所有的level3级都找到 - # url = 'http://10.189.2.78:8108/v1/edb/list?ClassifyId=1214' - #ParentId ":1160, 能源化工 - # ClassifyId ":1214,原油 - #ParentId ":1214,"，就是原油下所有的数据。 -ClassifyId = 1214 +# url = 'http://10.189.2.78:8108/v1/edb/list?ClassifyId=1214' +# ParentId ":1160, 能源化工 +# ClassifyId ":1214,原油 +# ParentId ":1214,"，就是原油下所有的数据。 +ClassifyId = 1214 - -############################################################################################################### 变量定义--测试环境 +# 变量定义--测试环境 server_host = '192.168.100.53' login_pushreport_url = f"http://{server_host}:8080/jingbo-dev/api/server/login" @@ -103,7 +101,7 @@ login_data = { "data": { "account": "api_test", # "password": "MmVmNzNlOWI0MmY0ZDdjZGUwNzE3ZjFiMDJiZDZjZWU=", # Shihua@123456 - "password": "ZTEwYWRjMzk0OWJhNTlhYmJlNTZlMDU3ZjIwZjg4M2U=", # 123456 + "password": "ZTEwYWRjMzk0OWJhNTlhYmJlNTZlMDU3ZjIwZjg4M2U=", # 123456 "tenantHashCode": "8a4577dbd919675758d57999a1e891fe", "terminal": "API" }, @@ -112,39 +110,39 @@ login_data = { } upload_data = { - "funcModule":'研究报告信息', - "funcOperation":'上传原油价格预测报告', - "data":{ - "ownerAccount":'arui', #报告所属用户账号 - "reportType":'OIL_PRICE_FORECAST', # 报告类型，固定为OIL_PRICE_FORECAST - "fileName": '2000-40-5-50--100-原油指标数据.xlsx-Brent活跃合约--2024-09-06-15-01-29-预测报告.pdf', #文件名称 - "fileBase64": '' ,#文件内容base64 - "categoryNo":'yyjgycbg', # 研究报告分类编码 - "smartBusinessClassCode":'YCJGYCBG', #分析报告分类编码 - "reportEmployeeCode":"E40116", # 报告人 - "reportDeptCode" :"D0044" ,# 报告部门 - "productGroupCode":"RAW_MATERIAL" # 商品分类 - } + "funcModule": '研究报告信息', + "funcOperation": '上传原油价格预测报告', + "data": { + "ownerAccount": 'arui', # 报告所属用户账号 + "reportType": 'OIL_PRICE_FORECAST', # 报告类型，固定为OIL_PRICE_FORECAST + "fileName": '2000-40-5-50--100-原油指标数据.xlsx-Brent活跃合约--2024-09-06-15-01-29-预测报告.pdf', # 文件名称 + "fileBase64": '', # 文件内容base64 + "categoryNo": 'yyjgycbg', # 研究报告分类编码 + "smartBusinessClassCode": 'YCJGYCBG', # 分析报告分类编码 + "reportEmployeeCode": "E40116", # 报告人 + "reportDeptCode": "D0044", # 报告部门 + "productGroupCode": "RAW_MATERIAL" # 商品分类 + } } warning_data = { - "funcModule":'原油特征停更预警', - "funcOperation":'原油特征停更预警', - "data":{ - 'WARNING_TYPE_NAME':'特征数据停更预警', - 'WARNING_CONTENT':'', - 'WARNING_DATE':'' - } + "funcModule": '原油特征停更预警', + "funcOperation": '原油特征停更预警', + "data": { + 'WARNING_TYPE_NAME': '特征数据停更预警', + 'WARNING_CONTENT': '', + 'WARNING_DATE': '' + } } query_data_list_item_nos_data = { - "funcModule": "数据项", - "funcOperation": "查询", + "funcModule": "数据项", + "funcOperation": "查询", "data": { - "dateStart":"20200101", - "dateEnd":"20241231", - "dataItemNoList":["Brentzdj","Brentzgj"] # 数据项编码,代表 brent最低价和最高价 + "dateStart": "20200101", + "dateEnd": "20241231", + "dataItemNoList": ["Brentzdj", "Brentzgj"] # 数据项编码,代表 brent最低价和最高价 } } @@ -152,96 +150,96 @@ query_data_list_item_nos_data = { # 北京环境数据库 host = '192.168.101.27' port = 3306 -dbusername ='root' +dbusername = 'root' password = '123456' dbname = 'jingbo_test' table_name = 'v_tbl_crude_oil_warning' -### 开关 -is_train = False # 是否训练 -is_debug = False # 是否调试 -is_eta = False # 是否使用eta接口 -is_market = True # 是否通过市场信息平台获取特征 ,在is_eta 为true 的情况下生效 -is_timefurture = True # 是否使用时间特征 -is_fivemodels = False # 是否使用之前保存的最佳的5个模型 -is_edbcode = False # 特征使用edbcoding列表中的 -is_edbnamelist = False # 自定义特征，对应上面的edbnamelist -is_update_eta = False # 预测结果上传到eta -is_update_report = False # 是否上传报告 -is_update_warning_data = False # 是否上传预警数据 -is_del_corr = 0.6 # 是否删除相关性高的特征,取值为 0-1 ，0 为不删除，0.6 表示删除相关性小于0.6的特征 -is_del_tow_month = True # 是否删除两个月不更新的特征 - +# 开关 +is_train = False # 是否训练 +is_debug = True # 是否调试 +is_eta = False # 是否使用eta接口 +is_market = True # 是否通过市场信息平台获取特征 ,在is_eta 为true 的情况下生效 +is_timefurture = True # 是否使用时间特征 +is_fivemodels = False # 是否使用之前保存的最佳的5个模型 +is_edbcode = False # 特征使用edbcoding列表中的 +is_edbnamelist = False # 自定义特征，对应上面的edbnamelist +is_update_eta = False # 预测结果上传到eta +is_update_report = False # 是否上传报告 +is_update_warning_data = False # 是否上传预警数据 +is_del_corr = 0.6 # 是否删除相关性高的特征,取值为 0-1 ，0 为不删除，0.6 表示删除相关性小于0.6的特征 +is_del_tow_month = True # 是否删除两个月不更新的特征 # 连接到数据库 -db_mysql = MySQLDB(host=host, user=dbusername, password=password, database=dbname) +db_mysql = MySQLDB(host=host, user=dbusername, + password=password, database=dbname) db_mysql.connect() -print("数据库连接成功",host,dbname,dbusername) +print("数据库连接成功", host, dbname, dbusername) # 数据截取日期 -start_year = 1993 # 数据开始年份 -end_time = '' # 数据截取日期 -freq = 'M' # 时间频率，"D": 天 "W": 周"M": 月"Q": 季度"A": 年 "H": 小时 "T": 分钟 "S": 秒 "B": 工作日 -delweekenday = True if freq == 'B' else False # 是否删除周末数据 -is_corr = False # 特征是否参与滞后领先提升相关系数 -add_kdj = False # 是否添加kdj指标 +start_year = 2005 # 数据开始年份 +end_time = '' # 数据截取日期 +freq = 'M' # 时间频率，"D": 天 "W": 周"M": 月"Q": 季度"A": 年 "H": 小时 "T": 分钟 "S": 秒 "B": 工作日 "WW" 自定义周 +delweekenday = True if freq == 'B' else False # 是否删除周末数据 +is_corr = False # 特征是否参与滞后领先提升相关系数 +add_kdj = False # 是否添加kdj指标 if add_kdj and is_edbnamelist: - edbnamelist = edbnamelist+['K','D','J'] + edbnamelist = edbnamelist+['K', 'D', 'J'] -### 模型参数 -y = 'Brent连1合约价格' # 原油指标数据的目标变量 Brent连1合约价格 Brent活跃合约 -horizon =3 # 预测的步长 -input_size = 9 # 输入序列长度 -train_steps = 50 if is_debug else 1000 # 训练步数,用来限定epoch次数 +# 模型参数 +y = 'Brent连1合约价格' # 原油指标数据的目标变量 Brent连1合约价格 Brent活跃合约 +horizon = 4 # 预测的步长 +input_size = 16 # 输入序列长度 +train_steps = 50 if is_debug else 1000 # 训练步数,用来限定epoch次数 val_check_steps = 30 # 评估频率 -early_stop_patience_steps = 5 # 早停的耐心步数 +early_stop_patience_steps = 5 # 早停的耐心步数 # --- 交叉验证用的参数 test_size = 100 # 测试集大小，定义100，后面使用的时候重新赋值 -val_size = test_size # 验证集大小,同测试集大小 +val_size = test_size # 验证集大小,同测试集大小 -### 特征筛选用到的参数 -k = 100 # 特征筛选数量，如果是0或者值比特征数量大，代表全部特征 -corr_threshold = 0.6 # 相关性大于0.6的特征 -rote = 0.06 # 绘图上下界阈值 +# 特征筛选用到的参数 +k = 100 # 特征筛选数量，如果是0或者值比特征数量大，代表全部特征 +corr_threshold = 0.6 # 相关性大于0.6的特征 +rote = 0.06 # 绘图上下界阈值 -### 计算准确率 -weight_dict = [0.4,0.15,0.1,0.1,0.25] # 权重 +# 计算准确率 +weight_dict = [0.4, 0.15, 0.1, 0.1, 0.25] # 权重 -### 文件 -data_set = '原油指标数据.xlsx' # 数据集文件 -dataset = 'yuanyouyuedudataset' # 数据集文件夹 +# 文件 +data_set = '原油指标数据.xlsx' # 数据集文件 +dataset = 'yuanyouyuedudataset' # 数据集文件夹 # 数据库名称 -db_name = os.path.join(dataset,'jbsh_yuanyou_yuedu.db') -sqlitedb = SQLiteHandler(db_name) +db_name = os.path.join(dataset, 'jbsh_yuanyou_yuedu.db') +sqlitedb = SQLiteHandler(db_name) sqlitedb.connect() -settings = f'{input_size}-{horizon}-{train_steps}--{k}-{data_set}-{y}' +settings = f'{input_size}-{horizon}-{train_steps}--{k}-{data_set}-{y}' # 获取日期时间 # now = datetime.datetime.now().strftime('%Y%m%d%H%M%S') # 获取当前日期时间 -now = datetime.datetime.now().strftime('%Y-%m-%d') # 获取当前日期时间 -reportname = f'Brent原油大模型预测--{end_time}.pdf' # 报告文件名 -reportname = reportname.replace(':', '-') # 替换冒号 +now = datetime.datetime.now().strftime('%Y-%m-%d') # 获取当前日期时间 +reportname = f'Brent原油大模型月度预测--{end_time}.pdf' # 报告文件名 +reportname = reportname.replace(':', '-') # 替换冒号 if end_time == '': end_time = now -### 邮件配置 -username='1321340118@qq.com' -passwd='wgczgyhtyyyyjghi' +# 邮件配置 +username = '1321340118@qq.com' +passwd = 'wgczgyhtyyyyjghi' # recv=['liurui_test@163.com','52585119@qq.com'] -recv=['liurui_test@163.com','jin.wang@chambroad.com'] +recv = ['liurui_test@163.com', 'jin.wang@chambroad.com'] # recv=['liurui_test@163.com'] -title='reportname' -content='brent价格预测报告请看附件' -file=os.path.join(dataset,'reportname') +title = 'reportname' +content = 'brent价格预测报告请看附件' +file = os.path.join(dataset, 'reportname') # file=os.path.join(dataset,'14-7-50--100-原油指标数据.xlsx-Brent连1合约价格--20240731175936-预测报告.pdf') -ssl=True +ssl = True -### 日志配置 +# 日志配置 # 创建日志目录（如果不存在） log_dir = 'logs' @@ -253,8 +251,10 @@ logger = logging.getLogger('my_logger') 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')) +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() @@ -265,4 +265,3 @@ logger.addHandler(file_handler) logger.addHandler(console_handler) # logger.info('当前配置：'+settings) - diff --git a/lib/dataread.py b/lib/dataread.py index 6d894b9..f3a385d 100644 --- a/lib/dataread.py +++ b/lib/dataread.py @@ -838,7 +838,9 @@ def datachuli(df_zhibiaoshuju, df_zhibiaoliebiao, datecol='date', end_time='', y df = df.resample('W', on='ds').mean().reset_index() elif config.freq == 'M': # 按月取样 - df = df.resample('M', on='ds').mean().reset_index() + if 'yearmonthweeks' in df.columns: + df.drop('yearmonthweeks', axis=1, inplace=True) + df = df.resample('ME', on='ds').mean().reset_index() # 删除预测列空值的行 ''' 工作日缺失，如果删除，会影响预测结果，导致统计准确率出错 ''' # df = df.dropna(subset=['y']) diff --git a/main_yuanyou_yuedu.py b/main_yuanyou_yuedu.py index c41ad85..22c48e4 100644 --- a/main_yuanyou_yuedu.py +++ b/main_yuanyou_yuedu.py @@ -1,12 +1,66 @@ # 读取配置 -from lib.dataread import * -from lib.tools import SendMail,exception_logger -from models.nerulforcastmodels import ex_Model,model_losss,model_losss_juxiting,brent_export_pdf,tansuanli_export_pdf,pp_export_pdf,model_losss_juxiting -import glob +from lib.dataread import * +from config_jingbo_yuedu import * +from lib.tools import SendMail, exception_logger +from models.nerulforcastmodels import ex_Model, model_losss, model_losss_juxiting, brent_export_pdf, tansuanli_export_pdf, pp_export_pdf, model_losss_juxiting +import datetime import torch torch.set_float32_matmul_precision("high") +global_config.update({ + # 核心参数 + 'logger': logger, + 'dataset': dataset, + 'y': y, + 'is_debug': is_debug, + 'is_train': is_train, + 'is_fivemodels': is_fivemodels, + 'settings': settings, + + + # 模型参数 + 'data_set': data_set, + 'input_size': input_size, + 'horizon': horizon, + 'train_steps': train_steps, + 'val_check_steps': val_check_steps, + 'val_size': val_size, + 'test_size': test_size, + 'modelsindex': modelsindex, + 'rote': rote, + + # 特征工程开关 + 'is_del_corr': is_del_corr, + 'is_del_tow_month': is_del_tow_month, + 'is_eta': is_eta, + 'is_update_eta': is_update_eta, + 'early_stop_patience_steps': early_stop_patience_steps, + + # 时间参数 + 'start_year': start_year, + 'end_time': end_time or datetime.datetime.now().strftime("%Y-%m-%d"), + 'freq': freq, # 保持列表结构 + + # 接口配置 + 'login_pushreport_url': login_pushreport_url, + 'login_data': login_data, + 'upload_url': upload_url, + 'upload_warning_url': upload_warning_url, + 'warning_data': warning_data, + + # 查询接口 + 'query_data_list_item_nos_url': query_data_list_item_nos_url, + 'query_data_list_item_nos_data': query_data_list_item_nos_data, + + # eta 配置 + 'APPID': APPID, + 'SECRET': SECRET, + 'etadata': data, + + # 数据库配置 + 'sqlitedb': sqlitedb, +}) def predict_main(): @@ -72,7 +126,8 @@ def predict_main(): edbdeleteurl=edbdeleteurl, edbbusinessurl=edbbusinessurl, ) - df_zhibiaoshuju, df_zhibiaoliebiao = etadata.get_eta_api_yuanyou_data(data_set=data_set, dataset=dataset) # 原始数据，未处理 + df_zhibiaoshuju, df_zhibiaoliebiao = etadata.get_eta_api_yuanyou_data( + data_set=data_set, dataset=dataset) # 原始数据，未处理 if is_market: logger.info('从市场信息平台获取数据...') @@ -83,26 +138,26 @@ def predict_main(): df_zhibiaoshuju = get_high_low_data(df_zhibiaoshuju) else: logger.info('从市场信息平台获取数据') - df_zhibiaoshuju = get_market_data(end_time,df_zhibiaoshuju) - - except : + df_zhibiaoshuju = get_market_data( + end_time, df_zhibiaoshuju) + + except: logger.info('最高最低价拼接失败') - + # 保存到xlsx文件的sheet表 - with pd.ExcelWriter(os.path.join(dataset,data_set)) as file: + with pd.ExcelWriter(os.path.join(dataset, data_set)) as file: df_zhibiaoshuju.to_excel(file, sheet_name='指标数据', index=False) df_zhibiaoliebiao.to_excel(file, sheet_name='指标列表', index=False) - - + # 数据处理 df = datachuli(df_zhibiaoshuju, df_zhibiaoliebiao, y=y, dataset=dataset, add_kdj=add_kdj, is_timefurture=is_timefurture, - end_time=end_time) + end_time=end_time) else: # 读取数据 logger.info('读取本地数据：' + os.path.join(dataset, data_set)) - df,df_zhibiaoliebiao = getdata(filename=os.path.join(dataset, data_set), y=y, dataset=dataset, add_kdj=add_kdj, - is_timefurture=is_timefurture, end_time=end_time) # 原始数据，未处理 + df, df_zhibiaoliebiao = getdata(filename=os.path.join(dataset, data_set), y=y, dataset=dataset, add_kdj=add_kdj, + is_timefurture=is_timefurture, end_time=end_time) # 原始数据，未处理 # 更改预测列名称 df.rename(columns={y: 'y'}, inplace=True) @@ -124,48 +179,65 @@ def predict_main(): else: for row in first_row.itertuples(index=False): row_dict = row._asdict() - row_dict['ds'] = row_dict['ds'].strftime('%Y-%m-%d %H:%M:%S') - check_query = sqlitedb.select_data('trueandpredict', where_condition=f"ds = '{row.ds}'") + config.logger.info(f'要保存的真实值：{row_dict}') + # 判断ds是否为字符串类型,如果不是则转换为字符串类型 + if isinstance(row_dict['ds'], (pd.Timestamp, datetime.datetime)): + row_dict['ds'] = row_dict['ds'].strftime('%Y-%m-%d') + elif not isinstance(row_dict['ds'], str): + try: + row_dict['ds'] = pd.to_datetime( + row_dict['ds']).strftime('%Y-%m-%d') + except: + logger.warning(f"无法解析的时间格式: {row_dict['ds']}") + # row_dict['ds'] = row_dict['ds'].strftime('%Y-%m-%d') + # row_dict['ds'] = row_dict['ds'].strftime('%Y-%m-%d %H:%M:%S') + check_query = sqlitedb.select_data( + 'trueandpredict', where_condition=f"ds = '{row.ds}'") if len(check_query) > 0: - set_clause = ", ".join([f"{key} = '{value}'" for key, value in row_dict.items()]) - sqlitedb.update_data('trueandpredict', set_clause, where_condition=f"ds = '{row.ds}'") + set_clause = ", ".join( + [f"{key} = '{value}'" for key, value in row_dict.items()]) + sqlitedb.update_data( + 'trueandpredict', set_clause, where_condition=f"ds = '{row.ds}'") continue - sqlitedb.insert_data('trueandpredict', tuple(row_dict.values()), columns=row_dict.keys()) + sqlitedb.insert_data('trueandpredict', tuple( + row_dict.values()), columns=row_dict.keys()) # 更新accuracy表的y值 if not sqlitedb.check_table_exists('accuracy'): pass else: - update_y = sqlitedb.select_data('accuracy',where_condition="y is null") + update_y = sqlitedb.select_data( + 'accuracy', where_condition="y is null") if len(update_y) > 0: logger.info('更新accuracy表的y值') # 找到update_y 中ds且df中的y的行 - update_y = update_y[update_y['ds']<=end_time] - + update_y = update_y[update_y['ds'] <= end_time] logger.info(f'要更新y的信息：{update_y}') # try: for row in update_y.itertuples(index=False): try: - row_dict = row._asdict() - yy = df[df['ds']==row_dict['ds']]['y'].values[0] - LOW = df[df['ds']==row_dict['ds']]['Brentzdj'].values[0] - HIGH = df[df['ds']==row_dict['ds']]['Brentzgj'].values[0] - sqlitedb.update_data('accuracy', f"y = {yy},LOW_PRICE = {LOW},HIGH_PRICE = {HIGH}", where_condition=f"ds = '{row_dict['ds']}'") + row_dict = row._asdict() + yy = df[df['ds'] == row_dict['ds']]['y'].values[0] + LOW = df[df['ds'] == row_dict['ds']]['Brentzdj'].values[0] + HIGH = df[df['ds'] == row_dict['ds']]['Brentzgj'].values[0] + sqlitedb.update_data( + 'accuracy', f"y = {yy},LOW_PRICE = {LOW},HIGH_PRICE = {HIGH}", where_condition=f"ds = '{row_dict['ds']}'") except: logger.info(f'更新accuracy表的y值失败：{row_dict}') # except Exception as e: # logger.info(f'更新accuracy表的y值失败：{e}') - import datetime # 判断当前日期是不是周一 is_weekday = datetime.datetime.now().weekday() == 0 if is_weekday: logger.info('今天是周一，更新预测模型') # 计算最近60天预测残差最低的模型名称 - model_results = sqlitedb.select_data('trueandpredict', order_by="ds DESC", limit="60") + model_results = sqlitedb.select_data( + 'trueandpredict', order_by="ds DESC", limit="60") # 删除空值率为90%以上的列 if len(model_results) > 10: - model_results = model_results.dropna(thresh=len(model_results)*0.1,axis=1) + model_results = model_results.dropna( + thresh=len(model_results)*0.1, axis=1) # 删除空行 model_results = model_results.dropna() modelnames = model_results.columns.to_list()[2:-1] @@ -173,47 +245,59 @@ def predict_main(): model_results[col] = model_results[col].astype(np.float32) # 计算每个预测值与真实值之间的偏差率 for model in modelnames: - model_results[f'{model}_abs_error_rate'] = abs(model_results['y'] - model_results[model]) / model_results['y'] + model_results[f'{model}_abs_error_rate'] = abs( + model_results['y'] - model_results[model]) / model_results['y'] # 获取每行对应的最小偏差率值 - min_abs_error_rate_values = model_results.apply(lambda row: row[[f'{model}_abs_error_rate' for model in modelnames]].min(), axis=1) + min_abs_error_rate_values = model_results.apply( + lambda row: row[[f'{model}_abs_error_rate' for model in modelnames]].min(), axis=1) # 获取每行对应的最小偏差率值对应的列名 - min_abs_error_rate_column_name = model_results.apply(lambda row: row[[f'{model}_abs_error_rate' for model in modelnames]].idxmin(), axis=1) + min_abs_error_rate_column_name = model_results.apply( + lambda row: row[[f'{model}_abs_error_rate' for model in modelnames]].idxmin(), axis=1) # 将列名索引转换为列名 - min_abs_error_rate_column_name = min_abs_error_rate_column_name.map(lambda x: x.split('_')[0]) + min_abs_error_rate_column_name = min_abs_error_rate_column_name.map( + lambda x: x.split('_')[0]) # 取出现次数最多的模型名称 most_common_model = min_abs_error_rate_column_name.value_counts().idxmax() logger.info(f"最近60天预测残差最低的模型名称：{most_common_model}") # 保存结果到数据库 if not sqlitedb.check_table_exists('most_model'): - sqlitedb.create_table('most_model', columns="ds datetime, most_common_model TEXT") - sqlitedb.insert_data('most_model', (datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), most_common_model,), columns=('ds', 'most_common_model',)) + sqlitedb.create_table( + 'most_model', columns="ds datetime, most_common_model TEXT") + sqlitedb.insert_data('most_model', (datetime.datetime.now().strftime( + '%Y-%m-%d %H:%M:%S'), most_common_model,), columns=('ds', 'most_common_model',)) try: if is_weekday: - # if True: + # if True: logger.info('今天是周一，发送特征预警') # 上传预警信息到数据库 warning_data_df = df_zhibiaoliebiao.copy() - warning_data_df = warning_data_df[warning_data_df['停更周期']> 3 ][['指标名称', '指标id', '频度','更新周期','指标来源','最后更新时间','停更周期']] + warning_data_df = warning_data_df[warning_data_df['停更周期'] > 3][[ + '指标名称', '指标id', '频度', '更新周期', '指标来源', '最后更新时间', '停更周期']] # 重命名列名 - warning_data_df = warning_data_df.rename(columns={'指标名称': 'INDICATOR_NAME', '指标id': 'INDICATOR_ID', '频度': 'FREQUENCY', '更新周期': 'UPDATE_FREQUENCY', '指标来源': 'DATA_SOURCE', '最后更新时间': 'LAST_UPDATE_DATE', '停更周期': 'UPDATE_SUSPENSION_CYCLE'}) + warning_data_df = warning_data_df.rename(columns={'指标名称': 'INDICATOR_NAME', '指标id': 'INDICATOR_ID', '频度': 'FREQUENCY', + '更新周期': 'UPDATE_FREQUENCY', '指标来源': 'DATA_SOURCE', '最后更新时间': 'LAST_UPDATE_DATE', '停更周期': 'UPDATE_SUSPENSION_CYCLE'}) from sqlalchemy import create_engine import urllib global password if '@' in password: password = urllib.parse.quote_plus(password) - engine = create_engine(f'mysql+pymysql://{dbusername}:{password}@{host}:{port}/{dbname}') - warning_data_df['WARNING_DATE'] = datetime.date.today().strftime("%Y-%m-%d %H:%M:%S") - warning_data_df['TENANT_CODE'] = 'T0004' + engine = create_engine( + f'mysql+pymysql://{dbusername}:{password}@{host}:{port}/{dbname}') + warning_data_df['WARNING_DATE'] = datetime.date.today().strftime( + "%Y-%m-%d %H:%M:%S") + warning_data_df['TENANT_CODE'] = 'T0004' # 插入数据之前查询表数据然后新增id列 existing_data = pd.read_sql(f"SELECT * FROM {table_name}", engine) if not existing_data.empty: max_id = existing_data['ID'].astype(int).max() - warning_data_df['ID'] = range(max_id + 1, max_id + 1 + len(warning_data_df)) + warning_data_df['ID'] = range( + max_id + 1, max_id + 1 + len(warning_data_df)) else: warning_data_df['ID'] = range(1, 1 + len(warning_data_df)) - warning_data_df.to_sql(table_name, con=engine, if_exists='append', index=False) + warning_data_df.to_sql( + table_name, con=engine, if_exists='append', index=False) if is_update_warning_data: upload_warning_info(len(warning_data_df)) except: @@ -228,72 +312,70 @@ def predict_main(): now = datetime.datetime.now().strftime('%Y%m%d%H%M%S') ex_Model(df, - horizon=horizon, - input_size=input_size, - train_steps=train_steps, - val_check_steps=val_check_steps, - early_stop_patience_steps=early_stop_patience_steps, - is_debug=is_debug, - dataset=dataset, - is_train=is_train, - is_fivemodels=is_fivemodels, - val_size=val_size, - test_size=test_size, - settings=settings, + horizon=global_config['horizon'], + input_size=global_config['input_size'], + train_steps=global_config['train_steps'], + val_check_steps=global_config['val_check_steps'], + early_stop_patience_steps=global_config['early_stop_patience_steps'], + is_debug=global_config['is_debug'], + dataset=global_config['dataset'], + is_train=global_config['is_train'], + is_fivemodels=global_config['is_fivemodels'], + val_size=global_config['val_size'], + test_size=global_config['test_size'], + settings=global_config['settings'], now=now, - etadata=etadata, - modelsindex=modelsindex, + etadata=global_config['etadata'], + modelsindex=global_config['modelsindex'], data=data, - is_eta=is_eta, - end_time=end_time, + is_eta=global_config['is_eta'], + end_time=global_config['end_time'], ) - - logger.info('模型训练完成') - - logger.info('训练数据绘图ing') - model_results3 = model_losss(sqlitedb,end_time=end_time) - logger.info('训练数据绘图end') - - # 模型报告 - # logger.info('制作报告ing') - # title = f'{settings}--{end_time}-预测报告' # 报告标题 - # reportname = f'Brent原油大模型月度预测--{end_time}.pdf' # 报告文件名 - # reportname = reportname.replace(':', '-') # 替换冒号 - # brent_export_pdf(dataset=dataset,num_models = 5 if is_fivemodels else 22,time=end_time, - # reportname=reportname,sqlitedb=sqlitedb), - - # logger.info('制作报告end') # logger.info('模型训练完成') + logger.info('训练数据绘图ing') + model_results3 = model_losss(sqlitedb, end_time=end_time) + logger.info('训练数据绘图end') + + # # 模型报告 + logger.info('制作报告ing') + title = f'{settings}--{end_time}-预测报告' # 报告标题 + reportname = f'Brent原油大模型周度预测--{end_time}.pdf' # 报告文件名 + reportname = reportname.replace(':', '-') # 替换冒号 + brent_export_pdf(dataset=dataset, num_models=5 if is_fivemodels else 22, time=end_time, + reportname=reportname, sqlitedb=sqlitedb), + + logger.info('制作报告end') + logger.info('模型训练完成') + # # LSTM 单变量模型 # ex_Lstm(df,input_seq_len=input_size,output_seq_len=horizon,is_debug=is_debug,dataset=dataset) - + # # lstm 多变量模型 # ex_Lstm_M(df,n_days=input_size,out_days=horizon,is_debug=is_debug,datasetpath=dataset) - + # # GRU 模型 # # ex_GRU(df) # 发送邮件 - m = SendMail( - username=username, - passwd=passwd, - recv=recv, - title=title, - content=content, - file=max(glob.glob(os.path.join(dataset,'*.pdf')), key=os.path.getctime), - ssl=ssl, - ) - # m.send_mail() + # m = SendMail( + # username=username, + # passwd=passwd, + # recv=recv, + # title=title, + # content=content, + # file=max(glob.glob(os.path.join(dataset,'*.pdf')), key=os.path.getctime), + # ssl=ssl, + # ) + # m.send_mail() if __name__ == '__main__': - global end_time - is_on = True - # 遍历2024-11-25 到 2024-12-3 之间的工作日日期 - for i_time in pd.date_range('2022-6-1', '2025-3-1', freq='ME'): - end_time = i_time.strftime('%Y-%m-%d') - predict_main() + # global end_time + # # 遍历2024-11-25 到 2024-12-3 之间的工作日日期 + # for i_time in pd.date_range('2024-12-1', '2025-2-26', freq='W'): + # end_time = i_time.strftime('%Y-%m-%d') + # predict_main() - # predict_main() \ No newline at end of file + predict_main() diff --git a/main_yuanyou_zhoudu.py b/main_yuanyou_zhoudu.py index 1e8e8b2..82df921 100644 --- a/main_yuanyou_zhoudu.py +++ b/main_yuanyou_zhoudu.py @@ -102,235 +102,235 @@ def predict_main(): 返回: None """ - # global end_time - # signature = BinanceAPI(APPID, SECRET) - # etadata = EtaReader(signature=signature, - # classifylisturl=classifylisturl, - # classifyidlisturl=classifyidlisturl, - # edbcodedataurl=edbcodedataurl, - # edbcodelist=edbcodelist, - # edbdatapushurl=edbdatapushurl, - # edbdeleteurl=edbdeleteurl, - # edbbusinessurl=edbbusinessurl - # ) - # # 获取数据 - # if is_eta: - # logger.info('从eta获取数据...') - # 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) # 原始数据，未处理 + global end_time + signature = BinanceAPI(APPID, SECRET) + etadata = EtaReader(signature=signature, + classifylisturl=classifylisturl, + classifyidlisturl=classifyidlisturl, + edbcodedataurl=edbcodedataurl, + edbcodelist=edbcodelist, + edbdatapushurl=edbdatapushurl, + edbdeleteurl=edbdeleteurl, + edbbusinessurl=edbbusinessurl + ) + # 获取数据 + if is_eta: + logger.info('从eta获取数据...') + 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) # 原始数据，未处理 - # if is_market: - # logger.info('从市场信息平台获取数据...') - # try: - # # 如果是测试环境，最高价最低价取excel文档 - # if server_host == '192.168.100.53': - # logger.info('从excel文档获取最高价最低价') - # df_zhibiaoshuju = get_high_low_data(df_zhibiaoshuju) - # else: - # logger.info('从市场信息平台获取数据') - # df_zhibiaoshuju = get_market_data( - # end_time, df_zhibiaoshuju) + if is_market: + logger.info('从市场信息平台获取数据...') + try: + # 如果是测试环境，最高价最低价取excel文档 + if server_host == '192.168.100.53': + logger.info('从excel文档获取最高价最低价') + df_zhibiaoshuju = get_high_low_data(df_zhibiaoshuju) + else: + logger.info('从市场信息平台获取数据') + df_zhibiaoshuju = get_market_data( + end_time, df_zhibiaoshuju) - # except: - # logger.info('最高最低价拼接失败') + except: + logger.info('最高最低价拼接失败') - # # 保存到xlsx文件的sheet表 - # with pd.ExcelWriter(os.path.join(dataset, data_set)) as file: - # df_zhibiaoshuju.to_excel(file, sheet_name='指标数据', index=False) - # df_zhibiaoliebiao.to_excel(file, sheet_name='指标列表', index=False) + # 保存到xlsx文件的sheet表 + with pd.ExcelWriter(os.path.join(dataset, data_set)) as file: + df_zhibiaoshuju.to_excel(file, sheet_name='指标数据', index=False) + df_zhibiaoliebiao.to_excel(file, sheet_name='指标列表', index=False) - # # 数据处理 - # df = datachuli(df_zhibiaoshuju, df_zhibiaoliebiao, y=y, dataset=dataset, add_kdj=add_kdj, is_timefurture=is_timefurture, - # end_time=end_time) + # 数据处理 + df = datachuli(df_zhibiaoshuju, df_zhibiaoliebiao, y=y, dataset=dataset, add_kdj=add_kdj, is_timefurture=is_timefurture, + end_time=end_time) - # else: - # # 读取数据 - # logger.info('读取本地数据：' + os.path.join(dataset, data_set)) - # df, df_zhibiaoliebiao = getdata(filename=os.path.join(dataset, data_set), y=y, dataset=dataset, add_kdj=add_kdj, - # is_timefurture=is_timefurture, end_time=end_time) # 原始数据，未处理 + else: + # 读取数据 + logger.info('读取本地数据：' + os.path.join(dataset, data_set)) + df, df_zhibiaoliebiao = getdata(filename=os.path.join(dataset, data_set), y=y, dataset=dataset, add_kdj=add_kdj, + is_timefurture=is_timefurture, end_time=end_time) # 原始数据，未处理 - # # 更改预测列名称 - # df.rename(columns={y: 'y'}, inplace=True) + # 更改预测列名称 + df.rename(columns={y: 'y'}, inplace=True) - # if is_edbnamelist: - # df = df[edbnamelist] - # df.to_csv(os.path.join(dataset, '指标数据.csv'), index=False) - # # 保存最新日期的y值到数据库 - # # 取第一行数据存储到数据库中 - # first_row = df[['ds', 'y']].tail(1) - # # 判断y的类型是否为float - # if not isinstance(first_row['y'].values[0], float): - # logger.info(f'{end_time}预测目标数据为空，跳过') - # return None + if is_edbnamelist: + df = df[edbnamelist] + df.to_csv(os.path.join(dataset, '指标数据.csv'), index=False) + # 保存最新日期的y值到数据库 + # 取第一行数据存储到数据库中 + first_row = df[['ds', 'y']].tail(1) + # 判断y的类型是否为float + if not isinstance(first_row['y'].values[0], float): + logger.info(f'{end_time}预测目标数据为空，跳过') + return None - # # 将最新真实值保存到数据库 - # if not sqlitedb.check_table_exists('trueandpredict'): - # first_row.to_sql('trueandpredict', sqlitedb.connection, index=False) - # else: - # for row in first_row.itertuples(index=False): - # row_dict = row._asdict() - # config.logger.info(f'要保存的真实值：{row_dict}') - # # 判断ds是否为字符串类型,如果不是则转换为字符串类型 - # if isinstance(row_dict['ds'], (pd.Timestamp, datetime.datetime)): - # row_dict['ds'] = row_dict['ds'].strftime('%Y-%m-%d') - # elif not isinstance(row_dict['ds'], str): - # try: - # row_dict['ds'] = pd.to_datetime( - # row_dict['ds']).strftime('%Y-%m-%d') - # except: - # logger.warning(f"无法解析的时间格式: {row_dict['ds']}") - # # row_dict['ds'] = row_dict['ds'].strftime('%Y-%m-%d') - # # row_dict['ds'] = row_dict['ds'].strftime('%Y-%m-%d %H:%M:%S') - # check_query = sqlitedb.select_data( - # 'trueandpredict', where_condition=f"ds = '{row.ds}'") - # if len(check_query) > 0: - # set_clause = ", ".join( - # [f"{key} = '{value}'" for key, value in row_dict.items()]) - # sqlitedb.update_data( - # 'trueandpredict', set_clause, where_condition=f"ds = '{row.ds}'") - # continue - # sqlitedb.insert_data('trueandpredict', tuple( - # row_dict.values()), columns=row_dict.keys()) + # 将最新真实值保存到数据库 + if not sqlitedb.check_table_exists('trueandpredict'): + first_row.to_sql('trueandpredict', sqlitedb.connection, index=False) + else: + for row in first_row.itertuples(index=False): + row_dict = row._asdict() + config.logger.info(f'要保存的真实值：{row_dict}') + # 判断ds是否为字符串类型,如果不是则转换为字符串类型 + if isinstance(row_dict['ds'], (pd.Timestamp, datetime.datetime)): + row_dict['ds'] = row_dict['ds'].strftime('%Y-%m-%d') + elif not isinstance(row_dict['ds'], str): + try: + row_dict['ds'] = pd.to_datetime( + row_dict['ds']).strftime('%Y-%m-%d') + except: + logger.warning(f"无法解析的时间格式: {row_dict['ds']}") + # row_dict['ds'] = row_dict['ds'].strftime('%Y-%m-%d') + # row_dict['ds'] = row_dict['ds'].strftime('%Y-%m-%d %H:%M:%S') + check_query = sqlitedb.select_data( + 'trueandpredict', where_condition=f"ds = '{row.ds}'") + if len(check_query) > 0: + set_clause = ", ".join( + [f"{key} = '{value}'" for key, value in row_dict.items()]) + sqlitedb.update_data( + 'trueandpredict', set_clause, where_condition=f"ds = '{row.ds}'") + continue + sqlitedb.insert_data('trueandpredict', tuple( + row_dict.values()), columns=row_dict.keys()) - # # 更新accuracy表的y值 - # if not sqlitedb.check_table_exists('accuracy'): - # pass - # else: - # update_y = sqlitedb.select_data( - # 'accuracy', where_condition="y is null") - # if len(update_y) > 0: - # logger.info('更新accuracy表的y值') - # # 找到update_y 中ds且df中的y的行 - # update_y = update_y[update_y['ds'] <= end_time] - # logger.info(f'要更新y的信息：{update_y}') - # # try: - # for row in update_y.itertuples(index=False): - # try: - # row_dict = row._asdict() - # yy = df[df['ds'] == row_dict['ds']]['y'].values[0] - # LOW = df[df['ds'] == row_dict['ds']]['Brentzdj'].values[0] - # HIGH = df[df['ds'] == row_dict['ds']]['Brentzgj'].values[0] - # sqlitedb.update_data( - # 'accuracy', f"y = {yy},LOW_PRICE = {LOW},HIGH_PRICE = {HIGH}", where_condition=f"ds = '{row_dict['ds']}'") - # except: - # logger.info(f'更新accuracy表的y值失败：{row_dict}') - # # except Exception as e: - # # logger.info(f'更新accuracy表的y值失败：{e}') + # 更新accuracy表的y值 + if not sqlitedb.check_table_exists('accuracy'): + pass + else: + update_y = sqlitedb.select_data( + 'accuracy', where_condition="y is null") + if len(update_y) > 0: + logger.info('更新accuracy表的y值') + # 找到update_y 中ds且df中的y的行 + update_y = update_y[update_y['ds'] <= end_time] + logger.info(f'要更新y的信息：{update_y}') + # try: + for row in update_y.itertuples(index=False): + try: + row_dict = row._asdict() + yy = df[df['ds'] == row_dict['ds']]['y'].values[0] + LOW = df[df['ds'] == row_dict['ds']]['Brentzdj'].values[0] + HIGH = df[df['ds'] == row_dict['ds']]['Brentzgj'].values[0] + sqlitedb.update_data( + 'accuracy', f"y = {yy},LOW_PRICE = {LOW},HIGH_PRICE = {HIGH}", where_condition=f"ds = '{row_dict['ds']}'") + except: + logger.info(f'更新accuracy表的y值失败：{row_dict}') + # except Exception as e: + # logger.info(f'更新accuracy表的y值失败：{e}') - # # 判断当前日期是不是周一 - # is_weekday = datetime.datetime.now().weekday() == 0 - # if is_weekday: - # logger.info('今天是周一，更新预测模型') - # # 计算最近60天预测残差最低的模型名称 - # model_results = sqlitedb.select_data( - # 'trueandpredict', order_by="ds DESC", limit="60") - # # 删除空值率为90%以上的列 - # if len(model_results) > 10: - # model_results = model_results.dropna( - # thresh=len(model_results)*0.1, axis=1) - # # 删除空行 - # model_results = model_results.dropna() - # modelnames = model_results.columns.to_list()[2:-1] - # for col in model_results[modelnames].select_dtypes(include=['object']).columns: - # model_results[col] = model_results[col].astype(np.float32) - # # 计算每个预测值与真实值之间的偏差率 - # for model in modelnames: - # model_results[f'{model}_abs_error_rate'] = abs( - # model_results['y'] - model_results[model]) / model_results['y'] - # # 获取每行对应的最小偏差率值 - # min_abs_error_rate_values = model_results.apply( - # lambda row: row[[f'{model}_abs_error_rate' for model in modelnames]].min(), axis=1) - # # 获取每行对应的最小偏差率值对应的列名 - # min_abs_error_rate_column_name = model_results.apply( - # lambda row: row[[f'{model}_abs_error_rate' for model in modelnames]].idxmin(), axis=1) - # # 将列名索引转换为列名 - # min_abs_error_rate_column_name = min_abs_error_rate_column_name.map( - # lambda x: x.split('_')[0]) - # # 取出现次数最多的模型名称 - # most_common_model = min_abs_error_rate_column_name.value_counts().idxmax() - # logger.info(f"最近60天预测残差最低的模型名称：{most_common_model}") - # # 保存结果到数据库 - # if not sqlitedb.check_table_exists('most_model'): - # sqlitedb.create_table( - # 'most_model', columns="ds datetime, most_common_model TEXT") - # sqlitedb.insert_data('most_model', (datetime.datetime.now().strftime( - # '%Y-%m-%d %H:%M:%S'), most_common_model,), columns=('ds', 'most_common_model',)) + # 判断当前日期是不是周一 + is_weekday = datetime.datetime.now().weekday() == 0 + if is_weekday: + logger.info('今天是周一，更新预测模型') + # 计算最近60天预测残差最低的模型名称 + model_results = sqlitedb.select_data( + 'trueandpredict', order_by="ds DESC", limit="60") + # 删除空值率为90%以上的列 + if len(model_results) > 10: + model_results = model_results.dropna( + thresh=len(model_results)*0.1, axis=1) + # 删除空行 + model_results = model_results.dropna() + modelnames = model_results.columns.to_list()[2:-1] + for col in model_results[modelnames].select_dtypes(include=['object']).columns: + model_results[col] = model_results[col].astype(np.float32) + # 计算每个预测值与真实值之间的偏差率 + for model in modelnames: + model_results[f'{model}_abs_error_rate'] = abs( + model_results['y'] - model_results[model]) / model_results['y'] + # 获取每行对应的最小偏差率值 + min_abs_error_rate_values = model_results.apply( + lambda row: row[[f'{model}_abs_error_rate' for model in modelnames]].min(), axis=1) + # 获取每行对应的最小偏差率值对应的列名 + min_abs_error_rate_column_name = model_results.apply( + lambda row: row[[f'{model}_abs_error_rate' for model in modelnames]].idxmin(), axis=1) + # 将列名索引转换为列名 + min_abs_error_rate_column_name = min_abs_error_rate_column_name.map( + lambda x: x.split('_')[0]) + # 取出现次数最多的模型名称 + most_common_model = min_abs_error_rate_column_name.value_counts().idxmax() + logger.info(f"最近60天预测残差最低的模型名称：{most_common_model}") + # 保存结果到数据库 + if not sqlitedb.check_table_exists('most_model'): + sqlitedb.create_table( + 'most_model', columns="ds datetime, most_common_model TEXT") + sqlitedb.insert_data('most_model', (datetime.datetime.now().strftime( + '%Y-%m-%d %H:%M:%S'), most_common_model,), columns=('ds', 'most_common_model',)) - # try: - # if is_weekday: - # # if True: - # logger.info('今天是周一，发送特征预警') - # # 上传预警信息到数据库 - # warning_data_df = df_zhibiaoliebiao.copy() - # warning_data_df = warning_data_df[warning_data_df['停更周期'] > 3][[ - # '指标名称', '指标id', '频度', '更新周期', '指标来源', '最后更新时间', '停更周期']] - # # 重命名列名 - # warning_data_df = warning_data_df.rename(columns={'指标名称': 'INDICATOR_NAME', '指标id': 'INDICATOR_ID', '频度': 'FREQUENCY', - # '更新周期': 'UPDATE_FREQUENCY', '指标来源': 'DATA_SOURCE', '最后更新时间': 'LAST_UPDATE_DATE', '停更周期': 'UPDATE_SUSPENSION_CYCLE'}) - # from sqlalchemy import create_engine - # import urllib - # global password - # if '@' in password: - # password = urllib.parse.quote_plus(password) + try: + if is_weekday: + # if True: + logger.info('今天是周一，发送特征预警') + # 上传预警信息到数据库 + warning_data_df = df_zhibiaoliebiao.copy() + warning_data_df = warning_data_df[warning_data_df['停更周期'] > 3][[ + '指标名称', '指标id', '频度', '更新周期', '指标来源', '最后更新时间', '停更周期']] + # 重命名列名 + warning_data_df = warning_data_df.rename(columns={'指标名称': 'INDICATOR_NAME', '指标id': 'INDICATOR_ID', '频度': 'FREQUENCY', + '更新周期': 'UPDATE_FREQUENCY', '指标来源': 'DATA_SOURCE', '最后更新时间': 'LAST_UPDATE_DATE', '停更周期': 'UPDATE_SUSPENSION_CYCLE'}) + from sqlalchemy import create_engine + import urllib + global password + if '@' in password: + password = urllib.parse.quote_plus(password) - # engine = create_engine( - # f'mysql+pymysql://{dbusername}:{password}@{host}:{port}/{dbname}') - # warning_data_df['WARNING_DATE'] = datetime.date.today().strftime( - # "%Y-%m-%d %H:%M:%S") - # warning_data_df['TENANT_CODE'] = 'T0004' - # # 插入数据之前查询表数据然后新增id列 - # existing_data = pd.read_sql(f"SELECT * FROM {table_name}", engine) - # if not existing_data.empty: - # max_id = existing_data['ID'].astype(int).max() - # warning_data_df['ID'] = range( - # max_id + 1, max_id + 1 + len(warning_data_df)) - # else: - # warning_data_df['ID'] = range(1, 1 + len(warning_data_df)) - # warning_data_df.to_sql( - # table_name, con=engine, if_exists='append', index=False) - # if is_update_warning_data: - # upload_warning_info(len(warning_data_df)) - # except: - # logger.info('上传预警信息到数据库失败') + engine = create_engine( + f'mysql+pymysql://{dbusername}:{password}@{host}:{port}/{dbname}') + warning_data_df['WARNING_DATE'] = datetime.date.today().strftime( + "%Y-%m-%d %H:%M:%S") + warning_data_df['TENANT_CODE'] = 'T0004' + # 插入数据之前查询表数据然后新增id列 + existing_data = pd.read_sql(f"SELECT * FROM {table_name}", engine) + if not existing_data.empty: + max_id = existing_data['ID'].astype(int).max() + warning_data_df['ID'] = range( + max_id + 1, max_id + 1 + len(warning_data_df)) + else: + warning_data_df['ID'] = range(1, 1 + len(warning_data_df)) + warning_data_df.to_sql( + table_name, con=engine, if_exists='append', index=False) + if is_update_warning_data: + upload_warning_info(len(warning_data_df)) + except: + logger.info('上传预警信息到数据库失败') - # if is_corr: - # df = corr_feature(df=df) + if is_corr: + df = corr_feature(df=df) - # df1 = df.copy() # 备份一下，后面特征筛选完之后加入ds y 列用 - # logger.info(f"开始训练模型...") - # row, col = df.shape + df1 = df.copy() # 备份一下，后面特征筛选完之后加入ds y 列用 + logger.info(f"开始训练模型...") + row, col = df.shape - # now = datetime.datetime.now().strftime('%Y%m%d%H%M%S') - # ex_Model(df, - # horizon=global_config['horizon'], - # input_size=global_config['input_size'], - # train_steps=global_config['train_steps'], - # val_check_steps=global_config['val_check_steps'], - # early_stop_patience_steps=global_config['early_stop_patience_steps'], - # is_debug=global_config['is_debug'], - # dataset=global_config['dataset'], - # is_train=global_config['is_train'], - # is_fivemodels=global_config['is_fivemodels'], - # val_size=global_config['val_size'], - # test_size=global_config['test_size'], - # settings=global_config['settings'], - # now=now, - # etadata=global_config['etadata'], - # modelsindex=global_config['modelsindex'], - # data=data, - # is_eta=global_config['is_eta'], - # end_time=global_config['end_time'], - # ) + now = datetime.datetime.now().strftime('%Y%m%d%H%M%S') + ex_Model(df, + horizon=global_config['horizon'], + input_size=global_config['input_size'], + train_steps=global_config['train_steps'], + val_check_steps=global_config['val_check_steps'], + early_stop_patience_steps=global_config['early_stop_patience_steps'], + is_debug=global_config['is_debug'], + dataset=global_config['dataset'], + is_train=global_config['is_train'], + is_fivemodels=global_config['is_fivemodels'], + val_size=global_config['val_size'], + test_size=global_config['test_size'], + settings=global_config['settings'], + now=now, + etadata=global_config['etadata'], + modelsindex=global_config['modelsindex'], + data=data, + is_eta=global_config['is_eta'], + end_time=global_config['end_time'], + ) # logger.info('模型训练完成')