PyLegend Stable V1.0.0

import os
import time
import random
import string
import gc
import pickle
import numpy as np
import cirq
import tensorflow as tf
from subprocess import Popen
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.linear_model import LinearRegression
from sklearn.metrics import make_scorer, mean_squared_error, r2_score
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense
from tensorflow.keras.callbacks import EarlyStopping
from tensorflow.keras.backend import clear_session
import keras_tuner as kt
from matplotlib import pyplot as plt
from matplotlib.animation import FuncAnimation
def configure_gpu_memory(log_file_path):
"""
Configure la mémoire GPU pour utiliser la croissance dynamique de la mémoire.

Args:
    log_file_path (str): Chemin du fichier journal pour les logs.
"""
try:
    physical_devices = tf.config.list_physical_devices('GPU')
    if physical_devices:
        tf.config.experimental.set_memory_growth(physical_devices[0], True)
        log_data("GPU configuré pour utiliser la croissance de mémoire dynamique.", log_file_path)
    else:
        log_data("Aucun GPU trouvé.", log_file_path)
except Exception as e:
    log_data(f"Erreur lors de la configuration du GPU: {e}", log_file_path)

import numpy as np

Définir la matrice de poids

def get_weight_matrix():
return np.array([
[1. , 0.9 , 0.81 , 0.729 , 0.6561 , 0.59049 , 0.531441 , 0.4782969 , 0.43046721, 0.38742049],
[0. , 1. , 0.9 , 0.81 , 0.729 , 0.6561 , 0.59049 , 0.531441 , 0.4782969 , 0.43046721],
[0. , 0. , 1. , 0.9 , 0.81 , 0.729 , 0.6561 , 0.59049 , 0.531441 , 0.4782969 ],
[0. , 0. , 0. , 1. , 0.9 , 0.81 , 0.729 , 0.6561 , 0.59049 , 0.531441 ],
[0. , 0. , 0. , 0. , 1. , 0.9 , 0.81 , 0.729 , 0.6561 , 0.59049 ],
[0. , 0. , 0. , 0. , 0. , 1. , 0.9 , 0.81 , 0.729 , 0.6561 ],
[0. , 0. , 0. , 0. , 0. , 0. , 1. , 0.9 , 0.81 , 0.729 ],
[0. , 0. , 0. , 0. , 0. , 0. , 0. , 1. , 0.9 , 0.81 ],
[0. , 0. , 0. , 0. , 0. , 0. , 0. , 0. , 1. , 0.9 ],
[0. , 0. , 0. , 0. , 0. , 0. , 0. , 0. , 0. , 1. ]
])

Définir la matrice de poids pour les parts en fonction du temps

def create_weight_matrix(num_shares, decay_rate=0.1):

weights = np.zeros((num_shares, num_shares))
for i in range(num_shares):
    weights[i, i:] = decay_rate ** np.arange(num_shares - i)
return weights

Calculer les récompenses basées sur la matrice des poids

def calculate_rewards(shares, weight_matrix):
"""
Calcule les récompenses pour les mineurs basées sur leurs parts et la matrice de poids.

Args:
    shares (numpy.ndarray): Matrice des parts où chaque ligne est un mineur et chaque colonne est une part spécifique.
    weight_matrix (numpy.ndarray): Matrice des poids.
    
Returns:
    numpy.ndarray: Récompenses pour chaque mineur.
"""
weighted_shares = np.dot(shares, weight_matrix)
total_rewards = weighted_shares.sum(axis=1)
return total_rewards

def simulate_quantum_circuit_optimized(num_qubits, depth, repetitions=1000):
"""
Simule un circuit quantique optimisé avec le nombre de qubits spécifié et la profondeur du circuit.

Args:
    num_qubits (int): Nombre de qubits dans le circuit.
    depth (int): Profondeur du circuit.
    repetitions (int): Nombre de répétitions pour l'expérience de simulation (défaut: 1000).

Returns:
    dict: Résultats de la simulation sous forme de dictionnaire.
"""
# Création du circuit quantique
qubits = [cirq.GridQubit(0, i) for i in range(num_qubits)]
circuit = cirq.Circuit()

# Ajout de portes au circuit
for _ in range(depth):
    circuit.append(cirq.H(q) for q in qubits)  # Ajout d'une porte Hadamard
    circuit.append(cirq.CNOT(q1, q2) for q1 in qubits for q2 in qubits if q1 != q2)  # Ajout de portes CNOT

# Optimisation du circuit
optimized_circuit = cirq.optimize_for_target_gateset(circuit, target_gateset={cirq.CNOT, cirq.X})

# Simulation
simulator = cirq.Simulator()
result = simulator.run(optimized_circuit, repetitions=repetitions)

return result.histogram(key='result')

Configuration des chemins et paramètres

MINER_PATH = "C:/Chemin/vers/le/dossier de l'exécutable du mineur/NBminer_Win"
MINER_EXECUTABLE = "nbminer.exe"
POOL_URL = "POOL"
USER = "USER"
PASSWORD = "x"

INTERCEPT_CONSTANT = 0.60
BATCH_SIZE = 32 # Taille des lots pour écrire sur disque (ajustée pour des performances optimales)

MODEL_FILE_PATH = "model.keras"
CIRCUIT_FILE_PATH = "quantum_circuit.pkl"

Fréquence initiale pour la simulation quantique

quantum_circuit_repetitions = 10
initial_qubits = 2

Initialiser les variables globales

process = None
X_data = None
y_data = None
batch_X = None
batch_y = None
rmse_data = []
intercept_data = []
r2_data = []
current_model = None

def generate_unique_filename(prefix="file", extension=".log"):
timestamp = time.strftime("%Y%m%d_%H%M%S")
random_id = ''.join(random.choices(string.ascii_letters + string.digits, k=6))
filename = f"{prefix}{timestamp}{random_id}{extension}"
return os.path.join(os.getcwd(), filename)

def log_data(data, log_file_path):
try:
with open(log_file_path, "a") as log_file:
log_file.write(data + "\n")
except Exception as e:
print(f"Erreur lors de l'enregistrement des données: {e}")

def save_file(obj, file_path, log_file_path, mode='wb'):
try:
with open(file_path, mode) as file:
pickle.dump(obj, file)
log_data(f"Fichier sauvegardé à {file_path}", log_file_path)
except Exception as e:
log_data(f"Erreur lors de la sauvegarde du fichier: {e}", log_file_path)

def load_file(file_path, log_file_path, mode='rb'):
try:
if os.path.isfile(file_path):
with open(file_path, mode) as file:
obj = pickle.load(file)
log_data(f"Fichier chargé depuis {file_path}", log_file_path)
return obj
else:
log_data(f"Le fichier n'existe pas à {file_path}", log_file_path)
return None
except Exception as e:
log_data(f"Erreur lors du chargement du fichier: {e}", log_file_path)
return None

def build_model(hp, input_shape):
model = Sequential([
Dense(hp.Int('units', min_value=64, max_value=128, step=32), activation='relu', input_shape=(input_shape,)),
Dense(32, activation='relu'),
Dense(1)
])
model.compile(optimizer='adam', loss='mean_squared_error', metrics=['mae'])
return model

def hyperparameter_tuning(X, y):
tuner = kt.Hyperband(
lambda hp: build_model(hp, X.shape[1]),
objective='val_loss',
max_epochs=3, # Réduit le nombre d'époques pour le tuning rapide
directory='tuner',
project_name='hyperparameter_tuning'
)

X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42)
tuner.search(X_train, y_train, epochs=3, validation_data=(X_val, y_val))  # Réduit les époques pour le tuning rapide

best_model = tuner.get_best_models(num_models=1)[0]
best_params = tuner.get_best_hyperparameters(num_trials=1)[0].values

return best_model, best_params

def retry_on_failure(func, max_attempts=3, *args, **kwargs):
attempts = 0
while attempts < max_attempts:
try:
return func(*args, **kwargs)
except Exception as e:
attempts += 1
log_data(f"Tentative {attempts}/{max_attempts} échouée pour {func.name}: {e}", kwargs.get('log_file_path', ''))
if attempts == max_attempts:
log_data(f"Échec permanent de {func.name} après {max_attempts} tentatives.", kwargs.get('log_file_path', ''))
raise e
time.sleep(1) # Réduire l'attente avant de réessayer

def create_dataset(X, y, batch_size=64):
dataset = tf.data.Dataset.from_tensor_slices((X, y))
dataset = dataset.shuffle(buffer_size=len(X))
dataset = dataset.batch(batch_size)
dataset = dataset.prefetch(buffer_size=tf.data.experimental.AUTOTUNE)
return dataset

def train_tf_model(X, y, log_file_path, existing_model=None, intercept_adjustment=7, params=None):
if params is None:
params = {}

# Utilisez des valeurs par défaut si les clés sont absentes
epochs = params.get('epochs', 7)  # Valeur par défaut : 10
batch_size = params.get('batch_size', 100000)  # Valeur par défaut : 32

# Affichez les paramètres pour le débogage
print(f"epochs: {epochs}, batch_size: {batch_size}")

# Assurez-vous que le modèle est défini
input_shape = X.shape[1]  # Déterminez la forme d'entrée basée sur X
model = existing_model if existing_model else create_model(input_shape)

# Ajustez le modèle si nécessaire avec l'intercept_adjustment
if intercept_adjustment is not None:
    # Vérifiez si une couche Lambda avec un nom similaire existe déjà
    if 'intercept_adjustment_layer' not in [layer.name for layer in model.layers]:
        # Ajoutez une couche Lambda avec un nom unique

...