MachineLearningCryptoModel / mltechnicalscanner.py
solanaexpert's picture
Update mltechnicalscanner.py
8870588 verified
import sys
import os
import ccxt
import pandas as pd
import numpy as np
from datetime import datetime
import ta
import argparse
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
import pickle
import warnings
import re
# Suppress warnings
warnings.filterwarnings('ignore')
# Configuration
pd.set_option('display.max_columns', None)
pd.set_option('display.max_rows', None)
pd.set_option('display.expand_frame_repr', True)
class MLTechnicalScanner:
def __init__(self, training_mode=False):
self.training_mode = training_mode
self.model = None
self.model_file = "technical_ml_model.pkl"
self.training_data_file = "training_data.csv"
self.results_file = "results.txt"
self.min_training_samples = 100
self.load_ml_model()
# Initialize exchanges
self.exchanges = {}
for id in ccxt.exchanges:
exchange = getattr(ccxt, id)
self.exchanges[id] = exchange()
# ML features configuration
self.feature_columns = [
'rsi', 'macd', 'bollinger_upper', 'bollinger_lower',
'volume_ma', 'ema_20', 'ema_50', 'adx'
]
# Performance tracking
self.performance_history = pd.DataFrame(columns=[
'timestamp', 'symbol', 'prediction', 'actual', 'profit'
])
# Training data collection
self.training_data = pd.DataFrame(columns=self.feature_columns + ['target'])
def init_results_file(self):
"""Initialize results file only when starting a new scan"""
with open(self.results_file, 'w') as f:
f.write("Scan Results Log\n")
f.write("="*50 + "\n")
f.write(f"Scan started at {datetime.now()}\n\n")
def log_result(self, message):
"""Log message to results file"""
try:
with open(self.results_file, 'a') as f:
f.write(message + '\n')
except Exception as e:
print(f"Error writing to results file: {str(e)}")
def load_ml_model(self):
"""Load trained ML model if exists"""
if os.path.exists(self.model_file):
with open(self.model_file, 'rb') as f:
self.model = pickle.load(f)
msg = "Loaded trained model from file"
print(msg)
self.log_result(msg)
else:
msg = "Initializing new model"
print(msg)
self.log_result(msg)
self.model = RandomForestClassifier(n_estimators=100, random_state=42)
def save_ml_model(self):
"""Save trained ML model"""
with open(self.model_file, 'wb') as f:
pickle.dump(self.model, f)
msg = "Saved model to file"
print(msg)
self.log_result(msg)
def load_training_data(self):
"""Load existing training data if available"""
if os.path.exists(self.training_data_file):
self.training_data = pd.read_csv(self.training_data_file)
msg = f"Loaded {len(self.training_data)} training samples"
print(msg)
self.log_result(msg)
def save_training_data(self):
"""Save training data to file"""
self.training_data.to_csv(self.training_data_file, index=False)
msg = f"Saved {len(self.training_data)} training samples"
print(msg)
self.log_result(msg)
def calculate_features(self, df):
"""Calculate technical indicators"""
try:
close = df['close'].astype(float)
high = df['high'].astype(float)
low = df['low'].astype(float)
volume = df['volume'].astype(float)
# Momentum Indicators
df['rsi'] = ta.momentum.rsi(close, window=14)
df['macd'] = ta.trend.macd_diff(close)
# Volatility Indicators
bollinger = ta.volatility.BollingerBands(close)
df['bollinger_upper'] = bollinger.bollinger_hband()
df['bollinger_lower'] = bollinger.bollinger_lband()
# Volume Indicators
df['volume_ma'] = volume.rolling(window=20).mean()
# Trend Indicators
df['ema_20'] = ta.trend.ema_indicator(close, window=20)
df['ema_50'] = ta.trend.ema_indicator(close, window=50)
df['adx'] = ta.trend.adx(high, low, close, window=14)
return df
except Exception as e:
error_msg = f"Error calculating features: {str(e)}"
print(error_msg)
self.log_result(error_msg)
return None
def train_initial_model(self):
"""Train initial model if we have enough data"""
self.load_training_data()
if len(self.training_data) >= self.min_training_samples:
X = self.training_data[self.feature_columns]
y = self.training_data['target']
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42
)
self.model.fit(X_train, y_train)
# Evaluate model
preds = self.model.predict(X_test)
accuracy = accuracy_score(y_test, preds)
msg = f"Initial model trained with accuracy: {accuracy:.2f}"
print(msg)
self.log_result(msg)
self.save_ml_model()
return True
else:
msg = f"Not enough training data ({len(self.training_data)} samples). Need at least {self.min_training_samples}."
print(msg)
self.log_result(msg)
return False
def predict_direction(self, features):
"""Predict price direction using ML model"""
try:
if self.model is None or not hasattr(self.model, 'classes_'):
return 0 # Neutral if no model
features = features[self.feature_columns].values.reshape(1, -1)
return self.model.predict(features)[0]
except Exception as e:
error_msg = f"Prediction error: {str(e)}"
print(error_msg)
self.log_result(error_msg)
return 0
def collect_training_sample(self, symbol, exchange, timeframe='1h'):
"""Collect data sample for training"""
try:
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=100)
if len(ohlcv) < 50:
return
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df = self.calculate_features(df)
if df is None:
return
current_price = df['close'].iloc[-1]
future_price = df['close'].iloc[-1] # Should be forward-looking in production
price_change = future_price - current_price
target = 1 if price_change > 0 else (-1 if price_change < 0 else 0)
features = df.iloc[-2].copy()
features['target'] = target
new_row = pd.DataFrame([features])
self.training_data = pd.concat([self.training_data, new_row], ignore_index=True)
msg = f"Collected training sample for {symbol}"
print(msg)
self.log_result(msg)
if len(self.training_data) % 10 == 0:
self.save_training_data()
except Exception as e:
error_msg = f"Error collecting training sample: {str(e)}"
print(error_msg)
self.log_result(error_msg)
def scan_symbol(self, symbol, exchange, timeframes):
"""Scan symbol for trading opportunities"""
try:
primary_tf = timeframes[0]
ohlcv = exchange.fetch_ohlcv(symbol, primary_tf, limit=100)
if len(ohlcv) < 50:
return
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df = self.calculate_features(df)
if df is None:
return
latest = df.iloc[-1].copy()
features = pd.DataFrame([latest[self.feature_columns]])
if self.training_mode:
self.collect_training_sample(symbol, exchange, primary_tf)
return
prediction = self.predict_direction(features)
# Simplified trend detection using EMA crossover
ema_20 = df['ema_20'].iloc[-1]
ema_50 = df['ema_50'].iloc[-1]
price = df['close'].iloc[-1]
uptrend = (ema_20 > ema_50) and (price > ema_20)
downtrend = (ema_20 < ema_50) and (price < ema_20)
if uptrend and prediction == 1:
self.alert(symbol, "STRONG UPTREND", timeframes, price)
elif downtrend and prediction == -1:
self.alert(symbol, "STRONG DOWNTREND", timeframes, price)
elif uptrend:
self.alert(symbol, "UPTREND", timeframes, price)
elif downtrend:
self.alert(symbol, "DOWNTREND", timeframes, price)
except Exception as e:
error_msg = f"Error scanning {symbol}: {str(e)}"
print(error_msg)
self.log_result(error_msg)
def alert(self, symbol, trend_type, timeframes, current_price):
"""Generate alert for detected trend"""
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
message = f"({trend_type}) detected for {symbol} at price {current_price} on {timeframes} at {timestamp}"
print(message)
self.log_result(message)
def compare_results(self, exchange_name):
"""Compare previous results with current prices"""
try:
if not os.path.exists(self.results_file):
print("No results file found to compare")
return
exchange = self.exchanges.get(exchange_name.lower())
if not exchange:
print(f"Exchange {exchange_name} not supported")
return
# Pattern to extract symbol and price from log entries
pattern = r"\((.*?)\) detected for (.*?) at price ([\d.]+) on"
with open(self.results_file, 'r') as f:
lines = f.readlines()
print("\n=== Price Comparison Report ===")
print(f"Generated at: {datetime.now()}\n")
for line in lines:
match = re.search(pattern, line)
if match:
trend_type = match.group(1)
symbol = match.group(2)
old_price = float(match.group(3))
timestamp = line.split(' at ')[-1].strip()
try:
ticker = exchange.fetch_ticker(symbol)
current_price = ticker['last']
price_change = current_price - old_price
percent_change = (price_change / old_price) * 100
print(f"Symbol: {symbol}")
print(f"Previous: {trend_type} at {old_price} ({timestamp})")
print(f"Current: {current_price} ({datetime.now().strftime('%Y-%m-%d %H:%M:%S')})")
print(f"Change: {price_change:.4f} ({percent_change:.2f}%)")
print("-" * 50)
except Exception as e:
print(f"Error fetching current price for {symbol}: {str(e)}")
continue
print("\n=== End of Report ===")
except Exception as e:
print(f"Error comparing results: {str(e)}")
# Main execution
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("-e", "--exchange", help="Exchange name", required=False)
parser.add_argument("-f", "--filter", help="Asset filter", required=False)
parser.add_argument("-tf", "--timeframes", help="Timeframes to scan (comma separated)", required=False)
parser.add_argument("--train", help="Run in training mode", action="store_true")
parser.add_argument("--compare", help="Compare previous results with current prices", action="store_true")
args = parser.parse_args()
if args.compare:
scanner = MLTechnicalScanner()
if args.exchange:
scanner.compare_results(args.exchange)
else:
print("Please specify an exchange with -e/--exchange when using --compare")
sys.exit(0)
if not all([args.exchange, args.filter, args.timeframes]):
print("Error: --exchange, --filter, and --timeframes are required when not using --compare")
sys.exit(1)
scanner = MLTechnicalScanner(training_mode=args.train)
# Initialize results file only for actual scans, not comparisons
scanner.init_results_file()
exchange = scanner.exchanges.get(args.exchange.lower())
if not exchange:
error_msg = f"Exchange {args.exchange} not supported"
print(error_msg)
scanner.log_result(error_msg)
sys.exit(1)
try:
markets = exchange.fetch_markets()
except Exception as e:
error_msg = f"Error fetching markets: {str(e)}"
print(error_msg)
scanner.log_result(error_msg)
sys.exit(1)
symbols = [
m['id'] for m in markets
if m['active'] and args.filter in m['id']
]
if not symbols:
error_msg = f"No symbols found matching filter {args.filter}"
print(error_msg)
scanner.log_result(error_msg)
sys.exit(1)
if args.train:
train_msg = f"Running in training mode for {len(symbols)} symbols"
print(train_msg)
scanner.log_result(train_msg)
for symbol in symbols:
scanner.collect_training_sample(symbol, exchange)
if scanner.train_initial_model():
success_msg = "Training completed successfully"
print(success_msg)
scanner.log_result(success_msg)
else:
fail_msg = "Not enough data collected for training"
print(fail_msg)
scanner.log_result(fail_msg)
sys.exit(0)
if not hasattr(scanner.model, 'classes_'):
warn_msg = "Warning: No trained model available. Running with basic scanning only."
print(warn_msg)
scanner.log_result(warn_msg)
timeframes = args.timeframes.split(',')
scan_msg = f"Scanning {len(symbols)} symbols on timeframes {timeframes}"
print(scan_msg)
scanner.log_result(scan_msg)
for symbol in symbols:
scanner.scan_symbol(symbol, exchange, timeframes)
# Add final summary
end_msg = f"\nScan completed at {datetime.now()}"
print(end_msg)
scanner.log_result(end_msg)