Mathematics Methods for Quant finance

import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns from arch import arch_model from scipy import stats, signal, optimize, fft from sklearn.preprocessing import RobustScaler, StandardScaler from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier, VotingClassifier, IsolationForest, RandomForestRegressor from sklearn.model_selection import train_test_split, TimeSeriesSplit, RandomizedSearchCV from sklearn.metrics import classification_report, confusion_matrix, precision_score, f1_score, roc_auc_score from sklearn.svm import SVC from sklearn.neural_network import MLPClassifier from sklearn.linear_model import LogisticRegression from sklearn.calibration import CalibratedClassifierCV from sklearn.decomposition import PCA from xgboost import XGBClassifier from lightgbm import LGBMClassifier from pykalman import KalmanFilter import pulp import mplfinance as mpf import matplotlib.gridspec as gridspec from matplotlib.patches import FancyBboxPatch, Rectangle, Circle import warnings import ta from datetime import datetime, timedelta from scipy.signal import savgol_filter from statsmodels.tsa.stattools import adfuller import hashlib import pickle import plotly.graph_objects as go from plotly.subplots import make_subplots warnings.filterwarnings('ignore') plt.style.use('dark_background') DARK_BG = '#0D1117' FIGURE_FACE = '#0D1117' AXES_FACE = '#161B22' TEXT_COLOR = '#E6EDF3' GRID_COLOR = '#30363D' LINE_COLORS = { 'price': '#79C0FF', 'signal': '#D2A8FF', 'buy': '#3FB950', 'sell': '#F85149', 'confidence': '#FFA657', 'rsi': '#FF7B72', 'volatility': '#A5D6FF', 'volume': '#8B949E', 'strategy': '#56D4C4', 'fourier': '#FFAB70', 'hyperbolic': '#A371F7', 'complex': '#F778BA', 'quantum': '#58A6FF', 'entropy': '#E3B341' } plt.rcParams.update({ 'figure.facecolor': FIGURE_FACE, 'axes.facecolor': AXES_FACE, 'axes.edgecolor': GRID_COLOR, 'axes.labelcolor': TEXT_COLOR, 'xtick.color': TEXT_COLOR, 'ytick.color': TEXT_COLOR, 'text.color': TEXT_COLOR, 'legend.facecolor': AXES_FACE, 'legend.edgecolor': GRID_COLOR, 'grid.color': GRID_COLOR, 'grid.alpha': 0.3, 'font.size': 11, 'axes.titlesize': 14, 'axes.labelsize': 12, 'legend.fontsize': 10 }) class AdvancedTrigonometryGeometry: @staticmethod def fourier_cycle_analysis(price_series, num_components=5): n = len(price_series) if n < 10: return np.zeros_like(price_series), np.array([0]), np.array([0]) t = np.arange(n) window = np.hanning(n) price_windowed = price_series * window fft_vals = fft.fft(price_windowed) frequencies = fft.fftfreq(n) magnitudes = np.abs(fft_vals) positive_freq_indices = np.where((frequencies > 0) & (frequencies < 0.5))[0] if len(positive_freq_indices) == 0: return np.zeros_like(price_series), np.array([0]), np.array([0]) num_components = min(num_components, len(positive_freq_indices)) dominant_indices = positive_freq_indices[np.argsort(magnitudes[positive_freq_indices])[-num_components:]] reconstructed = np.zeros(n, dtype=complex) for idx in dominant_indices: reconstructed += fft_vals[idx] * np.exp(2j * np.pi * frequencies[idx] * t) reconstructed_real = np.real(reconstructed) if len(reconstructed_real) > len(price_series): reconstructed_real = reconstructed_real[:len(price_series)] elif len(reconstructed_real) < len(price_series): reconstructed_real = np.pad(reconstructed_real, (0, len(price_series) - len(reconstructed_real))) return reconstructed_real, frequencies[dominant_indices], magnitudes[dominant_indices] @staticmethod def harmonic_pattern_detection(highs, lows, tolerance=0.1): patterns = [] n = len(highs) if n < 5: return patterns for i in range(n-4): try: X, A, B, C, D = i, i+1, i+2, i+3, i+4 if not (highs[A] > highs[X] and lows[B] < lows[A] and highs[C] > highs[B] and lows[D] < lows[C]): continue XA = highs[A] - lows[X] AB = highs[B] - lows[A] BC = highs[B] - lows[C] CD = highs[D] - lows[C] if XA <= 0 or AB <= 0 or BC <= 0 or CD <= 0: continue AB_ratio = AB / XA BC_ratio = BC / AB CD_ratio = CD / BC bullish_conditions = [ (0.618 - tolerance <= AB_ratio <= 0.618 + tolerance and 0.382 - tolerance <= BC_ratio <= 0.886 + tolerance and 1.272 - tolerance <= CD_ratio <= 1.618 + tolerance, 'Gartley_Bullish'), (0.786 - tolerance <= AB_ratio <= 0.786 + tolerance and 0.382 - tolerance <= BC_ratio <= 0.886 + tolerance and 1.618 - tolerance <= CD_ratio <= 2.618 + tolerance, 'Butterfly_Bullish'), (0.382 - tolerance <= AB_ratio <= 0.5 + tolerance and 0.382 - tolerance <= BC_ratio <= 0.886 + tolerance and 1.618 - tolerance <= CD_ratio <= 2.618 + tolerance, 'Bat_Bullish'), (0.382 - tolerance <= AB_ratio <= 0.618 + tolerance and 0.382 - tolerance <= BC_ratio <= 0.886 + tolerance and 2.618 - tolerance <= CD_ratio <= 3.618 + tolerance, 'Crab_Bullish') ] for condition, pattern_name in bullish_conditions: if condition: patterns.append((pattern_name, i, AB_ratio, BC_ratio, CD_ratio)) break except (IndexError, ZeroDivisionError, ValueError): continue return patterns @staticmethod def spherical_coordinates_mapping(price, volume, time): price = np.nan_to_num(price, nan=np.nanmedian(price)) volume = np.nan_to_num(volume, nan=np.nanmedian(volume)) time = np.nan_to_num(time, nan=np.nanmedian(time)) price_norm = (price - np.nanmedian(price)) / (np.nanstd(price) + 1e-8) volume_norm = (volume - np.nanmedian(volume)) / (np.nanstd(volume) + 1e-8) time_norm = (time - np.nanmedian(time)) / (np.nanstd(time) + 1e-8) price_norm = np.clip(price_norm, -3, 3) volume_norm = np.clip(volume_norm, -3, 3) time_norm = np.clip(time_norm, -3, 3) r = np.sqrt(price_norm**2 + volume_norm**2 + time_norm**2) theta = np.arccos(np.clip(price_norm / (r + 1e-8), -1, 1)) phi = np.arctan2(volume_norm, time_norm + 1e-8) return r, theta, phi @staticmethod def hyperbolic_geometry_momentum(price_series, period=20): if len(price_series) < period + 1: return np.zeros_like(price_series) price_series_clean = np.nan_to_num(price_series, nan=0.0) shifted_prices = np.roll(price_series_clean, 1) shifted_prices[0] = price_series_clean[0] log_returns = np.log(price_series_clean / shifted_prices) log_returns[0] = 0 def poincare_distance(x, y): x = np.clip(x, -0.999, 0.999) y = np.clip(y, -0.999, 0.999) numerator = 2 * ((x - y)**2) denominator = (1 - x**2) * (1 - y**2) + 1e-8 return np.arccosh(1 + numerator / denominator) price_min = np.nanpercentile(price_series_clean, 5) price_max = np.nanpercentile(price_series_clean, 95) if price_max == price_min: normalized = np.zeros_like(price_series_clean) else: normalized = 2 * (price_series_clean - price_min) / (price_max - price_min) - 1 normalized = np.clip(normalized, -0.99, 0.99) hyperbolic_mom = np.zeros_like(price_series_clean) for i in range(period, len(price_series_clean)): try: dist_current = poincare_distance(normalized[i], normalized[i-1]) dist_period = poincare_distance(normalized[i], normalized[i-period]) momentum = (dist_current * 0.3 + dist_period * 0.7) * np.sign(price_series_clean[i] - price_series_clean[i-period]) hyperbolic_mom[i] = momentum except (ValueError, FloatingPointError): hyperbolic_mom[i] = hyperbolic_mom[i-1] * 0.9 if i > 0 else 0 hyperbolic_mom_series = pd.Series(hyperbolic_mom) hyperbolic_mom_smoothed = hyperbolic_mom_series.rolling(window=5, min_periods=1, center=False).mean().values return hyperbolic_mom_smoothed @staticmethod def wavelet_transform_analysis(price_series, scales=None): if scales is None: scales = np.arange(1, min(50, len(price_series)//4)) n = len(price_series) if n < 10: return np.array([]) def morlet_wavelet(t, scale): if scale <= 0: scale = 1 w0 = 6 return np.pi**(-0.25) * np.exp(1j * w0 * t / scale) * np.exp(-0.5 * (t / scale)**2) wavelet_coeffs = np.zeros((len(scales), n), dtype=complex) for i, scale in enumerate(scales): if scale <= 0 or 8 * scale * 2 > n: continue try: wavelet_length = min(int(8 * scale), n//2) t = np.arange(-wavelet_length, wavelet_length) wavelet = morlet_wavelet(t, scale) wavelet = wavelet / (np.sqrt(scale) + 1e-8) price_series_clean = np.nan_to_num(price_series, nan=0.0) conv = np.convolve(price_series_clean, wavelet, mode='same') if len(conv) == n: wavelet_coeffs[i, :] = conv elif len(conv) > n: wavelet_coeffs[i, :] = conv[:n] else: wavelet_coeffs[i, :len(conv)] = conv except (ValueError, FloatingPointError, MemoryError): continue return np.abs(wavelet_coeffs) @staticmethod def topological_data_analysis(price_series, window_size=50): from scipy.spatial.distance import pdist, squareform n = len(price_series) window_size = min(window_size, n//2) if n < window_size: return np.array([0]), np.array([[0]]) n_windows = n - window_size + 1 embedding = np.zeros((n_windows, window_size)) for i in range(n_windows): window_data = price_series[i:i+window_size] window_data_clean = np.nan_to_num(window_data, nan=0.0) if np.std(window_data_clean) > 0: embedding[i, :] = (window_data_clean - np.mean(window_data_clean)) / np.std(window_data_clean) else: embedding[i, :] = window_data_clean try: dist_matrix = squareform(pdist(embedding, metric='chebyshev')) persistence = np.zeros(n_windows - 1) for i in range(n_windows - 1): persistence[i] = np.sqrt(np.mean((embedding[i+1] - embedding[i])**2)) except (ValueError, np.linalg.LinAlgError, MemoryError): persistence = np.zeros(max(1, n_windows - 1)) dist_matrix = np.eye(max(1, n_windows)) return persistence, dist_matrix @staticmethod def differential_geometry_features(price_series, volume_series, window=20): n = len(price_series) if n < window + 2: return np.zeros(n), np.zeros(n), np.zeros(n) price_series_clean = np.nan_to_num(price_series, nan=0.0) volume_series_clean = np.nan_to_num(volume_series, nan=0.0) dp = np.gradient(price_series_clean) dv = np.gradient(volume_series_clean) d2p = np.gradient(dp) d2v = np.gradient(dv) curvature = np.zeros(n) torsion = np.zeros(n) gaussian_curvature = np.zeros(n) for i in range(n): try: vel = np.array([1, dp[i]]) acc = np.array([0, d2p[i]]) cross_val = np.abs(vel[0]*acc[1] - vel[1]*acc[0]) vel_mag = np.sqrt(vel[0]**2 + vel[1]**2) if vel_mag > 0: curvature[i] = cross_val / (vel_mag**3) else: curvature[i] = 0 if i > 0 and i < n-1: window_indices = [i-1, i, i+1] window_prices = [price_series_clean[i-1], price_series_clean[i], price_series_clean[i+1]] if len(set(window_prices)) == len(window_prices): z = np.polyfit(window_indices, window_prices, 2) torsion[i] = abs(z[0]) else: torsion[i] = torsion[i-1] if i > 0 else 0 gaussian_curvature[i] = curvature[i] * torsion[i] if i > 0 else 0 except (ValueError, FloatingPointError, np.linalg.LinAlgError): curvature[i] = curvature[i-1] if i > 0 else 0 torsion[i] = torsion[i-1] if i > 0 else 0 gaussian_curvature[i] = gaussian_curvature[i-1] if i > 0 else 0 curvature_smooth = pd.Series(curvature).rolling(window=5, min_periods=1, center=False).mean().values torsion_smooth = pd.Series(torsion).rolling(window=5, min_periods=1, center=False).mean().values gaussian_curvature_smooth = pd.Series(gaussian_curvature).rolling(window=5, min_periods=1, center=False).mean().values return curvature_smooth, torsion_smooth, gaussian_curvature_smooth @staticmethod def manifold_learning_features(price_series, dimension=3): try: from sklearn.manifold import Isomap except ImportError: return np.zeros((len(price_series), dimension)) n = len(price_series) if n < 20: return np.zeros((n, dimension)) embedding_dim = min(10, n//3) features = np.zeros((n - embedding_dim + 1, embedding_dim)) for i in range(embedding_dim): features[:, i] = price_series[i:n - embedding_dim + i + 1] try: isomap = Isomap(n_components=min(dimension, features.shape[1]-1), n_neighbors=min(5, features.shape[0]-1)) manifold_features = isomap.fit_transform(features) result = np.zeros((n, dimension)) start_idx = embedding_dim-1 end_idx = start_idx + len(manifold_features) result[start_idx:end_idx] = manifold_features result[:start_idx] = result[start_idx] return result except (ValueError, Exception): try: from sklearn.decomposition import PCA pca = PCA(n_components=min(dimension, features.shape[1])) manifold_features = pca.fit_transform(features) result = np.zeros((n, dimension)) start_idx = embedding_dim-1 end_idx = start_idx + len(manifold_features) result[start_idx:end_idx] = manifold_features result[:start_idx] = result[start_idx] return result except: return np.zeros((n, dimension)) @staticmethod def quantum_wave_function(price_series, volume_series, lookback=20): n = len(price_series) if n < lookback: return np.zeros(n), np.zeros(n) price_series_clean = np.nan_to_num(price_series, nan=0.0) volume_series_clean = np.nan_to_num(volume_series, nan=0.0) price_mean = np.mean(price_series_clean) price_std = np.std(price_series_clean) + 1e-8 volume_mean = np.mean(volume_series_clean) volume_std = np.std(volume_series_clean) + 1e-8 price_norm = (price_series_clean - price_mean) / price_std volume_norm = (volume_series_clean - volume_mean) / volume_std wave_function_real = np.zeros(n) wave_function_imag = np.zeros(n) probability_density = np.zeros(n) for i in range(lookback, n): try: state_vector = price_norm[i-lookback:i] + 1j * volume_norm[i-lookback:i] H = np.outer(state_vector, np.conj(state_vector)) eigenvalues, eigenvectors = np.linalg.eig(H) dominant_idx = np.argmax(np.abs(eigenvalues)) dominant_state = eigenvectors[:, dominant_idx] wave_function_real[i] = np.real(dominant_state[-1]) wave_function_imag[i] = np.imag(dominant_state[-1]) probability_density[i] = np.abs(dominant_state[-1])**2 except (np.linalg.LinAlgError, ValueError): wave_function_real[i] = wave_function_real[i-1] if i > 0 else 0 wave_function_imag[i] = wave_function_imag[i-1] if i > 0 else 0 probability_density[i] = probability_density[i-1] if i > 0 else 0.5 return probability_density, wave_function_real + 1j * wave_function_imag @staticmethod def entropy_based_features(price_series, window=50): n = len(price_series) if n < window: return np.zeros(n), np.zeros(n), np.zeros(n) price_series_clean = np.nan_to_num(price_series, nan=0.0) shannon_entropy = np.zeros(n) approximate_entropy = np.zeros(n) sample_entropy = np.zeros(n) for i in range(window, n): window_data = price_series_clean[i-window:i] hist, bin_edges = np.histogram(window_data, bins=10, density=True) hist = hist[hist > 0] if len(hist) > 0: shannon_entropy[i] = -np.sum(hist * np.log(hist)) else: shannon_entropy[i] = 0 try: m = 2 r = 0.2 * np.std(window_data) vectors_m = [window_data[j:j+m] for j in range(len(window_data)-m+1)] if len(vectors_m) == 0: continue C_m = [] for vec in vectors_m: count = np.sum([np.max(np.abs(vec - other_vec)) <= r for other_vec in vectors_m]) C_m.append(count / len(vectors_m)) phi_m = np.mean(np.log(np.array(C_m) + 1e-8)) approximate_entropy[i] = phi_m - phi_m1 sample_entropy[i] = -np.log((np.sum(C_m1) + 1e-8) / (np.sum(C_m) + 1e-8)) except (ValueError, ZeroDivisionError): approximate_entropy[i] = approximate_entropy[i-1] if i > 0 else 0 sample_entropy[i] = sample_entropy[i-1] if i > 0 else 0 return shannon_entropy, approximate_entropy, sample_entropy @staticmethod def fractal_dimension(price_series, window=30): n = len(price_series) if n < window: return np.zeros(n) price_series_clean = np.nan_to_num(price_series, nan=0.0) fractal_dims = np.zeros(n) for i in range(window, n): window_data = price_series_clean[i-window:i] try: k_max = min(10, len(window_data)//2) if k_max < 2: fractal_dims[i] = 1.5 continue L = [] for k in range(1, k_max+1): Lk = 0 for m in range(k): indices = list(range(m, len(window_data), k)) if len(indices) > 1: segment = window_data[indices] Lkm = np.sum(np.abs(np.diff(segment))) normalization = (len(window_data) - 1) / (k * (len(indices) - 1)) if len(indices) > 1 else 1 Lk += Lkm * normalization if k > 0: L.append(np.log(Lk / k + 1e-8)) if len(L) < 2: fractal_dims[i] = 1.5 continue x = np.log(range(1, k_max+1)) y = np.array(L) if len(x) >= 2 and len(y) >= 2: slope = np.polyfit(x, y, 1)[0] fractal_dims[i] = max(1.0, min(2.0, slope)) else: fractal_dims[i] = 1.5 except (ValueError, np.linalg.LinAlgError): fractal_dims[i] = fractal_dims[i-1] if i > 0 else 1.5 return fractal_dims @staticmethod def spectral_entropy(price_series, nperseg=64): try: from scipy.signal import welch if len(price_series) < 10: return 0.0 nperseg = min(nperseg, len(price_series)//2) if nperseg < 4: return 0.0 f, Pxx = welch(price_series, fs=1, nperseg=nperseg) if Pxx.sum() == 0: return 0.0 Pxx = Pxx / (Pxx.sum() + 1e-12) se = -np.sum(Pxx * np.log(Pxx + 1e-12)) return se except Exception: return 0.0 class OptimizedHighAccuracyStrategy: def __init__(self): self.data = None self.performance_metrics = {} self.models = {} self.feature_importance = {} self.signals_df = None self.trade_log = [] self.signal_stability_threshold = 0.80 self.min_confidence = 0.75 self.risk_free_rate = 0.02 self.max_position_size = 0.08 self.stop_loss_pct = 0.03 self.take_profit_pct = 0.08 self.min_win_rate = 0.65 self.optimal_signal_ratio = 0.08 def _clean_and_validate_data(self, df): try: df = df[~df.index.duplicated(keep='first')] for col in ['open', 'high', 'low', 'close', 'volume']: if col in df.columns: df[col] = pd.to_numeric(df[col], errors='coerce') valid_data_mask = ( (df['high'] >= df['low']) & (df['high'] >= df['close']) & (df['low'] <= df['close']) & (df['volume'] >= 0) & (df['close'] > 0) & (abs(df['close'] / df['open'].replace(0, 1) - 1) < 0.15) ) df = df[valid_data_mask] for col in ['close', 'volume']: if col in df.columns: Q1 = df[col].quantile(0.01) Q3 = df[col].quantile(0.99) IQR = Q3 - Q1 lower_bound = Q1 - 3 * IQR upper_bound = Q3 + 3 * IQR df = df[(df[col] >= lower_bound) & (df[col] <= upper_bound)] df = df.ffill().bfill() df = df.replace([np.inf, -np.inf], np.nan).ffill() return df except Exception as e: return df def fetch_coinbase_data(self, symbol="BTC/USDT", timeframe="1h", limit=300): try: import ccxt except ImportError: return self._generate_optimized_synthetic_data() try: exchange = ccxt.coinbase({'enableRateLimit': True, 'timeout': 30000}) try: exchange.load_markets() except: pass symbols_to_try = [symbol, 'BTC/USD', 'BTC/USDT', 'BTC/EUR', 'BTC/USD'] successful_data = None for sym in symbols_to_try: try: ohlcv = exchange.fetch_ohlcv(sym, timeframe=timeframe, limit=min(limit, 300)) if ohlcv and len(ohlcv) > 100: df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') df.set_index('timestamp', inplace=True) df = self._clean_and_validate_data(df) if len(df) > 100: successful_data = df break except Exception as e: continue if successful_data is not None: return successful_data else: return self._generate_optimized_synthetic_data() except Exception as e: return self._generate_optimized_synthetic_data()