Hugging Face's logo

Spaces:
GaryLeckey
/
aureon-os
Running

App Files Community
aureon-os / aureon /intelligence /self_validating_predictor.py
GaryLeckey's picture
GaryLeckey
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 4)
b41bee3 verified
Raw History Blame Contribute Delete
17.8 kB
#!/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()