Blame for aureon/intelligence/self_validating_predictor.py

#!/usr/bin/env python3
"""

🎯🔮 AUREON SELF-VALIDATING PREDICTION ENGINE 🔮🎯

===================================================



This system PREDICTS market movements BEFORE they happen,

then VALIDATES predictions against actual results.



PROCESS:

1. Collect baseline data (10 seconds)

2. Generate PREDICTION for next 30 seconds

3. WAIT for prediction window

4. VALIDATE prediction against actual movement

5. Score accuracy and adjust



Only when predictions are CONSISTENTLY CORRECT do we trade.



Gary Leckey | December 2025

"Predict. Validate. Then Trade."

"""

from aureon.core.aureon_baton_link import link_system as _baton_link; _baton_link(__name__)
import os
import sys
import time
import math
import json
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass, field
from collections import deque

os.environ['LIVE'] = '1'

from aureon.exchanges.binance_client import BinanceClient, get_binance_client
from aureon.exchanges.kraken_client import KrakenClient, get_kraken_client

# Constants
PHI = (1 + math.sqrt(5)) / 2

FREQ_MAP = {
    'ROOT': 256.0,
    'TRANSFORMATION': 417.0,
    'NATURAL': 432.0,
    'DISTORTION': 440.0,
    'LOVE': 528.0,
    'CONNECTION': 639.0,
}


@dataclass
class Prediction:
    """A single prediction with validation"""
    symbol: str
    platform: str
    timestamp: datetime
    
    # Baseline
    baseline_price: float
    baseline_momentum: float
    baseline_frequency: float
    
    # Prediction (30 seconds ahead)
    predicted_direction: str  # UP, DOWN, FLAT
    predicted_change_pct: float
    predicted_price: float
    confidence: float
    
    # Validation (filled after window)
    actual_price: float = 0.0
    actual_change_pct: float = 0.0
    actual_direction: str = ""
    
    # Scoring
    direction_correct: bool = False
    magnitude_accuracy: float = 0.0  # How close was the prediction
    validated: bool = False
    
    def validate(self, actual_price: float):
        """Validate prediction against actual result"""
        self.actual_price = actual_price
        self.actual_change_pct = ((actual_price - self.baseline_price) / self.baseline_price) * 100
        
        # Determine actual direction
        if self.actual_change_pct > 0.01:
            self.actual_direction = "UP"
        elif self.actual_change_pct < -0.01:
            self.actual_direction = "DOWN"
        else:
            self.actual_direction = "FLAT"
        
        # Check direction accuracy
        self.direction_correct = (self.predicted_direction == self.actual_direction)
        
        # Calculate magnitude accuracy (how close was the prediction)
        if self.predicted_change_pct != 0:
            error = abs(self.actual_change_pct - self.predicted_change_pct)
            self.magnitude_accuracy = max(0, 1 - (error / abs(self.predicted_change_pct)))
        else:
            self.magnitude_accuracy = 1.0 if abs(self.actual_change_pct) < 0.02 else 0.0
        
        self.validated = True


class SelfValidatingPredictor:
    """

    Prediction engine that validates its own assumptions.

    """
    
    def __init__(self):
        print("\n🔮 Initializing Self-Validating Prediction Engine...")
        
        self.binance = get_binance_client()
        self.kraken = get_kraken_client()
        
        # Prediction history
        self.predictions: List[Prediction] = []
        self.accuracy_window = deque(maxlen=20)  # Last 20 predictions
        
        # Tracking
        self.total_predictions = 0
        self.correct_predictions = 0
        
        print("   ✅ Ready to predict and validate\n")
    
    def get_price(self, platform: str, symbol: str) -> float:
        """Get current price"""
        try:
            if platform == 'binance':
                ticker = self.binance.get_24h_ticker(symbol)
                return float(ticker.get('lastPrice', 0))
            elif platform == 'kraken':
                ticker = self.kraken.get_24h_ticker(symbol)
                return float(ticker.get('lastPrice', 0))
        except:
            pass
        return 0.0
    
    def collect_baseline(self, platform: str, symbol: str, 

                         duration_sec: int = 10) -> Tuple[List[float], float, float]:
        """

        Collect baseline data for prediction.

        Returns (prices, avg_momentum, frequency)

        """
        prices = []
        prev_price = None
        momentums = []
        
        for i in range(duration_sec * 2):  # 2 samples per second
            price = self.get_price(platform, symbol)
            if price > 0:
                prices.append(price)
                if prev_price:
                    momentum = ((price - prev_price) / prev_price) * 100
                    momentums.append(momentum)
                prev_price = price
            time.sleep(0.5)
        
        avg_momentum = np.mean(momentums) if momentums else 0
        
        # Calculate frequency
        if len(prices) >= 2:
            ratio = prices[-1] / prices[0]
            frequency = 432.0 * (ratio ** PHI)
            frequency = max(256, min(963, frequency))
        else:
            frequency = 432.0
        
        return prices, avg_momentum, frequency
    
    def generate_prediction(self, platform: str, symbol: str,

                           prices: List[float], momentum: float, 

                           frequency: float) -> Prediction:
        """

        Generate 30-second prediction based on:

        - Momentum trend

        - Frequency state

        - Pattern analysis

        """
        baseline_price = prices[-1] if prices else 0
        
        # Analyze momentum pattern
        if len(prices) >= 5:
            recent_prices = prices[-5:]
            price_trend = np.polyfit(range(len(recent_prices)), recent_prices, 1)[0]
            momentum_strength = abs(momentum) * 10
        else:
            price_trend = 0
            momentum_strength = 0
        
        # Frequency influence
        freq_boost = 0
        if abs(frequency - FREQ_MAP['LOVE']) < 30:
            freq_boost = 0.02  # Love frequency = bullish
        elif abs(frequency - FREQ_MAP['NATURAL']) < 15:
            freq_boost = 0.01  # Natural = slight bullish
        elif abs(frequency - FREQ_MAP['DISTORTION']) < 10:
            freq_boost = -0.02  # Distortion = bearish
        
        # Calculate predicted change
        # Base: momentum continuation with decay
        momentum_contribution = momentum * 0.5  # 50% momentum continuation
        trend_contribution = (price_trend / baseline_price) * 100 * 30 if baseline_price > 0 else 0  # Extrapolate 30 sec
        frequency_contribution = freq_boost
        
        predicted_change = momentum_contribution + trend_contribution * 0.3 + frequency_contribution
        
        # Clamp to reasonable range
        predicted_change = max(-0.5, min(0.5, predicted_change))
        
        # Determine direction
        if predicted_change > 0.01:
            direction = "UP"
        elif predicted_change < -0.01:
            direction = "DOWN"
        else:
            direction = "FLAT"
        
        # Calculate confidence
        momentum_consistency = 1.0 - min(1.0, np.std(prices[-5:]) / np.mean(prices[-5:]) * 100) if len(prices) >= 5 else 0.5
        confidence = min(0.9, (momentum_consistency * 0.5 + momentum_strength * 0.3 + 0.2))
        
        predicted_price = baseline_price * (1 + predicted_change / 100)
        
        return Prediction(
            symbol=symbol,
            platform=platform,
            timestamp=datetime.now(),
            baseline_price=baseline_price,
            baseline_momentum=momentum,
            baseline_frequency=frequency,
            predicted_direction=direction,
            predicted_change_pct=predicted_change,
            predicted_price=predicted_price,
            confidence=confidence
        )
    
    def run_prediction_cycle(self, platform: str, symbol: str,

                             prediction_window_sec: int = 30) -> Prediction:
        """

        Complete prediction cycle:

        1. Collect baseline

        2. Make prediction

        3. Wait for window

        4. Validate

        """
        print(f"\n{'='*60}")
        print(f"🔮 PREDICTION CYCLE: {platform.upper()} {symbol}")
        print(f"{'='*60}")
        
        # Step 1: Collect baseline
        print(f"\n📊 STEP 1: Collecting baseline (10 seconds)...")
        prices, momentum, frequency = self.collect_baseline(platform, symbol, 10)
        
        if not prices:
            print("   ❌ Failed to collect baseline data")
            return None
        
        print(f"   Baseline price: ${prices[-1]:.4f}")
        print(f"   Momentum: {momentum:+.4f}%/sample")
        print(f"   Frequency: {frequency:.1f}Hz")
        
        # Step 2: Generate prediction
        print(f"\n🎯 STEP 2: Generating prediction for next {prediction_window_sec} seconds...")
        prediction = self.generate_prediction(platform, symbol, prices, momentum, frequency)
        
        print(f"   ┌─────────────────────────────────────────┐")
        print(f"   │ PREDICTION:                             │")
        print(f"   │   Direction: {prediction.predicted_direction:8}                   │")
        print(f"   │   Change:    {prediction.predicted_change_pct:+.4f}%                  │")
        print(f"   │   Target:    ${prediction.predicted_price:.4f}               │")
        print(f"   │   Confidence: {prediction.confidence:.1%}                     │")
        print(f"   └─────────────────────────────────────────┘")
        
        # Step 3: Wait for prediction window
        print(f"\n⏳ STEP 3: Waiting {prediction_window_sec} seconds for validation...")
        
        for i in range(prediction_window_sec):
            current = self.get_price(platform, symbol)
            current_change = ((current - prediction.baseline_price) / prediction.baseline_price) * 100
            
            # Progress indicator
            bar_len = 30
            progress = int((i + 1) / prediction_window_sec * bar_len)
            bar = "█" * progress + "░" * (bar_len - progress)
            
            direction_icon = "↑" if current_change > 0 else "↓" if current_change < 0 else "→"
            print(f"   [{bar}] {i+1}/{prediction_window_sec}s | ${current:.4f} ({current_change:+.4f}%) {direction_icon}", end='\r')
            
            time.sleep(1)
        
        print()  # New line after progress
        
        # Step 4: Validate
        print(f"\n✓ STEP 4: Validating prediction...")
        actual_price = self.get_price(platform, symbol)
        prediction.validate(actual_price)
        
        # Update tracking
        self.predictions.append(prediction)
        self.total_predictions += 1
        if prediction.direction_correct:
            self.correct_predictions += 1
        self.accuracy_window.append(1 if prediction.direction_correct else 0)
        
        # Calculate running accuracy
        running_accuracy = sum(self.accuracy_window) / len(self.accuracy_window) if self.accuracy_window else 0
        
        print(f"\n   ┌─────────────────────────────────────────────────────┐")
        print(f"   │ VALIDATION RESULT:                                  │")
        print(f"   │   Predicted: {prediction.predicted_direction:8} ({prediction.predicted_change_pct:+.4f}%)            │")
        print(f"   │   Actual:    {prediction.actual_direction:8} ({prediction.actual_change_pct:+.4f}%)            │")
        print(f"   │   Direction: {'✅ CORRECT' if prediction.direction_correct else '❌ WRONG':18}                │")
        print(f"   │   Magnitude: {prediction.magnitude_accuracy:.1%} accurate                     │")
        print(f"   └─────────────────────────────────────────────────────┘")
        
        print(f"\n📈 RUNNING STATS:")
        print(f"   Total predictions: {self.total_predictions}")
        print(f"   Correct: {self.correct_predictions} ({self.correct_predictions/self.total_predictions:.1%})")
        print(f"   Last {len(self.accuracy_window)} accuracy: {running_accuracy:.1%}")
        
        # Check if we're on the right path
        if running_accuracy >= 0.7:
            print(f"\n   🎯 ON THE RIGHT PATH! Accuracy ≥70%")
        elif running_accuracy >= 0.5:
            print(f"\n   ⚠️  NEEDS IMPROVEMENT. Accuracy 50-70%")
        else:
            print(f"\n   ❌ MODEL NEEDS RECALIBRATION. Accuracy <50%")
        
        return prediction
    
    def run_validation_session(self, cycles: int = 5):
        """Run multiple prediction cycles to build confidence"""
        
        # 🔶 COMPREHENSIVE VALIDATION SYMBOLS
        symbols = [
            # TOP TIER
            ('binance', 'BTCUSDC'), ('binance', 'ETHUSDC'), ('binance', 'SOLUSDC'),
            ('binance', 'XRPUSDC'), ('binance', 'ADAUSDC'), ('binance', 'AVAXUSDC'),
            # LAYER 2s
            ('binance', 'ARBUSDC'), ('binance', 'OPUSDC'), ('binance', 'SUIUSDC'),
            # DEFI
            ('binance', 'UNIUSDC'), ('binance', 'LINKUSDC'), ('binance', 'AAVEUSDC'),
            # AI
            ('binance', 'FETUSDC'), ('binance', 'INJUSDC'),
            # MEMECOINS
            ('binance', 'DOGEUSDC'), ('binance', 'PEPEUSDC'), ('binance', 'BONKUSDC'),
        ]
        
        print("\n" + "="*70)
        print("🔮 SELF-VALIDATING PREDICTION SESSION")
        print("="*70)
        print(f"Running {cycles} prediction cycles across {len(symbols)} symbols")
        print("Each cycle: 10s baseline + 30s prediction window = 40s")
        print(f"Estimated time: {cycles * 40}s ({cycles * 40 / 60:.1f} min)")
        print("="*70)
        
        all_predictions = []
        
        for cycle in range(cycles):
            print(f"\n{'━'*70}")
            print(f"CYCLE {cycle + 1}/{cycles}")
            print(f"{'━'*70}")
            
            # Rotate through symbols
            platform, symbol = symbols[cycle % len(symbols)]
            
            prediction = self.run_prediction_cycle(platform, symbol, 30)
            if prediction:
                all_predictions.append(prediction)
        
        # Final summary
        print("\n" + "="*70)
        print("📊 FINAL VALIDATION SUMMARY")
        print("="*70)
        
        if all_predictions:
            correct = sum(1 for p in all_predictions if p.direction_correct)
            accuracy = correct / len(all_predictions)
            
            avg_magnitude_acc = np.mean([p.magnitude_accuracy for p in all_predictions])
            avg_confidence = np.mean([p.confidence for p in all_predictions])
            
            print(f"\n   Total Predictions: {len(all_predictions)}")
            print(f"   Direction Accuracy: {correct}/{len(all_predictions)} ({accuracy:.1%})")
            print(f"   Magnitude Accuracy: {avg_magnitude_acc:.1%}")
            print(f"   Avg Confidence: {avg_confidence:.1%}")
            
            print(f"\n   {'─'*50}")
            print(f"   │ {'PREDICTION':10} │ {'ACTUAL':10} │ {'RESULT':10} │")
            print(f"   {'─'*50}")
            
            for p in all_predictions:
                result = "✅ CORRECT" if p.direction_correct else "❌ WRONG"
                print(f"   │ {p.predicted_direction:10} │ {p.actual_direction:10} │ {result:10} │")
            
            print(f"   {'─'*50}")
            
            # Recommendation
            print(f"\n🎯 RECOMMENDATION:")
            if accuracy >= 0.7:
                print(f"   ✅ PREDICTIONS ARE RELIABLE ({accuracy:.0%} accuracy)")
                print(f"   ✅ System is ON THE RIGHT PATH")
                print(f"   ✅ Ready for live trading with confidence")
            elif accuracy >= 0.5:
                print(f"   ⚠️  PREDICTIONS NEED IMPROVEMENT ({accuracy:.0%} accuracy)")
                print(f"   ⚠️  Continue validation before trading")
            else:
                print(f"   ❌ PREDICTIONS ARE UNRELIABLE ({accuracy:.0%} accuracy)")
                print(f"   ❌ Model needs recalibration")
                print(f"   ❌ Do NOT trade until accuracy improves")
        
        print("\n" + "="*70)
        
        return all_predictions


def main():
    print("\n" + "="*70)
    print("🎯🔮 AUREON SELF-VALIDATING PREDICTION ENGINE 🔮🎯")
    print("="*70)
    print("\nThis system PREDICTS market movements BEFORE they happen,")
    print("then VALIDATES predictions against actual results.")
    print("\nOnly when predictions are CONSISTENTLY CORRECT do we trade.")
    print("="*70)
    
    predictor = SelfValidatingPredictor()
    
    # Run validation session
    predictions = predictor.run_validation_session(cycles=5)
    
    print("\nDone!")


if __name__ == "__main__":
    main()