""" CCT-ODE Next-Value Predictor ============================ Conditional Collapse Theory + ODE gradient descent for sequence prediction. Handles: numerical sequences, letter series, time series. Based on: Gap Probability Collapse, Taylor-Token Expansion, Question TSP, xFFT Antiresonance Detection """ import numpy as np from numpy.typing import NDArray from typing import Union, Optional, List, Tuple, Dict, Callable from dataclasses import dataclass, field from enum import Enum, auto import math from collections import Counter import warnings # ============================================================================ # PART 1: xFFT ANTIRESONANCE DETECTOR # ============================================================================ def detect_antiresonances_xfft( signal_main: NDArray, signal_ref: Optional[NDArray] = None, fs: float = 1.0, nperseg: Optional[int] = None, threshold_db: float = -6.0, min_freq: float = 0.0, max_freq: Optional[float] = None, mode: str = 'cross' ) -> Dict: """ Detect antiresonance frequencies using cross-spectral (xFFT) analysis. Parameters ---------- signal_main : array_like The main signal (output/loss/gap). signal_ref : array_like, optional Reference signal (input/state). If None and mode='cross', uses signal_main shifted. fs : float Sampling frequency (Hz). nperseg : int Segment length for Welch's method. threshold_db : float Minimum dip depth (dB) to qualify as antiresonance. min_freq, max_freq : float Frequency range to search. mode : {'cross', 'single'} 'cross' uses transfer function; 'single' uses power spectrum. Returns ------- results : dict - 'frequencies': ndarray of antiresonance frequencies - 'depths_db': Depth of each dip (dB) - 'coherences': Coherence at those frequencies - 'transfer_magnitude': (freqs, |H(f)|) """ signal_main = np.asarray(signal_main, dtype=np.float64) n = len(signal_main) if nperseg is None: nperseg = min(256, n // 2) nperseg = max(8, nperseg) if max_freq is None: max_freq = fs / 2.0 if mode == 'cross': if signal_ref is None: signal_ref = np.roll(signal_main, 1) # self-coherence fallback signal_ref = np.asarray(signal_ref, dtype=np.float64) if len(signal_ref) != n: raise ValueError("Main and reference signals must have same length.") # Compute auto and cross spectra via Welch's method (manual implementation) freqs, Pxx = _welch_psd(signal_ref, fs, nperseg) _, Pxy = _welch_csd(signal_ref, signal_main, fs, nperseg) H_mag = np.abs(Pxy) / (np.abs(Pxx) + 1e-12) Coh = _coherence(signal_ref, signal_main, fs, nperseg) elif mode == 'single': freqs = np.fft.rfftfreq(n, d=1.0/fs) spectrum = np.abs(np.fft.rfft(signal_main)) H_mag = spectrum / (np.max(spectrum) + 1e-12) Coh = np.ones_like(freqs) else: raise ValueError("mode must be 'cross' or 'single'.") # Apply frequency mask mask = (freqs >= min_freq) & (freqs <= max_freq) freqs = freqs[mask] H_mag = H_mag[mask] Coh = Coh[mask] # Convert to dB H_db = 20 * np.log10(H_mag + 1e-12) # Find local minima (antiresonance candidates) local_min_idx = _argrelmin(H_db) antires_freqs = [] antires_depths = [] antires_coherence = [] for idx in local_min_idx: left = max(0, idx - 5) right = min(len(H_db) - 1, idx + 5) neighbor_vals = np.concatenate([H_db[left:idx], H_db[idx+1:right+1]]) if len(neighbor_vals) == 0: continue local_avg = np.mean(neighbor_vals) dip_depth = local_avg - H_db[idx] if dip_depth >= -threshold_db: antires_freqs.append(freqs[idx]) antires_depths.append(dip_depth) antires_coherence.append(Coh[idx]) return { 'frequencies': np.array(antires_freqs), 'depths_db': np.array(antires_depths), 'coherences': np.array(antires_coherence), 'transfer_magnitude': (freqs, H_mag), 'mode': mode } def _welch_psd(x: NDArray, fs: float, nperseg: int) -> Tuple[NDArray, NDArray]: """Estimate power spectral density using Welch's method.""" n = len(x) n_fft = nperseg window = np.hanning(nperseg) # Zero-pad if necessary if n < nperseg: x = np.pad(x, (0, nperseg - n)) n = nperseg n_overlap = nperseg // 2 n_frames = 1 + (n - nperseg) // (nperseg - n_overlap) psd = np.zeros(n_fft // 2 + 1) for i in range(n_frames): start = i * (nperseg - n_overlap) segment = x[start:start + nperseg] * window fft_result = np.fft.rfft(segment) psd += np.abs(fft_result) ** 2 psd /= n_frames freqs = np.fft.rfftfreq(n_fft, d=1.0/fs) return freqs, psd def _welch_csd(x: NDArray, y: NDArray, fs: float, nperseg: int) -> Tuple[NDArray, NDArray]: """Estimate cross-spectral density using Welch's method.""" n = len(x) n_fft = nperseg window = np.hanning(nperseg) n_overlap = nperseg // 2 n_frames = 1 + (n - nperseg) // (nperseg - n_overlap) csd = np.zeros(n_fft // 2 + 1, dtype=complex) for i in range(n_frames): start = i * (nperseg - n_overlap) x_segment = x[start:start + nperseg] * window y_segment = y[start:start + nperseg] * window X = np.fft.rfft(x_segment) Y = np.fft.rfft(y_segment) csd += X * np.conj(Y) csd /= n_frames freqs = np.fft.rfftfreq(n_fft, d=1.0/fs) return freqs, csd def _coherence(x: NDArray, y: NDArray, fs: float, nperseg: int) -> NDArray: """Compute magnitude-squared coherence.""" _, Pxx = _welch_psd(x, fs, nperseg) _, Pyy = _welch_psd(y, fs, nperseg) _, Pxy = _welch_csd(x, y, fs, nperseg) coh = np.abs(Pxy) ** 2 / (Pxx * Pyy + 1e-12) return np.clip(coh, 0, 1) def _argrelmin(x: NDArray) -> List[int]: """Find indices of relative minima.""" minima = [] for i in range(1, len(x) - 1): if x[i] < x[i-1] and x[i] < x[i+1]: minima.append(i) return minima # ============================================================================ # PART 2: PATTERN DETECTORS # ============================================================================ class PatternType(Enum): """Types of patterns the predictor can detect.""" ARITHMETIC = auto() # Constant difference GEOMETRIC = auto() # Constant ratio PERIODIC = auto() # Repeating cycle FIBONACCI = auto() # Sum of previous two POLYNOMIAL = auto() # Polynomial fit MARKOV = auto() # Transition probabilities MIXED = auto() # Multiple patterns UNKNOWN = auto() # No clear pattern @dataclass class PatternInfo: """Information about a detected pattern.""" pattern_type: PatternType parameters: Dict confidence: float # 0-1 score: float # Collapse potential description: str class PatternDetector: """Detects pattern types in sequences.""" def __init__(self, tolerance: float = 1e-6): self.tolerance = tolerance def detect(self, sequence: Union[List, NDArray]) -> List[PatternInfo]: """Detect all applicable patterns in the sequence.""" seq = np.asarray(sequence, dtype=np.float64) patterns = [] # 1. Arithmetic pattern arith = self._detect_arithmetic(seq) if arith is not None: patterns.append(arith) # 2. Geometric pattern (only for positive sequences) if np.all(seq > 0): geom = self._detect_geometric(seq) if geom is not None: patterns.append(geom) # 3. Periodic pattern periodic = self._detect_periodic(seq) if periodic is not None: patterns.append(periodic) # 4. Fibonacci-like pattern fib = self._detect_fibonacci(seq) if fib is not None: patterns.append(fib) # 5. Polynomial pattern poly = self._detect_polynomial(seq) if poly is not None: patterns.append(poly) # Sort by score (collapse potential) patterns.sort(key=lambda p: p.score, reverse=True) return patterns def _detect_arithmetic(self, seq: NDArray) -> Optional[PatternInfo]: """Detect arithmetic sequence: x_{n+1} - x_n = constant.""" if len(seq) < 2: return None diffs = np.diff(seq) if len(diffs) < 2: return PatternInfo( pattern_type=PatternType.ARITHMETIC, parameters={'difference': diffs[0]}, confidence=1.0, score=1.0, description=f"x_{{n+1}} - x_n = {diffs[0]:.4g}" ) variance = np.var(diffs) if variance < self.tolerance: return PatternInfo( pattern_type=PatternType.ARITHMETIC, parameters={'difference': np.mean(diffs)}, confidence=1.0 - variance, score=1.0 * (1.0 - variance), description=f"Arithmetic: d = {np.mean(diffs):.4g}" ) return None def _detect_geometric(self, seq: NDArray) -> Optional[PatternInfo]: """Detect geometric sequence: x_{n+1} / x_n = constant.""" if len(seq) < 2 or np.any(seq == 0): return None ratios = seq[1:] / seq[:-1] variance = np.var(ratios) if variance < self.tolerance: return PatternInfo( pattern_type=PatternType.GEOMETRIC, parameters={'ratio': np.mean(ratios)}, confidence=1.0 - variance, score=1.0 * (1.0 - variance), description=f"Geometric: r = {np.mean(ratios):.4g}" ) return None def _detect_periodic(self, seq: NDArray) -> Optional[PatternInfo]: """Detect periodic sequence: x_{n+k} = x_n.""" n = len(seq) # Try periods from 2 to n//2 for period in range(2, n // 2 + 1): if n % period != 0: continue # Check if sequence repeats repeats = n // period base = seq[:period] is_periodic = True for i in range(1, repeats): if not np.allclose(seq[i*period:(i+1)*period], base, atol=self.tolerance): is_periodic = False break if is_periodic: return PatternInfo( pattern_type=PatternType.PERIODIC, parameters={'period': period, 'cycle': base.tolist()}, confidence=1.0, score=1.0, description=f"Periodic with period {period}: {base}" ) return None def _detect_fibonacci(self, seq: NDArray) -> Optional[PatternInfo]: """Detect Fibonacci-like recurrence: x_n = a*x_{n-1} + b*x_{n-2}.""" if len(seq) < 4: return None # Try to find a, b such that x_n ≈ a*x_{n-1} + b*x_{n-2} X = np.column_stack([seq[2:-1], seq[1:-2]]) y = seq[2:] try: coeffs, residuals, _, _ = np.linalg.lstsq(X, y, rcond=None) a, b = coeffs # Check fit quality predicted = X @ coeffs error = np.mean((predicted - y) ** 2) if error < self.tolerance: return PatternInfo( pattern_type=PatternType.FIBONACCI, parameters={'a': a, 'b': b}, confidence=1.0 - error, score=0.9 * (1.0 - error), description=f"Fibonacci-like: x_n = {a:.4g}*x_{{n-1}} + {b:.4g}*x_{{n-2}}" ) except: pass return None def _detect_polynomial(self, seq: NDArray) -> Optional[PatternInfo]: """Detect polynomial sequence using finite differences.""" if len(seq) < 4: return None diffs = seq.copy() degree = 0 # Count how many times we need to differentiate to get constant while len(diffs) > 1 and np.var(diffs[1:]) > self.tolerance: diffs = np.diff(diffs) degree += 1 if degree > 5: # Cap at degree 5 return None if degree >= 1 and degree <= 5: return PatternInfo( pattern_type=PatternType.POLYNOMIAL, parameters={'degree': degree}, confidence=0.8, score=0.7, description=f"Polynomial of degree {degree}" ) return None # ============================================================================ # PART 3: QUESTION TSP - Optimal Pattern Selection # ============================================================================ @dataclass class Question: """A question that can collapse pattern uncertainty.""" id: str text: str collapse_potential: float # Δ_i cost: float # W_i applies_to: List[PatternType] = field(default_factory=list) class QuestionTSP: """Finds optimal question path to collapse pattern space.""" def __init__(self): self.questions = self._build_question_lattice() def _build_question_lattice(self) -> List[Question]: """Build the lattice of possible questions.""" return [ Question("Q1", "Is this arithmetic?", 0.9, 0.1, [PatternType.ARITHMETIC]), Question("Q2", "Is this geometric?", 0.85, 0.15, [PatternType.GEOMETRIC]), Question("Q3", "Is this periodic?", 0.8, 0.2, [PatternType.PERIODIC]), Question("Q4", "Is this Fibonacci-like?", 0.75, 0.25, [PatternType.FIBONACCI]), Question("Q5", "Is this polynomial?", 0.7, 0.3, [PatternType.POLYNOMIAL]), Question("Q6", "Are gaps constant?", 0.85, 0.1, [PatternType.ARITHMETIC]), Question("Q7", "Do gaps follow a pattern?", 0.8, 0.2, [PatternType.PERIODIC, PatternType.FIBONACCI]), Question("Q8", "Is there a hidden period?", 0.75, 0.25, [PatternType.PERIODIC]), ] def find_optimal_path(self, candidate_patterns: List[PatternInfo], current_entropy: float) -> Tuple[List[Question], float]: """ Find the optimal sequence of questions to collapse uncertainty. Returns ------- path: List of questions to ask total_cost: Total W_i for the path """ if not candidate_patterns: return [], 0.0 # Filter questions relevant to candidate patterns pattern_types = {p.pattern_type for p in candidate_patterns} relevant_qs = [q for q in self.questions if any(pt in q.applies_to for pt in pattern_types)] # Sort by efficiency ratio Δ_i / W_i relevant_qs.sort(key=lambda q: q.collapse_potential / (q.cost + 1e-9), reverse=True) # Greedy selection until entropy collapses path = [] remaining_entropy = current_entropy total_cost = 0.0 for q in relevant_qs: if remaining_entropy < 0.1: # Threshold reached break path.append(q) total_cost += q.cost remaining_entropy *= (1.0 - q.collapse_potential) return path, total_cost def collapse_to_best_pattern(self, patterns: List[PatternInfo]) -> PatternInfo: """Select the best pattern based on confidence and score.""" if not patterns: return PatternInfo( pattern_type=PatternType.UNKNOWN, parameters={}, confidence=0.0, score=0.0, description="No pattern detected" ) # Combine confidence and score with bias toward higher scores best = max(patterns, key=lambda p: p.score * p.confidence) return best # ============================================================================ # PART 4: ODE GRADIENT DESCENT ON GAP # ============================================================================ class GapODEOptimizer: """Gradient descent on the gap probability loss function.""" def __init__(self, lr: float = 0.01, max_iter: int = 1000, tol: float = 1e-8): self.lr = lr self.max_iter = max_iter self.tol = tol def optimize(self, prior_probability: float, gap_history: Optional[NDArray] = None, antiresonance_freqs: Optional[NDArray] = None) -> Tuple[float, float]: """ Find optimal gap using ODE gradient descent. Parameters ---------- prior_probability: P_prior from pattern analysis gap_history: Historical gaps for initialization antiresonance_freqs: Frequencies to avoid Returns ------- (optimal_gap, final_loss) """ # Initialize gap from history or random if gap_history is not None and len(gap_history) > 0: gap = np.mean(gap_history[-10:]) # Use recent average gap += np.random.randn() * np.std(gap_history[-10:]) * 0.1 else: gap = np.random.randn() * 0.5 P_prior = max(prior_probability, 1e-6) # Simple gradient descent (avoid PyTorch dependency) for iteration in range(self.max_iter): # Compute loss: L(g) = sin²(π * s) where s = sqrt(g² + 4*P_prior) s = np.sqrt(gap ** 2 + 4.0 * P_prior) loss = np.sin(np.pi * s) ** 2 # Gradient: dL/dg = π * sin(2πs) * (d/ds)(s) * (ds/dg) # s = sqrt(g² + 4P), ds/dg = g/s # dL/dg = π * sin(2πs) * (g/s) grad = np.pi * np.sin(2 * np.pi * s) * (gap / (s + 1e-12)) # Antiresonance penalty if antiresonance_freqs is not None and len(antiresonance_freqs) > 0: # Frequency of current gap relative to history length freq_g = np.abs(gap) / (len(gap_history) + 1) if gap_history else np.abs(gap) for f_ar in antiresonance_freqs: penalty = np.exp(-((freq_g - f_ar) ** 2) / (2 * 0.01)) grad += 0.1 * penalty * (freq_g - f_ar) # Update with momentum if iteration == 0: velocity = 0.0 velocity = 0.9 * velocity - self.lr * grad gap = gap + velocity # Ensure positive gap if gap < 0: gap = np.abs(gap) # Check convergence if loss < self.tol: break return gap, loss # ============================================================================ # PART 5: LETTER SERIES PROCESSOR # ============================================================================ class LetterSeriesProcessor: """Specialized processor for letter/categorical sequences.""" # Standard alphabet mapping ALPHABET = {chr(ord('a') + i): i for i in range(26)} ALPHABET.update({chr(ord('A') + i): i for i in range(26)}) @classmethod def encode(cls, sequence: List[str]) -> NDArray: """Encode letter sequence to numeric ordinals.""" encoded = [] for char in sequence: if char in cls.ALPHABET: encoded.append(cls.ALPHABET[char]) else: # Unknown character - assign next available ordinal encoded.append(len(cls.ALPHABET) + len([c for c in encoded if c >= 26])) return np.array(encoded) @classmethod def decode(cls, ordinals: NDArray, original_seq: Optional[List[str]] = None) -> List[str]: """Decode ordinals back to letters.""" result = [] for ord_val in ordinals: if ord_val < 26: result.append(chr(ord('a') + int(ord_val))) elif original_seq: # Map back to original sequence's unknown chars idx = int(ord_val) - 26 if idx < len([c for c in (original_seq or []) if c not in cls.ALPHABET]): unknown_chars = [c for c in (original_seq or []) if c not in cls.ALPHABET] result.append(unknown_chars[idx]) else: result.append('?') else: result.append('?') return result @classmethod def compute_gaps(cls, encoded: NDArray) -> NDArray: """Compute ordinal gaps between consecutive letters.""" return np.diff(encoded) @classmethod def detect_periodicity(cls, encoded: NDArray) -> Tuple[Optional[int], float]: """Detect period in letter sequence using autocorrelation.""" n = len(encoded) if n < 4: return None, 0.0 # Compute autocorrelation mean = np.mean(encoded) var = np.var(encoded) if var < 1e-10: return 1, 1.0 # Constant sequence autocorr = np.correlate(encoded - mean, encoded - mean, mode='full') autocorr = autocorr[n-1:] / (var * np.arange(n, 0, -1)) # Find first significant peak after lag 0 best_period = None best_corr = 0.0 for lag in range(2, n // 2 + 1): if autocorr[lag] > best_corr and autocorr[lag] > 0.5: best_corr = autocorr[lag] best_period = lag return best_period, best_corr @classmethod def predict_next(cls, sequence: List[str]) -> Tuple[str, Dict]: """Predict the next letter in the sequence.""" encoded = cls.encode(sequence) # Detect periodicity period, confidence = cls.detect_periodicity(encoded) if period is not None and confidence > 0.7: # Periodic prediction next_ordinal = encoded[-(period - len(encoded) % period)] else: # Use gap prediction gaps = cls.compute_gaps(encoded) if len(gaps) > 0: # Most common gap gap_counter = Counter(gaps) most_common_gap = gap_counter.most_common(1)[0][0] next_ordinal = encoded[-1] + most_common_gap else: next_ordinal = encoded[-1] # Decode result = cls.decode(np.array([next_ordinal]), sequence)[0] return result, { 'period': period, 'confidence': confidence, 'encoded': encoded.tolist(), 'method': 'periodic' if period else 'gap_average' } # ============================================================================ # PART 6: CCT-ODE PREDICTOR (MAIN CLASS) # ============================================================================ @dataclass class PredictionResult: """Result of a prediction.""" next_value: Union[float, str] confidence: float pattern: PatternInfo method: str entropy_reduction: float antiresonances: NDArray question_path: List[str] energy_cost: float class CCTODEPredictor: """ CCT-ODE Next-Value Predictor ---------------------------- Predicts next values in sequences using: - Pattern detection (arithmetic, geometric, periodic, etc.) - xFFT antiresonance detection - Question TSP for optimal pattern selection - ODE gradient descent on gap probability """ def __init__(self, threshold: float = 0.01, lr: float = 0.01, max_iter: int = 1000, antiresonance_threshold_db: float = -6.0): self.threshold = threshold self.lr = lr self.max_iter = max_iter self.pattern_detector = PatternDetector() self.question_tsp = QuestionTSP() self.gap_optimizer = GapODEOptimizer(lr=lr, max_iter=max_iter) self.antiresonance_threshold_db = antiresonance_threshold_db # History self.sequence_history: List[float] = [] self.gap_history: List[float] = [] self.pattern_history: List[PatternInfo] = [] self.entropy_history: List[float] = [] self.letter_processor = LetterSeriesProcessor() def fit_predict(self, sequence: Union[List, NDArray, List[str]], return_full_result: bool = False) -> Union[PredictionResult, Union[float, str]]: """ Fit to sequence and predict next value. Parameters ---------- sequence: Input sequence (numeric, or letter strings) return_full_result: If True, return PredictionResult; else just the value Returns ------- PredictionResult or next value """ # Detect sequence type is_letter_sequence = all(isinstance(x, str) and len(x) == 1 for x in sequence) if is_letter_sequence: return self._predict_letter(sequence, return_full_result) else: return self._predict_numeric(sequence, return_full_result) def _predict_letter(self, sequence: List[str], return_full_result: bool) -> Union[PredictionResult, str]: """Predict next letter using specialized processor.""" next_letter, details = self.LetterSeriesProcessor.predict_next(sequence) if return_full_result: return PredictionResult( next_value=next_letter, confidence=details['confidence'], pattern=PatternInfo( pattern_type=PatternType.PERIODIC if details['period'] else PatternType.UNKNOWN, parameters={'period': details['period']}, confidence=details['confidence'], score=details['confidence'], description=f"Period {details['period']}" if details['period'] else "Unknown" ), method=details['method'], entropy_reduction=details['confidence'], antiresonances=np.array([]), question_path=['Periodicity Detection'], energy_cost=1.0 ) return next_letter def _predict_numeric(self, sequence: Union[List, NDArray], return_full_result: bool) -> Union[PredictionResult, float]: """Predict next numeric value using full CCT-ODE machinery.""" seq = np.asarray(sequence, dtype=np.float64) # Update history self.sequence_history.extend(seq.tolist()) # Compute gaps if len(seq) >= 2: gaps = np.diff(seq) self.gap_history.extend(gaps.tolist()) # === STEP 1: Stationary Pattern Detection === patterns = self.pattern_detector.detect(seq) # === STEP 2: Antiresonance Detection === if len(seq) >= 16: ar_result = detect_antiresonances_xfft( signal_main=seq, signal_ref=np.roll(seq, 1), fs=1.0, threshold_db=self.antiresonance_threshold_db, mode='cross' ) antiresonance_freqs = ar_result['frequencies'] else: antiresonance_freqs = np.array([]) # === STEP 3: Question TSP === current_entropy = self._estimate_entropy(seq) question_path, energy_cost = self.question_tsp.find_optimal_path( patterns, current_entropy ) # === STEP 4: Collapse to Best Pattern === best_pattern = self.question_tsp.collapse_to_best_pattern(patterns) self.pattern_history.append(best_pattern) # === STEP 5: ODE Gradient Descent on Gap === prior_prob = best_pattern.confidence if best_pattern.confidence > 0 else 0.5 gap_history_arr = np.array(self.gap_history[-50:]) if self.gap_history else None optimal_gap, final_loss = self.gap_optimizer.optimize( prior_probability=prior_prob, gap_history=gap_history_arr, antiresonance_freqs=antiresonance_freqs ) # === STEP 6: Predict Next Value === if best_pattern.pattern_type == PatternType.ARITHMETIC: next_value = seq[-1] + best_pattern.parameters.get('difference', optimal_gap) method = 'arithmetic' elif best_pattern.pattern_type == PatternType.GEOMETRIC: ratio = best_pattern.parameters.get('ratio', 1.0) next_value = seq[-1] * ratio method = 'geometric' elif best_pattern.pattern_type == PatternType.PERIODIC: period = best_pattern.parameters['period'] cycle = best_pattern.parameters['cycle'] idx = (len(seq) - period) % period next_value = cycle[idx] method = 'periodic' elif best_pattern.pattern_type == PatternType.FIBONACCI: a = best_pattern.parameters.get('a', 1.0) b = best_pattern.parameters.get('b', 1.0) if len(seq) >= 2: next_value = a * seq[-1] + b * seq[-2] else: next_value = seq[-1] + optimal_gap method = 'fibonacci' elif best_pattern.pattern_type == PatternType.POLYNOMIAL: # Extrapolate using finite differences next_value = self._extrapolate_polynomial(seq, best_pattern.parameters.get('degree', 2)) method = 'polynomial' else: # Fallback to ODE gap optimization next_value = seq[-1] + optimal_gap method = 'ode_gap' # === STEP 7: Compute Confidence === entropy_reduction = current_entropy - self._estimate_entropy(np.append(seq, next_value)) confidence = min(1.0, max(0.0, 1.0 - entropy_reduction)) if return_full_result: return PredictionResult( next_value=next_value, confidence=confidence, pattern=best_pattern, method=method, entropy_reduction=entropy_reduction, antiresonances=antiresonance_freqs, question_path=[q.text for q in question_path], energy_cost=energy_cost ) return next_value def _estimate_entropy(self, seq: NDArray) -> float: """Estimate entropy of the sequence (uncertainty measure).""" if len(seq) < 2: return 1.0 gaps = np.diff(seq) if len(gaps) < 2: return 1.0 # Shannon entropy of gap distribution gap_bins = np.digitize(gaps, np.linspace(gaps.min(), gaps.max(), 10)) counts = np.bincount(gap_bins, minlength=10) probs = counts / len(gaps) probs = probs[probs > 0] entropy = -np.sum(probs * np.log2(probs + 1e-12)) return min(1.0, entropy / np.log2(10)) # Normalize to [0, 1] def _extrapolate_polynomial(self, seq: NDArray, degree: int) -> float: """Extrapolate next value using polynomial fit.""" x = np.arange(len(seq)) try: coeffs = np.polyfit(x, seq, min(degree, len(seq) - 1)) return np.polyval(coeffs, len(seq)) except: return seq[-1] + np.mean(np.diff(seq)) def predict_n_steps(self, sequence: Union[List, NDArray, List[str]], n: int) -> List: """Predict next n values autoregressively.""" seq = list(sequence) predictions = [] for _ in range(n): next_val = self.fit_predict(seq) predictions.append(next_val) seq.append(next_val) return predictions def get_statistics(self) -> Dict: """Get predictor statistics and history.""" return { 'sequence_length': len(self.sequence_history), 'gap_history_length': len(self.gap_history), 'patterns_detected': len(self.pattern_history), 'entropy_trajectory': self.entropy_history, 'current_entropy': self.entropy_history[-1] if self.entropy_history else 1.0, 'antiresonance_count': sum(1 for p in self.pattern_history if len(p.description) > 0) } # ============================================================================ # PART 7: DEMONSTRATION # ============================================================================ def demo(): """Demonstrate CCT-ODE Predictor on various sequence types.""" print("=" * 70) print("CCT-ODE Next-Value Predictor Demo") print("=" * 70) predictor = CCTODEPredictor(threshold=0.01) # --- Demo 1: Arithmetic Sequence --- print("\n[1] Arithmetic Sequence") arith_seq = [3, 7, 11, 15, 19, 23] pred = predictor.fit_predict(arith_seq, return_full_result=True) print(f" Sequence: {arith_seq}") print(f" Prediction: {pred.next_value} (expected: 27)") print(f" Pattern: {pred.pattern.description}") print(f" Method: {pred.method}") print(f" Confidence: {pred.confidence:.2%}") print(f" Energy Cost: {pred.energy_cost:.4f}") # --- Demo 2: Geometric Sequence --- print("\n[2] Geometric Sequence") geom_seq = [2, 6, 18, 54, 162] pred = predictor.fit_predict(geom_seq, return_full_result=True) print(f" Sequence: {geom_seq}") print(f" Prediction: {pred.next_value} (expected: 486)") print(f" Pattern: {pred.pattern.description}") print(f" Method: {pred.method}") # --- Demo 3: Periodic Sequence --- print("\n[3] Periodic Sequence") periodic_seq = [1, 3, 5, 7, 1, 3, 5, 7, 1] pred = predictor.fit_predict(periodic_seq, return_full_result=True) print(f" Sequence: {periodic_seq}") print(f" Prediction: {pred.next_value} (expected: 3)") print(f" Pattern: {pred.pattern.description}") # --- Demo 4: Fibonacci Sequence --- print("\n[4] Fibonacci-like Sequence") fib_seq = [1, 1, 2, 3, 5, 8, 13, 21] pred = predictor.fit_predict(fib_seq, return_full_result=True) print(f" Sequence: {fib_seq}") print(f" Prediction: {pred.next_value} (expected: 34)") print(f" Pattern: {pred.pattern.description}") # --- Demo 5: Letter Series --- print("\n[5] Letter Series") letter_seq = ['A', 'B', 'A', 'B', 'A', 'B'] pred = predictor.fit_predict(letter_seq, return_full_result=True) print(f" Sequence: {letter_seq}") print(f" Prediction: {pred.next_value} (expected: A)") print(f" Pattern: {pred.pattern.description}") # --- Demo 6: Complex Letter Series --- print("\n[6] Complex Letter Series") complex_letter_seq = ['A', 'B', 'C', 'A', 'B', 'D', 'A', 'B', 'E'] pred = predictor.fit_predict(complex_letter_seq, return_full_result=True) print(f" Sequence: {complex_letter_seq}") print(f" Prediction: {pred.next_value}") print(f" Method: {pred.method}") # --- Demo 7: Noisy Sequence (uncertain prediction) --- print("\n[7] Noisy Sequence (high entropy)") noisy_seq = [1.1, 2.3, 1.9, 4.2, 3.8, 5.1, 5.9] pred = predictor.fit_predict(noisy_seq, return_full_result=True) print(f" Sequence: {noisy_seq}") print(f" Prediction: {pred.next_value:.4f}") print(f" Pattern: {pred.pattern.description}") print(f" Confidence: {pred.confidence:.2%}") print(f" Antiresonances: {pred.antiresonances}") # --- Demo 8: Multi-step Prediction --- print("\n[8] Multi-step Prediction (Fibonacci)") fib_full = [1, 1, 2, 3, 5] predictions = predictor.predict_n_steps(fib_full, 5) print(f" Given: {fib_full}") print(f" Next 5: {predictions}") print(f" Expected: [8, 13, 21, 34, 55]") # --- Demo 9: Antiresonance Detection --- print("\n[9] Antiresonance Detection in Periodic Signal") import numpy as np t = np.linspace(0, 10, 200) periodic_signal = np.sin(2 * np.pi * 0.5 * t) + 0.3 * np.sin(2 * np.pi * 0.9 * t) ar_result = detect_antiresonances_xfft(periodic_signal, np.roll(periodic_signal, 1), fs=20.0, threshold_db=-6.0, mode='cross') print(f" Detected antiresonances: {ar_result['frequencies']}") print(f" Depths (dB): {ar_result['depths_db']}") # --- Demo 10: Question TSP Path --- print("\n[10] Question TSP Path Analysis") predictor2 = CCTODEPredictor() patterns = predictor2.pattern_detector.detect(np.array([1, 3, 5, 7, 9])) path, cost = predictor2.question_tsp.find_optimal_path(patterns, current_entropy=0.8) print(f" Patterns found: {[p.pattern_type.name for p in patterns]}") print(f" Optimal question path: {[q.text for q in path]}") print(f" Total energy cost: {cost:.4f}") print("\n" + "=" * 70) print("Demo Complete") print("=" * 70) if __name__ == "__main__": demo()