import numpy as np import pandas as pd import sympy as sp import MetaTrader5 as mt5 from datetime import datetime, timedelta from sklearn.preprocessing import StandardScaler, PolynomialFeatures from sklearn.linear_model import LinearRegression, Ridge, LogisticRegression from sklearn.metrics import mean_squared_error, r2_score, accuracy_score, precision_score, recall_score, f1_score, confusion_matrix, classification_report from sklearn.model_selection import train_test_split, cross_val_score from scipy.optimize import minimize import warnings warnings.filterwarnings('ignore') class SymbolicPricePredictor: """ Symbolic price forecasting model after 24 H1 bars using SymPy for creating an accurate mathematic equation + binary classification for price movement direction """ def __init__(self, symbol: str = "EURUSD", terminal_path: str = None): """ Initialize predictor Args: symbol: Currency pair for analysis terminal_path: Path to MT5 terminal """ self.symbol = symbol self.terminal_path = terminal_path self.prediction_horizon = 24 # 24 H1 bars self.lookback_bars = 10000 # 10,000 historical bars # Symbolic SymPy variables self.t = sp.Symbol('t', real=True) # time self.p = sp.Symbol('p', real=True) # price self.v = sp.Symbol('v', real=True) # volume self.r = sp.Symbol('r', real=True) # return self.vol = sp.Symbol('vol', real=True) # volatility self.mom = sp.Symbol('mom', real=True) # momentum self.rsi = sp.Symbol('rsi', real=True) # RSI self.ma = sp.Symbol('ma', real=True) # moving average self.bb = sp.Symbol('bb', real=True) # bollinger bands self.macd = sp.Symbol('macd', real=True) # MACD # Models and equations self.price_equation = None self.binary_equation = None self.price_function = None self.binary_function = None # Machine learning models self.ridge_model = None self.logistic_model = None self.poly_features = None self.poly_features_binary = None # Data and metrics self.data = None self.features = None self.scaler = StandardScaler() self.scaler_binary = StandardScaler() # Binary classification metrics self.binary_metrics = {} print(f"SymbolicPricePredictor initialized for {symbol}") print(f"Prediction horizon: {self.prediction_horizon} H1 bars") print(f"Historical data: {self.lookback_bars} bars") print("Features: Price prediction + Binary direction classification") def connect_mt5(self) -> bool: """Connect to MT5""" if self.terminal_path: if not mt5.initialize(path=self.terminal_path): print(f"MT5 initialization failed: {mt5.last_error()}") return False else: if not mt5.initialize(): print(f"MT5 initialization failed: {mt5.last_error()}") return False return True def disconnect_mt5(self): """Disconnect from MT5""" mt5.shutdown() def get_historical_data(self) -> pd.DataFrame: """Get historical data""" try: if not self.connect_mt5(): print("Failed to connect to MT5") return None # Get data rates = mt5.copy_rates_from_pos( self.symbol, mt5.TIMEFRAME_H1, 0, self.lookback_bars + self.prediction_horizon + 100 ) self.disconnect_mt5() if rates is None or len(rates) == 0: print(f"No data received for {self.symbol}") return None # Create DataFrame df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') df.set_index('time', inplace=True) print(f"Received {len(df)} historical bars") return df except Exception as e: print(f"Error getting historical data: {e}") self.disconnect_mt5() return None def calculate_technical_indicators(self, df: pd.DataFrame) -> pd.DataFrame: """Calculate technical indicators""" try: data = df.copy() # Basic calculations data['returns'] = data['close'].pct_change() data['log_returns'] = np.log(data['close'] / data['close'].shift(1)) data['price_change'] = data['close'] - data['close'].shift(1) data['high_low_ratio'] = data['high'] / data['low'] data['open_close_ratio'] = data['open'] / data['close'] # Volatility (different periods) for period in [5, 10, 20, 50]: data[f'volatility_{period}'] = data['returns'].rolling(period).std() * np.sqrt(24) data[f'realized_vol_{period}'] = data['log_returns'].rolling(period).std() * np.sqrt(24) # Moving averages for period in [5, 10, 20, 50, 100, 200]: data[f'sma_{period}'] = data['close'].rolling(period).mean() data[f'ema_{period}'] = data['close'].ewm(span=period).mean() data[f'price_to_sma_{period}'] = data['close'] / data[f'sma_{period}'] - 1 # RSI def calculate_rsi(prices, period=14): delta = prices.diff() gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi data['rsi_14'] = calculate_rsi(data['close'], 14) data['rsi_7'] = calculate_rsi(data['close'], 7) data['rsi_21'] = calculate_rsi(data['close'], 21) # MACD ema_12 = data['close'].ewm(span=12).mean() ema_26 = data['close'].ewm(span=26).mean() data['macd'] = ema_12 - ema_26 data['macd_signal'] = data['macd'].ewm(span=9).mean() data['macd_histogram'] = data['macd'] - data['macd_signal'] # Bollinger Bands for period in [20, 50]: sma = data['close'].rolling(period).mean() std = data['close'].rolling(period).std() data[f'bb_upper_{period}'] = sma + 2 * std data[f'bb_lower_{period}'] = sma - 2 * std data[f'bb_position_{period}'] = (data['close'] - data[f'bb_lower_{period}']) / (data[f'bb_upper_{period}'] - data[f'bb_lower_{period}']) # Momentum indicators for period in [5, 10, 20]: data[f'momentum_{period}'] = data['close'] / data['close'].shift(period) - 1 data[f'roc_{period}'] = (data['close'] - data['close'].shift(period)) / data['close'].shift(period) * 100 # Volume indicators data['volume_sma_20'] = data['tick_volume'].rolling(20).mean() data['volume_ratio'] = data['tick_volume'] / data['volume_sma_20'] data['price_volume'] = data['close'] * data['tick_volume'] # Stochastic for period in [14, 21]: low_min = data['low'].rolling(period).min() high_max = data['high'].rolling(period).max() data[f'stoch_k_{period}'] = 100 * (data['close'] - low_min) / (high_max - low_min) data[f'stoch_d_{period}'] = data[f'stoch_k_{period}'].rolling(3).mean() # ATR data['tr'] = np.maximum( data['high'] - data['low'], np.maximum( np.abs(data['high'] - data['close'].shift(1)), np.abs(data['low'] - data['close'].shift(1)) ) ) data['atr_14'] = data['tr'].rolling(14).mean() data['atr_ratio'] = data['tr'] / data['atr_14'] # Fractal indicators data['fractal_dimension'] = self._calculate_fractal_dimension(data['close']) data['hurst_exponent'] = self._calculate_hurst_exponent(data['close']) # Temporary features data['hour'] = data.index.hour data['day_of_week'] = data.index.dayofweek data['month'] = data.index.month data['hour_sin'] = np.sin(2 * np.pi * data['hour'] / 24) data['hour_cos'] = np.cos(2 * np.pi * data['hour'] / 24) data['dow_sin'] = np.sin(2 * np.pi * data['day_of_week'] / 7) data['dow_cos'] = np.cos(2 * np.pi * data['day_of_week'] / 7) # Lag features for lag in [1, 2, 3, 5, 10, 20]: data[f'price_lag_{lag}'] = data['close'].shift(lag) data[f'returns_lag_{lag}'] = data['returns'].shift(lag) data[f'volume_lag_{lag}'] = data['tick_volume'].shift(lag) print(f"Calculated {len([col for col in data.columns if col not in df.columns])} technical indicators") return data except Exception as e: print(f"Error calculating technical indicators: {e}") return df def _calculate_fractal_dimension(self, prices, max_lag=50): """Calculate fractal dimensionality""" try: prices_array = prices.dropna().values n = len(prices_array) if n < max_lag * 2: return pd.Series([1.5] * len(prices), index=prices.index) lags = range(2, min(max_lag, n//4)) rs_values = [] for lag in lags: segments = n // lag local_rs = [] for i in range(segments): start_idx = i * lag end_idx = start_idx + lag segment = prices_array[start_idx:end_idx] if len(segment) < lag: continue mean_val = np.mean(segment) deviations = segment - mean_val cumulative_deviations = np.cumsum(deviations) R = np.max(cumulative_deviations) - np.min(cumulative_deviations) S = np.std(segment) if S > 0: local_rs.append(R / S) if local_rs: rs_values.append(np.mean(local_rs)) else: rs_values.append(1.0) if len(rs_values) > 1: log_lags = np.log(lags[:len(rs_values)]) log_rs = np.log(rs_values) valid_idx = np.isfinite(log_rs) & (log_rs > 0) if np.sum(valid_idx) > 1: hurst = np.polyfit(log_lags[valid_idx], log_rs[valid_idx], 1)[0] fractal_dim = 2 - hurst else: fractal_dim = 1.5 else: fractal_dim = 1.5 return pd.Series([max(1.0, min(2.0, fractal_dim))] * len(prices), index=prices.index) except Exception as e: print(f"Error calculating fractal dimension: {e}") return pd.Series([1.5] * len(prices), index=prices.index) def _calculate_hurst_exponent(self, prices, max_lag=100): """Calculate Hurst exponent""" try: prices_clean = prices.dropna() if len(prices_clean) < max_lag * 2: return pd.Series([0.5] * len(prices), index=prices.index) lags = range(2, min(max_lag, len(prices_clean)//4)) variances = [] for lag in lags: differences = prices_clean.diff(lag).dropna() if len(differences) > 0: variances.append(np.var(differences)) else: variances.append(1.0) if len(variances) > 1: log_lags = np.log(lags[:len(variances)]) log_vars = np.log(variances) valid_idx = np.isfinite(log_vars) if np.sum(valid_idx) > 1: hurst = np.polyfit(log_lags[valid_idx], log_vars[valid_idx], 1)[0] / 2 else: hurst = 0.5 else: hurst = 0.5 return pd.Series([max(0.0, min(1.0, hurst))] * len(prices), index=prices.index) except Exception as e: print(f"Error calculating Hurst exponent: {e}") return pd.Series([0.5] * len(prices), index=prices.index) def prepare_features_and_targets(self, data: pd.DataFrame): """Prepare features and target variables""" try: # Select numerical columns (excluding datetime and object) numeric_columns = data.select_dtypes(include=[np.number]).columns.tolist() # Exclude basic OHLCV columns from features (keep the derivatives) feature_columns = [col for col in numeric_columns if col not in ['open', 'high', 'low', 'close', 'tick_volume', 'spread', 'real_volume']] # Create target variables data['target_price'] = data['close'].shift(-self.prediction_horizon) data['target_return'] = (data['target_price'] / data['close'] - 1) * 100 # Binary target variable: 1 = rise, 0 = fall data['target_direction'] = (data['target_price'] > data['close']).astype(int) # Remove NaN data_clean = data.dropna() if len(data_clean) < 100: print("Insufficient clean data after removing NaN") return None, None # Form features and targets X = data_clean[feature_columns].values y_price = data_clean['target_price'].values y_return = data_clean['target_return'].values y_direction = data_clean['target_direction'].values # Normalize features for regression X_scaled = self.scaler.fit_transform(X) # Separate normalization for binary classification X_scaled_binary = self.scaler_binary.fit_transform(X) print(f"Features prepared: {X_scaled.shape}") print(f"Feature columns: {len(feature_columns)}") print(f"Clean samples: {len(data_clean)}") print(f"Direction distribution: UP={np.sum(y_direction)}, DOWN={len(y_direction)-np.sum(y_direction)}") print(f"Direction balance: {np.mean(y_direction)*100:.1f}% UP, {(1-np.mean(y_direction))*100:.1f}% DOWN") return { 'X': X_scaled, 'X_binary': X_scaled_binary, 'y_price': y_price, 'y_return': y_return, 'y_direction': y_direction, 'feature_names': feature_columns, 'data_clean': data_clean }, data_clean except Exception as e: print(f"Error preparing features: {e}") return None, None def create_symbolic_equations(self, features_data: dict): """Create symbolic equations for price and direction""" try: X = features_data['X'] X_binary = features_data['X_binary'] y_price = features_data['y_price'] y_direction = features_data['y_direction'] feature_names = features_data['feature_names'] print("Creating symbolic equations...") # Create symbolic variables for each feature symbols = {} for i, name in enumerate(feature_names): short_name = f"x{i}" symbols[short_name] = sp.Symbol(short_name, real=True) # === PRICE EQUATION === print("Training price regression model...") poly_features = PolynomialFeatures(degree=2, include_bias=True, interaction_only=False) X_poly = poly_features.fit_transform(X) ridge = Ridge(alpha=1.0) ridge.fit(X_poly, y_price) print(f"Price model R² score: {ridge.score(X_poly, y_price):.4f}") # Create symbolic features for price symbol_list = list(symbols.values()) # Create polynomial terms symbol by symbol poly_terms = [] poly_terms.append(sp.S(1)) # constant # Lineat terms for sym in symbol_list: poly_terms.append(sym) # Quadratic terms for sym in symbol_list: poly_terms.append(sym**2) # Interactions for i in range(len(symbol_list)): for j in range(i+1, len(symbol_list)): poly_terms.append(symbol_list[i] * symbol_list[j]) # Limit the number of terms poly_terms = poly_terms[:len(ridge.coef_)] # Create symbolic price equation price_equation = sp.S(0) for coef, term in zip(ridge.coef_, poly_terms): if abs(coef) > 1e-6: price_equation += coef * term price_equation = sp.simplify(price_equation) # === DIRECTION EQUATION === print("Training binary classification model...") # Divide data for validating binary model X_train, X_test, y_train, y_test = train_test_split( X_binary, y_direction, test_size=0.3, random_state=42, stratify=y_direction ) poly_features_binary = PolynomialFeatures(degree=2, include_bias=True, interaction_only=False) X_train_poly = poly_features_binary.fit_transform(X_train) X_test_poly = poly_features_binary.transform(X_test) logistic = LogisticRegression(random_state=42, max_iter=1000, C=1.0) logistic.fit(X_train_poly, y_train) # Evaluate model train_accuracy = logistic.score(X_train_poly, y_train) test_accuracy = logistic.score(X_test_poly, y_test) y_pred = logistic.predict(X_test_poly) y_pred_proba = logistic.predict_proba(X_test_poly)[:, 1] # Calculate metrics precision = precision_score(y_test, y_pred) recall = recall_score(y_test, y_pred) f1 = f1_score(y_test, y_pred) # Cross-validation cv_scores = cross_val_score(logistic, X_train_poly, y_train, cv=5, scoring='accuracy') self.binary_metrics = { 'train_accuracy': train_accuracy, 'test_accuracy': test_accuracy, 'precision': precision, 'recall': recall, 'f1_score': f1, 'cv_mean_accuracy': cv_scores.mean(), 'cv_std_accuracy': cv_scores.std(), 'confusion_matrix': confusion_matrix(y_test, y_pred), 'classification_report': classification_report(y_test, y_pred) } print(f"Binary classification test accuracy: {test_accuracy:.4f}") print(f"Cross-validation accuracy: {cv_scores.mean():.4f} (±{cv_scores.std():.4f})") # Create symbolic equation for logistic regression # Logistic function: P(y=1) = 1 / (1 + exp(-z)), where z = linear combination # Limit the number of terms for readability binary_terms = poly_terms[:len(logistic.coef_[0])] # Create a linear combination linear_combination = sp.S(0) for coef, term in zip(logistic.coef_[0], binary_terms): if abs(coef) > 1e-6: linear_combination += coef * term # Create full logistic equation binary_equation = 1 / (1 + sp.exp(-linear_combination)) binary_equation = sp.simplify(binary_equation) print("Symbolic equations created successfully") print(f"Price equation complexity: {len(price_equation.args)} terms") print(f"Binary equation complexity: {len(binary_equation.args)} main terms") # Save results self.price_equation = price_equation self.binary_equation = binary_equation self.symbols = symbols self.feature_names = feature_names self.poly_features = poly_features self.poly_features_binary = poly_features_binary self.ridge_model = ridge self.logistic_model = logistic # Create numerical functions self.price_function = sp.lambdify( list(symbols.values()), price_equation, 'numpy' ) self.binary_function = sp.lambdify( list(symbols.values()), binary_equation, 'numpy' ) return price_equation, binary_equation except Exception as e: print(f"Error creating symbolic equations: {e}") return None, None def display_equations(self): """Display equations in readable form""" if self.price_equation is None or self.binary_equation is None: print("No equations available") return print("\n" + "="*80) print("SYMBOLIC PRICE PREDICTION EQUATIONS") print("="*80) # Display price equation print(f"\n1. PRICE PREDICTION EQUATION:") print(f"P(t+24) = {self.price_equation}") # Display direction equation print(f"\n2. BINARY DIRECTION EQUATION:") print(f"Prob(UP) = {self.binary_equation}") print("where Prob(UP) > 0.5 indicates price will go UP, < 0.5 indicates DOWN") # Display the match of variables print(f"\nVariable mapping:") for i, name in enumerate(self.feature_names): print(f" x{i} = {name}") # Display price statistics if hasattr(self, 'ridge_model'): print(f"\nPrice Model Statistics:") print(f" R² Score: {self.ridge_model.score(self.poly_features.transform(self.features['X']), self.features['y_price']):.4f}") print(f" Number of features: {len(self.feature_names)}") print(f" Polynomial degree: 2") # Display binary classification statistics if self.binary_metrics: print(f"\nBinary Classification Statistics:") print(f" Test Accuracy: {self.binary_metrics['test_accuracy']:.4f}") print(f" Cross-Val Accuracy: {self.binary_metrics['cv_mean_accuracy']:.4f} (±{self.binary_metrics['cv_std_accuracy']:.4f})") print(f" Precision: {self.binary_metrics['precision']:.4f}") print(f" Recall: {self.binary_metrics['recall']:.4f}") print(f" F1-Score: {self.binary_metrics['f1_score']:.4f}") print(f"\nConfusion Matrix:") cm = self.binary_metrics['confusion_matrix'] print(f" Predicted") print(f"Actual DOWN UP") print(f"DOWN {cm[0,0]:4d} {cm[0,1]:4d}") print(f"UP {cm[1,0]:4d} {cm[1,1]:4d}") print("="*80) def predict_future_price(self, current_data: dict) -> dict: """Forecast future price and direction""" try: if self.price_function is None or self.binary_function is None: print("No trained models available") return None # Prepare inputs input_values = [] for name in self.feature_names: if name in current_data: input_values.append(current_data[name]) else: input_values.append(0.0) # Default value # Normalize data for regression input_array = np.array(input_values).reshape(1, -1) input_scaled = self.scaler.transform(input_array)[0] # Normalize data for classification input_scaled_binary = self.scaler_binary.transform(input_array)[0] # Price forecast via symbolic function price_prediction = self.price_function(*input_scaled) # Forecast direction via symbolic function direction_probability = self.binary_function(*input_scaled_binary) direction_predicted = 1 if direction_probability > 0.5 else 0 # Also forecasts via ML models for comparing X_poly = self.poly_features.transform(input_scaled.reshape(1, -1)) ridge_prediction = self.ridge_model.predict(X_poly)[0] X_poly_binary = self.poly_features_binary.transform(input_scaled_binary.reshape(1, -1)) logistic_probability = self.logistic_model.predict_proba(X_poly_binary)[0, 1] logistic_direction = self.logistic_model.predict(X_poly_binary)[0] current_price = current_data.get('close', 0) result = { # Price 'predicted_price_24h': float(price_prediction), 'ridge_prediction': float(ridge_prediction), 'current_price': current_price, 'price_change_pct': ((price_prediction - current_price) / current_price * 100) if current_price > 0 else 0, # Direction 'direction_probability': float(direction_probability), 'direction_predicted': int(direction_predicted), 'direction_text': 'UP' if direction_predicted == 1 else 'DOWN', 'logistic_probability': float(logistic_probability), 'logistic_direction': int(logistic_direction), # Meta data 'prediction_horizon_hours': self.prediction_horizon, 'confidence_score': min(abs(price_prediction - ridge_prediction) / current_price, 1.0) if current_price > 0 else 0, 'direction_confidence': abs(direction_probability - 0.5) * 2 # Normalize to [0,1] } return result except Exception as e: print(f"Error predicting future price: {e}") return None def evaluate_models(self, features_data: dict): """Evaluate model quality on test data""" try: print("Evaluating models performance...") X = features_data['X'] X_binary = features_data['X_binary'] y_price = features_data['y_price'] y_direction = features_data['y_direction'] # Divide data by train/test test_size = 0.2 split_idx = int(len(X) * (1 - test_size)) # For price X_train, X_test = X[:split_idx], X[split_idx:] y_price_train, y_price_test = y_price[:split_idx], y_price[split_idx:] # For direction X_binary_train, X_binary_test = X_binary[:split_idx], X_binary[split_idx:] y_dir_train, y_dir_test = y_direction[:split_idx], y_direction[split_idx:] # Evaluate price model X_test_poly = self.poly_features.transform(X_test) price_predictions = self.ridge_model.predict(X_test_poly) price_mse = mean_squared_error(y_price_test, price_predictions) price_r2 = r2_score(y_price_test, price_predictions) # Evalauate direction model X_binary_test_poly = self.poly_features_binary.transform(X_binary_test) direction_predictions = self.logistic_model.predict(X_binary_test_poly) direction_probabilities = self.logistic_model.predict_proba(X_binary_test_poly)[:, 1] dir_accuracy = accuracy_score(y_dir_test, direction_predictions) dir_precision = precision_score(y_dir_test, direction_predictions) dir_recall = recall_score(y_dir_test, direction_predictions) dir_f1 = f1_score(y_dir_test, direction_predictions) evaluation_results = { 'price_model': { 'mse': price_mse, 'rmse': np.sqrt(price_mse), 'r2_score': price_r2, 'test_samples': len(y_price_test) }, 'direction_model': { 'accuracy': dir_accuracy, 'precision': dir_precision, 'recall': dir_recall, 'f1_score': dir_f1, 'test_samples': len(y_dir_test) } } print(f"\nModel Evaluation Results:") print(f"Price Model - RMSE: {np.sqrt(price_mse):.6f}, R²: {price_r2:.4f}") print(f"Direction Model - Accuracy: {dir_accuracy:.4f}, F1: {dir_f1:.4f}") return evaluation_results except Exception as e: print(f"Error evaluating models: {e}") return None def run_full_analysis(self): """Launch full analysis""" try: print(f"Starting full analysis for {self.symbol}") print("-" * 60) # 1. Get data print("1. Getting historical data...") self.data = self.get_historical_data() if self.data is None: return None # 2. Calculate indicators print("2. Calculating technical indicators...") self.data = self.calculate_technical_indicators(self.data) # 3. Preparte features print("3. Preparing features and targets...") self.features, data_clean = self.prepare_features_and_targets(self.data) if self.features is None: return None # 4. Create symbolic equations print("4. Creating symbolic equations...") price_eq, binary_eq = self.create_symbolic_equations(self.features) if price_eq is None or binary_eq is None: return None # 5. Evaluate models print("5. Evaluating models...") evaluation = self.evaluate_models(self.features) # 6. Display results self.display_equations() # 7. Test forecast print("\n6. Testing predictions...") if len(data_clean) > 0: # Take the last string as current data current_data = data_clean.iloc[-1].to_dict() prediction = self.predict_future_price(current_data) if prediction: print(f"\nTest Prediction Results:") print(f" Current price: {prediction['current_price']:.5f}") print(f" Predicted price (24H): {prediction['predicted_price_24h']:.5f}") print(f" Expected change: {prediction['price_change_pct']:.2f}%") print(f" Direction: {prediction['direction_text']} (Prob: {prediction['direction_probability']:.3f})") print(f" Direction confidence: {prediction['direction_confidence']:.3f}") print(f" Price confidence: {prediction['confidence_score']:.3f}") print("\nAnalysis completed successfully!") return { 'price_equation': price_eq, 'binary_equation': binary_eq, 'model': self, 'data': data_clean, 'features': self.features, 'evaluation': evaluation, 'binary_metrics': self.binary_metrics } except Exception as e: print(f"Error in full analysis: {e}") return None def save_results(self, filename: str = None): """Save results to file""" try: if filename is None: filename = f"{self.symbol}_prediction_equations.txt" with open(filename, "w", encoding='utf-8') as f: f.write(f"Symbolic Price Prediction Equations\n") f.write(f"Symbol: {self.symbol}\n") f.write(f"Prediction Horizon: {self.prediction_horizon} hours\n") f.write(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write("="*80 + "\n\n") # Equations f.write("PRICE PREDICTION EQUATION:\n") f.write(f"P(t+24) = {self.price_equation}\n\n") f.write("BINARY DIRECTION EQUATION:\n") f.write(f"Prob(UP) = {self.binary_equation}\n\n") # Variables f.write("Variable Mapping:\n") for i, name in enumerate(self.feature_names): f.write(f"x{i} = {name}\n") f.write("\n") # Metrics if hasattr(self, 'ridge_model'): f.write("Price Model Statistics:\n") f.write(f"R² Score: {self.ridge_model.score(self.poly_features.transform(self.features['X']), self.features['y_price']):.4f}\n") f.write(f"Number of features: {len(self.feature_names)}\n\n") if self.binary_metrics: f.write("Binary Classification Statistics:\n") f.write(f"Test Accuracy: {self.binary_metrics['test_accuracy']:.4f}\n") f.write(f"Cross-Val Accuracy: {self.binary_metrics['cv_mean_accuracy']:.4f} (±{self.binary_metrics['cv_std_accuracy']:.4f})\n") f.write(f"Precision: {self.binary_metrics['precision']:.4f}\n") f.write(f"Recall: {self.binary_metrics['recall']:.4f}\n") f.write(f"F1-Score: {self.binary_metrics['f1_score']:.4f}\n\n") f.write("Classification Report:\n") f.write(self.binary_metrics['classification_report']) print(f"Results saved to '{filename}'") except Exception as e: print(f"Error saving results: {e}") # Sample use if __name__ == "__main__": # Create and launch analysis predictor = SymbolicPricePredictor(symbol="EURUSD") result = predictor.run_full_analysis() if result: print("\nSymbolic equations successfully created!") print("Features:") print("- Price prediction equation with R² score") print("- Binary direction classification with accuracy metrics") print("- Cross-validation and confusion matrix analysis") print("- Comprehensive performance evaluation") # Save results predictor.save_results() # Display key metrics if predictor.binary_metrics: print(f"\nKey Binary Classification Metrics:") print(f"- Test Accuracy: {predictor.binary_metrics['test_accuracy']:.1%}") print(f"- Precision: {predictor.binary_metrics['precision']:.1%}") print(f"- Recall: {predictor.binary_metrics['recall']:.1%}") print(f"- F1-Score: {predictor.binary_metrics['f1_score']:.1%}") print(f"- Cross-Val Accuracy: {predictor.binary_metrics['cv_mean_accuracy']:.1%} (±{predictor.binary_metrics['cv_std_accuracy']:.1%})") print("\nYou can now use:") print("- predictor.predict_future_price(current_data) for predictions") print("- Both price and direction will be predicted simultaneously") else: print("Failed to create symbolic equations")