import pandas as pd import numpy as np import math from datetime import datetime, timedelta import matplotlib.pyplot as plt from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score, classification_report, roc_auc_score, confusion_matrix from sklearn.ensemble import RandomForestClassifier import warnings import MetaTrader5 as mt5 warnings.filterwarnings('ignore') # Check CatBoost availability try: from catboost import CatBoostClassifier CATBOOST_AVAILABLE = True print("CatBoost is available") except ImportError: CATBOOST_AVAILABLE = False print("CatBoost is not available, using RandomForest") np.random.seed(42) class UniversalForexAstroMT5: """Universal lunar-solar analysis of currency pairs with MT5 data""" def __init__(self, pair_name="EURUSD"): self.pair_name = pair_name self.mt5_connected = False # Astronomical cycles self.lunar_cycle = 29.530588 # Synodic month self.solar_cycle = 365.25636 # Tropical year # Base dates for different currencies self.birth_dates = { 'USD': datetime(1792, 4, 2), # US Dollar 'EUR': datetime(1999, 1, 1), # Euro 'GBP': datetime(1694, 7, 27), # British Pound 'JPY': datetime(1871, 5, 10), # Japanese Yen 'CHF': datetime(1850, 5, 7), # Swiss Franc 'CAD': datetime(1858, 8, 2), # Canadian Dollar 'AUD': datetime(1966, 2, 14), # Australian Dollar 'NZD': datetime(1967, 7, 10), # New Zealand Dollar } # Define pair currencies self.base_currency = pair_name[:3] self.quote_currency = pair_name[3:] self.birth_date = self.birth_dates.get(self.quote_currency, datetime(1900, 1, 1)) # Moon phases self.moon_phases = { 'new_moon': (0, 15), 'waxing_crescent': (15, 75), 'first_quarter': (75, 105), 'waxing_gibbous': (105, 165), 'full_moon': (165, 195), 'waning_gibbous': (195, 255), 'last_quarter': (255, 285), 'waning_crescent': (285, 345), 'balsamic': (345, 360) } # Lags self.lag_periods = [1, 2, 4, 8, 13] def connect_mt5(self): """Connect to MetaTrader 5""" if not mt5.initialize(): print("MT5 initialization failed") print(mt5.last_error()) return False self.mt5_connected = True print("MT5 connected successfully") print(f"MT5 version: {mt5.version()}") return True def disconnect_mt5(self): """Disconnect from MetaTrader 5""" if self.mt5_connected: mt5.shutdown() self.mt5_connected = False print("MT5 disconnected") def get_mt5_data(self, start_date, end_date, timeframe=mt5.TIMEFRAME_W1): """Get real data from MetaTrader 5""" if not self.mt5_connected: print("MT5 not connected. Connecting...") if not self.connect_mt5(): return None # Check symbol availability symbol_info = mt5.symbol_info(self.pair_name) if symbol_info is None: print(f"Symbol {self.pair_name} not found") return None if not symbol_info.visible: print(f"Symbol {self.pair_name} is not visible, trying to enable...") if not mt5.symbol_select(self.pair_name, True): print(f"Failed to select {self.pair_name}") return None print(f"Getting data for {self.pair_name}...") # Get rates rates = mt5.copy_rates_range(self.pair_name, timeframe, start_date, end_date) if rates is None or len(rates) == 0: print(f"Failed to get data for {self.pair_name}") print(mt5.last_error()) return None # Convert to DataFrame df = pd.DataFrame(rates) df['date'] = pd.to_datetime(df['time'], unit='s') print(f"Successfully loaded {len(df)} bars") print(f"Date range: {df['date'].min()} to {df['date'].max()}") return df def process_mt5_data(self, df): """Process MT5 data and calculate returns""" if df is None or len(df) == 0: return None # Calculate returns df['return'] = df['close'].pct_change() df['volatility'] = abs(df['return']) # Calculate additional metrics df['price_range'] = (df['high'] - df['low']) / df['close'] df['body'] = abs(df['close'] - df['open']) / df['close'] # Remove first row with NaN return df = df.dropna(subset=['return']).reset_index(drop=True) # Add astrological effects as features df['lunar_effect'] = df.apply(lambda row: self.calculate_lunar_effect(row['date']), axis=1) df['solar_effect'] = df.apply(lambda row: self.calculate_solar_effect(row['date']), axis=1) df['currency_effect'] = df.apply(lambda row: self.calculate_currency_effect(row['date']), axis=1) return df def get_moon_phase_angle(self, date): """Moon phase angle""" days_since_birth = (date - self.birth_date).days lunar_position = (days_since_birth % self.lunar_cycle) / self.lunar_cycle * 360 solar_position = (days_since_birth % self.solar_cycle) / self.solar_cycle * 360 phase_angle = (lunar_position - solar_position) % 360 return phase_angle def get_moon_phase_name(self, angle): """Moon phase name""" for phase_name, (start, end) in self.moon_phases.items(): if start <= angle < end: return phase_name return 'new_moon' def calculate_lunar_effect(self, date): """Lunar effects for currency pair - pure astronomical calculation""" moon_phase_angle = self.get_moon_phase_angle(date) phase_name = self.get_moon_phase_name(moon_phase_angle) effect = 0 # Phase effects - based only on moon phase if phase_name in ['new_moon', 'full_moon']: effect += 0.01 elif phase_name in ['waxing_crescent', 'first_quarter', 'waxing_gibbous']: effect += 0.002 elif phase_name in ['waning_gibbous', 'last_quarter', 'waning_crescent']: effect -= 0.002 # Harmonics effect += math.sin(math.radians(moon_phase_angle)) * 0.001 effect += math.sin(math.radians(moon_phase_angle * 2)) * 0.0005 return effect def calculate_solar_effect(self, date): """Solar/seasonal effects - pure astronomical calculation""" day_of_year = date.timetuple().tm_yday effect = 0 # Seasonal cycles seasonal_cycle = math.sin(2 * math.pi * day_of_year / 365.25) effect += seasonal_cycle * 0.001 # Quarterly effects month = date.month if month in [3, 6, 9, 12]: effect += 0.002 # New Year effect if month == 1 and date.day <= 7: effect += 0.001 return effect def calculate_currency_effect(self, date): """Specific currency effects - based on currency pair only""" effect = 0 # Carry trade effects if self.base_currency in ['AUD', 'NZD'] and self.quote_currency in ['JPY', 'CHF']: risk_cycle = math.sin(2 * math.pi * date.timetuple().tm_yday / 365.25 * 3) effect += risk_cycle * 0.001 # USD effects if 'USD' in self.pair_name: usd_strength_cycle = math.cos(2 * math.pi * date.timetuple().tm_yday / 365.25) if self.base_currency == 'USD': effect += usd_strength_cycle * 0.0005 else: effect -= usd_strength_cycle * 0.0005 return effect def create_features_with_lags(self, df): """Create features with lags""" features_list = [] for i in range(len(df)): date = df.iloc[i]['date'] features = {} # Current astrological features moon_phase_angle = self.get_moon_phase_angle(date) day_of_year = date.timetuple().tm_yday # Lunar features features['moon_phase_angle'] = moon_phase_angle features['moon_phase_sin'] = math.sin(math.radians(moon_phase_angle)) features['moon_phase_cos'] = math.cos(math.radians(moon_phase_angle)) features['moon_phase_sin2'] = math.sin(math.radians(moon_phase_angle * 2)) features['moon_phase_cos2'] = math.cos(math.radians(moon_phase_angle * 2)) features['moon_phase_sin4'] = math.sin(math.radians(moon_phase_angle * 4)) features['moon_phase_cos4'] = math.cos(math.radians(moon_phase_angle * 4)) # Moon phases phase_name = self.get_moon_phase_name(moon_phase_angle) for phase in self.moon_phases.keys(): features[f'is_{phase}'] = 1 if phase_name == phase else 0 # Critical points features['is_critical_lunar'] = 1 if (moon_phase_angle <= 15 or moon_phase_angle >= 345 or 165 <= moon_phase_angle <= 195) else 0 features['is_waxing'] = 1 if 15 < moon_phase_angle < 165 else 0 features['is_waning'] = 1 if 195 < moon_phase_angle < 345 else 0 # Solar features features['day_of_year'] = day_of_year features['solar_sin'] = math.sin(2 * math.pi * day_of_year / 365.25) features['solar_cos'] = math.cos(2 * math.pi * day_of_year / 365.25) features['solar_sin2'] = math.sin(4 * math.pi * day_of_year / 365.25) features['solar_cos2'] = math.cos(4 * math.pi * day_of_year / 365.25) # Seasons if 80 <= day_of_year <= 172: season = 'spring' elif 172 < day_of_year <= 266: season = 'summer' elif 266 < day_of_year <= 355: season = 'autumn' else: season = 'winter' for s in ['spring', 'summer', 'autumn', 'winter']: features[f'is_{s}'] = 1 if season == s else 0 # Monthly cycles features['month'] = date.month features['month_sin'] = math.sin(2 * math.pi * date.month / 12) features['month_cos'] = math.cos(2 * math.pi * date.month / 12) # Price features from MT5 features['price_range'] = df.iloc[i]['price_range'] features['body'] = df.iloc[i]['body'] # Lag features for lag in self.lag_periods: if i >= lag: lag_date = df.iloc[i - lag]['date'] lag_moon = self.get_moon_phase_angle(lag_date) lag_day = lag_date.timetuple().tm_yday features[f'lag_{lag}_moon_angle'] = lag_moon features[f'lag_{lag}_moon_sin'] = math.sin(math.radians(lag_moon)) features[f'lag_{lag}_moon_cos'] = math.cos(math.radians(lag_moon)) features[f'lag_{lag}_is_waxing'] = 1 if 15 < lag_moon < 165 else 0 features[f'lag_{lag}_is_critical'] = 1 if (lag_moon <= 15 or lag_moon >= 345 or 165 <= lag_moon <= 195) else 0 features[f'lag_{lag}_solar_sin'] = math.sin(2 * math.pi * lag_day / 365.25) features[f'lag_{lag}_solar_cos'] = math.cos(2 * math.pi * lag_day / 365.25) # Price lags features[f'lag_{lag}_return'] = df.iloc[i - lag]['return'] features[f'lag_{lag}_volatility'] = df.iloc[i - lag]['volatility'] features[f'lag_{lag}_price_range'] = df.iloc[i - lag]['price_range'] else: for key in ['moon_angle', 'moon_sin', 'moon_cos', 'is_waxing', 'is_critical', 'solar_sin', 'solar_cos', 'return', 'volatility', 'price_range']: features[f'lag_{lag}_{key}'] = 0 # Moon phase changes for lag in [1, 2, 4]: if i >= lag: current_moon = moon_phase_angle lag_moon = self.get_moon_phase_angle(df.iloc[i - lag]['date']) moon_diff = (current_moon - lag_moon) % 360 if moon_diff > 180: moon_diff -= 360 features[f'moon_change_{lag}'] = moon_diff features[f'moon_velocity_{lag}'] = moon_diff / lag else: features[f'moon_change_{lag}'] = 0 features[f'moon_velocity_{lag}'] = 0 # Statistical features if i >= 8: recent_returns = [df.iloc[j]['return'] for j in range(max(0, i-8), i)] recent_moon_phases = [self.get_moon_phase_angle(df.iloc[j]['date']) for j in range(max(0, i-8), i)] features['return_volatility_8w'] = np.std(recent_returns) features['moon_phase_std_8w'] = np.std(recent_moon_phases) features['critical_count_8w'] = sum(1 for moon in recent_moon_phases if moon <= 15 or moon >= 345 or 165 <= moon <= 195) else: features['return_volatility_8w'] = 0 features['moon_phase_std_8w'] = 0 features['critical_count_8w'] = 0 features_list.append(features) return pd.DataFrame(features_list) def create_binary_target(self, df, threshold_percentile=60): """Binary target variable""" returns = df['return'].values threshold = np.percentile(np.abs(returns), threshold_percentile) binary_target = [] for ret in returns: if ret > threshold: binary_target.append(1) # Growth else: binary_target.append(0) # Decline/stagnation return binary_target, threshold def train_binary_model(self, features_df, binary_target): """Train binary model""" print(f"\n=== TRAINING BINARY MODEL FOR {self.pair_name} ===") cat_features = [col for col in features_df.columns if col.startswith('is_')] split_idx = int(len(features_df) * 0.7) X_train = features_df.iloc[:split_idx] X_test = features_df.iloc[split_idx:] y_train = binary_target[:split_idx] y_test = binary_target[split_idx:] print(f"Training: {len(X_train)}, Test: {len(X_test)}") print(f"Positive class in training: {sum(y_train)}/{len(y_train)} ({sum(y_train)/len(y_train)*100:.1f}%)") print(f"Positive class in test: {sum(y_test)}/{len(y_test)} ({sum(y_test)/len(y_test)*100:.1f}%)") if CATBOOST_AVAILABLE: model = CatBoostClassifier( iterations=600, learning_rate=0.08, depth=13, cat_features=cat_features, random_seed=42, verbose=100, early_stopping_rounds=150, eval_metric='AUC' ) model.fit(X_train, y_train, eval_set=(X_test, y_test)) else: model = RandomForestClassifier( n_estimators=300, max_depth=12, random_state=42, class_weight='balanced' ) model.fit(X_train, y_train) y_pred = model.predict(X_test) y_pred_proba = model.predict_proba(X_test)[:, 1] accuracy = accuracy_score(y_test, y_pred) auc_score = roc_auc_score(y_test, y_pred_proba) print(f"\nAccuracy: {accuracy:.3f}") print(f"AUC Score: {auc_score:.3f}") print("\nClassification Report:") print(classification_report(y_test, y_pred, target_names=['Decline/Stagnation', 'Growth'])) if hasattr(model, 'feature_importances_'): feature_importance = pd.DataFrame({ 'feature': features_df.columns, 'importance': model.feature_importances_ }).sort_values('importance', ascending=False) print(f"\nTop 15 Important Features:") print(feature_importance.head(15)) return { 'model': model, 'accuracy': accuracy, 'auc_score': auc_score, 'feature_importance': feature_importance, 'predictions': y_pred, 'probabilities': y_pred_proba, 'split_idx': split_idx } return {'model': model, 'accuracy': accuracy, 'auc_score': auc_score, 'split_idx': split_idx} def run_analysis(self, start_date, end_date, timeframe=mt5.TIMEFRAME_W1, threshold_percentile=60): """Complete analysis with MT5 data""" print(f"=== ASTROLOGICAL ANALYSIS {self.pair_name} WITH MT5 DATA ===") print(f"Base currency: {self.base_currency}") print(f"Quote currency: {self.quote_currency}") print(f"Quote currency birth date: {self.birth_date.strftime('%d.%m.%Y')}") # Get MT5 data df_raw = self.get_mt5_data(start_date, end_date, timeframe) if df_raw is None: print("Failed to get MT5 data") self.disconnect_mt5() return None, None, None # Process data df = self.process_mt5_data(df_raw) print(f"Processed {len(df)} bars") # Create features print("Creating astrological features...") features_df = self.create_features_with_lags(df) print(f"Created {len(features_df.columns)} features") # Binary target binary_target, threshold = self.create_binary_target(df, threshold_percentile) print(f"Growth threshold: {threshold:.4f}") print(f"Positive weeks: {sum(binary_target)}/{len(binary_target)} ({sum(binary_target)/len(binary_target)*100:.1f}%)") # Training ml_results = self.train_binary_model(features_df, binary_target) # Results results = [] split_idx = ml_results['split_idx'] for i in range(len(df)): row = df.iloc[i] ml_prediction = 0 ml_probability = 0.5 if i >= split_idx and 'predictions' in ml_results: test_idx = i - split_idx if test_idx < len(ml_results['predictions']): ml_prediction = ml_results['predictions'][test_idx] ml_probability = ml_results['probabilities'][test_idx] moon_phase_angle = self.get_moon_phase_angle(row['date']) result = { 'date': row['date'], 'price': row['close'], 'actual_return': row['return'], 'binary_actual': binary_target[i], 'ml_prediction': ml_prediction, 'ml_probability': round(ml_probability, 3), 'moon_phase_angle': round(moon_phase_angle, 1), 'moon_phase_name': self.get_moon_phase_name(moon_phase_angle), 'lunar_effect': row['lunar_effect'], 'solar_effect': row['solar_effect'], 'currency_effect': row['currency_effect'], 'is_test_period': i >= split_idx } results.append(result) results_df = pd.DataFrame(results) # Save filename = f'{self.pair_name.lower()}_astro_binary_mt5.csv' results_df.to_csv(filename, index=False) print(f"\nResults saved: {filename}") # Statistics self.print_statistics(results_df, ml_results, threshold) # Plots self.create_plots(results_df, ml_results) # Disconnect MT5 self.disconnect_mt5() return results_df, features_df, ml_results def print_statistics(self, df, ml_results, threshold): """Analysis statistics""" print(f"\n{'='*60}") print(f"ANALYSIS STATISTICS {self.pair_name}") print(f"{'='*60}") print(f"Period: {len(df)} weeks") print(f"Average return: {df['actual_return'].mean()*100:.3f}%") print(f"Volatility: {df['actual_return'].std()*100:.3f}%") print(f"Growth threshold: {threshold:.4f}") if 'accuracy' in ml_results: baseline = max(sum(df['binary_actual']), len(df) - sum(df['binary_actual'])) / len(df) improvement = (ml_results['accuracy'] - baseline) / baseline * 100 print(f"\nML accuracy: {ml_results['accuracy']:.3f}") print(f"Improvement over baseline: {improvement:.1f}%") def create_plots(self, df, ml_results): """Create plots with 700px width""" # Convert pixels to inches (assuming 100 DPI) width_inches = 700 / 100 height_inches = 10 fig = plt.figure(figsize=(width_inches, height_inches), dpi=100) # Price chart plt.subplot(3, 1, 1) plt.plot(df['date'], df['price'], label='Price', linewidth=1.5) plt.title(f'{self.pair_name} Price (MT5 Data)', fontsize=12, fontweight='bold') plt.ylabel('Price') plt.grid(True, alpha=0.3) plt.legend() # Moon phase chart plt.subplot(3, 1, 2) plt.plot(df['date'], df['moon_phase_angle'], label='Moon Phase Angle', color='orange', linewidth=1.5) plt.axhline(y=0, color='r', linestyle='--', alpha=0.3, label='New Moon') plt.axhline(y=180, color='b', linestyle='--', alpha=0.3, label='Full Moon') plt.title('Moon Phase Cycle', fontsize=12, fontweight='bold') plt.ylabel('Phase Angle (degrees)') plt.ylim(0, 360) plt.grid(True, alpha=0.3) plt.legend() # Predictions chart plt.subplot(3, 1, 3) test_df = df[df['is_test_period']] plt.plot(test_df['date'], test_df['binary_actual'], label='Actual', color='blue', linewidth=1.5) plt.plot(test_df['date'], test_df['ml_prediction'], label='Predicted', color='red', linestyle='--', linewidth=1.5) plt.fill_between(test_df['date'], 0, test_df['binary_actual'], alpha=0.2, color='blue') plt.title('Actual vs Predicted (Test Period)', fontsize=12, fontweight='bold') plt.ylabel('Growth Signal') plt.xlabel('Date') plt.legend() plt.grid(True, alpha=0.3) plt.tight_layout() filename = f'{self.pair_name.lower()}_plots_mt5.png' plt.savefig(filename, dpi=100, bbox_inches='tight') print(f"Plots saved: {filename}") plt.close() if __name__ == "__main__": # Usage example with MT5 pair_name = "EURUSD" # Currency pair start_date = datetime(2010, 1, 1) # Start date end_date = datetime(2025, 1, 1) # End date # Timeframe options: # mt5.TIMEFRAME_M1 - 1 minute # mt5.TIMEFRAME_M5 - 5 minutes # mt5.TIMEFRAME_M15 - 15 minutes # mt5.TIMEFRAME_M30 - 30 minutes # mt5.TIMEFRAME_H1 - 1 hour # mt5.TIMEFRAME_H4 - 4 hours # mt5.TIMEFRAME_D1 - 1 day # mt5.TIMEFRAME_W1 - 1 week # mt5.TIMEFRAME_MN1 - 1 month # Create class instance analyzer = UniversalForexAstroMT5(pair_name) # Run analysis with real MT5 data results_df, features_df, ml_results = analyzer.run_analysis( start_date, end_date, timeframe=mt5.TIMEFRAME_W1, # Weekly timeframe threshold_percentile=60 )