final_working_test.py•8.64 kB
# final_working_test.py - All bugs fixed
import warnings
warnings.filterwarnings('ignore')
def test_imports():
"""Test which packages are working."""
print("Testing Package Imports...")
print("-" * 30)
# Test basic packages
try:
import yfinance as yf
print("✅ yfinance: OK")
except ImportError as e:
print(f"❌ yfinance: {e}")
try:
import pandas as pd
import numpy as np
print("✅ pandas/numpy: OK")
except ImportError as e:
print(f"❌ pandas/numpy: {e}")
try:
import sklearn
print("✅ scikit-learn: OK")
except ImportError as e:
print(f"❌ scikit-learn: {e}")
# Test ARIMA options
arima_option = None
try:
import pmdarima as pm
print("✅ pmdarima: OK")
arima_option = "pmdarima"
except ImportError:
try:
from sktime.forecasting.arima import ARIMA
print("✅ sktime ARIMA: OK")
arima_option = "sktime"
except ImportError:
try:
from statsmodels.tsa.arima.model import ARIMA
print("✅ statsmodels ARIMA: OK")
arima_option = "statsmodels"
except ImportError:
print("❌ No ARIMA implementation available")
# Test ML options
ml_option = None
try:
import xgboost as xgb
print("✅ xgboost: OK")
ml_option = "xgboost"
except ImportError as e:
print(f"❌ xgboost: {e}")
try:
import lightgbm as lgb
print("✅ lightgbm: OK")
ml_option = "lightgbm"
except ImportError:
print("✅ sklearn RandomForest: OK (fallback)")
ml_option = "sklearn"
return arima_option, ml_option
def simple_forecast_test(arima_option, ml_option):
"""Test basic forecasting with available packages."""
print(f"\nTesting Basic Forecast (ARIMA: {arima_option}, ML: {ml_option})")
print("-" * 50)
try:
import yfinance as yf
import pandas as pd
import numpy as np
# Get data
ticker = 'AAPL'
print(f"Downloading {ticker} data...")
df = yf.download(ticker, period="3mo", interval="1d", progress=False)
if df.empty:
print("❌ No data downloaded")
return False
print(f"✅ Downloaded {len(df)} days of data")
close_prices = df['Close'].dropna()
# Simple ARIMA forecast with proper error handling
arima_forecast = None
residuals = None
if arima_option == "pmdarima":
try:
import pmdarima as pm
model = pm.auto_arima(close_prices, seasonal=False, suppress_warnings=True)
arima_forecast = float(model.predict(n_periods=1)[0])
residuals = close_prices - model.predict_in_sample()
print("✅ Using pmdarima for ARIMA")
except Exception as e:
print(f"pmdarima failed: {e}")
arima_option = "fallback"
if arima_option == "statsmodels" and arima_forecast is None:
try:
from statsmodels.tsa.arima.model import ARIMA
# Fix: Extract values properly and ensure 1D array
close_values = close_prices.values.flatten() # Ensure 1D
model = ARIMA(close_values, order=(1, 1, 1))
fitted = model.fit()
# Get forecast properly
forecast_result = fitted.forecast(steps=1)
arima_forecast = float(forecast_result.iloc[0] if hasattr(forecast_result, 'iloc') else forecast_result[0])
residuals = pd.Series(fitted.resid, index=close_prices.index[:len(fitted.resid)])
print("✅ Using statsmodels for ARIMA")
except Exception as e:
print(f"statsmodels failed: {e}")
arima_option = "fallback"
# Fallback to simple moving average if ARIMA fails
if arima_forecast is None:
print("✅ Using simple moving average fallback")
arima_forecast = float(close_prices.rolling(window=5).mean().iloc[-1])
residuals = close_prices - close_prices.rolling(window=5).mean()
residuals = residuals.dropna()
# Ensure arima_forecast is a float
if hasattr(arima_forecast, 'iloc'):
arima_forecast = float(arima_forecast.iloc[0])
elif hasattr(arima_forecast, 'values'):
arima_forecast = float(arima_forecast.values[0])
else:
arima_forecast = float(arima_forecast)
print(f"✅ ARIMA forecast: ${arima_forecast:.2f}")
# Simple feature engineering
df_features = df.copy()
for i in range(1, 4):
df_features[f'Lag_{i}'] = df_features['Close'].shift(i)
df_features = df_features.dropna()
# Ensure we have enough data
if len(df_features) < 10:
print("❌ Insufficient data for ML model")
return False
X = df_features[['Lag_1', 'Lag_2', 'Lag_3']]
# Align residuals with features
min_len = min(len(X), len(residuals))
X = X.tail(min_len)
y = residuals.tail(min_len) if hasattr(residuals, 'tail') else residuals[-min_len:]
# Convert to numpy arrays to avoid pandas issues
X_values = X.values
y_values = y.values if hasattr(y, 'values') else np.array(y)
# Simple ML model
from sklearn.model_selection import train_test_split
if len(X_values) < 4:
print("❌ Insufficient data for train/test split")
return False
X_train, X_test, y_train, y_test = train_test_split(
X_values, y_values, test_size=0.3, shuffle=False
)
if ml_option == "xgboost":
from xgboost import XGBRegressor
model = XGBRegressor(n_estimators=50, max_depth=3, random_state=42)
print("✅ Using XGBoost for ML")
elif ml_option == "lightgbm":
import lightgbm as lgb
model = lgb.LGBMRegressor(n_estimators=50, max_depth=3, random_state=42, verbose=-1)
print("✅ Using LightGBM for ML")
else:
from sklearn.ensemble import RandomForestRegressor
model = RandomForestRegressor(n_estimators=50, max_depth=3, random_state=42)
print("✅ Using RandomForest for ML")
model.fit(X_train, y_train)
# Make prediction on last available features
last_features = X_values[[-1]] # Use numpy array
residual_correction = float(model.predict(last_features)[0])
# Calculate final forecast
hybrid_forecast = arima_forecast + residual_correction
last_price = float(close_prices.iloc[-1])
change = (hybrid_forecast - last_price) / last_price
print(f"✅ Last price: ${last_price:.2f}")
print(f"✅ ARIMA forecast: ${arima_forecast:.2f}")
print(f"✅ Residual correction: {residual_correction:.4f}")
print(f"✅ Hybrid forecast: ${hybrid_forecast:.2f}")
print(f"✅ Expected change: {change:+.2%}")
# Generate signal
if change > 0.02:
signal = "BUY"
elif change < -0.02:
signal = "SELL"
else:
signal = "HOLD"
print(f"✅ Trading Signal: {signal}")
print("\n🎉 Basic forecast test SUCCESSFUL!")
# Test model accuracy
if len(y_test) > 0:
y_pred = model.predict(X_test)
from sklearn.metrics import mean_squared_error, r2_score
mse = mean_squared_error(y_test, y_pred)
r2 = r2_score(y_test, y_pred)
print(f"✅ Model MSE: {mse:.6f}")
print(f"✅ Model R²: {r2:.4f}")
return True
except Exception as e:
print(f"❌ Forecast test failed: {e}")
import traceback
traceback.print_exc()
return False
def test_original_approach():
"""Test the original simple approach that should definitely work."""
print("\nTesting Original Simple Approach...")
print("-" * 40)
try:
import yfinance as yf
import pandas as pd
import numpy as np