83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
import os
|
|
import sys
|
|
import joblib
|
|
import numpy as np
|
|
os.environ["CUDA_VISIBLE_DEVICES"] = "-1"
|
|
import yfinance as yf
|
|
import features
|
|
import matplotlib
|
|
matplotlib.use("Agg")
|
|
|
|
def Predict():
|
|
|
|
# Get the Symbol from ARGV
|
|
Symbol = sys.argv[1]
|
|
|
|
# Define paths (consistent with your previous script)
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
DATA_DIR = os.path.join(SCRIPT_DIR, "data")
|
|
MODEL_PATH = os.path.join(DATA_DIR, "model.keras")
|
|
|
|
# Pull 1 month of current data to make prediction against | for volatility 20
|
|
df = yf.download(Symbol, period="2mo", auto_adjust=True)
|
|
if not df.empty:
|
|
# Remove the horizontal ticker column
|
|
df.columns = df.columns.get_level_values(0)
|
|
# Add in the Vertical ticker column
|
|
df['Ticker'] = Symbol
|
|
# Make the feature set
|
|
df = features.MakeFeatures(df)
|
|
|
|
# Drop our predictor
|
|
df.drop('Target_Close', axis=1, inplace=True)
|
|
|
|
# Lazy load this so it doesnt interfere with yfinance
|
|
from keras.models import load_model
|
|
|
|
# Load the model
|
|
reconstructed_model = load_model(MODEL_PATH)
|
|
|
|
# Verify it loaded correctly
|
|
reconstructed_model.summary()
|
|
|
|
# Load the scalers
|
|
feature_scaler = joblib.load(os.path.join(DATA_DIR, "feature_scaler.pkl"))
|
|
target_scaler = joblib.load(os.path.join(DATA_DIR, "target_scaler.pkl"))
|
|
|
|
# Scale the data
|
|
scaled_data = feature_scaler.transform(df)
|
|
|
|
# Predict
|
|
scaled_predictions = reconstructed_model.predict(scaled_data)
|
|
|
|
# Use the loaded target scaler to get back to % change
|
|
actual_prediction = target_scaler.inverse_transform(scaled_predictions.reshape(-1, 1)).flatten()
|
|
|
|
# 'predictions' will be a 2D array, flatten it if you want a simple list
|
|
flat_predictions = actual_prediction
|
|
|
|
# Get the overall trend to pull predictions from
|
|
predictionTrend = 0
|
|
with open("Target_Close_Average.txt", "r") as f:
|
|
predictionTrend = float(f.read().strip())
|
|
|
|
# Set the movement indicator
|
|
movement_indicator = 0
|
|
averagePrediction = np.mean(flat_predictions) + predictionTrend
|
|
if (averagePrediction > 0.3): # as in 3% swing up
|
|
movement_indicator = 1
|
|
elif (averagePrediction < -0.3): # as in 3% swing down
|
|
movement_indicator = -1
|
|
else:
|
|
movement_indicator = 0
|
|
|
|
# Debug data
|
|
print(f"averagePrediction: {averagePrediction}")
|
|
|
|
# Return to C# via stdout
|
|
print(f"---RESULT_START---")
|
|
print(movement_indicator)
|
|
print(f"---RESULT_END---")
|
|
|
|
if __name__ == "__main__":
|
|
Predict() |