184 lines
5.6 KiB
Python
184 lines
5.6 KiB
Python
"""
|
|
System Test Script
|
|
Verifies that the trading system is properly installed and configured.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
def test_imports():
|
|
"""Test that all required modules can be imported"""
|
|
print("Testing imports...")
|
|
|
|
try:
|
|
import pandas as pd
|
|
print("✓ pandas imported successfully")
|
|
except ImportError as e:
|
|
print(f"✗ pandas import failed: {e}")
|
|
return False
|
|
|
|
try:
|
|
import alpaca_trade_api as tradeapi
|
|
print("✓ alpaca_trade_api imported successfully")
|
|
except ImportError as e:
|
|
print(f"✗ alpaca_trade_api import failed: {e}")
|
|
return False
|
|
|
|
# Test local imports
|
|
sys.path.append('src')
|
|
sys.path.append('config')
|
|
|
|
try:
|
|
from config.trading_config import trading_config, alpaca_config
|
|
print("✓ trading configuration imported successfully")
|
|
except ImportError as e:
|
|
print(f"✗ trading configuration import failed: {e}")
|
|
return False
|
|
|
|
try:
|
|
from src.data_handler import DataHandler
|
|
print("✓ DataHandler imported successfully")
|
|
except ImportError as e:
|
|
print(f"✗ DataHandler import failed: {e}")
|
|
return False
|
|
|
|
try:
|
|
from src.strategy import TrendFollowingStrategy
|
|
print("✓ TrendFollowingStrategy imported successfully")
|
|
except ImportError as e:
|
|
print(f"✗ TrendFollowingStrategy import failed: {e}")
|
|
return False
|
|
|
|
try:
|
|
from src.risk_manager import RiskManager
|
|
print("✓ RiskManager imported successfully")
|
|
except ImportError as e:
|
|
print(f"✗ RiskManager import failed: {e}")
|
|
return False
|
|
|
|
try:
|
|
from src.trading_engine import TradingEngine
|
|
print("✓ TradingEngine imported successfully")
|
|
except ImportError as e:
|
|
print(f"✗ TradingEngine import failed: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def test_configuration():
|
|
"""Test configuration settings"""
|
|
print("\nTesting configuration...")
|
|
|
|
try:
|
|
from config.trading_config import trading_config, alpaca_config
|
|
|
|
print(f"✓ Trading symbol: {trading_config.symbol}")
|
|
print(f"✓ Risk per trade: {trading_config.risk_per_trade}")
|
|
print(f"✓ Max position size: {trading_config.max_position_size}")
|
|
print(f"✓ Alpaca base URL: {alpaca_config.base_url}")
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"✗ Configuration test failed: {e}")
|
|
return False
|
|
|
|
def test_data_connection():
|
|
"""Test connection to data source"""
|
|
print("\nTesting data connection...")
|
|
|
|
try:
|
|
from src.data_handler import DataHandler
|
|
|
|
data_handler = DataHandler()
|
|
print("✓ DataHandler initialized")
|
|
|
|
# Test getting a small amount of recent data
|
|
try:
|
|
bars = data_handler.get_historical_data(days=1)
|
|
if not bars.empty:
|
|
print(f"✓ Data connection successful - got {len(bars)} data points")
|
|
print(f"✓ Latest price: ${bars['close'].iloc[-1]:.2f}")
|
|
return True
|
|
else:
|
|
print("✗ No data received")
|
|
return False
|
|
except Exception as e:
|
|
print(f"✗ Data fetch failed: {e}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"✗ DataHandler initialization failed: {e}")
|
|
return False
|
|
|
|
def test_directory_structure():
|
|
"""Test that required directories exist"""
|
|
print("\nTesting directory structure...")
|
|
|
|
required_dirs = ['src', 'config', 'logs']
|
|
|
|
for dir_name in required_dirs:
|
|
if os.path.exists(dir_name):
|
|
print(f"✓ {dir_name}/ directory exists")
|
|
else:
|
|
print(f"✗ {dir_name}/ directory missing")
|
|
return False
|
|
|
|
return True
|
|
|
|
def main():
|
|
"""Run all tests"""
|
|
print("=" * 50)
|
|
print("TRADING SYSTEM INSTALLATION TEST")
|
|
print("=" * 50)
|
|
|
|
tests = [
|
|
("Directory Structure", test_directory_structure),
|
|
("Module Imports", test_imports),
|
|
("Configuration", test_configuration),
|
|
("Data Connection", test_data_connection)
|
|
]
|
|
|
|
results = []
|
|
|
|
for test_name, test_func in tests:
|
|
print(f"\n--- {test_name} ---")
|
|
try:
|
|
result = test_func()
|
|
results.append((test_name, result))
|
|
except Exception as e:
|
|
print(f"✗ {test_name} failed with exception: {e}")
|
|
results.append((test_name, False))
|
|
|
|
print("\n" + "=" * 50)
|
|
print("TEST SUMMARY")
|
|
print("=" * 50)
|
|
|
|
all_passed = True
|
|
for test_name, result in results:
|
|
status = "PASS" if result else "FAIL"
|
|
symbol = "✓" if result else "✗"
|
|
print(f"{symbol} {test_name}: {status}")
|
|
if not result:
|
|
all_passed = False
|
|
|
|
print("=" * 50)
|
|
|
|
if all_passed:
|
|
print("🎉 All tests passed! The trading system is ready to use.")
|
|
print("\nNext steps:")
|
|
print("1. Review configuration in config/trading_config.py")
|
|
print("2. Run backtesting: python main.py --mode backtest")
|
|
print("3. For live trading: python main.py --mode live")
|
|
else:
|
|
print("❌ Some tests failed. Please fix the issues before proceeding.")
|
|
print("\nCommon solutions:")
|
|
print("1. Install dependencies: pip install -r requirements.txt")
|
|
print("2. Check API credentials in .env file")
|
|
print("3. Verify internet connection")
|
|
|
|
return 0 if all_passed else 1
|
|
|
|
if __name__ == "__main__":
|
|
exit_code = main()
|
|
sys.exit(exit_code)
|