# quanta_sim_v4_correlated_outputs.py import neat import numpy as np import os import logging import pickle import random import math import datetime import itertools # --- Loglama Ayarları --- log_filename = f"quanta_log_v4_corr_{datetime.datetime.now():%Y%m%d_%H%M%S}.log" # <-- V4 için dosya adı logging.basicConfig( level=logging.INFO, # DEBUG seviyesi daha fazla bilgi verir ama yavaşlatabilir format='%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s', handlers=[ logging.FileHandler(log_filename), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) logger.info("="*70) logger.info("Quanta Simülatörü Başlatılıyor (Sürüm 4 - Korelasyonlu Çıktılar)") # <-- GÜNCELLENDİ logger.info("="*70) # --- Simülasyon Parametreleri --- NUM_TEST_VALUES_PER_AXIS = 3 # Değerlendirme için ızgara boyutu (3x3=9 nokta) NUM_TRIALS_PER_PAIR = 30 # Her girdi çifti için deneme sayısı (artırıldı) MAX_GENERATIONS = 250 # <-- Problem çok daha zor, nesil sayısı artırıldı FITNESS_THRESHOLD = 0.97 # <-- Hedef zor, eşik ayarlanabilir num_test_points = NUM_TEST_VALUES_PER_AXIS ** 2 total_trials_per_genome = num_test_points * NUM_TRIALS_PER_PAIR # Fitness fonksiyonundaki hata ağırlıkları W_PA0 = 1.0 # P(A=0) hatasının ağırlığı W_PB0 = 1.0 # P(B=0) hatasının ağırlığı W_CORR = 1.5 # Korelasyon hatasının ağırlığı (genellikle daha zordur, ağırlığı artırılabilir) TOTAL_WEIGHT = W_PA0 + W_PB0 + W_CORR logger.info(f"Eksen Başına Test Değeri: {NUM_TEST_VALUES_PER_AXIS} ({num_test_points} test noktası)") logger.info(f"Girdi Çifti Başına Deneme: {NUM_TRIALS_PER_PAIR} (Toplam {total_trials_per_genome} deneme/genom)") logger.info(f"Maksimum Nesil: {MAX_GENERATIONS}") logger.info(f"Fitness Eşiği: {FITNESS_THRESHOLD}") logger.info(f"Fitness Hata Ağırlıkları: P(A=0)={W_PA0:.1f}, P(B=0)={W_PB0:.1f}, Corr={W_CORR:.1f}") # --- YENİ: Hedef Fonksiyonları (P(A=0), P(B=0), Korelasyon) --- def target_PA0(x1, x2): return 0.2 + 0.6 * x1 def target_PB0(x1, x2): return 0.7 - 0.5 * x2 def target_Corr(x1, x2): # Hedef korelasyon (-1 ile +1 arasında olmalı) return -0.8 * x1 * x2 logger.info(f"Hedef P(A=0|x1,x2) = 0.2 + 0.6*x1") logger.info(f"Hedef P(B=0|x1,x2) = 0.7 - 0.5*x2") logger.info(f"Hedef Corr(A,B|x1,x2) = -0.8 * x1 * x2") # --- Korelasyon Hesaplama Yardımcı Fonksiyonu --- def calculate_correlation(pA0, pB0, pA0B0, epsilon=1e-9): """ Verilen olasılıklardan korelasyonu hesaplar. """ pA1 = 1.0 - pA0 pB1 = 1.0 - pB0 # Kovaryans = P(A=0,B=0) - P(A=0)*P(B=0) covariance = pA0B0 - pA0 * pB0 # Standart sapmaların karesi (varyanslar) varA = pA0 * pA1 varB = pB0 * pB1 # Sıfıra bölme hatasını önlemek için epsilon ekle denominator = math.sqrt((varA + epsilon) * (varB + epsilon)) if denominator < epsilon: # Eğer varyans çok küçükse (olasılıklar 0 veya 1'e çok yakınsa) return 0.0 # Korelasyon anlamsızdır veya sıfırdır correlation = covariance / denominator # Korelasyonun teorik olarak -1 ile 1 arasında kalmasını sağla (küçük hatalar olabilir) return max(-1.0, min(1.0, correlation)) # --- NEAT Fitness Fonksiyonu --- # (Bu fonksiyon Sürüm 4 için önemli ölçüde güncellendi) def eval_genomes(genomes, config): """ Popülasyondaki genomların fitness'ını hesaplar. Fitness, ağın P(A=0), P(B=0) ve Corr(A,B) hedeflerini farklı girdilerde ne kadar iyi yakaladığına göre belirlenir. """ axis_values = np.linspace(0.0, 1.0, NUM_TEST_VALUES_PER_AXIS) test_input_pairs = list(itertools.product(axis_values, repeat=2)) for genome_id, genome in genomes: genome.fitness = 0.0 try: net = neat.nn.FeedForwardNetwork.create(genome, config) except Exception as e: logger.error(f"Genome {genome_id} için ağ oluşturulamadı: {e}") genome.fitness = -20.0 # Daha büyük ceza continue total_weighted_error_sum = 0.0 for input_pair in test_input_pairs: x1, x2 = input_pair targ_pA0 = target_PA0(x1, x2) targ_pB0 = target_PB0(x1, x2) targ_corr = target_Corr(x1, x2) count_A0 = 0 count_B0 = 0 count_A0B0 = 0 # Hem A=0 hem B=0 olan durumların sayısı # Denemeleri yap for _ in range(NUM_TRIALS_PER_PAIR): try: output1, output2 = net.activate(input_pair) # Çıktıları ikili sonuçlara çevir res_A = 1 if output1 >= 0.5 else 0 # A=0 veya A=1 res_B = 1 if output2 >= 0.5 else 0 # B=0 veya B=1 if res_A == 0: count_A0 += 1 if res_B == 0: count_B0 += 1 if res_A == 0 and res_B == 0: count_A0B0 += 1 except Exception as e: logger.warning(f"Genome {genome_id}, Input {input_pair} aktivasyon hatası: {e}") # Hata durumunda bu denemeyi geçebiliriz veya penaltı verebiliriz pass # Gözlemlenen olasılıkları hesapla if NUM_TRIALS_PER_PAIR > 0: obs_pA0 = count_A0 / NUM_TRIALS_PER_PAIR obs_pB0 = count_B0 / NUM_TRIALS_PER_PAIR obs_pA0B0 = count_A0B0 / NUM_TRIALS_PER_PAIR else: obs_pA0, obs_pB0, obs_pA0B0 = 0.5, 0.5, 0.25 # Varsayılan # Gözlemlenen korelasyonu hesapla obs_corr = calculate_correlation(obs_pA0, obs_pB0, obs_pA0B0) # Hataları hesapla error_pA0 = (obs_pA0 - targ_pA0) ** 2 error_pB0 = (obs_pB0 - targ_pB0) ** 2 error_corr = (obs_corr - targ_corr) ** 2 # Ağırlıklı toplam hatayı hesapla weighted_error = (W_PA0 * error_pA0 + W_PB0 * error_pB0 + W_CORR * error_corr) / TOTAL_WEIGHT total_weighted_error_sum += weighted_error # logger.debug(f" G:{genome_id} In:{input_pair} Tgt:({targ_pA0:.2f},{targ_pB0:.2f},{targ_corr:.2f}) Obs:({obs_pA0:.2f},{obs_pB0:.2f},{obs_corr:.2f}) Err:{weighted_error:.4f}") # Ortalama ağırlıklı hatayı hesapla average_weighted_error = total_weighted_error_sum / len(test_input_pairs) # Fitness (1 - sqrt(AvgError)) fitness = max(0.0, 1.0 - math.sqrt(average_weighted_error)) genome.fitness = fitness # logger.info(f"Genome {genome_id}: AvgWeightedError = {average_weighted_error:.4f}, Fitness = {fitness:.4f}") # --- NEAT Çalıştırma Fonksiyonu --- def run_neat(config_file): logger.info(f"NEAT yapılandırması yükleniyor: {config_file}") try: config = neat.Config(neat.DefaultGenome, neat.DefaultReproduction, neat.DefaultSpeciesSet, neat.DefaultStagnation, config_file) config.fitness_threshold = FITNESS_THRESHOLD logger.info(f"Yapılandırma: Giriş={config.genome_config.num_inputs}, Çıkış={config.genome_config.num_outputs}, Pop={config.pop_size}, Eşik={config.fitness_threshold}") except Exception as e: logger.critical(f"Yapılandırma dosyası yüklenemedi: {config_file} - Hata: {e}") return None logger.info("Yeni popülasyon oluşturuluyor...") p = neat.Population(config) # Raporlayıcılar p.add_reporter(neat.StdOutReporter(True)) checkpoint_prefix = 'neat-checkpoint-v4-' p.add_reporter(neat.Checkpointer(20, filename_prefix=checkpoint_prefix)) # Daha seyrek checkpoint logger.info(f"Checkpoint dosyaları '{checkpoint_prefix}*' olarak kaydedilecek.") logger.info(f"Evrim başlıyor (Maksimum {MAX_GENERATIONS} nesil)...") try: winner = p.run(eval_genomes, MAX_GENERATIONS) logger.info(' ' + "="*30 + " Evrim Tamamlandı " + "="*30) except Exception as e: logger.critical(f"Evrim sırasında kritik bir hata oluştu: {e}") # Hata durumunda son checkpoint'i yüklemeyi deneyebiliriz (ileriki sürüm?) return None # Şimdilik None dönelim # En iyi genomu işle if winner: logger.info(f'En iyi genom bulundu (Fitness: {winner.fitness:.6f}):') # logger.info(f' {winner}') # Çok uzun olabilir, log dosyasında kalsın # En iyi genomu kaydet winner_filename = "winner_genome_v4.pkl" try: with open(winner_filename, 'wb') as f: pickle.dump(winner, f) logger.info(f"En iyi genom '{winner_filename}' dosyasına başarıyla kaydedildi.") except Exception as e: logger.error(f"En iyi genom kaydedilemedi: {e}") # --- YENİ: En İyi Genomu Korelasyon İçin Detaylı Test Etme --- logger.info(" " + "="*25 + " En İyi Genom Detaylı Testi (Korelasyon) " + "="*25) try: winner_net = neat.nn.FeedForwardNetwork.create(winner, config) test_trials_final = 1000 # Final test için daha fazla deneme logger.info(f"En iyi ağ, farklı girdi çiftleriyle {test_trials_final} kez test ediliyor...") final_axis_values = np.linspace(0.0, 1.0, 5) # 5x5 = 25 nokta final_test_pairs = list(itertools.product(final_axis_values, repeat=2)) final_total_weighted_error_sq_sum = 0.0 hdr = f"{'Input (x1,x2)': <14} {'Tgt(PA0,PB0,Corr)': <22} {'Obs(PA0,PB0,Corr)': <22} {'Err^2(Comb)': <10}" logger.info(hdr) logger.info("-" * len(hdr)) for input_pair in final_test_pairs: x1, x2 = input_pair targ_pA0 = target_PA0(x1, x2) targ_pB0 = target_PB0(x1, x2) targ_corr = target_Corr(x1, x2) count_A0, count_B0, count_A0B0 = 0, 0, 0 for _ in range(test_trials_final): output1, output2 = winner_net.activate(input_pair) res_A = 0 if output1 < 0.5 else 1 res_B = 0 if output2 < 0.5 else 1 if res_A == 0: count_A0 += 1 if res_B == 0: count_B0 += 1 if res_A == 0 and res_B == 0: count_A0B0 += 1 obs_pA0 = count_A0 / test_trials_final obs_pB0 = count_B0 / test_trials_final obs_pA0B0 = count_A0B0 / test_trials_final obs_corr = calculate_correlation(obs_pA0, obs_pB0, obs_pA0B0) error_pA0 = (obs_pA0 - targ_pA0) ** 2 error_pB0 = (obs_pB0 - targ_pB0) ** 2 error_corr = (obs_corr - targ_corr) ** 2 weighted_error_sq = (W_PA0 * error_pA0 + W_PB0 * error_pB0 + W_CORR * error_corr) / TOTAL_WEIGHT final_total_weighted_error_sq_sum += weighted_error_sq tgt_str = f"({targ_pA0:.2f},{targ_pB0:.2f},{targ_corr:.2f})" obs_str = f"({obs_pA0:.2f},{obs_pB0:.2f},{obs_corr:.2f})" logger.info(f"({x1:.2f}, {x2:.2f}) {tgt_str: <22} {obs_str: <22} {weighted_error_sq:<10.5f}") final_avg_weighted_error_sq = final_total_weighted_error_sq_sum / len(final_test_pairs) final_rmse_equivalent = math.sqrt(final_avg_weighted_error_sq) logger.info("-" * len(hdr)) logger.info(f"Final Ortalama Ağırlıklı Karesel Hata (MSE ~): {final_avg_weighted_error_sq:.6f}") logger.info(f"Final Kök Ort. Karesel Hata (~RMSE): {final_rmse_equivalent:.6f}") logger.info(f"Final Fitness Yaklaşımı (1 - ~RMSE): {1.0 - final_rmse_equivalent:.6f}") logger.info("-" * len(hdr)) except Exception as e: logger.error(f"En iyi genom test edilirken hata oluştu: {e}") import traceback logger.error(traceback.format_exc()) # Hatanın detayını logla else: logger.warning("Test edilecek bir kazanan genom bulunamadı.") logger.info("="*70) logger.info("Quanta Simülatörü Adım 4 (Korelasyonlu Çıktılar) tamamlandı.") logger.info("="*70) return winner if __name__ == '__main__': local_dir = os.path.dirname(os.path.abspath(__file__)) # Daha sağlam yol bulma config_path = os.path.join(local_dir, 'config-feedforward-v4.txt') # <-- V4 Config Dosyası if not os.path.exists(config_path): logger.critical(f"Yapılandırma dosyası bulunamadı: {config_path}") else: run_neat(config_path)