Update app.py

app.py

@@ -8,7 +8,7 @@ import pandas as pd
 import numpy as np
 from aiohttp import web
 from sklearn.ensemble import RandomForestRegressor
-from sklearn.metrics import mean_squared_error
+from concurrent.futures import ThreadPoolExecutor
 
 SYMBOL_KRAKEN = "BTC/USD"
 PORT = 7860
@@ -30,6 +30,7 @@ market_state = {
 }
 
 connected_clients = set()
+executor = ThreadPoolExecutor(max_workers=1)
 
 def calculate_indicators(candles):
     if len(candles) < 100:
@@ -72,7 +73,11 @@ def calculate_indicators(candles):
     df['vol_change'] = df['volume'].pct_change()
     df['log_ret'] = np.log(df['close'] / df['close'].shift(1))
 
-    for lag in [1, 2, 3]:
+    df['datetime'] = pd.to_datetime(df['time'], unit='s')
+    df['hour_sin'] = np.sin(2 * np.pi * df['datetime'].dt.hour / 24)
+    df['hour_cos'] = np.cos(2 * np.pi * df['datetime'].dt.hour / 24)
+
+    for lag in [1, 2, 3, 5, 8]:
         df[f'rsi_lag{lag}'] = df['rsi'].shift(lag)
         df[f'macd_hist_lag{lag}'] = df['macd_hist'].shift(lag)
         df[f'log_ret_lag{lag}'] = df['log_ret'].shift(lag)
@@ -88,32 +93,38 @@ def train_model(df):
         'dist_ema20', 'dist_ema50', 
         'bb_width', 'bb_pos', 
         'vol_change', 'log_ret',
-        'rsi_lag1', 'rsi_lag2', 'rsi_lag3',
-        'macd_hist_lag1', 'macd_hist_lag2', 'macd_hist_lag3',
-        'log_ret_lag1', 'log_ret_lag2', 'log_ret_lag3',
-        'vol_change_lag1', 'vol_change_lag2', 'vol_change_lag3'
+        'hour_sin', 'hour_cos'
     ]
     
+    for lag in [1, 2, 3, 5, 8]:
+        feature_cols.extend([
+            f'rsi_lag{lag}', f'macd_hist_lag{lag}', 
+            f'log_ret_lag{lag}', f'vol_change_lag{lag}'
+        ])
+    
     data = df.dropna().copy()
-    targets = []
+    
+    target_cols_dict = {}
+    target_names = []
     
     for i in range(1, PREDICTION_HORIZON + 1):
         col_name = f'target_return_{i}'
-        data[col_name] = (data['close'].shift(-i) - data['close']) / data['close']
-        targets.append(col_name)
-        
-    data = data.dropna()
+        target_cols_dict[col_name] = (data['close'].shift(-i) - data['close']) / data['close']
+        target_names.append(col_name)
+    
+    targets_df = pd.DataFrame(target_cols_dict, index=data.index)
+    data = pd.concat([data, targets_df], axis=1).dropna()
 
     if len(data) < 200:
         return None, None
 
     X = data[feature_cols].values
-    y = data[targets].values
+    y = data[target_names].values
 
     model = RandomForestRegressor(
-        n_estimators=150, 
-        max_depth=20, 
-        min_samples_split=4,
+        n_estimators=200, 
+        max_depth=25, 
+        min_samples_split=5,
         min_samples_leaf=2,
         max_features='sqrt',
         n_jobs=-1, 
@@ -136,12 +147,15 @@ def get_prediction(df, model, residual_std):
         'dist_ema20', 'dist_ema50', 
         'bb_width', 'bb_pos', 
         'vol_change', 'log_ret',
-        'rsi_lag1', 'rsi_lag2', 'rsi_lag3',
-        'macd_hist_lag1', 'macd_hist_lag2', 'macd_hist_lag3',
-        'log_ret_lag1', 'log_ret_lag2', 'log_ret_lag3',
-        'vol_change_lag1', 'vol_change_lag2', 'vol_change_lag3'
+        'hour_sin', 'hour_cos'
     ]
     
+    for lag in [1, 2, 3, 5, 8]:
+        feature_cols.extend([
+            f'rsi_lag{lag}', f'macd_hist_lag{lag}', 
+            f'log_ret_lag{lag}', f'vol_change_lag{lag}'
+        ])
+    
     last_row = df.iloc[[-1]][feature_cols]
     
     if last_row.isnull().values.any():
@@ -171,7 +185,7 @@ def get_prediction(df, model, residual_std):
     
     return pred_data
 
-def process_market_data():
+async def process_market_data():
     if not market_state['ready'] or not market_state['ohlc_history']:
         return {"error": "Initializing..."}
 
@@ -181,7 +195,8 @@ def process_market_data():
 
     if market_state['model'] is None or (time.time() - market_state['last_training_time'] > TRAIN_INTERVAL):
         try:
-            model, res_std = train_model(df)
+            loop = asyncio.get_running_loop()
+            model, res_std = await loop.run_in_executor(executor, train_model, df)
             if model is not None:
                 market_state['model'] = model
                 market_state['model_residuals'] = res_std
@@ -196,7 +211,8 @@ def process_market_data():
         logging.error(f"Prediction failed: {e}")
 
     df_clean = df.replace([np.inf, -np.inf], np.nan)
-    df_clean = df_clean.astype(object).where(pd.notnull(df_clean), None)
+    cols_to_keep = ['time', 'open', 'high', 'low', 'close', 'volume', 'ema20', 'bb_upper', 'bb_lower', 'rsi', 'macd_hist']
+    df_clean = df_clean[cols_to_keep].where(pd.notnull(df_clean), None)
     
     last_close = float(df['close'].iloc[-1]) if len(df) > 0 else 0
     first_close = float(df['close'].iloc[0]) if len(df) > 0 else 0
@@ -479,7 +495,7 @@ document.addEventListener('DOMContentLoaded', () => {
             document.getElementById('ema-val').textContent = lastData.ema20 ? lastData.ema20.toFixed(2) : '--';
             document.getElementById('bb-upper').textContent = lastData.bb_upper ? lastData.bb_upper.toFixed(2) : '--';
             document.getElementById('bb-lower').textContent = lastData.bb_lower ? lastData.bb_lower.toFixed(2) : '--';
-            const macdVal = lastData.macd;
+            const macdVal = lastData.macd_hist;
             const macdEl = document.getElementById('macd-val');
             if (macdVal !== null && macdVal !== undefined) {
                 macdEl.textContent = macdVal.toFixed(2);
@@ -633,16 +649,18 @@ async def kraken_rest_worker():
 async def broadcast_worker():
     while True:
         if connected_clients and market_state['ready']:
-            payload = process_market_data()
+            payload = await process_market_data()
             if payload and "data" in payload:
                 msg = json.dumps(payload)
+                current_clients = connected_clients.copy()
                 disconnected = set()
-                for ws in connected_clients:
+                for ws in current_clients:
                     try:
                         await ws.send_str(msg)
                     except Exception:
                         disconnected.add(ws)
-                connected_clients.difference_update(disconnected)
+                if disconnected:
+                    connected_clients.difference_update(disconnected)
         await asyncio.sleep(BROADCAST_RATE)
 
 async def websocket_handler(request):