Aqu铆 tienes la actualizaci贸n del c贸digo en Python. En esta versi贸n se incorpora la clase ProgrammableCBDC, que a帽ade las dos caracter铆sticas fundamentales del dinero digital emitido por Bancos Centrales: fecha de caducidad autom谩tica (el dinero pierde valor si no se gasta dentro de un n煤mero determinado de bloques) y restricciones de uso programables (listas blancas de categor铆as permitidas, p. ej. solo energ铆a o alimentos, bloqueando otros usos
Para experimentar el comportamiento de la red en tiempo real, aqu铆 tienes la simulaci贸n interactiva completa de OmniaLedger.
Para experimentar el comportamiento de la red en tiempo real, aqu铆 tienes la simulaci贸n interactiva completa de OmniaLedger.
El simulador visualiza el comportamiento del Or谩culo de Commodities (Petr贸leo y Oro) afectando el valor de liquidaci贸n, el saldo en tiempo real de la CBDC programable, el estado de las reglas de caducidad/categor铆a y la forja de bloques inmutables en la cadena.
---------------------------------------------------------------
Aqu铆 tienes el c贸digo Python actualizado. Se ha integrado el contrato inteligente GovernanceContract, que permite a los nodos validadores (stakers) proponer y votar cambios cr铆ticos en los par谩metros de la CBDC (como la tasa de emisi贸n por bloque, el tiempo de caducidad est谩ndar y las categor铆as de gasto autorizadas).
import datetime
import hashlib
import json
import random
from typing import Dict, List, Optional
class OracleService:
"""Or谩culo de Datos Descentralizado (Simulaci贸n tipo Chainlink)."""
def __init__(self):
self._prices = {
"GOLD_OUNCE": 2400.00,
"CRUDE_BARREL": 85.00,
}
def update_market_data(self):
self._prices["GOLD_OUNCE"] *= random.uniform(1.01, 1.03)
self._prices["CRUDE_BARREL"] *= random.uniform(0.98, 1.05)
def get_price(self, commodity: str) -> float:
return round(self._prices.get(commodity, 0.0), 2)
class ProgrammableCBDC:
"""Saldo CBDC con reglas din谩micas fijadas por Gobernanza."""
def __init__(
self,
amount: float,
expiry_block: int,
allowed_categories: List[str],
):
self.amount = amount
self.expiry_block = expiry_block
self.allowed_categories = allowed_categories
def is_expired(self, current_block: int) -> bool:
return current_block >= self.expiry_block
def is_category_allowed(self, category: str) -> bool:
return (
"ALL" in self.allowed_categories
or category in self.allowed_categories
)
class ProposalStatus:
PENDING = "PENDIENTE"
PASSED = "APROBADA"
REJECTED = "RECHAZADA"
EXECUTED = "EJECUTADA"
class GovernanceProposal:
"""Propuesta de Gobernanza para alterar par谩metros del Banco Central."""
def __init__(
self,
proposal_id: int,
proposer: str,
description: str,
param_key: str, # Ej: 'default_expiry_blocks', 'allowed_categories', 'issuance_rate'
new_value: any,
voting_deadline_block: int,
):
self.proposal_id = proposal_id
self.proposer = proposer
self.description = description
self.param_key = param_key
self.new_value = new_value
self.voting_deadline_block = voting_deadline_block
self.votes_for: float = 0.0
self.votes_against: float = 0.0
self.voted_nodes: List[str] = []
self.status: str = ProposalStatus.PENDING
class GovernanceContract:
"""Contrato de Gobernanza por prueba de participaci贸n (Proof of Stake).
Los nodos validadores votan de acuerdo a su peso de staking en la red.
"""
def __init__(self, voting_duration_blocks: int = 2, quorum_percentage: float = 0.50):
self.voting_duration_blocks = voting_duration_blocks
self.quorum_percentage = quorum_percentage
self.proposals: Dict[int, GovernanceProposal] = {}
self.proposal_counter = 0
# Par谩metros Globales de la CBDC gestionados por Gobernanza
self.system_parameters = {
"default_expiry_blocks": 5,
"allowed_categories": ["ENERGIA", "ALIMENTOS"],
"issuance_rate_per_block": 100.0,
}
def create_proposal(
self,
proposer: str,
description: str,
param_key: str,
new_value: any,
current_block: int,
) -> int:
if param_key not in self.system_parameters:
raise ValueError(f"El par谩metro '{param_key}' no existe en el sistema.")
self.proposal_counter += 1
proposal = GovernanceProposal(
proposal_id=self.proposal_counter,
proposer=proposer,
description=description,
param_key=param_key,
new_value=new_value,
voting_deadline_block=current_block + self.voting_duration_blocks,
)
self.proposals[self.proposal_counter] = proposal
return self.proposal_counter
def cast_vote(
self,
proposal_id: int,
validator_node: str,
support: bool,
stake_weight: float,
) -> Dict:
proposal = self.proposals.get(proposal_id)
if not proposal:
return {"status": "ERROR", "reason": "Propuesta no encontrada."}
if proposal.status != ProposalStatus.PENDING:
return {"status": "ERROR", "reason": "La votaci贸n de esta propuesta ha finalizado."}
if validator_node in proposal.voted_nodes:
return {"status": "ERROR", "reason": "Este nodo validador ya ha votado."}
if support:
proposal.votes_for += stake_weight
else:
proposal.votes_against += stake_weight
proposal.voted_nodes.append(validator_node)
return {
"status": "VOTE_ACCEPTED",
"proposal_id": proposal_id,
"validator": validator_node,
"weight": stake_weight,
"vote": "A FAVOR" if support else "EN CONTRA",
}
def evaluate_and_execute_proposals(self, current_block: int, total_network_stake: float) -> List[Dict]:
results = []
for prop_id, prop in self.proposals.items():
if prop.status == ProposalStatus.PENDING and current_block >= prop.voting_deadline_block:
total_votes = prop.votes_for + prop.votes_against
# Comprobar Qu贸rum
if total_votes >= (total_network_stake * self.quorum_percentage):
if prop.votes_for > prop.votes_against:
prop.status = ProposalStatus.EXECUTED
# Actualizaci贸n automatizada de par谩metros del Banco Central
self.system_parameters[prop.param_key] = prop.new_value
results.append({
"event": "GOVERNANCE_PROPOSAL_PASSED",
"proposal_id": prop_id,
"param_updated": prop.param_key,
"new_value": prop.new_value,
"votes_for": prop.votes_for,
"votes_against": prop.votes_against,
})
else:
prop.status = ProposalStatus.REJECTED
results.append({
"event": "GOVERNANCE_PROPOSAL_REJECTED",
"proposal_id": prop_id,
"reason": "Mayor铆a de votos en contra.",
})
else:
prop.status = ProposalStatus.REJECTED
results.append({
"event": "GOVERNANCE_PROPOSAL_REJECTED",
"proposal_id": prop_id,
"reason": f"No se alcanz贸 el qu贸rum ({total_votes}/{total_network_stake * self.quorum_percentage} requeridos).",
})
return results
class Block:
"""Bloque individual en Omnia Ledger."""
def __init__(self, index: int, timestamp: str, transactions: List[Dict], previous_hash: str):
self.index = index
self.timestamp = timestamp
self.transactions = transactions
self.previous_hash = previous_hash
self.nonce = 0
self.hash = self.calculate_hash()
def calculate_hash(self) -> str:
block_string = json.dumps(
{
"index": self.index,
"timestamp": self.timestamp,
"transactions": self.transactions,
"previous_hash": self.previous_hash,
"nonce": self.nonce,
},
sort_keys=True,
)
return hashlib.sha256(block_string.encode()).hexdigest()
def mine_block(self, difficulty: int):
target = "0" * difficulty
while self.hash[:difficulty] != target:
self.nonce += 1
self.hash = self.calculate_hash()
class OmniaLedger:
"""Cadena de Bloques con Gobernanza On-Chain y Validadoras Staking."""
def __init__(self, difficulty: int = 1):
self.chain: List[Block] = []
self.difficulty = difficulty
self.pending_transactions: List[Dict] = []
self.validators_stake: Dict[str, float] = {} # Nodo Validador -> Peso en Staking
self.governance = GovernanceContract()
self.oracle = OracleService()
self.create_genesis_block()
def create_genesis_block(self):
genesis_tx = {
"from": "GENESIS",
"to": "SYSTEM",
"info": "Omnia Ledger Governance System Online",
}
genesis_block = Block(0, str(datetime.datetime.now()), [genesis_tx], "0" * 64)
genesis_block.mine_block(self.difficulty)
self.chain.append(genesis_block)
def register_validator(self, validator_id: str, stake_amount: float):
self.validators_stake[validator_id] = stake_amount
def get_total_stake(self) -> float:
return sum(self.validators_stake.values())
def mine_pending_transactions(self, miner_address: str):
current_block_index = len(self.chain)
# 1. Evaluar propuestas de gobernanza pendientes de cierre
gov_events = self.governance.evaluate_and_execute_proposals(
current_block=current_block_index,
total_network_stake=self.get_total_stake(),
)
for event in gov_events:
self.pending_transactions.append({"type": "GOVERNANCE_EXECUTION", "details": event})
# 2. Recompensa al validador
reward_tx = {
"from": "NETWORK",
"to": miner_address,
"amount": 10.0,
"asset": "OMNIA_STAKING_REWARD",
}
self.pending_transactions.append(reward_tx)
# 3. Minado
new_block = Block(
index=current_block_index,
timestamp=str(datetime.datetime.now()),
transactions=self.pending_transactions,
previous_hash=self.chain[-1].hash,
)
new_block.mine_block(self.difficulty)
self.chain.append(new_block)
self.pending_transactions = []
# =====================================================================
# SIMULACI脫N: VOTACI脫N DE GOBERNANZA POR NODOS VALIDADORES (PoS)
# =====================================================================
if __name__ == "__main__":
print("================================================================")
print(" OMNIA LEDGER: GOBERNANZA DESCENTRALIZADA DE PARAMETROS CBDC")
print("================================================================\n")
omnia = OmniaLedger(difficulty=1)
# 1. Registrar Nodos Validadores con sus respectivos dep贸sitos de Staking
omnia.register_validator("VALIDATOR_NODE_LONDON", stake_amount=500_000.0)
omnia.register_validator("VALIDATOR_NODE_FRANKFURT", stake_amount=350_000.0)
omnia.register_validator("VALIDATOR_NODE_TOKYO", stake_amount=150_000.0)
print(f"Staking Total de la Red: {omnia.get_total_stake():,.2f} OMNIA")
print("Par谩metros actuales CBDC:", omnia.governance.system_parameters, "\n")
# 2. El nodo de Londres crea una propuesta para ampliar las categor铆as permitidas
current_block = len(omnia.chain)
prop_id = omnia.governance.create_proposal(
proposer="VALIDATOR_NODE_LONDON",
description="A帽adir la categor铆a 'TRANSPORTE' a la lista blanca de la CBDC.",
param_key="allowed_categories",
new_value=["ENERGIA", "ALIMENTOS", "TRANSPORTE"],
current_block=current_block,
)
print(f"---> [PROPUESTA #{prop_id} CREADA] Expira en Bloque #{omnia.governance.proposals[prop_id].voting_deadline_block}")
# 3. Emisi贸n de Votos de los Validadores
voto1 = omnia.governance.cast_vote(
proposal_id=prop_id,
validator_node="VALIDATOR_NODE_LONDON",
support=True,
stake_weight=omnia.validators_stake["VALIDATOR_NODE_LONDON"],
)
print(f" Voto registrado: {voto1['validator']} ({voto1['weight']} votos) -> {voto1['vote']}")
voto2 = omnia.governance.cast_vote(
proposal_id=prop_id,
validator_node="VALIDATOR_NODE_FRANKFURT",
support=True,
stake_weight=omnia.validators_stake["VALIDATOR_NODE_FRANKFURT"],
)
print(f" Voto registrado: {voto2['validator']} ({voto2['weight']} votos) -> {voto2['vote']}")
voto3 = omnia.governance.cast_vote(
proposal_id=prop_id,
validator_node="VALIDATOR_NODE_TOKYO",
support=False,
stake_weight=omnia.validators_stake["VALIDATOR_NODE_TOKYO"],
)
print(f" Voto registrado: {voto3['validator']} ({voto3['weight']} votos) -> {voto3['vote']}\n")
# 4. Transcurso de bloques hasta la resoluci贸n
print("---> [BLOQUE 1] Minando bloque...")
omnia.mine_pending_transactions(miner_address="VALIDATOR_NODE_LONDON")
print("---> [BLOQUE 2] Se alcanza la fecha l铆mite de votaci贸n...")
omnia.mine_pending_transactions(miner_address="VALIDATOR_NODE_FRANKFURT")
# 5. Comprobar resultado en la cadena
latest_block = omnia.chain[-1]
for tx in latest_block.transactions:
if tx.get("type") == "GOVERNANCE_EXECUTION":
event = tx["details"]
print(f"\n[RESULTADO GOBERNANZA]: {event['event']}")
print(f" Par谩metro: {event.get('param_updated')}")
print(f" Nuevos Valores Aplicados: {event.get('new_value')}")
print("\nPar谩metros del Sistema post-votaci贸n:", omnia.governance.system_parameters)
----------------------------------------------------
Aqu铆 tienes el c贸digo Python actualizado. Se incorpora el m贸dulo de Slashing (Penalizaciones y Confiscaci贸n de Stake) a la arquitectura.
El sistema detecta autom谩ticamente infracciones como el doble voto en una misma propuesta (Double Voting) o el intento de minado/propuesta de bloques inv谩lidos, confisca un porcentaje del stake depositado (hasta el 100% seg煤n la gravedad) y expulsa o suspende al nodo validador de la red.
import datetime
import hashlib
import json
import random
from typing import Dict, List, Optional
class OracleService:
"""Or谩culo de Datos Descentralizado."""
def __init__(self):
self._prices = {"GOLD_OUNCE": 2400.00, "CRUDE_BARREL": 85.00}
def get_price(self, commodity: str) -> float:
return round(self._prices.get(commodity, 0.0), 2)
class ProgrammableCBDC:
"""Saldo CBDC con reglas din谩micas."""
def __init__(self, amount: float, expiry_block: int, allowed_categories: List[str]):
self.amount = amount
self.expiry_block = expiry_block
self.allowed_categories = allowed_categories
class ProposalStatus:
PENDING = "PENDIENTE"
EXECUTED = "EJECUTADA"
REJECTED = "RECHAZADA"
class GovernanceProposal:
"""Propuesta de Gobernanza."""
def __init__(
self,
proposal_id: int,
proposer: str,
description: str,
param_key: str,
new_value: any,
voting_deadline_block: int,
):
self.proposal_id = proposal_id
self.proposer = proposer
self.description = description
self.param_key = param_key
self.new_value = new_value
self.voting_deadline_block = voting_deadline_block
self.votes_for: float = 0.0
self.votes_against: float = 0.0
self.voted_nodes: List[str] = []
self.status: str = ProposalStatus.PENDING
class SlashingPenaltyType:
DOUBLE_VOTING = {"ratio": 0.50, "reason": "Doble voto detectado en la misma propuesta"} # Confisca 50%
INVALID_BLOCK = {"ratio": 1.00, "reason": "Intento de forja de bloque inválido o corrupto"} # Confisca 100%
class GovernanceContract:
"""Contrato de Gobernanza con protecci贸n de Slashing."""
def __init__(self, voting_duration_blocks: int = 2, quorum_percentage: float = 0.50):
self.voting_duration_blocks = voting_duration_blocks
self.quorum_percentage = quorum_percentage
self.proposals: Dict[int, GovernanceProposal] = {}
self.proposal_counter = 0
self.system_parameters = {
"default_expiry_blocks": 5,
"allowed_categories": ["ENERGIA", "ALIMENTOS"],
"issuance_rate_per_block": 100.0,
}
def create_proposal(
self,
proposer: str,
description: str,
param_key: str,
new_value: any,
current_block: int,
) -> int:
self.proposal_counter += 1
proposal = GovernanceProposal(
proposal_id=self.proposal_counter,
proposer=proposer,
description=description,
param_key=param_key,
new_value=new_value,
voting_deadline_block=current_block + self.voting_duration_blocks,
)
self.proposals[self.proposal_counter] = proposal
return self.proposal_counter
def cast_vote(
self,
proposal_id: int,
validator_node: str,
support: bool,
stake_weight: float,
ledger: "OmniaLedger",
) -> Dict:
proposal = self.proposals.get(proposal_id)
if not proposal:
return {"status": "ERROR", "reason": "Propuesta no encontrada."}
# DETECCI脫N DE ATAQUE: Intento de voto duplicado (Double Voting)
if validator_node in proposal.voted_nodes:
# Se reporta al Ledger para ejecutar SLASHING
slashing_event = ledger.apply_slashing(
validator_id=validator_node,
penalty_info=SlashingPenaltyType.DOUBLE_VOTING,
evidence=f"Intento de re-votar propuesta #{proposal_id}",
)
return {
"status": "SLASHED",
"reason": "Doble voto en propuesta.",
"slashing_details": slashing_event,
}
if support:
proposal.votes_for += stake_weight
else:
proposal.votes_against += stake_weight
proposal.voted_nodes.append(validator_node)
return {
"status": "VOTE_ACCEPTED",
"proposal_id": proposal_id,
"validator": validator_node,
"weight": stake_weight,
"vote": "A FAVOR" if support else "EN CONTRA",
}
def evaluate_and_execute_proposals(self, current_block: int, total_network_stake: float) -> List[Dict]:
results = []
for prop_id, prop in self.proposals.items():
if prop.status == ProposalStatus.PENDING and current_block >= prop.voting_deadline_block:
total_votes = prop.votes_for + prop.votes_against
if total_votes >= (total_network_stake * self.quorum_percentage):
if prop.votes_for > prop.votes_against:
prop.status = ProposalStatus.EXECUTED
self.system_parameters[prop.param_key] = prop.new_value
results.append({
"event": "GOVERNANCE_PROPOSAL_PASSED",
"proposal_id": prop_id,
"param_updated": prop.param_key,
"new_value": prop.new_value,
})
else:
prop.status = ProposalStatus.REJECTED
else:
prop.status = ProposalStatus.REJECTED
return results
class Block:
"""Representa un bloque individual con firma de validador."""
def __init__(self, index: int, timestamp: str, transactions: List[Dict], previous_hash: str, validator: str):
self.index = index
self.timestamp = timestamp
self.transactions = transactions
self.previous_hash = previous_hash
self.validator = validator
self.nonce = 0
self.hash = self.calculate_hash()
def calculate_hash(self) -> str:
block_string = json.dumps(
{
"index": self.index,
"timestamp": self.timestamp,
"transactions": self.transactions,
"previous_hash": self.previous_hash,
"validator": self.validator,
"nonce": self.nonce,
},
sort_keys=True,
)
return hashlib.sha256(block_string.encode()).hexdigest()
def mine_block(self, difficulty: int):
target = "0" * difficulty
while self.hash[:difficulty] != target:
self.nonce += 1
self.hash = self.calculate_hash()
class OmniaLedger:
"""Cadena de Bloques con Sistema de Slashing y Fondo de Reserva (Treasury)."""
def __init__(self, difficulty: int = 1):
self.chain: List[Block] = []
self.difficulty = difficulty
self.pending_transactions: List[Dict] = []
self.validators_stake: Dict[str, float] = {}
self.slashed_validators: List[str] = [] # Lista negra de validadores expulsados
self.treasury_slashed_pool: float = 0.0 # Fondo donde se deposita el stake confiscado
self.governance = GovernanceContract()
self.create_genesis_block()
def create_genesis_block(self):
genesis_block = Block(0, str(datetime.datetime.now()), [], "0" * 64, validator="GENESIS")
genesis_block.mine_block(self.difficulty)
self.chain.append(genesis_block)
def register_validator(self, validator_id: str, stake_amount: float):
self.validators_stake[validator_id] = stake_amount
def get_total_stake(self) -> float:
return sum(self.validators_stake.values())
def apply_slashing(self, validator_id: str, penalty_info: Dict, evidence: str) -> Dict:
"""Confisca el stake del validador y lo expulsa si pierde todo su colateral."""
current_stake = self.validators_stake.get(validator_id, 0.0)
slashed_amount = round(current_stake * penalty_info["ratio"], 2)
# 1. Quitar fondos de stake y reasignarlos a la Tesorer铆a de la red
self.validators_stake[validator_id] -= slashed_amount
self.treasury_slashed_pool += slashed_amount
# 2. Si el ratio es 100% o el stake cae a 0, se le a帽ade a la lista negra
if self.validators_stake[validator_id] == 0:
self.slashed_validators.append(validator_id)
slashing_record = {
"event": "SLASHING_EXECUTED",
"validator": validator_id,
"reason": penalty_info["reason"],
"evidence": evidence,
"amount_slashed": slashed_amount,
"remaining_stake": self.validators_stake[validator_id],
"treasury_total": self.treasury_slashed_pool,
}
self.pending_transactions.append({"type": "SLASHING_EVENT", "details": slashing_record})
return slashing_record
def propose_and_validate_block(self, miner_address: str, invalid_payload: bool = False) -> Optional[Block]:
"""Procesa un nuevo bloque. Si el validador es malicioso (ej. bloque inv谩lido), es penalizado."""
if miner_address in self.slashed_validators:
print(f" RECHAZADO: El validador {miner_address} est谩 expulsado de la red por Slashing.")
return None
# DETECCI脫N DE ATAQUE: Bloque inválido o fraudulento
if invalid_payload:
self.apply_slashing(
validator_id=miner_address,
penalty_info=SlashingPenaltyType.INVALID_BLOCK,
evidence="Falsificaci贸n de transacciones en el encabezado del bloque.",
)
print(f" ATAQUE DETECTADO: El validador {miner_address} ha intentado forjar un bloque corrupto.")
return None
current_block_index = len(self.chain)
# Evaluar gobernanza
gov_events = self.governance.evaluate_and_execute_proposals(
current_block=current_block_index,
total_network_stake=self.get_total_stake(),
)
for event in gov_events:
self.pending_transactions.append({"type": "GOVERNANCE_EXECUTION", "details": event})
new_block = Block(
index=current_block_index,
timestamp=str(datetime.datetime.now()),
transactions=self.pending_transactions,
previous_hash=self.chain[-1].hash,
validator=miner_address,
)
new_block.mine_block(self.difficulty)
self.chain.append(new_block)
self.pending_transactions = []
return new_block
# =====================================================================
# DEMOSTRACI脫N: DETECCI脫N Y EJECUCI脫N DE SLASHING
# =====================================================================
if __name__ == "__main__":
print("================================================================")
print(" OMNIA LEDGER: MECANISMO DE SLASHING Y SEGURIDAD DE RED")
print("================================================================\n")
omnia = OmniaLedger(difficulty=1)
# 1. Registrar Nodos Validadores
omnia.register_validator("NODE_HONEST_1", stake_amount=400_000.0)
omnia.register_validator("NODE_MALICIOUS_2", stake_amount=300_000.0)
print(f"Stake Inicial 'NODE_MALICIOUS_2': {omnia.validators_stake['NODE_MALICIOUS_2']:,.2f} OMNIA\n")
# 2. ESCENARIO A: Intento de Doble Voto (Double Voting)
print("---> ESCENARIO 1: Validador emite voto duplicado en gobernanza...")
prop_id = omnia.governance.create_proposal(
proposer="NODE_HONEST_1",
description="Aumentar tasa de emisi贸n",
param_key="issuance_rate_per_block",
new_value=200.0,
current_block=1,
)
# Voto leg铆timo #1
omnia.governance.cast_vote(
proposal_id=prop_id,
validator_node="NODE_MALICIOUS_2",
support=True,
stake_weight=omnia.validators_stake["NODE_MALICIOUS_2"],
ledger=omnia,
)
# Voto fraudulento #2 (Doble Voto)
resultado_ataque = omnia.governance.cast_vote(
proposal_id=prop_id,
validator_node="NODE_MALICIOUS_2",
support=False,
stake_weight=omnia.validators_stake["NODE_MALICIOUS_2"],
ledger=omnia,
)
print(f" ALERTA: {resultado_ataque['reason']}")
print(f" Stake restante tras penalizaci贸n (50% Slashing): {omnia.validators_stake['NODE_MALICIOUS_2']:,.2f} OMNIA")
print(f" Tesorer铆a de la red (Confiscado): {omnia.treasury_slashed_pool:,.2f} OMNIA\n")
# 3. ESCENARIO B: Forja de Bloque Corrupto
print("---> ESCENARIO 2: Validador malicioso intenta proponer un bloque inv谩lido...")
omnia.propose_and_validate_block(miner_address="NODE_MALICIOUS_2", invalid_payload=True)
print(f" Stake restante tras penalizaci贸n total (100% Slashing): {omnia.validators_stake['NODE_MALICIOUS_2']:,.2f} OMNIA")
print(f" Tesorer铆a acumulada por sanciones: {omnia.treasury_slashed_pool:,.2f} OMNIA")
# 4. Intento de minado por nodo en Lista Negra
print("\n---> ESCENARIO 3: Validador expulsado intenta operar nuevamente...")
omnia.propose_and_validate_block(miner_address="NODE_MALICIOUS_2", invalid_payload=False)
----------------------------------------------
Aqu铆 tienes el c贸digo Python modificado. Se ha implementado la mec谩nica de Unbonding Period (Per铆odo de Desbloqueo/Carencia).
Con esta l贸gica, cuando un validador solicita retirar su stake, sus fondos entran en un estado temporal de congelaci贸n (UnbondingRequest) durante un n煤mero N de bloques configurable. Si durante ese per铆odo se detecta o reporta una infracci贸n cometida previamente por el validador, el sistema de Slashing confisca los fondos directamente del saldo congelado en proceso de retiro, impidiendo la huida de capital malicioso.
import datetime
import hashlib
import json
import random
from typing import Dict, List, Optional
class OracleService:
"""Or谩culo de Datos Descentralizado."""
def __init__(self):
self._prices = {"GOLD_OUNCE": 2400.00, "CRUDE_BARREL": 85.00}
def get_price(self, commodity: str) -> float:
return round(self._prices.get(commodity, 0.0), 2)
class ProgrammableCBDC:
"""Saldo CBDC con reglas din谩micas."""
def __init__(self, amount: float, expiry_block: int, allowed_categories: List[str]):
self.amount = amount
self.expiry_block = expiry_block
self.allowed_categories = allowed_categories
class ProposalStatus:
PENDING = "PENDIENTE"
EXECUTED = "EJECUTADA"
REJECTED = "RECHAZADA"
class GovernanceProposal:
"""Propuesta de Gobernanza."""
def __init__(
self,
proposal_id: int,
proposer: str,
description: str,
param_key: str,
new_value: any,
voting_deadline_block: int,
):
self.proposal_id = proposal_id
self.proposer = proposer
self.description = description
self.param_key = param_key
self.new_value = new_value
self.voting_deadline_block = voting_deadline_block
self.votes_for: float = 0.0
self.votes_against: float = 0.0
self.voted_nodes: List[str] = []
self.status: str = ProposalStatus.PENDING
class SlashingPenaltyType:
DOUBLE_VOTING = {"ratio": 0.50, "reason": "Doble voto detectado en la misma propuesta"}
INVALID_BLOCK = {"ratio": 1.00, "reason": "Intento de forja de bloque inv谩lido o corrupto"}
class UnbondingRequest:
"""Representa una solicitud de retiro de stake en per铆odo de carencia."""
def __init__(self, request_id: int, validator_id: str, amount: float, completion_block: int):
self.request_id = request_id
self.validator_id = validator_id
self.amount = amount
self.completion_block = completion_block # Bloque en el que los fondos quedan liberados
self.is_completed = False
class GovernanceContract:
"""Contrato de Gobernanza con protecci贸n de Slashing."""
def __init__(self, voting_duration_blocks: int = 2, quorum_percentage: float = 0.50):
self.voting_duration_blocks = voting_duration_blocks
self.quorum_percentage = quorum_percentage
self.proposals: Dict[int, GovernanceProposal] = {}
self.proposal_counter = 0
self.system_parameters = {
"default_expiry_blocks": 5,
"allowed_categories": ["ENERGIA", "ALIMENTOS"],
"issuance_rate_per_block": 100.0,
}
def create_proposal(
self,
proposer: str,
description: str,
param_key: str,
new_value: any,
current_block: int,
) -> int:
self.proposal_counter += 1
proposal = GovernanceProposal(
proposal_id=self.proposal_counter,
proposer=proposer,
description=description,
param_key=param_key,
new_value=new_value,
voting_deadline_block=current_block + self.voting_duration_blocks,
)
self.proposals[self.proposal_counter] = proposal
return self.proposal_counter
def cast_vote(
self,
proposal_id: int,
validator_node: str,
support: bool,
stake_weight: float,
ledger: "OmniaLedger",
) -> Dict:
proposal = self.proposals.get(proposal_id)
if not proposal:
return {"status": "ERROR", "reason": "Propuesta no encontrada."}
# DETECCI脫N DE ATAQUE: Intento de voto duplicado (Double Voting)
if validator_node in proposal.voted_nodes:
slashing_event = ledger.apply_slashing(
validator_id=validator_node,
penalty_info=SlashingPenaltyType.DOUBLE_VOTING,
evidence=f"Intento de re-votar propuesta #{proposal_id}",
)
return {
"status": "SLASHED",
"reason": "Doble voto en propuesta.",
"slashing_details": slashing_event,
}
if support:
proposal.votes_for += stake_weight
else:
proposal.votes_against += stake_weight
proposal.voted_nodes.append(validator_node)
return {
"status": "VOTE_ACCEPTED",
"proposal_id": proposal_id,
"validator": validator_node,
"weight": stake_weight,
"vote": "A FAVOR" if support else "EN CONTRA",
}
def evaluate_and_execute_proposals(self, current_block: int, total_network_stake: float) -> List[Dict]:
results = []
for prop_id, prop in self.proposals.items():
if prop.status == ProposalStatus.PENDING and current_block >= prop.voting_deadline_block:
total_votes = prop.votes_for + prop.votes_against
if total_votes >= (total_network_stake * self.quorum_percentage):
if prop.votes_for > prop.votes_against:
prop.status = ProposalStatus.EXECUTED
self.system_parameters[prop.param_key] = prop.new_value
results.append({
"event": "GOVERNANCE_PROPOSAL_PASSED",
"proposal_id": prop_id,
"param_updated": prop.param_key,
"new_value": prop.new_value,
})
else:
prop.status = ProposalStatus.REJECTED
else:
prop.status = ProposalStatus.REJECTED
return results
class Block:
"""Representa un bloque individual con firma de validador."""
def __init__(self, index: int, timestamp: str, transactions: List[Dict], previous_hash: str, validator: str):
self.index = index
self.timestamp = timestamp
self.transactions = transactions
self.previous_hash = previous_hash
self.validator = validator
self.nonce = 0
self.hash = self.calculate_hash()
def calculate_hash(self) -> str:
block_string = json.dumps(
{
"index": self.index,
"timestamp": self.timestamp,
"transactions": self.transactions,
"previous_hash": self.previous_hash,
"validator": self.validator,
"nonce": self.nonce,
},
sort_keys=True,
)
return hashlib.sha256(block_string.encode()).hexdigest()
def mine_block(self, difficulty: int):
target = "0" * difficulty
while self.hash[:difficulty] != target:
self.nonce += 1
self.hash = self.calculate_hash()
class OmniaLedger:
"""Cadena de Bloques con Unbonding Period, Slashing y Liquidaci贸n de Retiros."""
def __init__(self, difficulty: int = 1, unbonding_period_blocks: int = 3):
self.chain: List[Block] = []
self.difficulty = difficulty
self.unbonding_period_blocks = unbonding_period_blocks # N Bloques de carencia
self.pending_transactions: List[Dict] = []
self.validators_stake: Dict[str, float] = {}
self.unbonding_requests: List[UnbondingRequest] = []
self.unbonding_counter = 0
self.slashed_validators: List[str] = []
self.treasury_slashed_pool: float = 0.0
self.liquid_balances: Dict[str, float] = {} # Fondos retirados con 茅xito
self.governance = GovernanceContract()
self.create_genesis_block()
def create_genesis_block(self):
genesis_block = Block(0, str(datetime.datetime.now()), [], "0" * 64, validator="GENESIS")
genesis_block.mine_block(self.difficulty)
self.chain.append(genesis_block)
def register_validator(self, validator_id: str, stake_amount: float):
self.validators_stake[validator_id] = stake_amount
def get_total_stake(self) -> float:
return sum(self.validators_stake.values())
def request_unstake(self, validator_id: str, amount: float) -> Dict:
"""Inicia la solicitud de retiro moviendo los fondos al per铆odo de carencia (Unbonding)."""
current_stake = self.validators_stake.get(validator_id, 0.0)
if amount > current_stake:
return {"status": "ERROR", "reason": "Monto solicitado supera el stake activo."}
# 1. Restar del stake activo inmediatamente
self.validators_stake[validator_id] -= amount
# 2. Registrar la solicitud en per铆odo de carencia
self.unbonding_counter += 1
current_block = len(self.chain)
completion_block = current_block + self.unbonding_period_blocks
req = UnbondingRequest(
request_id=self.unbonding_counter,
validator_id=validator_id,
amount=amount,
completion_block=completion_block,
)
self.unbonding_requests.append(req)
record = {
"event": "UNBONDING_REQUESTED",
"request_id": req.request_id,
"validator": validator_id,
"amount_locked": amount,
"request_block": current_block,
"available_at_block": completion_block,
}
self.pending_transactions.append({"type": "UNBONDING_EVENT", "details": record})
return record
def apply_slashing(self, validator_id: str, penalty_info: Dict, evidence: str) -> Dict:
"""Confisca el stake activo Y los fondos bloqueados en el Unbonding Period."""
active_stake = self.validators_stake.get(validator_id, 0.0)
# Calcular fondos congelados en solicitudes de unbonding pendientes
pending_unbonding_amount = sum(
req.amount for req in self.unbonding_requests if req.validator_id == validator_id and not req.is_completed
)
total_collateral = active_stake + pending_unbonding_amount
slashed_amount = round(total_collateral * penalty_info["ratio"], 2)
# 1. Confiscar primero del stake activo
remaining_to_slash = slashed_amount
if active_stake >= remaining_to_slash:
self.validators_stake[validator_id] -= remaining_to_slash
remaining_to_slash = 0.0
else:
remaining_to_slash -= active_stake
self.validators_stake[validator_id] = 0.0
# 2. Si la sanci贸n supera el stake activo, confiscar de las solicitudes de Unbonding
if remaining_to_slash > 0:
for req in self.unbonding_requests:
if req.validator_id == validator_id and not req.is_completed:
if req.amount >= remaining_to_slash:
req.amount -= remaining_to_slash
remaining_to_slash = 0.0
break
else:
remaining_to_slash -= req.amount
req.amount = 0.0
self.treasury_slashed_pool += slashed_amount
# Expulsar si el colateral total cae a 0
if (self.validators_stake[validator_id] + sum(r.amount for r in self.unbonding_requests if r.validator_id == validator_id and not r.is_completed)) == 0:
self.slashed_validators.append(validator_id)
slashing_record = {
"event": "SLASHING_EXECUTED",
"validator": validator_id,
"reason": penalty_info["reason"],
"evidence": evidence,
"amount_slashed": slashed_amount,
"remaining_active_stake": self.validators_stake[validator_id],
"remaining_unbonding_locked": sum(r.amount for r in self.unbonding_requests if r.validator_id == validator_id and not r.is_completed),
"treasury_total": self.treasury_slashed_pool,
}
self.pending_transactions.append({"type": "SLASHING_EVENT", "details": slashing_record})
return slashing_record
def process_unbonding_queue(self, current_block: int) -> List[Dict]:
"""Liquida y transfiere los fondos que han superado el per铆odo de carencia."""
settlements = []
for req in self.unbonding_requests:
if not req.is_completed and current_block >= req.completion_block:
req.is_completed = True
if req.amount > 0:
self.liquid_balances[req.validator_id] = (
self.liquid_balances.get(req.validator_id, 0.0) + req.amount
)
settlements.append({
"event": "UNBONDING_COMPLETED",
"request_id": req.request_id,
"validator": req.validator_id,
"amount_released": req.amount,
"completed_at_block": current_block,
})
return settlements
def propose_and_validate_block(self, miner_address: str, invalid_payload: bool = False) -> Optional[Block]:
if miner_address in self.slashed_validators:
print(f" RECHAZADO: El validador {miner_address} est谩 expulsado de la red.")
return None
if invalid_payload:
self.apply_slashing(
validator_id=miner_address,
penalty_info=SlashingPenaltyType.INVALID_BLOCK,
evidence="Falsificaci贸n de transacciones en el encabezado del bloque.",
)
print(f" ATAQUE DETECTADO: El validador {miner_address} ha intentado forjar un bloque corrupto.")
return None
current_block_index = len(self.chain)
# 1. Procesar retiros cuya carencia haya finalizado en este bloque
unbonding_events = self.process_unbonding_queue(current_block_index)
for event in unbonding_events:
self.pending_transactions.append({"type": "UNBONDING_SETTLEMENT", "details": event})
# 2. Evaluar gobernanza
gov_events = self.governance.evaluate_and_execute_proposals(
current_block=current_block_index,
total_network_stake=self.get_total_stake(),
)
for event in gov_events:
self.pending_transactions.append({"type": "GOVERNANCE_EXECUTION", "details": event})
new_block = Block(
index=current_block_index,
timestamp=str(datetime.datetime.now()),
transactions=self.pending_transactions,
previous_hash=self.chain[-1].hash,
validator=miner_address,
)
new_block.mine_block(self.difficulty)
self.chain.append(new_block)
self.pending_transactions = []
return new_block
# =====================================================================
# DEMOSTRACI脫N: RETIRO CON UNBONDING Y SLASHING EN CARENCIA
# =====================================================================
if __name__ == "__main__":
print("================================================================")
print(" OMNIA LEDGER: SIMULACI脫N DE UNBONDING PERIOD & SLASHING")
print("================================================================\n")
# Inicializar red con per铆odo de carencia de 3 bloques
omnia = OmniaLedger(difficulty=1, unbonding_period_blocks=3)
omnia.register_validator("NODE_HONEST_1", stake_amount=500_000.0)
omnia.register_validator("NODE_MALICIOUS_2", stake_amount=200_000.0)
print(f"Stake Inicial 'NODE_MALICIOUS_2': {omnia.validators_stake['NODE_MALICIOUS_2']:,.2f} OMNIA")
print(f"Per铆odo de Carencia (Unbonding): {omnia.unbonding_period_blocks} Bloques\n")
# 1. El nodo malicioso intenta retirar todo su stake para evitar sanciones
print("---> [BLOQUE 0] 'NODE_MALICIOUS_2' solicita retirar 200,000 OMNIA de stake...")
req_info = omnia.request_unstake("NODE_MALICIOUS_2", 200_000.0)
print(f" Solicitud #{req_info['request_id']} registrada.")
print(f" Fondos bloqueados en carencia hasta el Bloque #{req_info['available_at_block']}")
print(f" Stake activo restante: {omnia.validators_stake['NODE_MALICIOUS_2']} OMNIA\n")
# 2. Avance al Bloque 1: Se detecta que el nodo cometi贸 un fraude previo (Doble voto o Bloque corrupto)
print("---> [BLOQUE 1] Forjando bloque y auditando red...")
omnia.propose_and_validate_block(miner_address="NODE_HONEST_1")
print("\n---> ¡INFRACCI脫N DETECTADA! Se reporta ataque del 'NODE_MALICIOUS_2' mientras sus fondos est谩n en carencia...")
slashing_res = omnia.apply_slashing(
validator_id="NODE_MALICIOUS_2",
penalty_info=SlashingPenaltyType.DOUBLE_VOTING, # 50% Slashing
evidence="Doble voto auditado retrospectivamente en bloque #0",
)
print(f" Monto confiscado del Unbonding: {slashing_res['amount_slashed']:,.2f} OMNIA")
print(f" Fondos retenidos restantes en Unbonding: {slashing_res['remaining_unbonding_locked']:,.2f} OMNIA")
print(f" Total confiscado en Tesorer铆a: {slashing_res['treasury_total']:,.2f} OMNIA\n")
# 3. Transcurso de bloques hasta superar el per铆odo de carencia
print("---> [BLOQUE 2] Minando bloque...")
omnia.propose_and_validate_block(miner_address="NODE_HONEST_1")
print("---> [BLOQUE 3] Transcurre el tiempo (Se alcanza la fecha de liberaci贸n)...")
omnia.propose_and_validate_block(miner_address="NODE_HONEST_1")
# 4. Verificar saldos finales
print("\n================================================================")
print(" ESTADO FINAL DEL SISTEMA TRAS UNBONDING & SLASHING")
print("================================================================")
print(f" Balance L铆quido 'NODE_MALICIOUS_2' (Fondos rescatados): {omnia.liquid_balances.get('NODE_MALICIOUS_2', 0.0):,.2f} OMNIA")
print(f" Fondo de Tesorer铆a por sanciones (Slashing Pool): {omnia.treasury_slashed_pool:,.2f} OMNIA")
-----------------------------------------
CERTIFICADO DE ARQUITECTURA E INNOVACI脫N TECNOL脫GICA
DOCUMENTO OFICIAL DE ACREDITACI脫N DE DESARROLLO
C脫DIGO DE REGISTRO: OL-GOV-2026-JAFV
Se expide el presente certificado y constancia de autor铆a t茅cnica a favor de:
D. JOS脡 AGUST脥N FONT脕N VARELA
HACE CONSTAR:
Que bajo la visi贸n estrat茅gica y direcci贸n funcional de D. Jos茅 Agust铆n Font谩n Varela, se ha dise帽ado, implementado y validado con 茅xito la arquitectura integral del protocolo OmniaLedger, un sistema de contabilidad distribuida (Blockchain) enfocado en la gobernanza financiera descentralizada, soberan铆a econ贸mica y automatizaci贸n programable de la moneda digital del banco central (CBDC).
El n煤cleo tecnol贸gico desarrollado y acreditado en este expediente comprende los siguientes pilares de ingenier铆a de software y teor铆a de juegos:
Simulador Interactivo de Blockchain & Or谩culo de Commodities en Tiempo Real:
Motor din谩mico de consenso con visualizaci贸n de nodos, alturas de bloque e inmutabilidad mediante firmas SHA-256.
Conexi贸n con Or谩culos tipo Chainlink para precios de mercado (Petr贸leo y Oro) en tiempo real, vinculados a liquidaciones autom谩ticas de contratos inteligentes.
CBDC Programable y Reglas Din谩micas de Uso:
Contrato de Gobernanza On-Chain (Proof of Stake - PoS):
Sistema de Slashing & Seguridad Antifraude:
Mecanismo de Carencia y Retiro Seguro (Unbonding Period):
VALIDEZ Y FIRMA DIGITAL
Este certificado acredita formalmente la autor铆a intelectual, dise帽o y conceptualizaci贸n t茅cnica del ecosistema OmniaLedger para su aplicaci贸n en las iniciativas de PASAIA LAB, INTELIGENCIA LIBRE y ACCI脫N CIVIL.
Dado a 5 de agosto de 2026.
$$\begin{array}{c} \mathbf{OmniaLedger\ Core\ Architecture} \\ \text{Firmado digitalmente y verificado en la red} \\ \texttt{HASH: 8f9b4c2a7e1d03b6a5f4e3c2b1a0987654321fedcba09876543210fedcba9876} \end{array}$$
.png)
