Spaces:
Sleeping
Sleeping
File size: 4,303 Bytes
e99527b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | import numpy as np
import pandas as pd
from fastapi import FastAPI, HTTPException, Request
from keras.layers import LSTM, Dense, Dropout
from keras.models import Sequential
from pydantic import BaseModel
from sklearn.preprocessing import MinMaxScaler
import json
import os
import tempfile
app = FastAPI()
class StockRequest(BaseModel):
stock_name: str
stock_data: list = None # Allow passing stock data directly
# We'll use in-memory storage instead of file paths
STOCK_DATA = {}
@app.post("/LSTM_Predict")
async def predict(request: Request):
# Parse the request body manually
body = await request.json()
# Create StockRequest object
if isinstance(body, dict):
stock_request = StockRequest(**body)
else:
raise HTTPException(status_code=422, detail="Invalid request format")
stock_name = stock_request.stock_name
# Check if data was passed directly in the request
if hasattr(stock_request, 'stock_data') and stock_request.stock_data:
# Use the data provided in the request
data = pd.DataFrame(stock_request.stock_data)
else:
# Try to get data from the request body directly
try:
data = pd.DataFrame(body.get('stock_data', []))
if data.empty:
raise HTTPException(status_code=422, detail="No stock data provided")
except Exception:
raise HTTPException(status_code=422, detail="Invalid stock data format")
# Convert columns to numeric, handling any non-numeric values
if 'Close' not in data.columns:
raise HTTPException(status_code=422, detail="Missing 'Close' column in data")
data['Close'] = pd.to_numeric(data['Close'], errors='coerce')
# Drop any rows with NaN values after conversion
data.dropna(subset=['Close'], inplace=True)
# Ensure we have enough data
if len(data) < 100:
raise HTTPException(status_code=400, detail="Not enough data for prediction (need at least 100 points)")
dataset = data[['Close']].values
train_len = int(np.ceil(len(dataset) * 0.8))
# Normalize the data
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(dataset)
# Prepare training data
train_data = scaled_data[:train_len, :]
seq_len = min(60, len(train_data) // 2) # Make sequence length adaptive
x_train, y_train = [], []
for i in range(seq_len, len(train_data)):
x_train.append(train_data[i - seq_len : i, 0])
y_train.append(train_data[i, 0])
x_train, y_train = np.array(x_train), np.array(y_train)
x_train = np.reshape(x_train, (x_train.shape[0], x_train.shape[1], 1))
# Build LSTM model
model = Sequential()
model.add(LSTM(64, return_sequences=True, input_shape=(x_train.shape[1], 1)))
model.add(Dropout(0.2))
model.add(LSTM(32, return_sequences=False))
model.add(Dropout(0.2))
model.add(Dense(1, activation="linear"))
model.compile(
optimizer="adam", loss="mean_squared_error", metrics=["mean_squared_error"]
)
# Train the model (reduced epochs for faster response)
model.fit(x_train, y_train, batch_size=32, epochs=5, verbose=0)
# Prepare data for prediction (we'll predict for the entire dataset)
x_all = []
for i in range(seq_len, len(scaled_data)):
x_all.append(scaled_data[i - seq_len : i, 0])
x_all = np.array(x_all)
x_all = np.reshape(x_all, (x_all.shape[0], x_all.shape[1], 1))
# Make predictions
predictions = model.predict(x_all)
predictions = scaler.inverse_transform(predictions)
# Create full predictions array (including the initial sequence length)
full_predictions = np.zeros((len(dataset), 1))
# Fill the first seq_len positions with actual values
full_predictions[:seq_len, 0] = dataset[:seq_len, 0]
# Fill the rest with predictions
full_predictions[seq_len:, 0] = predictions.flatten()
# Convert to list of floats for JSON response
predict_prices = [float(price) for price in full_predictions.flatten()]
return {"prediction": predict_prices}
# For testing locally
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000) |