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)
