Update app.py

app.py

@@ -1,105 +1,46 @@
-from fastapi import FastAPI, Request
-from fastapi.responses import JSONResponse
-import os
-import pandas as pd
-import numpy as np
-from tensorflow.keras.models import Sequential, load_model
-from tensorflow.keras.layers import Dense, LSTM
-from tensorflow.keras.callbacks import EarlyStopping
-from sklearn.preprocessing import MinMaxScaler
-import time
-import datetime
-
-app = FastAPI()
-
-MODEL_DIR = 'saved_models'
-DATA_DIR = 'stock_data'
-os.makedirs(MODEL_DIR, exist_ok=True)
-os.makedirs(DATA_DIR, exist_ok=True)
-
-def load_data(stock_name):
-    path = os.path.join(DATA_DIR, f"{stock_name}.csv")
-    if not os.path.exists(path):
-        return None
-    df = pd.read_csv(path)
-    df['Date'] = pd.to_datetime(df['Date'])
-    df.sort_values('Date', inplace=True)
-    df = df[['Date', 'Close']]
-    return df
-
-def train_model(stock_name):
-    df = load_data(stock_name)
-    if df is None:
-        print(f"No data found for {stock_name}")
-        return
-
-    today = datetime.datetime.today()
-    df = df[df['Date'] >= today - datetime.timedelta(days=5)]
-
-    scaler = MinMaxScaler()
-    scaled_data = scaler.fit_transform(df[['Close']])
-
-    X, y = [], []
-    for i in range(3, len(scaled_data)):
-        X.append(scaled_data[i-3:i])
-        y.append(scaled_data[i])
-    X, y = np.array(X), np.array(y)
-
-    model = Sequential()
-    model.add(LSTM(50, return_sequences=True, input_shape=(X.shape[1], X.shape[2])))
-    model.add(LSTM(50))
-    model.add(Dense(1))
-    model.compile(optimizer='adam', loss='mean_squared_error')
-
-    model.fit(X, y, epochs=10, batch_size=1, verbose=0, callbacks=[EarlyStopping(patience=2)])
-
-    model.save(os.path.join(MODEL_DIR, f"{stock_name}.h5"))
-    print(f"Model trained and saved for {stock_name}")
-
-def predict_next(stock_name, steps=1):
-    df = load_data(stock_name)
-    if df is None:
-        return {"error": "No data"}
-
-    model_path = os.path.join(MODEL_DIR, f"{stock_name}.h5")
-    if not os.path.exists(model_path):
-        train_model(stock_name)
-
-    model = load_model(model_path)
-
-    scaler = MinMaxScaler()
-    scaled_data = scaler.fit_transform(df[['Close']])
-    last_data = scaled_data[-3:]
-
-    predictions = []
-    input_seq = np.array([last_data])
-
-    for _ in range(steps):
-        pred = model.predict(input_seq, verbose=0)
-        predictions.append(pred[0][0])
-        input_seq = np.append(input_seq[:, 1:], [[pred]], axis=1)
-
-    predictions = scaler.inverse_transform(np.array(predictions).reshape(-1, 1)).flatten()
-    return {"predictions": predictions.tolist()}
-
-@app.get("/predict")
-def predict(stock: str, steps: int = 1):
-    return JSONResponse(predict_next(stock, steps=steps))
-
-def daily_trainer(scheduled_hour=9, scheduled_minute=12):
-    trained_today = False
-    while True:
-        now = datetime.datetime.now()
-        if now.hour == scheduled_hour and now.minute == scheduled_minute and not trained_today:
-            for file in os.listdir(DATA_DIR):
-                if file.endswith('.csv'):
-                    stock_name = file.replace('.csv', '')
-                    train_model(stock_name)
-            trained_today = True
-        elif now.minute != scheduled_minute:
-            trained_today = False
-        time.sleep(30)
-
-# Start scheduler in background
-import threading
-threading.Thread(target=daily_trainer, daemon=True).start()
+# worker_app.py
+from flask import Flask, request, jsonify
+import ray
+
+app = Flask(__name__)
+
+# ====== Yeh apna FIXED head node address hai ======
+HEAD_NODE_ADDRESS = "ray://192.168.1.100:10001"
+# ====================================================
+
+connected = False  # Worker connection status
+
+@app.route('/worker', methods=['POST'])
+def connect_worker():
+    global connected
+    
+    if connected:
+        return jsonify({"message": "Already connected to Ray head node."}), 200
+
+    try:
+        ray.init(address=HEAD_NODE_ADDRESS)  # Use the fixed head node address
+        connected = True
+        return jsonify({"message": f"Worker connected successfully to head node at {HEAD_NODE_ADDRESS}."}), 200
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+
+@app.route('/noworker', methods=['POST'])
+def disconnect_worker():
+    global connected
+    
+    if not connected:
+        return jsonify({"message": "Worker is already disconnected."}), 200
+
+    try:
+        ray.shutdown()
+        connected = False
+        return jsonify({"message": "Worker disconnected successfully."}), 200
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+
+@app.route('/')
+def home():
+    return "Worker Flask App Running."
+
+if __name__ == '__main__':
+    app.run(host='0.0.0.0', port=5000)