predictor.py

# This file is part of the Number Sequence Predictor Activity.
# Copyright (C) 2025 
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

import math
import numpy as np
from datetime import datetime
from sklearn.linear_model import SGDRegressor
from collections import defaultdict
from sklearn.exceptions import NotFittedError
import os
import json

class NumberSequencePredictor:
    
    def __init__(self, save_file="sequence_predictor_data.json"):
        self.save_file = save_file
        self.models = defaultdict(lambda: SGDRegressor(learning_rate='adaptive', eta0=0.01))
        self.pattern_features = []
        self.pattern_targets = []
        self.correction_history = []
        self.sequences_database = []
        self.memorized_sequences = {}
        self.is_fitted = False
        self.stats = {
            'total_predictions': 0,
            'correct_predictions': 0,
            'total_corrections': 0,
            'rule_based_correct': 0,
            'memorized_correct': 0,
            'ml_correct': 0,
            'fallback_used': 0,
            'created_date': datetime.now().isoformat(),
            'last_updated': datetime.now().isoformat(),
            'sessions_played': 0,
            'sequences_played': 0
        }
        
        self.load_data()
        self.stats['sessions_played'] += 1

    def predict_next(self, sequence):
        self.stats['total_predictions'] += 1
        
        rule_pred = self._rule_based_predict(sequence)
        if rule_pred is not None:
            pattern_name = self._detect_pattern_name(sequence)
            return rule_pred, "Rule-based", pattern_name
        
        memorized_pred, match_type = self._check_memorized_sequence(sequence)
        if memorized_pred is not None:
            return memorized_pred, f"Memorized ({match_type})", "Exact match"
        
        if self.is_fitted and len(self.pattern_features) > 0:
            try:
                features = self.extract_features(sequence)
                ml_pred = self.models['general'].predict([features])[0]
                return ml_pred, "ML Model", "Complex pattern"
            except (NotFittedError, Exception):
                self.is_fitted = False
        
        self.stats['fallback_used'] += 1
        if len(sequence) >= 2:
            fallback_pred = sequence[-1] + (sequence[-1] - sequence[-2])
            return fallback_pred, "Fallback", "Simple difference"
        else:
            fallback_pred = sequence[-1] if sequence else 1
            return fallback_pred, "Fallback", "Single value"

    def learn_from_correction(self, sequence, correct_next, predicted_next, method_used):
        """Learn from user corrections"""
        self.memorize_sequence(sequence, correct_next)
        
        self.correction_history.append({
            'sequence': sequence.copy(),
            'predicted': float(predicted_next),
            'correct': float(correct_next),
            'method_used': method_used,
            'timestamp': datetime.now().isoformat(),
            'pattern': self._detect_pattern_name(sequence)
        })
        
        self.stats['total_corrections'] += 1
        
        if method_used not in ["Rule-based"]:
            features = self.extract_features(sequence)
            self.pattern_features.append(features)
            self.pattern_targets.append(correct_next)
            
            if len(self.pattern_features) >= 2:
                X = np.array(self.pattern_features)
                y = np.array(self.pattern_targets)
                
                try:
                    if not self.is_fitted:
                        self.models['general'].fit(X, y)
                        self.is_fitted = True
                    else:
                        self.models['general'].partial_fit(X[-1:], y[-1:])
                except Exception:
                    pass

    def extract_features(self, sequence):
        """Extract features for ML model"""
        if len(sequence) < 2:
            return np.array([0] * 12)
        
        features = [
            len(sequence),
            np.mean(sequence),
            np.std(sequence) if len(sequence) > 1 else 0,
            sequence[-1],
            sequence[-1] - sequence[-2] if len(sequence) > 1 else 0,
        ]
        
        if len(sequence) > 1:
            diffs = np.diff(sequence)
            features.extend([np.mean(diffs), np.std(diffs) if len(diffs) > 1 else 0])
        else:
            features.extend([0, 0])
        
        if len(sequence) > 2:
            second_diffs = np.diff(np.diff(sequence))
            features.extend([np.mean(second_diffs), np.std(second_diffs) if len(second_diffs) > 1 else 0])
        else:
            features.extend([0, 0])
        
        features.extend([
            1 if self._is_arithmetic(sequence) else 0,
            1 if self._is_geometric(sequence) else 0,
            1 if self._is_powers(sequence) else 0,
        ])
        
        return np.array(features)

    def _is_fibonacci(self, seq):
        if len(seq) < 3:
            return False
        for i in range(2, len(seq)):
            if abs(seq[i] - (seq[i-1] + seq[i-2])) > 1e-10:
                return False
        return True

    def _is_arithmetic(self, seq):
        if len(seq) < 3:
            return True
        diffs = np.diff(seq)
        return np.std(diffs) < 1e-6

    def _is_geometric(self, seq):
        if len(seq) < 3 or any(abs(x) < 1e-10 for x in seq[:-1]):
            return False
        ratios = [seq[i+1] / seq[i] for i in range(len(seq)-1)]
        return np.std(ratios) < 1e-6

    def _is_quadratic(self, seq):
        if len(seq) < 4:
            return False
        first_diffs = np.diff(seq)
        second_diffs = np.diff(first_diffs)
        return np.std(second_diffs) < 1e-6

    def _is_factorial(self, seq):
        if len(seq) < 3:
            return False
        for i, val in enumerate(seq):
            expected = math.factorial(i + 1)
            if abs(val - expected) > 1e-10:
                return False
        return True

    def _is_powers(self, seq):
        if len(seq) < 3:
            return False
        for power in range(1, 8):
            expected = [(i+1)**power for i in range(len(seq))]
            if all(abs(seq[i] - expected[i]) < 1e-10 for i in range(len(seq))):
                return True
        return False

    def _detect_power(self, seq):
        for power in range(1, 8):
            expected = [(i+1)**power for i in range(len(seq))]
            if all(abs(seq[i] - expected[i]) < 1e-10 for i in range(len(seq))):
                return power
        return None
    
    def memorize_sequence(self, sequence, correct_next):
        key = str([round(x, 6) for x in sequence])
        self.memorized_sequences[key] = correct_next

    def _check_memorized_sequence(self, sequence):
        key = str([round(x, 6) for x in sequence])
        
        if key in self.memorized_sequences:
            return self.memorized_sequences[key], "Exact match"
        
        for memorized_key, next_val in self.memorized_sequences.items():
            try:
                memorized_seq = eval(memorized_key)
                if len(memorized_seq) > len(sequence):
                    if memorized_seq[:len(sequence)] == sequence:
                        if len(sequence) + 1 < len(memorized_seq):
                            return memorized_seq[len(sequence)], "Partial match"
            except:
                continue
                
        return None, None
    
    def _rule_based_predict(self, sequence):
        if len(sequence) < 2:
            return None
        
        if self._is_fibonacci(sequence):
            return sequence[-1] + sequence[-2]
        
        if self._is_arithmetic(sequence):
            diff = sequence[-1] - sequence[-2]
            return sequence[-1] + diff
        
        if self._is_geometric(sequence):
            if sequence[-2] != 0:
                ratio = sequence[-1] / sequence[-2]
                return sequence[-1] * ratio
        
        if self._is_powers(sequence):
            power = self._detect_power(sequence)
            if power:
                base = len(sequence) + 1
                return base ** power
        
        if self._is_factorial(sequence):
            return math.factorial(len(sequence) + 1)
        
        if self._is_quadratic(sequence):
            first_diffs = np.diff(sequence)
            second_diffs = np.diff(first_diffs)
            next_first_diff = first_diffs[-1] + second_diffs[-1]
            return sequence[-1] + next_first_diff
        
        return None

    def _detect_pattern_name(self, sequence):
        if len(sequence) < 2:
            return "Too short"
        
        if len(sequence) >= 3 and self._is_fibonacci(sequence):
            return "Fibonacci"
        elif self._is_arithmetic(sequence):
            diff = sequence[-1] - sequence[-2] if len(sequence) >= 2 else 0
            return f"Arithmetic (+{diff})"
        elif self._is_geometric(sequence):
            ratio = sequence[-1] / sequence[-2] if len(sequence) >= 2 and sequence[-2] != 0 else 1
            return f"Geometric (×{ratio:.2f})"
        elif self._is_powers(sequence):
            power = self._detect_power(sequence)
            return f"Powers (n^{power})" if power else "Powers"
        elif self._is_quadratic(sequence):
            return "Quadratic"
        elif self._is_factorial(sequence):
            return "Factorial"
        else:
            return "Custom/Complex"

    def extract_features(self, sequence):
        if len(sequence) < 2:
            return np.array([0] * 12)
        
        features = [
            len(sequence),
            np.mean(sequence),
            np.std(sequence) if len(sequence) > 1 else 0,
            sequence[-1],
            sequence[-1] - sequence[-2] if len(sequence) > 1 else 0,
        ]
        
        if len(sequence) > 1:
            diffs = np.diff(sequence)
            features.extend([np.mean(diffs), np.std(diffs) if len(diffs) > 1 else 0])
        else:
            features.extend([0, 0])
        
        if len(sequence) > 2:
            second_diffs = np.diff(np.diff(sequence))
            features.extend([np.mean(second_diffs), np.std(second_diffs) if len(second_diffs) > 1 else 0])
        else:
            features.extend([0, 0])
        
        features.extend([
            1 if self._is_arithmetic(sequence) else 0,
            1 if self._is_geometric(sequence) else 0,
            1 if self._is_powers(sequence) else 0,
        ])
        
        return np.array(features)

    def save_data(self):
        try:
            self.stats['last_updated'] = datetime.now().isoformat()
            
            data = {
                'pattern_features': [feature.tolist() if isinstance(feature, np.ndarray) 
                                   else feature for feature in self.pattern_features],
                'pattern_targets': self.pattern_targets,
                'correction_history': self.correction_history,
                'sequences_database': self.sequences_database,
                'memorized_sequences': self.memorized_sequences,
                'is_fitted': self.is_fitted,
                'stats': self.stats
            }
            
            with open(self.save_file, 'w') as f:
                json.dump(data, f, indent=2, default=str)
                
        except Exception as e:
            print(f"Error saving data: {e}")

    def load_data(self):
        try:
            if os.path.exists(self.save_file):
                with open(self.save_file, 'r') as f:
                    data = json.load(f)
                
                self.pattern_features = [np.array(feature) for feature in data.get('pattern_features', [])]
                self.pattern_targets = data.get('pattern_targets', [])
                self.correction_history = data.get('correction_history', [])
                self.sequences_database = data.get('sequences_database', [])
                self.memorized_sequences = data.get('memorized_sequences', {})
                self.is_fitted = data.get('is_fitted', False)
                self.stats = data.get('stats', self.stats)
                
                if self.is_fitted and len(self.pattern_features) >= 2:
                    try:
                        X = np.array(self.pattern_features)
                        y = np.array(self.pattern_targets)
                        self.models['general'].fit(X, y)
                    except Exception:
                        self.is_fitted = False
                        
        except Exception as e:
            print(f"Could not load previous data: {e}")