DB functions and auto handle time of day

dbConn.py

@@ -0,0 +1,42 @@
+from dotenv import load_dotenv
+from sqlalchemy import create_engine
+import pandas as pd
+import os
+import MySQLdb
+load_dotenv()
+
+engine = create_engine(
+        f"mysql+mysqldb://{os.getenv('DATABASE_USERNAME')}:" \
+        f"{os.getenv('DATABASE_PASSWORD')}@{os.getenv('DATABASE_HOST')}/" \
+        f"{os.getenv('DATABASE')}?ssl_ca=ca-certificates.crt&ssl_mode=VERIFY_IDENTITY"
+)
+
+connection = MySQLdb.connect(
+  host=os.getenv("DATABASE_HOST"),
+  user=os.getenv("DATABASE_USERNAME"),
+  passwd=os.getenv("DATABASE_PASSWORD"),
+  db=os.getenv("DATABASE"),
+  autocommit=True,
+  ssl_mode="VERIFY_IDENTITY",
+  ssl={ "ca": "ca-certificates.crt" }
+)
+
+# Function to write dataframe to SQL
+def insert_dataframe_to_sql(table_name, dataframe, cursor):
+    # Prepare the SQL insert statement
+    placeholders = ', '.join(['%s'] * len(dataframe.columns))
+    columns = ', '.join(dataframe.columns)
+
+    # Prepare the ON DUPLICATE KEY UPDATE part of the query
+    update_columns = ', '.join([f"{col} = VALUES({col})" for col in dataframe.columns])
+
+    sql = f"""INSERT INTO {table_name} ({columns}) VALUES ({placeholders})
+              ON DUPLICATE KEY UPDATE {update_columns}"""
+
+    # Convert dataframe to a list of tuples, handling NaN values
+    data = [tuple(row) if not any(pd.isna(val) for val in row) 
+            else tuple(None if pd.isna(val) else val for val in row) 
+            for row in dataframe.values]
+
+    # Execute the SQL command for each row
+    cursor.executemany(sql, data)

getDailyData.py

@@ -2,11 +2,8 @@ import pandas as pd
 import numpy as np
 import yfinance as yf
 from tqdm import tqdm
-import os
 from pandas.tseries.offsets import BDay
-from sqlalchemy import create_engine
-from dotenv import load_dotenv
-load_dotenv()
+from dbConn import engine
 
 data_start_date = '2018-07-01'
 
@@ -21,13 +18,6 @@ def get_daily(mode='daily', periods_30m=None):
     vvix = yf.Ticker('^VVIX')
     spx = yf.Ticker('^GSPC')
 
-    # Grab data from db
-    engine = create_engine(
-        f"mysql+mysqldb://{os.getenv('DATABASE_USERNAME')}:" \
-        f"{os.getenv('DATABASE_PASSWORD')}@{os.getenv('DATABASE_HOST')}/" \
-        f"{os.getenv('DATABASE')}?ssl_ca=ca-certificates.crt&ssl_mode=VERIFY_IDENTITY"
-    )
-
     query = f'''SELECT
         spx.Datetime AS Datetime,
         spx.Open AS Open,
@@ -266,12 +256,6 @@ def get_daily(mode='daily', periods_30m=None):
 
         data['GreenProbas'] = probas
 
-    engine = create_engine(
-        f"mysql+mysqldb://{os.getenv('DATABASE_USERNAME')}:" \
-        f"{os.getenv('DATABASE_PASSWORD')}@{os.getenv('DATABASE_HOST')}/" \
-        f"{os.getenv('DATABASE')}?ssl_ca=ca-certificates.crt&ssl_mode=VERIFY_IDENTITY"
-    )
-
     df_releases = pd.read_sql_query('select * from releases', con=engine)
     df_releases = df_releases.set_index('Datetime')
     data = data.merge(df_releases, how = 'left', left_index=True, right_index=True)

getIntraData.py

@@ -1,24 +1,13 @@
 import pandas as pd
 import yfinance as yf
 import datetime
-# from datasets import load_dataset
-from sqlalchemy import create_engine
-import os
 from getDailyData import data_start_date
-from dotenv import load_dotenv
-
-# Load environment variables from the .env file
-load_dotenv()
+from dbConn import engine
 
 def get_intra(periods_30m = 1):
     '''
     Method to get historical 30 minute data and append live data to it, if exists. 
     '''
-    engine = create_engine(
-        f"mysql+mysqldb://{os.getenv('DATABASE_USERNAME')}:" \
-        f"{os.getenv('DATABASE_PASSWORD')}@{os.getenv('DATABASE_HOST')}/" \
-        f"{os.getenv('DATABASE')}?ssl_ca=ca-certificates.crt&ssl_mode=VERIFY_IDENTITY"
-    )
 
     query = f'''SELECT
         spx30.Datetime AS Datetime,
@@ -44,42 +33,11 @@ def get_intra(periods_30m = 1):
         spx30.Datetime > '{data_start_date}'
 
     '''
-    # spx30 = pd.read_sql_query(f'SELECT * FROM SPX_full_30min WHERE Datetime > {data_start_date}', con=engine)
-    # vix30 = pd.read_sql_query(f'SELECT * FROM VIX_full_30min WHERE Datetime > {data_start_date}', con=engine)
-    # vvix30 = pd.read_sql_query(f'SELECT * FROM VVIX_full_30min WHERE Datetime > {data_start_date}', con=engine)
-    # dfs = []
 
     df_30m = pd.read_sql_query(sql=query, con=engine.connect())
     df_30m['Datetime'] = df_30m['Datetime'].dt.tz_localize('America/New_York')
     df_30m = df_30m.set_index('Datetime',drop=True)
 
-    # for fr in [spx30, vix30, vvix30]:
-    #     # fr['Datetime'] = fr['Datetime'].apply(lambda x: datetime.datetime.strptime(x[:-6], dt_format))
-    #     fr['Datetime'] = fr['Datetime'].dt.tz_localize('America/New_York')
-    #     fr = fr.set_index('Datetime')
-    #     fr['Open'] = pd.to_numeric(fr['Open'])
-    #     fr['High'] = pd.to_numeric(fr['High'])
-    #     fr['Low'] = pd.to_numeric(fr['Low'])
-    #     fr['Close'] = pd.to_numeric(fr['Close'])
-    #     dfs.append(fr[['Open','High','Low','Close']])
-
-    # df_30m = pd.concat(dfs, axis=1)
-
-    # df_30m.columns = [
-    #     'Open30',
-    #     'High30',
-    #     'Low30',
-    #     'Close30',
-    #     'Open_VIX30',
-    #     'High_VIX30',
-    #     'Low_VIX30',
-    #     'Close_VIX30',
-    #     'Open_VVIX30',
-    #     'High_VVIX30',
-    #     'Low_VVIX30',
-    #     'Close_VVIX30'
-    # ]
-
     # Get incremental date
     last_date = df_30m.index.date[-1]
     last_date = last_date + datetime.timedelta(days=1)
@@ -134,6 +92,4 @@ def get_intra(periods_30m = 1):
     vvix_intra = df_30m.groupby('Datetime')['VVIX30IntraPerf'].tail(1)
 
     df_intra = pd.concat([opens_intra, highs_intra, lows_intra, closes_intra, spx_intra, vix_intra, vvix_intra], axis=1)
-    return df_intra
-
-    
+    return df_intra

lambda_function.py

@@ -1,17 +1,74 @@
 # Function should get the data and run the whole model, return a single prediction based on the time
 from getDailyData import get_daily
 from model_intra_v3 import walk_forward_validation
+from model_day_v2 import walk_forward_validation_seq as walk_forward_validation_daily
+from datetime import timedelta
+import pandas as pd
 import json
+from dbConn import connection, engine, insert_dataframe_to_sql
+import numpy as np
+import datetime
+from datetime import time
+import datetime
+from pandas.tseries.offsets import BDay
 
 def lambda_handler(periods_30m):
-    data, df_final, final_row = get_daily(mode='intra', periods_30m=periods_30m)
-    res, _ = walk_forward_validation(df_final.drop(columns=['Target']).dropna(), 'Target_clf', 1, mode='single')
-    return json.loads(json.dumps({
-        'date': str(res.index[-1]),
-        'prediction': res['Predicted'].iloc[-1],
-        'time':periods_30m
-    }))
+    if periods_30m > 0:
+        data, df_final, final_row = get_daily(mode='intra', periods_30m=periods_30m)
+        res, _ = walk_forward_validation(df_final.drop(columns=['Target']).dropna(), 'Target_clf', 1, mode='single')
+
+    elif periods_30m == 0:
+        data, df_final, final_row = get_daily()
+        res, _, _ = walk_forward_validation_daily(df_final.dropna(), 'Target_clf', 'Target', 200, 1)
+    
+    # Get results, run calibration and pvalue    
+    df_results = pd.read_sql_query(f'select * from results where ModelNum = {str(periods_30m)}', con = engine)
+
+    # Calibrate Probabilities
+    def get_quantiles(df, col_name, q):
+        return df.groupby(pd.cut(df[col_name], q))['IsTrue'].mean()
+
+    pct = res['Predicted'].iloc[-1]
+
+    df_q = get_quantiles(df_results, 'Predicted', 10)
+    for q in df_q.index:
+        if q.left <= pct <= q.right:
+            p = df_q[q]
+
+    calib_scores = np.abs(df_results['Predicted'].iloc[:-1] - 0.5)
+    score = abs(pct - 0.5)
+    pv = np.mean(calib_scores >= score)
+    asof = datetime.combine(data.index[-1], time(9,30)) + (periods_30m * timedelta(minutes=30)) 
+
+    blob = {
+        'Datetime': str(res.index[-1]),
+        'IsTrue':df_final['Target_clf'].iloc[-1],
+        'Predicted': pct,
+        'CalibPredicted': p,
+        'Pvalue':pv,
+        'ModelNum':periods_30m,
+        'AsOf':str(asof)
+    }
+
+    # Write to DB
+    df_write = pd.DataFrame.from_dict({k:[v] for k, v in blob.items()})
+    cursor = connection.cursor()
+    insert_dataframe_to_sql('results', df_write, cursor)
+    cursor.close()
+    connection.close()
+
+    return json.loads(json.dumps(blob))
 
 if __name__ == '__main__':
-    j = lambda_handler(1)
-    print(j)
+    # Code that, based on the time of the day, return which data/model to run
+    from datetime import datetime, time
+    now = datetime.now()
+    morning_start = datetime.combine(now.date(), time(6, 30))
+    delta = now - morning_start
+    intervals = max(1,min((delta.total_seconds() / 60 / 30) // 1, 12))
+    print(f'running for {str(intervals)}')
+    j = lambda_handler(intervals)
+    # times_list = np.arange(0,13)
+    # for i in times_list: 
+    #     j = lambda_handler(i)
+    #     print(j)

model_day_v2.py

@@ -3,6 +3,7 @@ from tqdm import tqdm
 from sklearn import linear_model
 import lightgbm as lgb
 from dailyCols import model_cols
+import numpy as np
 
 def walk_forward_validation(df, target_column, num_training_rows, num_periods):
     
@@ -84,22 +85,26 @@ def walk_forward_validation_seq(df, target_column_clf, target_column_regr, num_t
         return df.groupby(pd.cut(df[col_name], q))['True'].mean()
 
     greenprobas = []
-    meanprobas = []
-    for i, pct in tqdm(enumerate(df_results['Predicted']), desc='Calibrating Probas'):
+    pvals = []
+    for i, pct in tqdm(enumerate(df_results['Predicted']), desc='Calibrating Probas',total=len(df_results)):
         try:
-            df_q = get_quantiles(df_results.iloc[:i], 'Predicted', 7)
+            df_q = get_quantiles(df_results.iloc[:i], 'Predicted', 10)
             for q in df_q.index:
                 if q.left <= pct <= q.right:
                     p = df_q[q]
-                    c = (q.left + q.right) / 2
+
+            calib_scores = np.abs(df_results['Predicted'].iloc[:i] - 0.5)
+            score = abs(df_results['Predicted'].iloc[i] - 0.5)
+            pv = np.mean(calib_scores >= score)
         except:
             p = None
-            c = None
+            pv = None
 
         greenprobas.append(p)
-        meanprobas.append(c)
-
+        pvals.append(pv)
+    
     df_results['CalibPredicted'] = greenprobas
+    df_results['Pvalue'] = pvals
     
     return df_results, model1, model2
 

model_intra_v2.py

@@ -1,4 +1,5 @@
 import pandas as pd
+import numpy as np
 from tqdm import tqdm
 import lightgbm as lgb
 from sklearn.model_selection import TimeSeriesSplit
@@ -29,29 +30,36 @@ def walk_forward_validation(df, target_column, num_periods, mode='full'):
             predictions = model.predict_proba(X_test)[:,-1]
                 
             # Create a DataFrame to store the true and predicted values
-            result_df = pd.DataFrame({'True': y_test, 'Predicted': predictions}, index=y_test.index)
+            result_df = pd.DataFrame({'IsTrue': y_test, 'Predicted': predictions}, index=y_test.index)
             overall_results.append(result_df)
-        df_results = pd.concat(overall_results)
 
+        df_results = pd.concat(overall_results)
+        
         # Calibrate Probabilities
         def get_quantiles(df, col_name, q):
-            return df.groupby(pd.cut(df[col_name], q))['True'].mean()
+            return df.groupby(pd.cut(df[col_name], q))['IsTrue'].mean()
 
         greenprobas = []
+        pvals = []
         for i, pct in tqdm(enumerate(df_results['Predicted']), desc='Calibrating Probas',total=len(df_results)):
             try:
-                df_q = get_quantiles(df_results.iloc[:i], 'Predicted', 7)
+                df_q = get_quantiles(df_results.iloc[:i], 'Predicted', 10)
                 for q in df_q.index:
                     if q.left <= pct <= q.right:
                         p = df_q[q]
+
+                calib_scores = np.abs(df_results['Predicted'].iloc[:i] - 0.5)
+                score = abs(df_results['Predicted'].iloc[i] - 0.5)
+                pv = np.mean(calib_scores >= score)
             except:
                 p = None
+                pv = None
 
             greenprobas.append(p)
-
+            pvals.append(pv)
+        
         df_results['CalibPredicted'] = greenprobas
-
-        return df_results, model
+        df_results['Pvalue'] = pvals
 
     elif mode == 'single':
         X_train = df.drop(target_column, axis=1).iloc[:-1]
@@ -62,13 +70,11 @@ def walk_forward_validation(df, target_column, num_periods, mode='full'):
         model = lgb.LGBMClassifier(n_estimators=10, random_state=42, verbosity=-1)
         model.fit(X_train, y_train)
         predictions = model.predict_proba(X_test.values.reshape(1, -1))[:,-1]
-        result_df = pd.DataFrame({'True': y_test, 'Predicted': predictions}, index=[df.index[-1]])
+        df_results = pd.DataFrame({'IsTrue': y_test, 'Predicted': predictions}, index=[df.index[-1]])
 
-        return result_df, model
+    return df_results, model
         
 
-    
-
 def seq_predict_proba(df, trained_clf_model):
     clf_pred_proba = trained_clf_model.predict_proba(df[model_cols])[:,-1]
     return clf_pred_proba