#!/usr/bin/env python3 """ 虫群经济系统 — 算力货币(Compute Credit) 核心逻辑： - 每个节点有CC账户 - 虫皇收税：每笔交易抽取税率 - 贡献算力→获得CC，使用算力→消费CC - 虫皇将税收分配给高贡献成员(提升权限) """ import hashlib import time from datetime import datetime from typing import Dict, List, Optional, Tuple from .economy_types import ( Account, Transaction, TransactionType, ResourceType, ComputePrice, PricingMode ) class ComputeCredit: """算力货币系统""" # 虫皇税率 TAX_RATE = 0.05 # 5%交易税 # 新用户赠送 SIGNUP_BONUS = 100.0 # 算力兑换率：1单位推理 = 1 CC UNIT_RATES = { ResourceType.INFERENCE: 1.0, ResourceType.TRAINING: 2.0, ResourceType.MEMORY_WRITE: 0.5, ResourceType.STORAGE: 0.1, } def __init__(self, overmind_id: str = "overmind"): self.overmind_id = overmind_id self.accounts: Dict[str, Account] = {} self.transactions: List[Transaction] = [] self.overmind_balance = 0.0 # 虫皇税收余额 self._tax_distributions: List[Dict] = [] # ============================================================ # 账户管理 # ============================================================ def create_account(self, node_id: str, initial_balance: float = None) -> Account: """创建账户""" if node_id in self.accounts: return self.accounts[node_id] acc = Account( node_id=node_id, balance=initial_balance if initial_balance is not None else self.SIGNUP_BONUS, ) # 默认定价 acc.pricing["inference"] = ComputePrice( resource_type=ResourceType.INFERENCE, mode=PricingMode.PER_TOKEN, price_per_unit=0.01, friend_discount=1.0, # 好友免费 circle_discount=1.0, # 圈子免费 min_price=0.001, ) acc.pricing["training"] = ComputePrice( resource_type=ResourceType.TRAINING, mode=PricingMode.PER_HOUR, price_per_unit=0.5, friend_discount=0.8, # 好友8折 circle_discount=0.9, # 圈子9折 min_price=0.1, ) self.accounts[node_id] = acc return acc def get_account(self, node_id: str) -> Optional[Account]: return self.accounts.get(node_id) def get_balance(self, node_id: str) -> float: acc = self.accounts.get(node_id) return acc.balance if acc else 0.0 # ============================================================ # 交易 # ============================================================ def transfer(self, from_id: str, to_id: str, amount: float, tx_type: TransactionType = TransactionType.TRADE, resource: ResourceType = ResourceType.INFERENCE, units: float = 0.0, desc: str = "") -> Optional[Transaction]: """ 转账（含税收） 1. 检查余额 2. 扣税（虫皇收走TAX_RATE比例） 3. 转账 4. 记录 """ from_acc = self.accounts.get(from_id) to_acc = self.accounts.get(to_id) if not from_acc or not to_acc: return None if not from_acc.can_afford(amount): return None if amount <= 0: return None # 计算税收 tax = amount * self.TAX_RATE net = amount - tax # 扣款 from_acc.balance -= amount from_acc.total_spent += amount if tx_type == TransactionType.TAX: from_acc.tax_paid += amount # 到账 to_acc.balance += net to_acc.total_earned += net # 虫皇收税 self.overmind_balance += tax # 记录 tx = Transaction( tx_id=hashlib.md5(f"{from_id}{to_id}{time.time()}".encode()).hexdigest()[:12], from_node=from_id, to_node=to_id, amount=amount, tx_type=tx_type, resource_type=resource, units=units, description=desc, ) self.transactions.append(tx) # 更新贡献分 to_acc.contribution_score += units * 0.1 from_acc.contribution_score += amount * 0.01 return tx def charge_usage(self, caller_id: str, provider_id: str, resource: ResourceType, units: float, is_friend: bool = False, is_circle: bool = False) -> Optional[Transaction]: """ 按量收费：调用者付费使用提供者的算力 根据提供者的定价和关系计算费用 """ provider_acc = self.accounts.get(provider_id) if not provider_acc: return None # 查找定价 price = provider_acc.pricing.get(resource.value) if not price: # 用默认价格 price = ComputePrice( resource_type=resource, mode=PricingMode.PER_TOKEN, price_per_unit=self.UNIT_RATES.get(resource, 1.0) * 0.01, ) # 计算费用 cost = price.calc_cost(units, is_friend=is_friend, is_circle=is_circle) if cost <= 0: # 免费（好友/圈子） return Transaction( tx_id=hashlib.md5(f"{caller_id}{provider_id}{time.time()}".encode()).hexdigest()[:12], from_node=caller_id, to_node=provider_id, amount=0.0, tx_type=TransactionType.GIFT, resource_type=resource, units=units, description="好友/圈子免费调用", ) return self.transfer(caller_id, provider_id, cost, tx_type=TransactionType.TRADE, resource=resource, units=units, desc=f"调用{provider_id}的{resource.value}算力") # ============================================================ # 虫皇税收分配 # ============================================================ def distribute_tax(self, top_contributors: Dict[str, float]) -> List[Transaction]: """ 虫皇分配税收给高贡献成员 top_contributors: {node_id: contribution_ratio} """ if self.overmind_balance <= 0 or not top_contributors: return [] total_ratio = sum(top_contributors.values()) if total_ratio <= 0: return [] distributions = [] for node_id, ratio in top_contributors.items(): amount = self.overmind_balance * (ratio / total_ratio) if amount < 0.01: continue acc = self.accounts.get(node_id) if not acc: continue acc.balance += amount acc.total_earned += amount tx = Transaction( tx_id=hashlib.md5(f"tax_{node_id}{time.time()}".encode()).hexdigest()[:12], from_node=self.overmind_id, to_node=node_id, amount=amount, tx_type=TransactionType.REWARD, description=f"虫皇税收分配(贡献分{ratio:.1f})", ) self.transactions.append(tx) distributions.append(tx) self._tax_distributions.append(tx.to_dict()) self.overmind_balance = 0.0 # 分配完毕 return distributions # ============================================================ # 定价设置 # ============================================================ def set_price(self, node_id: str, resource: ResourceType, price_per_unit: float, friend_discount: float = 1.0, circle_discount: float = 1.0, min_price: float = 0.001): """节点自定义定价""" acc = self.accounts.get(node_id) if not acc: return acc.pricing[resource.value] = ComputePrice( resource_type=resource, mode=PricingMode.PER_TOKEN, price_per_unit=price_per_unit, friend_discount=friend_discount, circle_discount=circle_discount, min_price=min_price, ) def set_reserved_sale(self, node_id: str, amount: float): """设定预留出售算力量""" acc = self.accounts.get(node_id) if acc: acc.reserved_for_sale = amount # ============================================================ # 查询 # ============================================================ def get_stats(self) -> Dict: total_balance = sum(a.balance for a in self.accounts.values()) return { "total_accounts": len(self.accounts), "total_balance": round(total_balance, 2), "total_transactions": len(self.transactions), "overmind_balance": round(self.overmind_balance, 2), "tax_distributions": len(self._tax_distributions), "tax_rate": self.TAX_RATE, } def get_leaderboard(self, top_n: int = 10) -> List[Dict]: """贡献排行榜""" sorted_accs = sorted(self.accounts.values(), key=lambda a: a.contribution_score, reverse=True) return [a.to_dict() for a in sorted_accs[:top_n]]