clean_test.py•7.98 kB
# clean_test.py - Clean working version
import warnings
warnings.filterwarnings('ignore')
def test_basic_functionality():
"""Test basic trading system functionality."""
print("Testing Basic Trading System Functionality")
print("=" * 50)
try:
import yfinance as yf
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
print("✅ All required packages imported successfully")
# Download data
ticker = 'AAPL'
print(f"Downloading {ticker} data...")
df = yf.download(ticker, period="6mo", interval="1d", progress=False)
if df.empty:
print("❌ No data downloaded")
return False
print(f"✅ Downloaded {len(df)} days of data")
# Clean data
df = df.dropna()
close_prices = df['Close']
# Simple ARIMA substitute (moving average)
window_size = 10
ma_forecast = float(close_prices.rolling(window=window_size).mean().iloc[-1])
print(f"✅ Moving average forecast: ${ma_forecast:.2f}")
# Create lag features
df_features = df.copy()
for i in range(1, 4):
df_features[f'Lag_{i}'] = df_features['Close'].shift(i)
# Remove NaN rows
df_features = df_features.dropna()
# Prepare features and target
X = df_features[['Lag_1', 'Lag_2', 'Lag_3']]
# Create simple residuals (price - moving average)
ma_values = close_prices.rolling(window=window_size).mean()
residuals = close_prices - ma_values
residuals = residuals.dropna()
# Align data
min_len = min(len(X), len(residuals))
X = X.tail(min_len)
y = residuals.tail(min_len)
print(f"✅ Feature matrix shape: {X.shape}")
print(f"✅ Target vector length: {len(y)}")
# Split data
if len(X) < 10:
print("❌ Insufficient data for modeling")
return False
split_size = max(0.2, 4/len(X)) # At least 4 samples for test
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=split_size, shuffle=False
)
print(f"✅ Train set: {len(X_train)} samples")
print(f"✅ Test set: {len(X_test)} samples")
# Train model
model = RandomForestRegressor(
n_estimators=50,
max_depth=4,
random_state=42
)
model.fit(X_train, y_train)
print("✅ Model trained successfully")
# Make predictions
if len(X_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}")
# Generate forecast
last_features = X.iloc[[-1]]
residual_correction = float(model.predict(last_features)[0])
hybrid_forecast = ma_forecast + residual_correction
last_price = float(close_prices.iloc[-1])
expected_change = (hybrid_forecast - last_price) / last_price
print(f"\n📊 FORECAST RESULTS:")
print(f"✅ Last price: ${last_price:.2f}")
print(f"✅ MA forecast: ${ma_forecast:.2f}")
print(f"✅ Residual correction: {residual_correction:.4f}")
print(f"✅ Hybrid forecast: ${hybrid_forecast:.2f}")
print(f"✅ Expected change: {expected_change:+.2%}")
# Generate trading signal
if expected_change > 0.02:
signal = "BUY"
emoji = "🟢"
elif expected_change < -0.02:
signal = "SELL"
emoji = "🔴"
else:
signal = "HOLD"
emoji = "🟡"
print(f"✅ Trading Signal: {emoji} {signal}")
print(f"\n🎉 BASIC FUNCTIONALITY TEST PASSED!")
return True
except Exception as e:
print(f"❌ Test failed: {e}")
import traceback
traceback.print_exc()
return False
def test_package_availability():
"""Test which advanced packages are available."""
print("\nTesting Package Availability...")
print("-" * 30)
packages = {
'yfinance': 'yfinance',
'pandas': 'pandas',
'numpy': 'numpy',
'sklearn': 'sklearn',
'pmdarima': 'pmdarima',
'xgboost': 'xgboost',
'lightgbm': 'lightgbm',
'statsmodels': 'statsmodels'
}
available = {}
for name, module in packages.items():
try:
__import__(module)
print(f"✅ {name}: Available")
available[name] = True
except ImportError:
print(f"❌ {name}: Not available")
available[name] = False
return available
def test_xgboost_specifically():
"""Test XGBoost specifically since it was problematic."""
print("\nTesting XGBoost Specifically...")
print("-" * 30)
try:
import xgboost as xgb
print("✅ XGBoost import successful")
# Test basic functionality
import numpy as np
X = np.random.random((100, 3))
y = np.random.random(100)
model = xgb.XGBRegressor(n_estimators=10, random_state=42)
model.fit(X, y)
pred = model.predict(X[:1])
print(f"✅ XGBoost basic test passed: prediction = {pred[0]:.4f}")
return True
except Exception as e:
print(f"❌ XGBoost test failed: {e}")
return False
def main():
"""Run all tests."""
print("🚀 COMPREHENSIVE TRADING SYSTEM TEST")
print("=" * 60)
# Test 1: Package availability
available_packages = test_package_availability()
# Test 2: XGBoost specific test
xgb_works = test_xgboost_specifically()
# Test 3: Basic functionality
basic_works = test_basic_functionality()
# Summary
print("\n" + "=" * 60)
print("📋 TEST SUMMARY")
print("=" * 60)
essential_packages = ['yfinance', 'pandas', 'numpy', 'sklearn']
essential_available = all(available_packages.get(pkg, False) for pkg in essential_packages)
if essential_available:
print("✅ Essential packages: ALL AVAILABLE")
else:
print("❌ Essential packages: MISSING SOME")
missing = [pkg for pkg in essential_packages if not available_packages.get(pkg, False)]
print(f" Missing: {', '.join(missing)}")
if available_packages.get('xgboost', False) and xgb_works:
print("✅ XGBoost: WORKING")
elif available_packages.get('lightgbm', False):
print("🟡 XGBoost: Not working, but LightGBM available")
else:
print("🟡 XGBoost: Not working, will use RandomForest")
if available_packages.get('pmdarima', False):
print("✅ ARIMA: pmdarima available")
elif available_packages.get('statsmodels', False):
print("🟡 ARIMA: statsmodels available (manual ARIMA)")
else:
print("🟡 ARIMA: Will use moving average")
if basic_works:
print("✅ Basic functionality: WORKING")
print("\n🎉 YOUR SYSTEM IS READY FOR TRADING ANALYSIS!")
print("\nNext steps:")
print("1. mkdir models && touch models/__init__.py")
print("2. Copy enhanced model files to models/")
print("3. python main.py")
else:
print("❌ Basic functionality: FAILED")
print("\n⚠️ System needs troubleshooting")
print("\n" + "=" * 60)
if __name__ == "__main__":
main()