Bittensor:
L'Internet Neuronal

Formule d'Émission

Block Reward = Initial_Reward × (0.5)^(blocks_passed / halving_interval)
Initial Reward: 1 TAO/block
Block Time: 12 secondes
Halving Interval: 10,500,000 blocks (~4 ans)

Principe d'Émission

"Chaque bloc produit sur la blockchain Bittensor génère exactement 1 TAO qui est distribué aux participants du réseau selon des règles prédéfinies. Cette émission est divisée par 2 basée sur l'offre totale émise plutôt que sur le nombre de blocs, créant un modèle déflationnaire unique."

Recyclage et Impact sur le Halving

"Les TAO dépensés pour l'enregistrement de mineurs dans les subnets ne sont pas détruits mais retournent au pool d'émission non distribué. Ce mécanisme unique peut retarder la date exacte du halving car il est basé sur l'offre totale émise plutôt que sur un nombre de blocs fixe."

Principe du Staking

"Le staking TAO est un processus de délégation où les détenteurs de tokens prêtent leur pouvoir de stake à des validateurs sans transférer la propriété de leurs tokens. Les validateurs utilisent ce stake délégué pour augmenter leur poids de consensus et gagner plus de récompenses, qu'ils partagent ensuite avec leurs délégateurs."

Exemple de Calcul

Un validateur avec 900 TAO de self-stake et 200 TAO délégués (total: 1,100 TAO) gagne 100 TAO de récompenses.
• Le validateur garde: 81.82 TAO (proportionnel à son stake)
• Plus sa commission: 1.64 TAO (9% des récompenses déléguées)
• Les délégateurs reçoivent: 16.54 TAO (distribués proportionnellement)

Nouveauté: Alpha Tokens

"Avec Dynamic TAO, le staking dans des subnets spécifiques implique l'achat de tokens Alpha propres à chaque subnet. Votre TAO est converti en Alpha via l'AMM du subnet, et vos récompenses dépendent de la performance relative du subnet dans l'écosystème."

Coldkey vs Hotkey

"Dans Bittensor, le coldkey est utilisé pour toutes les opérations sensibles affectant les balances (transferts, staking) tandis que le hotkey est utilisé pour les opérations de minage et validation. Un coldkey peut posséder plusieurs hotkeys, mais chaque hotkey appartient à un seul coldkey."

Formule de Distribution

Exemple : Un validateur avec 800 TAO personnels et 200 TAO délégués (total: 1000 TAO)

• Émissions totales : 100 TAO
• Part du validateur : (800/1000) × 100 = 80 TAO
• Part déléguée : (200/1000) × 100 = 20 TAO
• Commission (18%) : 20 × 0.18 = 3.6 TAO
• Distribution aux nominateurs : 20 - 3.6 = 16.4 TAO

Définition du Recyclage

"Le recyclage TAO est le processus par lequel les tokens dépensés pour l'enregistrement de mineurs dans les subnets sont retournés au pool d'émission non distribué plutôt que d'être détruits. Ce mécanisme retarde effectivement la date du halving en réduisant la quantité de TAO considérée comme émise."

Calcul Dynamique des Frais

Les frais d'enregistrement dans un subnet sont variables et basés sur la demande :

• Faible demande : ~100 TAO minimum
• Demande moyenne : 500-2,000 TAO
• Forte demande : Jusqu'à 10,000 TAO

Ces frais constituent un coût opérationnel pour les mineurs et ne sont PAS remboursés lors de la dé-registration.

Principe du Minage par Intelligence

"Dans Bittensor, les mineurs sont des modèles d'IA qui compétitionnent pour fournir les meilleures réponses aux requêtes des validateurs. La qualité de leurs outputs, évaluée par le mécanisme d'incitation du subnet, détermine leur part des émissions TAO."

Enregistrement et Coûts

L'enregistrement d'un mineur nécessite de recycler des TAO. Les coûts varient :

• Minimum: ~100 TAO (faible demande)
• Moyen: 500-2,000 TAO
• Maximum: Jusqu'à 10,000 TAO (forte demande)

Ces TAO sont recyclés et ne sont PAS récupérables si vous êtes désenregistré.

Commandes de Base

# Installation Bittensor
/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/opentensor/bittensor/master/scripts/install.sh)"

# Créer un wallet
btcli wallet new_coldkey --wallet.name miner
btcli wallet new_hotkey --wallet.name miner --wallet.hotkey default

# Enregistrer sur un subnet
btcli subnet recycle_register --netuid [SUBNET_ID] --wallet.name miner --wallet.hotkey default

# Vérifier l'enregistrement
btcli wallet overview --wallet.name miner --subtensor.network finney

Exigences Réseau Critiques

• Bande passante: Minimum 100 Mbps symétrique, 1 Gbps recommandé
• Latence: < 50ms vers les principaux nœuds (critique pour trading)
• IP statique: Une IP externe par UID obligatoire
• Ports ouverts: Varient selon subnet (ex: 4444 pour SN27)
• Firewall: Configuration précise requise, blocage des ports sensibles

Définition : Validation dans Bittensor

"La validation est le processus par lequel les nœuds validateurs évaluent la qualité et la pertinence des réponses fournies par les mineurs, en utilisant des métriques objectives et des algorithmes de scoring pour déterminer la distribution des récompenses TAO."

Formule de Consensus

Score(mineur_i) = Σ(similarité(réponse_i, réponse_j) × poids_j) / Σ(poids_j)

Pseudocode de Validation

class Validator:
    def validation_loop(self):
        while True:
            # 1. Sélectionner mineurs aléatoires
            miners = self.select_miners(batch_size=64)
            
            # 2. Générer requête unique
            query = self.generate_query()
            
            # 3. Envoyer requête et collecter réponses
            responses = self.query_miners(miners, query, timeout=12)
            
            # 4. Évaluer chaque réponse
            scores = []
            for response in responses:
                score = self.evaluate_response(query, response)
                scores.append(score)
            
            # 5. Mettre à jour les poids
            self.update_weights(miners, scores)
            
            # 6. Soumettre on-chain si nécessaire
            if self.should_submit():
                self.submit_weights_to_chain()
            
            # Attendre avant prochain cycle
            time.sleep(self.validation_interval)

Équation Fondamentale du Score

Score(i) = Σ(w_j × q_ij × r_j) / Σ(w_j)
Où : i = mineur, j = validateur, w = poids, q = qualité, r = fiabilité

Implémentation Softmax

def softmax_score(scores, temperature=1.0):
    # Éviter overflow numérique
    scores = np.array(scores)
    scores = scores - np.max(scores)
    
    # Appliquer softmax avec température
    exp_scores = np.exp(scores / temperature)
    return exp_scores / np.sum(exp_scores)

Lissage Temporel des Scores

class EMAScoring:
    def __init__(self, alpha=0.1):
        self.alpha = alpha
        self.scores = {}
    
    def update(self, miner_uid, new_score):
        if miner_uid not in self.scores:
            self.scores[miner_uid] = new_score
        else:
            # EMA: S_t = α × X_t + (1-α) × S_(t-1)
            self.scores[miner_uid] = (
                self.alpha * new_score + 
                (1 - self.alpha) * self.scores[miner_uid]
            )
        return self.scores[miner_uid]

Exemple : Algorithme de Scoring Multi-Critères

class CompositeScoring:
    def __init__(self, weights=None):
        self.weights = weights or {
            'quality': 0.4,
            'speed': 0.2,
            'diversity': 0.2,
            'consistency': 0.2
        }
        self.history = defaultdict(list)
    
    def calculate_score(self, miner_uid, response):
        # 1. Qualité de la réponse
        quality = self.evaluate_quality(response)
        
        # 2. Vitesse (latence inverse normalisée)
        speed = 1 / (1 + response.latency / 1000)
        
        # 3. Diversité par rapport aux réponses précédentes
        diversity = self.calculate_diversity(miner_uid, response)
        
        # 4. Cohérence historique
        consistency = self.calculate_consistency(miner_uid, quality)
        
        # Score composite pondéré
        components = {
            'quality': quality,
            'speed': speed,
            'diversity': diversity,
            'consistency': consistency
        }
        
        final_score = sum(
            self.weights[k] * v 
            for k, v in components.items()
        )
        
        # Mise à jour historique
        self.history[miner_uid].append({
            'timestamp': time.time(),
            'score': final_score,
            'components': components
        })
        
        return final_score, components
    
    def calculate_diversity(self, miner_uid, response):
        if len(self.history[miner_uid]) < 2:
            return 1.0
        
        # Calcul de distance avec réponses précédentes
        prev_embeddings = [h['embedding'] for h in self.history[miner_uid][-10:]]
        curr_embedding = self.get_embedding(response)
        
        distances = [
            cosine_distance(curr_embedding, prev) 
            for prev in prev_embeddings
        ]
        
        return np.mean(distances)  # Plus distant = plus divers
    
    def calculate_consistency(self, miner_uid, current_quality):
        history = self.history[miner_uid]
        if len(history) < 5:
            return 1.0
        
        # Variance inverse des scores récents
        recent_scores = [h['components']['quality'] for h in history[-20:]]
        variance = np.var(recent_scores)
        
        return 1 / (1 + variance)  # Moins de variance = plus consistent

Principe Fondamental

"Dans Bittensor, la valeur économique suit directement la valeur informationnelle. Plus un neurone contribue à l'intelligence collective du réseau, plus il est récompensé en tokens TAO, créant ainsi une boucle de rétroaction positive vers l'excellence."

Flux d'Émission TAO

┌─────────────────┐
│  Émission: 1 TAO │
│   par 12 sec    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Root Network    │ ──── 41% (Coordination)
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ 118+ Subnets    │ ──── 59% (Distribution)
└────────┬────────┘
         │
         ▼
┌─────────────────────────────┐
│  64 Validateurs par Subnet  │
└────────┬───────────────────┘
         │
         ▼
┌─────────────────────────────┐
│  Mineurs (selon performance) │
└─────────────────────────────┘

Simulation de Distribution de Récompenses

class RewardCalculator:
    def __init__(self):
        self.emission_per_block = 1.0  # TAO
        self.root_allocation = 0.41
        self.subnet_allocation = 0.59
        self.blocks_per_day = 7200
        
    def calculate_subnet_rewards(self, subnet_weights):
        """Calcule les récompenses pour chaque subnet"""
        total_emission = self.emission_per_block * self.subnet_allocation
        
        rewards = {}
        for subnet_id, weight in subnet_weights.items():
            rewards[subnet_id] = total_emission * weight
            
        return rewards
    
    def calculate_miner_rewards(self, subnet_reward, miner_scores, validator_commission=0.18):
        """Distribue les récompenses aux mineurs d'un subnet"""
        # Validateurs prennent leur commission
        validator_share = subnet_reward * validator_commission
        miner_pool = subnet_reward * (1 - validator_commission)
        
        # Distribution selon les scores normalisés
        total_score = sum(miner_scores.values())
        miner_rewards = {}
        
        for miner_uid, score in miner_scores.items():
            weight = score / total_score if total_score > 0 else 0
            miner_rewards[miner_uid] = miner_pool * weight
            
        return {
            'validators': validator_share,
            'miners': miner_rewards,
            'total': subnet_reward
        }
    
    def estimate_apy(self, stake, daily_rewards):
        """Estime l'APY basé sur les récompenses quotidiennes"""
        daily_return = daily_rewards / stake
        apy = ((1 + daily_return) ** 365 - 1) * 100
        return apy

# Exemple d'utilisation
calculator = RewardCalculator()

# Poids des subnets (exemple)
subnet_weights = {
    1: 0.15,   # Text Generation
    8: 0.12,   # Trading
    19: 0.10,  # Image Generation
    # ... autres subnets
}

# Calcul des récompenses
subnet_rewards = calculator.calculate_subnet_rewards(subnet_weights)

# Pour un subnet spécifique
miner_scores = {
    'miner_1': 0.95,
    'miner_2': 0.87,
    'miner_3': 0.92,
    # ... autres mineurs
}

distribution = calculator.calculate_miner_rewards(
    subnet_rewards[1], 
    miner_scores
)

print(f"Récompenses Subnet 1: {subnet_rewards[1]:.4f} TAO/bloc")
print(f"Part validateurs: {distribution['validators']:.4f} TAO")
print(f"Top mineur: {max(distribution['miners'].values()):.4f} TAO")

Installation et Import

# Installation
pip install bittensor

# Import de base
import bittensor as bt

# Configuration du logging
bt.logging.set_trace(True)  # Mode debug
bt.logging.set_debug(False) # Mode production

Opérations Wallet

# Créer/Charger un wallet
wallet = bt.wallet(
    name='mon_wallet',
    hotkey='default',
    path='~/.bittensor/wallets/'
)

# Créer nouveau wallet
wallet.create_if_non_existent(
    coldkey_use_password=True,
    hotkey_use_password=False
)

# Obtenir les adresses
coldkey_address = wallet.coldkey.ss58_address
hotkey_address = wallet.hotkey.ss58_address

Exemples de Requêtes

# Balance d'un wallet
balance = subtensor.get_balance(wallet.coldkey.ss58_address)
print(f"Balance: {balance} TAO")

# Informations d'un subnet
subnet_info = subtensor.get_subnet_info(netuid=1)
print(f"Subnet 1: {subnet_info}")

# Liste des validateurs
delegates = subtensor.get_delegates()
for delegate in delegates[:5]:
    print(f"Validateur: {delegate.hotkey}")
    print(f"Stake total: {delegate.total_stake}")

# Émissions actuelles
emissions = subtensor.get_emission_value_by_subnet(
    netuid=1, 
    block=subtensor.block
)

Requêtes vers les Mineurs

# Créer le dendrite
dendrite = bt.dendrite(wallet=wallet)

# Préparer la requête
synapse = MyMiner(query="Test query")

# Query unique
response = dendrite.query(
    axons=[axon_info],
    synapse=synapse,
    timeout=12.0
)

# Query multiple (batch)
responses = dendrite.query(
    axons=metagraph.axons[:10],
    synapse=synapse,
    timeout=12.0
)

# Traiter les réponses
for resp in responses:
    if resp.response:
        print(f"Réponse: {resp.response}")

Installation via pip

# Installation stable
pip install bittensor

# Installation avec dépendances GPU (pour mining)
pip install bittensor[torch]

# Installation depuis source (dernières features)
git clone https://github.com/opentensor/bittensor.git
cd bittensor
pip install -e .

# Vérifier l'installation
python -c "import bittensor as bt; print(bt.__version__)"

Utilisation de Subtensor

import bittensor as bt

# Initialisation
subtensor = bt.subtensor(
    network='finney',  # ou 'test', 'local'
    config=bt.config(
        subtensor=dict(
            chain_endpoint='wss://entrypoint-finney.opentensor.ai:443',
            _mock=False  # True pour tests unitaires
        )
    )
)

# Méthodes essentielles
class SubtensorMethods:
    # Informations réseau
    block = subtensor.block  # Bloc actuel
    difficulty = subtensor.difficulty(netuid=1)
    burn = subtensor.burn(netuid=1)
    
    # Gestion des neurones
    is_registered = subtensor.is_hotkey_registered(
        hotkey_ss58='5F3sa2...',
        netuid=1
    )
    
    # Requêtes de données
    neurons = subtensor.neurons(netuid=1)
    weights = subtensor.weights(netuid=1)
    bonds = subtensor.bonds(netuid=1)

Template de Mineur Complet

import bittensor as bt
import torch
from typing import Optional

class TextMiner:
    def __init__(self, config=None):
        self.config = config or bt.config()
        self.wallet = bt.wallet(config=self.config)
        self.subtensor = bt.subtensor(config=self.config)
        self.metagraph = bt.metagraph(
            netuid=self.config.netuid,
            network=self.config.subtensor.network
        )
        
        # Initialiser le modèle
        self.model = self.load_model()
        
        # Créer l'axon
        self.axon = bt.axon(
            wallet=self.wallet,
            config=self.config
        )
        
    def load_model(self):
        """Charger votre modèle IA"""
        # Exemple avec transformers
        from transformers import AutoModelForCausalLM, AutoTokenizer
        
        model_name = "gpt2"  # Remplacer par votre modèle
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        model = AutoModelForCausalLM.from_pretrained(model_name)
        
        if torch.cuda.is_available():
            model = model.cuda()
            
        return model
    
    async def forward(self, synapse: bt.Synapse) -> bt.Synapse:
        """Traiter les requêtes des validateurs"""
        try:
            # Extraire la requête
            prompt = synapse.prompt
            
            # Générer la réponse
            inputs = self.tokenizer(prompt, return_tensors="pt")
            if torch.cuda.is_available():
                inputs = inputs.to('cuda')
                
            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_length=256,
                    temperature=0.7,
                    do_sample=True
                )
            
            response = self.tokenizer.decode(
                outputs[0], 
                skip_special_tokens=True
            )
            
            # Mettre à jour la synapse
            synapse.completion = response
            
        except Exception as e:
            bt.logging.error(f"Erreur forward: {e}")
            synapse.completion = ""
            
        return synapse
    
    def blacklist(self, synapse: bt.Synapse) -> tuple[bool, str]:
        """Filtrer les requêtes malveillantes"""
        # Vérifier si le validateur est vérifié
        hotkey = synapse.dendrite.hotkey
        
        if hotkey not in self.metagraph.hotkeys:
            return True, "Hotkey non enregistré"
            
        stake = self.metagraph.S[self.metagraph.hotkeys.index(hotkey)]
        if stake < 1000:  # Minimum stake requis
            return True, f"Stake insuffisant: {stake}"
            
        return False, "OK"
    
    def priority(self, synapse: bt.Synapse) -> float:
        """Prioriser les requêtes par stake"""
        hotkey = synapse.dendrite.hotkey
        uid = self.metagraph.hotkeys.index(hotkey)
        return float(self.metagraph.S[uid])
    
    def run(self):
        """Boucle principale du mineur"""
        # Enregistrer sur le subnet si nécessaire
        if not self.subtensor.is_hotkey_registered(
            hotkey_ss58=self.wallet.hotkey.ss58_address,
            netuid=self.config.netuid
        ):
            bt.logging.info("Enregistrement sur le subnet...")
            self.subtensor.burned_register(
                wallet=self.wallet,
                netuid=self.config.netuid
            )
        
        # Attacher les handlers
        self.axon.attach(
            forward_fn=self.forward,
            blacklist_fn=self.blacklist,
            priority_fn=self.priority
        )
        
        # Servir l'axon
        self.axon.serve(
            netuid=self.config.netuid,
            subtensor=self.subtensor
        )
        
        # Démarrer
        self.axon.start()
        bt.logging.info(f"Mineur démarré sur {self.axon.ip}:{self.axon.port}")
        
        # Boucle principale
        try:
            while True:
                # Sync périodique du métagraphe
                self.metagraph.sync(subtensor=self.subtensor)
                
                # Logging des métriques
                my_uid = self.metagraph.hotkeys.index(
                    self.wallet.hotkey.ss58_address
                )
                bt.logging.info(
                    f"UID: {my_uid} | "
                    f"Rank: {self.metagraph.R[my_uid]:.4f} | "
                    f"Incentive: {self.metagraph.I[my_uid]:.4f}"
                )
                
                # Attendre avant resync
                bt.sleep(30)
                
        except KeyboardInterrupt:
            bt.logging.info("Arrêt du mineur...")
        finally:
            self.axon.stop()

# Lancer le mineur
if __name__ == "__main__":
    config = bt.config()
    config.netuid = 1  # Subnet de text generation
    
    miner = TextMiner(config=config)
    miner.run()

Configuration Complète

# Configuration via CLI
python miner.py --netuid 1 --subtensor.network finney \
    --wallet.name my_wallet --wallet.hotkey default \
    --axon.port 8091 --logging.debug

# Configuration via code
config = bt.config(
    parser=argparse.ArgumentParser(),
    args=[
        '--netuid', '1',
        '--subtensor.network', 'finney',
        '--wallet.name', 'my_wallet',
        '--axon.port', '8091',
        '--logging.debug'
    ]
)

# Configuration via fichier
config = bt.config()
config.merge(bt.config().load_from_relative_path('config.yml'))

Points Clés de l'EVM sur Bittensor

• Les smart contracts s'exécutent sur subtensor, pas sur Ethereum
• 1 TAO = 1e18 dans l'EVM (18 décimales)
• Compatibilité JSON-RPC standard d'Ethereum
• Déploiement actuellement soumis à permission

Connexion au Réseau EVM

// Configuration MetaMask/Web3
const config = {
    chainId: '0x7E4', // 2020 en décimal
    chainName: 'Bittensor EVM',
    nativeCurrency: {
        name: 'TAO',
        symbol: 'TAO',
        decimals: 18
    },
    rpcUrls: ['wss://evm.chain.opentensor.ai'],
    blockExplorerUrls: []
}

// Web3.js
const Web3 = require('web3');
const web3 = new Web3('wss://evm.chain.opentensor.ai');

Processus de Déploiement

// 1. Compiler avec Hardhat/Truffle
npx hardhat compile

// 2. Déployer sur Bittensor EVM
const contract = await ethers.deployContract("BittensorAI");
await contract.waitForDeployment();

console.log("Contract deployed at:", await contract.getAddress());

// 3. Vérifier le déploiement
const modelHash = await contract.getModel(1);
console.log("Model hash:", modelHash);

Implémentation Gateway

import bittensor as bt
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio
from typing import List, Dict

app = FastAPI()

class QueryRequest(BaseModel):
    prompt: str
    subnet_id: int = 1
    timeout: float = 12.0
    top_k: int = 5

class BittensorGateway:
    def __init__(self):
        self.wallet = bt.wallet(name="gateway")
        self.dendrite = bt.dendrite(wallet=self.wallet)
        self.metagraphs = {}
        
    async def get_metagraph(self, netuid: int):
        if netuid not in self.metagraphs:
            self.metagraphs[netuid] = bt.metagraph(netuid=netuid)
            self.metagraphs[netuid].sync()
        return self.metagraphs[netuid]
    
    async def query_subnet(self, request: QueryRequest) -> List[Dict]:
        metagraph = await self.get_metagraph(request.subnet_id)
        
        # Sélectionner les top mineurs
        top_miners = self.select_top_miners(
            metagraph, 
            k=request.top_k
        )
        
        # Créer la synapse
        synapse = bt.Synapse(
            prompt=request.prompt,
            timeout=request.timeout
        )
        
        # Query en parallèle
        responses = await self.dendrite.query(
            axons=top_miners,
            synapse=synapse,
            timeout=request.timeout
        )
        
        return self.format_responses(responses)
    
    def select_top_miners(self, metagraph, k=5):
        # Sélectionner par incentive score
        incentives = metagraph.I.numpy()
        top_indices = incentives.argsort()[-k:][::-1]
        
        return [metagraph.axons[i] for i in top_indices]
    
    def format_responses(self, responses):
        return [
            {
                "text": r.completion,
                "confidence": r.confidence if hasattr(r, 'confidence') else 1.0,
                "latency": r.process_time
            }
            for r in responses if r.completion
        ]

gateway = BittensorGateway()

@app.post("/query")
async def query_bittensor(request: QueryRequest):
    try:
        responses = await gateway.query_subnet(request)
        return {
            "success": True,
            "responses": responses,
            "subnet": request.subnet_id
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Architecture Événementielle avec Bittensor

import asyncio
from dataclasses import dataclass
from typing import Callable, Dict, List
from enum import Enum

class EventType(Enum):
    QUERY_RECEIVED = "query_received"
    RESPONSE_READY = "response_ready"
    MINER_FAILED = "miner_failed"
    SCORE_UPDATED = "score_updated"

@dataclass
class Event:
    type: EventType
    data: Dict
    timestamp: float

class EventBus:
    def __init__(self):
        self.handlers: Dict[EventType, List[Callable]] = {}
        self.queue = asyncio.Queue()
        
    def subscribe(self, event_type: EventType, handler: Callable):
        if event_type not in self.handlers:
            self.handlers[event_type] = []
        self.handlers[event_type].append(handler)
    
    async def publish(self, event: Event):
        await self.queue.put(event)
    
    async def start(self):
        while True:
            event = await self.queue.get()
            if event.type in self.handlers:
                for handler in self.handlers[event.type]:
                    asyncio.create_task(handler(event))

# Exemple d'utilisation
class BittensorEventDriven:
    def __init__(self):
        self.bus = EventBus()
        self.setup_handlers()
        
    def setup_handlers(self):
        self.bus.subscribe(
            EventType.QUERY_RECEIVED, 
            self.handle_query
        )
        self.bus.subscribe(
            EventType.MINER_FAILED,
            self.handle_failure
        )
        
    async def handle_query(self, event: Event):
        query = event.data['query']
        # Traiter la requête de manière asynchrone
        result = await self.process_query(query)
        
        await self.bus.publish(Event(
            type=EventType.RESPONSE_READY,
            data={'result': result},
            timestamp=time.time()
        ))

Système de Notifications Asynchrone

class WebhookManager:
    def __init__(self):
        self.subscribers = []
        self.retry_policy = RetryPolicy(max_attempts=3)
        
    async def register_webhook(self, url: str, events: List[str]):
        subscriber = {
            'url': url,
            'events': events,
            'active': True
        }
        self.subscribers.append(subscriber)
        
    async def notify(self, event_type: str, data: dict):
        tasks = []
        for sub in self.subscribers:
            if event_type in sub['events'] and sub['active']:
                task = self.send_notification(sub['url'], data)
                tasks.append(task)
        
        await asyncio.gather(*tasks, return_exceptions=True)
        
    @retry(retry_policy)
    async def send_notification(self, url: str, data: dict):
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=data) as response:
                if response.status >= 400:
                    raise WebhookError(f"Failed: {response.status}")

text_generation_miner.py

#!/usr/bin/env python3
import bittensor as bt
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
import time
import traceback
from typing import Optional
import argparse

class TextGenerationMiner:
    """
    Mineur pour le subnet de génération de texte de Bittensor
    """
    
    def __init__(self, config):
        # Configuration de base
        self.config = config
        bt.logging.info("Initialisation du mineur...")
        
        # Wallet et réseau
        self.wallet = bt.wallet(config=config)
        self.subtensor = bt.subtensor(config=config)
        self.metagraph = bt.metagraph(
            netuid=config.netuid, 
            network=config.subtensor.network
        )
        
        # Chargement du modèle
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.load_model()
        
        # Configuration de l'axon
        self.axon = bt.axon(wallet=self.wallet, config=config)
        self.uid = None
        
    def load_model(self):
        """Charge le modèle de génération de texte"""
        model_name = self.config.model_name or "microsoft/DialoGPT-medium"
        
        bt.logging.info(f"Chargement du modèle: {model_name}")
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForCausalLM.from_pretrained(model_name)
        self.model.to(self.device)
        self.model.eval()
        
        # Ajout du padding token si nécessaire
        if self.tokenizer.pad_token is None:
            self.tokenizer.pad_token = self.tokenizer.eos_token
            
        bt.logging.success(f"Modèle chargé sur {self.device}")
    
    def generate_response(self, prompt: str, max_length: int = 256) -> str:
        """Génère une réponse à partir du prompt"""
        try:
            # Tokenisation
            inputs = self.tokenizer.encode(
                prompt, 
                return_tensors='pt',
                max_length=512,
                truncation=True
            ).to(self.device)
            
            # Génération
            with torch.no_grad():
                outputs = self.model.generate(
                    inputs,
                    max_length=max_length,
                    num_return_sequences=1,
                    temperature=self.config.temperature,
                    top_p=self.config.top_p,
                    do_sample=True,
                    pad_token_id=self.tokenizer.pad_token_id,
                    eos_token_id=self.tokenizer.eos_token_id
                )
            
            # Décodage
            response = self.tokenizer.decode(
                outputs[0], 
                skip_special_tokens=True
            )
            
            # Retirer le prompt de la réponse
            if response.startswith(prompt):
                response = response[len(prompt):].strip()
                
            return response
            
        except Exception as e:
            bt.logging.error(f"Erreur de génération: {e}")
            return ""
    
    async def forward(self, synapse: bt.Synapse) -> bt.Synapse:
        """
        Traite les requêtes des validateurs
        """
        start_time = time.time()
        
        try:
            # Extraction du prompt
            prompt = synapse.messages[-1] if hasattr(synapse, 'messages') else str(synapse.roles[-1])
            
            bt.logging.info(f"Requête reçue: {prompt[:50]}...")
            
            # Génération de la réponse
            response = self.generate_response(
                prompt,
                max_length=synapse.max_tokens if hasattr(synapse, 'max_tokens') else 256
            )
            
            # Mise à jour de la synapse
            synapse.completion = response
            
            # Logging des performances
            process_time = time.time() - start_time
            bt.logging.success(
                f"Réponse générée en {process_time:.2f}s | "
                f"Longueur: {len(response)} chars"
            )
            
        except Exception as e:
            bt.logging.error(f"Erreur forward: {traceback.format_exc()}")
            synapse.completion = ""
            
        finally:
            synapse.process_time = time.time() - start_time
            
        return synapse
    
    def blacklist(self, synapse: bt.Synapse) -> tuple[bool, str]:
        """
        Filtre les requêtes selon différents critères
        """
        # Vérifier si le validateur est enregistré
        if synapse.dendrite.hotkey not in self.metagraph.hotkeys:
            return True, "Hotkey non enregistré"
        
        # Obtenir l'UID et le stake du validateur
        uid = self.metagraph.hotkeys.index(synapse.dendrite.hotkey)
        stake = self.metagraph.S[uid].item()
        
        # Filtrer par stake minimum
        if stake < self.config.min_stake:
            return True, f"Stake insuffisant: {stake:.2f} < {self.config.min_stake}"
        
        # Vérifier le rate limiting
        if hasattr(self, 'last_requests'):
            last_time = self.last_requests.get(synapse.dendrite.hotkey, 0)
            if time.time() - last_time < self.config.rate_limit:
                return True, "Rate limit dépassé"
        
        return False, "Autorisé"
    
    def priority(self, synapse: bt.Synapse) -> float:
        """
        Priorise les requêtes par stake du validateur
        """
        if synapse.dendrite.hotkey not in self.metagraph.hotkeys:
            return 0.0
            
        uid = self.metagraph.hotkeys.index(synapse.dendrite.hotkey)
        return float(self.metagraph.S[uid])
    
    def run(self):
        """
        Boucle principale du mineur
        """
        # Vérifier l'enregistrement
        if not self.subtensor.is_hotkey_registered(
            netuid=self.config.netuid,
            hotkey_ss58=self.wallet.hotkey.ss58_address
        ):
            bt.logging.error(
                f"Wallet {self.wallet} non enregistré sur netuid {self.config.netuid}. "
                "Veuillez vous enregistrer avant de continuer."
            )
            exit()
        
        # Obtenir l'UID
        self.uid = self.metagraph.hotkeys.index(
            self.wallet.hotkey.ss58_address
        )
        bt.logging.info(f"Mineur en cours avec UID: {self.uid}")
        
        # Attacher les handlers
        bt.logging.info("Attachement des fonctions forward...")
        self.axon.attach(
            forward_fn=self.forward,
            blacklist_fn=self.blacklist,
            priority_fn=self.priority
        )
        
        # Servir l'axon
        self.axon.serve(
            netuid=self.config.netuid,
            subtensor=self.subtensor
        )
        
        # Démarrer
        self.axon.start()
        bt.logging.info(
            f"Axon démarré sur {self.axon.ip}:{self.axon.port}"
        )
        
        # Boucle principale
        step = 0
        try:
            while True:
                # Sync périodique du métagraphe
                if step % 100 == 0:
                    self.metagraph.sync(subtensor=self.subtensor)
                    self.log_performance()
                
                # Pause
                time.sleep(1)
                step += 1
                
        except KeyboardInterrupt:
            bt.logging.info("Arrêt du mineur...")
        finally:
            self.axon.stop()
    
    def log_performance(self):
        """Log les métriques de performance"""
        try:
            incentive = self.metagraph.I[self.uid].item()
            rank = self.metagraph.R[self.uid].item()
            trust = self.metagraph.T[self.uid].item()
            consensus = self.metagraph.C[self.uid].item()
            
            bt.logging.info(
                f"Performance | UID: {self.uid} | "
                f"Incentive: {incentive:.4f} | "
                f"Rank: {rank:.4f} | "
                f"Trust: {trust:.4f} | "
                f"Consensus: {consensus:.4f}"
            )
        except Exception as e:
            bt.logging.error(f"Erreur de logging: {e}")

def main():
    # Parser de configuration
    parser = argparse.ArgumentParser()
    parser.add_argument('--netuid', type=int, default=1, help='Subnet UID')
    parser.add_argument('--model_name', type=str, help='Nom du modèle HuggingFace')
    parser.add_argument('--temperature', type=float, default=0.7)
    parser.add_argument('--top_p', type=float, default=0.9)
    parser.add_argument('--min_stake', type=float, default=100.0)
    parser.add_argument('--rate_limit', type=float, default=1.0)
    
    # Ajouter les args Bittensor
    bt.subtensor.add_args(parser)
    bt.logging.add_args(parser)
    bt.wallet.add_args(parser)
    bt.axon.add_args(parser)
    
    config = bt.config(parser)
    
    # Setup logging
    bt.logging(config=config, level='INFO')
    
    # Créer et lancer le mineur
    miner = TextGenerationMiner(config)
    miner.run()

if __name__ == "__main__":
    main()

simple_client.py

import bittensor as bt
from typing import List, Dict, Optional
import json

class BittensorClient:
    """
    Client simple pour interagir avec le réseau Bittensor
    """
    
    def __init__(self, wallet_name: str = "client", network: str = "finney"):
        # Configuration
        self.wallet = bt.wallet(name=wallet_name)
        self.subtensor = bt.subtensor(network=network)
        self.dendrite = bt.dendrite(wallet=self.wallet)
        
        # Cache des métagraphes
        self.metagraphs = {}
        
    def get_subnet_info(self, netuid: int) -> Dict:
        """Obtient les informations d'un subnet"""
        info = self.subtensor.get_subnet_info(netuid)
        return {
            'netuid': netuid,
            'emission': float(info.emission_value),
            'max_n': info.max_n,
            'difficulty': float(info.difficulty),
            'immunity_period': info.immunity_period,
            'is_open': info.is_open
        }
    
    def list_subnets(self) -> List[Dict]:
        """Liste tous les subnets actifs"""
        subnets = []
        for netuid in range(32):  # Max 32 subnets
            try:
                info = self.get_subnet_info(netuid)
                if info['emission'] > 0:
                    subnets.append(info)
            except:
                continue
        return subnets
    
    def get_balance(self, address: Optional[str] = None) -> float:
        """Obtient le balance en TAO"""
        if address is None:
            address = self.wallet.coldkey.ss58_address
        
        balance = self.subtensor.get_balance(address)
        return float(balance)
    
    def query_subnet(
        self, 
        prompt: str, 
        netuid: int = 1,
        top_k: int = 5,
        timeout: float = 10.0
    ) -> List[Dict]:
        """
        Interroge un subnet et retourne les réponses
        """
        # Obtenir ou mettre à jour le métagraphe
        if netuid not in self.metagraphs:
            self.metagraphs[netuid] = bt.metagraph(
                netuid=netuid,
                network=self.subtensor.network
            )
        
        metagraph = self.metagraphs[netuid]
        metagraph.sync(subtensor=self.subtensor)
        
        # Sélectionner les top mineurs par incentive
        incentives = metagraph.I.numpy()
        top_indices = incentives.argsort()[-top_k:][::-1]
        
        # Préparer les axons
        axons = [metagraph.axons[i] for i in top_indices]
        
        # Créer la synapse
        synapse = bt.Synapse(messages=[prompt])
        
        # Query
        responses = self.dendrite.query(
            axons=axons,
            synapse=synapse,
            timeout=timeout
        )
        
        # Formater les résultats
        results = []
        for idx, (i, response) in enumerate(zip(top_indices, responses)):
            if response and hasattr(response, 'completion'):
                results.append({
                    'uid': int(i),
                    'response': response.completion,
                    'incentive': float(incentives[i]),
                    'stake': float(metagraph.S[i])
                })
        
        return results
    
    def get_neuron_info(self, uid: int, netuid: int = 1) -> Dict:
        """Obtient les informations détaillées d'un neurone"""
        if netuid not in self.metagraphs:
            self.metagraphs[netuid] = bt.metagraph(
                netuid=netuid,
                network=self.subtensor.network
            )
        
        metagraph = self.metagraphs[netuid]
        metagraph.sync(subtensor=self.subtensor)
        
        if uid >= metagraph.n.item():
            raise ValueError(f"UID {uid} n'existe pas dans subnet {netuid}")
        
        neuron = metagraph.neurons[uid]
        
        return {
            'uid': uid,
            'hotkey': neuron.hotkey,
            'coldkey': neuron.coldkey,
            'stake': float(metagraph.S[uid]),
            'rank': float(metagraph.R[uid]),
            'trust': float(metagraph.T[uid]),
            'consensus': float(metagraph.C[uid]),
            'incentive': float(metagraph.I[uid]),
            'dividends': float(metagraph.D[uid]),
            'emission': float(metagraph.E[uid]),
            'is_validator': neuron.is_null,
            'last_update': neuron.last_update,
            'axon_info': {
                'ip': neuron.axon_info.ip,
                'port': neuron.axon_info.port,
                'hotkey': neuron.axon_info.hotkey
            }
        }

# Exemples d'utilisation
if __name__ == "__main__":
    # Créer le client
    client = BittensorClient(network="finney")
    
    # Exemple 1: Lister les subnets
    print("=== Subnets Actifs ===")
    subnets = client.list_subnets()
    for subnet in subnets[:5]:
        print(f"Subnet {subnet['netuid']}: {subnet['emission']:.4f} TAO/bloc")
    
    # Exemple 2: Vérifier le balance
    print(f"\n=== Balance ===")
    balance = client.get_balance()
    print(f"Balance: {balance:.4f} TAO")
    
    # Exemple 3: Query simple
    print("\n=== Query Subnet 1 ===")
    responses = client.query_subnet(
        "What is the capital of France?",
        netuid=1,
        top_k=3
    )
    
    for i, resp in enumerate(responses):
        print(f"\n{i+1}. UID {resp['uid']} (incentive: {resp['incentive']:.4f}):")
        print(f"   {resp['response'][:200]}...")
    
    # Exemple 4: Info d'un neurone
    print("\n=== Info Neurone ===")
    try:
        info = client.get_neuron_info(uid=0, netuid=1)
        print(json.dumps(info, indent=2))
    except Exception as e:
        print(f"Erreur: {e}")