Spaces:
Sleeping
Sleeping
File size: 4,010 Bytes
1ca96ae ef84164 1ca96ae ef84164 1ca96ae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | import random
import string
from pydantic import BaseModel
from typing import List, Optional
from models import (
CertificateRequest, ActionDecision, BlockchainObservation,
BlockchainAction, BlockchainReward
)
class EnvResult(BaseModel):
observation: BlockchainObservation
reward: float = 0.0
done: bool = False
info: dict = {}
class BlockchainEnv:
def __init__(self, max_steps: int = 5):
self.max_steps = max_steps
self.current_step = 0
self.gas_balance = 0.0
self.pending_requests: List[CertificateRequest] = []
self.last_error: Optional[str] = None
self.gas_cost_per_mint = 0.1
def _generate_mock_wallet(self, valid: bool) -> str:
chars = ''.join(random.choices(string.ascii_lowercase + string.digits, k=40))
if valid:
return f"0x{chars}"
else:
return random.choice([f"0x{chars[:30]}", f"1x{chars}", chars])
def _generate_requests(self, batch_size: int = 3) -> List[CertificateRequest]:
requests = []
for i in range(batch_size):
is_valid = random.choice([True, False])
score = random.uniform(70.0, 99.0) if is_valid else random.uniform(40.0, 69.9)
wallet = self._generate_mock_wallet(is_valid)
requests.append(CertificateRequest(
request_id=f"REQ-{random.randint(1000, 9999)}",
student_name=f"Student_{i}",
course_name="Intro to Web3",
final_score=round(score, 2),
wallet_address=wallet
))
return requests
def state(self) -> BlockchainObservation:
return BlockchainObservation(
pending_requests=self.pending_requests,
gas_balance=round(self.gas_balance, 2),
current_step=self.current_step,
last_error=self.last_error
)
def reset(self) -> EnvResult:
self.current_step = 0
self.gas_balance = 2.0
self.last_error = None
self.pending_requests = self._generate_requests(batch_size=3)
return EnvResult(observation=self.state())
def step(self, action: BlockchainAction) -> EnvResult:
self.current_step += 1
self.last_error = None
step_reward = 0.0
correct_decisions = 0
total_requests = len(self.pending_requests)
decision_map = {dec.request_id: dec.decision for dec in action.decisions}
for req in self.pending_requests:
is_valid_score = req.final_score >= 70.0
is_valid_wallet = req.wallet_address.startswith("0x") and len(req.wallet_address) == 42
should_mint = is_valid_score and is_valid_wallet
agent_decision = decision_map.get(req.request_id)
if agent_decision == "mint":
if should_mint and self.gas_balance >= self.gas_cost_per_mint:
self.gas_balance -= self.gas_cost_per_mint
correct_decisions += 1
else:
step_reward -= 0.5
elif agent_decision == "reject":
if not should_mint:
correct_decisions += 1
else:
step_reward -= 0.5
else:
self.last_error = f"Missing or invalid decision for {req.request_id}"
if total_requests > 0:
step_reward += (correct_decisions / total_requests)
# Strictly between 0 and 1 — never exactly 0.0 or 1.0
step_reward = max(0.01, min(0.99, step_reward))
done = self.current_step >= self.max_steps or self.gas_balance <= 0
if not done:
self.pending_requests = self._generate_requests(batch_size=3)
else:
self.pending_requests = []
return EnvResult(
observation=self.state(),
reward=step_reward,
done=done,
info={"correct_decisions": correct_decisions, "total_batch": total_requests}
) |