economics.py
1 """ 2 Attention Economics - Bridge Between Attention System and Currency 3 4 Integrates the attention tracking system with Bitcoin-anchored economics. 5 Key responsibilities: 6 1. Track daily attention budget in sats 7 2. Allocate sats to sessions based on duration 8 3. Attribute value to work items via V-scores 9 4. Maintain cross-session economic state 10 11 Architecture: 12 - Reads config from ~/.sovereign/economics.yaml 13 - Maintains daily ledger at ~/.sovereign/attention-ledger/YYYY-MM-DD.yaml 14 - Integrates with CrossSessionTracker for session lifecycle 15 - Reports to FO-STATE.json for dashboard visibility 16 """ 17 18 from dataclasses import dataclass, field 19 from datetime import datetime, date, timedelta 20 from typing import Optional, Dict, List, Any, Tuple # Tuple for TrustMetrics.f_readings 21 from pathlib import Path 22 import json 23 24 # Optional yaml import with fallback 25 try: 26 import yaml 27 HAS_YAML = True 28 except ImportError: 29 HAS_YAML = False 30 31 32 def parse_yaml_simple(content: str) -> Dict[str, Any]: 33 """Simple YAML parser for economics.yaml (fallback when PyYAML not installed).""" 34 result = {} 35 current_section = None 36 current_subsection = None 37 38 for line in content.split('\n'): 39 stripped = line.strip() 40 if not stripped or stripped.startswith('#'): 41 continue 42 43 # Count indentation 44 indent = len(line) - len(line.lstrip()) 45 46 if ':' in stripped: 47 key, _, value = stripped.partition(':') 48 key = key.strip() 49 value = value.strip().strip('"').strip("'") 50 51 if indent == 0: 52 # Top-level section 53 if value: 54 result[key] = value 55 else: 56 result[key] = {} 57 current_section = key 58 current_subsection = None 59 elif indent == 2 and current_section: 60 # Second level 61 if value: 62 # Try to convert to number 63 try: 64 if '.' in value: 65 result[current_section][key] = float(value) 66 else: 67 result[current_section][key] = int(value) 68 except ValueError: 69 result[current_section][key] = value 70 else: 71 result[current_section][key] = {} 72 current_subsection = key 73 elif indent == 4 and current_section and current_subsection: 74 # Third level 75 if value: 76 try: 77 if '.' in value: 78 result[current_section][current_subsection][key] = float(value) 79 else: 80 result[current_section][current_subsection][key] = int(value) 81 except ValueError: 82 result[current_section][current_subsection][key] = value 83 84 return result 85 86 87 @dataclass 88 class TrustMetrics: 89 """ 90 Trust tracking - Trust = Alignment = Low Free Energy over time. 91 92 Trust is the integral of (1 - F) over time intervals. 93 High trust = consistently low F = compounding returns. 94 Trust destruction is non-ergodic (A4) - one failure can undo gains. 95 """ 96 # F readings over time (timestamp, F value) 97 f_readings: List[Tuple[datetime, float]] = field(default_factory=list) 98 99 # System health events (crashes, misses, successes) 100 events: List[Dict[str, Any]] = field(default_factory=list) 101 102 # Running trust score (accrues when F low, destroyed when F spikes) 103 trust_score: float = 1.0 # Start at 1.0 (full trust) 104 105 # Leverage multiplier (high trust = more leverage) 106 leverage_multiplier: float = 1.0 107 108 def record_f_reading(self, f_value: float, timestamp: datetime = None): 109 """Record an F reading. Trust accrues when F is low.""" 110 ts = timestamp or datetime.now() 111 self.f_readings.append((ts, f_value)) 112 113 # Trust accrual/destruction 114 # F < 0.10 = trust gain (+0.01 per reading) 115 # F 0.10-0.25 = neutral 116 # F > 0.25 = trust loss (-0.05 per reading) 117 # F > 0.50 = major trust loss (-0.15 per reading) 118 if f_value < 0.10: 119 self.trust_score = min(2.0, self.trust_score + 0.01) 120 elif f_value > 0.50: 121 self.trust_score = max(0.1, self.trust_score - 0.15) 122 elif f_value > 0.25: 123 self.trust_score = max(0.1, self.trust_score - 0.05) 124 125 # Update leverage based on trust 126 self._update_leverage() 127 128 def record_event(self, event_type: str, description: str, impact: float): 129 """ 130 Record a trust event. 131 event_type: 'success', 'crash', 'miss', 'save' 132 impact: positive = trust gain, negative = trust loss 133 """ 134 self.events.append({ 135 'type': event_type, 136 'description': description, 137 'impact': impact, 138 'timestamp': datetime.now().isoformat() 139 }) 140 self.trust_score = max(0.1, min(2.0, self.trust_score + impact)) 141 self._update_leverage() 142 143 def _update_leverage(self): 144 """Update leverage multiplier based on trust score.""" 145 # Trust < 0.5 = 0.5x leverage (need lots of verification) 146 # Trust 0.5-1.0 = 1x leverage (normal) 147 # Trust 1.0-1.5 = 2x leverage (can rely on system) 148 # Trust > 1.5 = 3x leverage (high confidence) 149 if self.trust_score < 0.5: 150 self.leverage_multiplier = 0.5 151 elif self.trust_score < 1.0: 152 self.leverage_multiplier = 1.0 153 elif self.trust_score < 1.5: 154 self.leverage_multiplier = 2.0 155 else: 156 self.leverage_multiplier = 3.0 157 158 @property 159 def trust_value_sats(self) -> int: 160 """ 161 Value of current trust in sats. 162 Trust enables leverage = more value per unit attention. 163 """ 164 # Base: 172,110 sats/hr × 8 hrs = 1,376,880 daily attention budget 165 # Trust value = how much MORE we can do with same attention 166 base_daily = 1_376_880 167 leverage_gain = self.leverage_multiplier - 1.0 # 0 if 1x, 1 if 2x, etc. 168 return int(base_daily * leverage_gain * self.trust_score) 169 170 171 @dataclass 172 class SessionEconomics: 173 """Economic state for a single session.""" 174 session_id: str 175 context: str # 'sovereign_os', 'deep_work', 'meetings', etc. 176 177 # Time tracking 178 started_at: datetime 179 ended_at: Optional[datetime] = None 180 duration_hours: float = 0.0 181 182 # Sat allocation 183 sats_allocated: int = 0 184 185 # V-score weighted value (sum of work item V-scores * time) 186 weighted_value: float = 0.0 187 188 # Work items with V-scores 189 work_items: List[Dict[str, Any]] = field(default_factory=list) 190 191 # Trust metrics for this session 192 trust: TrustMetrics = field(default_factory=TrustMetrics) 193 194 @property 195 def is_active(self) -> bool: 196 return self.ended_at is None 197 198 @property 199 def effective_hourly_rate(self) -> float: 200 """Effective sat rate based on V-weighted value.""" 201 if self.duration_hours == 0: 202 return 0.0 203 return self.sats_allocated / self.duration_hours 204 205 206 @dataclass 207 class DailyAttentionLedger: 208 """ 209 Daily budget and allocation of attention across sessions. 210 211 This is the bridge between time and sats - it tracks how the 212 daily attention budget is distributed across all activities. 213 """ 214 date: date 215 216 # Budget from economics.yaml 217 total_hours_budget: float = 8.0 218 sats_per_hour: int = 172110 219 total_sats_budget: int = 0 # = hours * sats_per_hour 220 221 # Actual usage 222 hours_used: float = 0.0 223 sats_used: int = 0 224 225 # Sessions for the day 226 sessions: Dict[str, SessionEconomics] = field(default_factory=dict) 227 228 # Non-session attention (general focus time) 229 ambient_hours: float = 0.0 230 ambient_sats: int = 0 231 232 # Created/updated timestamps 233 created_at: datetime = field(default_factory=datetime.now) 234 updated_at: datetime = field(default_factory=datetime.now) 235 236 @property 237 def remaining_hours(self) -> float: 238 return max(0, self.total_hours_budget - self.hours_used) 239 240 @property 241 def remaining_sats(self) -> int: 242 return max(0, self.total_sats_budget - self.sats_used) 243 244 @property 245 def utilization(self) -> float: 246 """Fraction of daily budget used.""" 247 if self.total_sats_budget == 0: 248 return 0.0 249 return self.sats_used / self.total_sats_budget 250 251 def to_dict(self) -> Dict[str, Any]: 252 """Convert to dictionary for YAML serialization.""" 253 return { 254 'date': self.date.isoformat(), 255 'budget': { 256 'total_hours': self.total_hours_budget, 257 'sats_per_hour': self.sats_per_hour, 258 'total_sats': self.total_sats_budget, 259 }, 260 'usage': { 261 'hours_used': round(self.hours_used, 2), 262 'sats_used': self.sats_used, 263 'remaining_hours': round(self.remaining_hours, 2), 264 'remaining_sats': self.remaining_sats, 265 'utilization': round(self.utilization, 3), 266 }, 267 'sessions': { 268 sid: { 269 'context': s.context, 270 'started_at': s.started_at.isoformat(), 271 'ended_at': s.ended_at.isoformat() if s.ended_at else None, 272 'duration_hours': round(s.duration_hours, 2), 273 'sats_allocated': s.sats_allocated, 274 'weighted_value': round(s.weighted_value, 2), 275 'work_items_count': len(s.work_items), 276 'work_items': s.work_items, # Include full work items for value attribution 277 'active': s.is_active, 278 } 279 for sid, s in self.sessions.items() 280 }, 281 'ambient': { 282 'hours': round(self.ambient_hours, 2), 283 'sats': self.ambient_sats, 284 }, 285 'timestamps': { 286 'created_at': self.created_at.isoformat(), 287 'updated_at': self.updated_at.isoformat(), 288 } 289 } 290 291 292 class AttentionEconomics: 293 """ 294 Bridge between attention tracking and Bitcoin-anchored economics. 295 296 This class: 297 1. Loads config from economics.yaml 298 2. Maintains daily attention ledger 299 3. Integrates with CrossSessionTracker 300 4. Reports economic state to FO-STATE 301 302 Usage: 303 economics = AttentionEconomics() 304 305 # Start a session 306 session = economics.start_session( 307 session_id="abc123", 308 context="sovereign_os" 309 ) 310 311 # Record work with V-score 312 economics.record_work_item( 313 session_id="abc123", 314 description="Built attention economics bridge", 315 v_score=0.75, 316 duration_minutes=30 317 ) 318 319 # End session 320 economics.end_session("abc123") 321 322 # Get daily summary 323 ledger = economics.get_today_ledger() 324 print(f"Used: {ledger.sats_used} sats ({ledger.utilization:.0%})") 325 """ 326 327 # Default paths 328 ECONOMICS_CONFIG = Path.home() / ".sovereign" / "economics.yaml" 329 LEDGER_DIR = Path.home() / ".sovereign" / "attention-ledger" 330 FO_STATE_PATH = Path.home() / "repos" / "Sovereign_OS" / "sessions" / "FO-STATE.json" 331 332 def __init__( 333 self, 334 economics_config: Path = None, 335 ledger_dir: Path = None 336 ): 337 self.economics_config = economics_config or self.ECONOMICS_CONFIG 338 self.ledger_dir = ledger_dir or self.LEDGER_DIR 339 340 # Ensure ledger directory exists 341 self.ledger_dir.mkdir(parents=True, exist_ok=True) 342 343 # Load config 344 self._config = self._load_config() 345 346 # Current day's ledger 347 self._current_ledger: Optional[DailyAttentionLedger] = None 348 349 def _load_config(self) -> Dict[str, Any]: 350 """Load economics configuration.""" 351 if not self.economics_config.exists(): 352 # Default config 353 return { 354 'operator': { 355 'hourly_rate_usd': 172.11, 356 'sats_per_hour': 172110, 357 'btc_price_usd': 100000, 358 }, 359 'budget': { 360 'default_hours_per_day': 8.0, 361 } 362 } 363 364 with open(self.economics_config) as f: 365 content = f.read() 366 367 if HAS_YAML: 368 return yaml.safe_load(content) 369 else: 370 return parse_yaml_simple(content) 371 372 def _get_ledger_path(self, for_date: date) -> Path: 373 """Get path to ledger file for a specific date.""" 374 return self.ledger_dir / f"{for_date.isoformat()}.yaml" 375 376 def get_today_ledger(self) -> DailyAttentionLedger: 377 """Get or create today's attention ledger.""" 378 today = date.today() 379 380 # Check if we have it cached 381 if self._current_ledger and self._current_ledger.date == today: 382 return self._current_ledger 383 384 # Try to load from file 385 ledger_path = self._get_ledger_path(today) 386 if ledger_path.exists(): 387 self._current_ledger = self._load_ledger(ledger_path) 388 else: 389 # Create new ledger 390 sats_per_hour = self._config.get('operator', {}).get('sats_per_hour', 172110) 391 hours_budget = self._config.get('budget', {}).get('default_hours_per_day', 8.0) 392 393 self._current_ledger = DailyAttentionLedger( 394 date=today, 395 total_hours_budget=hours_budget, 396 sats_per_hour=sats_per_hour, 397 total_sats_budget=int(hours_budget * sats_per_hour), 398 ) 399 self._save_ledger(self._current_ledger) 400 401 return self._current_ledger 402 403 def _load_ledger(self, path: Path) -> DailyAttentionLedger: 404 """Load ledger from YAML/JSON file.""" 405 with open(path) as f: 406 content = f.read() 407 408 # Try YAML first, then JSON 409 if HAS_YAML: 410 data = yaml.safe_load(content) 411 else: 412 # Try JSON (ledger files may be saved as JSON when yaml unavailable) 413 try: 414 data = json.loads(content) 415 except json.JSONDecodeError: 416 # Fallback to simple YAML parser 417 data = parse_yaml_simple(content) 418 419 ledger = DailyAttentionLedger( 420 date=date.fromisoformat(data['date']), 421 total_hours_budget=data['budget']['total_hours'], 422 sats_per_hour=data['budget']['sats_per_hour'], 423 total_sats_budget=data['budget']['total_sats'], 424 hours_used=data['usage']['hours_used'], 425 sats_used=data['usage']['sats_used'], 426 ambient_hours=data.get('ambient', {}).get('hours', 0), 427 ambient_sats=data.get('ambient', {}).get('sats', 0), 428 created_at=datetime.fromisoformat(data['timestamps']['created_at']), 429 updated_at=datetime.fromisoformat(data['timestamps']['updated_at']), 430 ) 431 432 # Reconstruct sessions 433 for sid, sdata in data.get('sessions', {}).items(): 434 session = SessionEconomics( 435 session_id=sid, 436 context=sdata['context'], 437 started_at=datetime.fromisoformat(sdata['started_at']), 438 ended_at=datetime.fromisoformat(sdata['ended_at']) if sdata.get('ended_at') else None, 439 duration_hours=sdata['duration_hours'], 440 sats_allocated=sdata['sats_allocated'], 441 weighted_value=sdata.get('weighted_value', 0), 442 ) 443 # Restore work items if present 444 session.work_items = sdata.get('work_items', []) 445 ledger.sessions[sid] = session 446 447 return ledger 448 449 def _save_ledger(self, ledger: DailyAttentionLedger) -> None: 450 """Save ledger to YAML/JSON file.""" 451 ledger.updated_at = datetime.now() 452 path = self._get_ledger_path(ledger.date) 453 454 with open(path, 'w') as f: 455 if HAS_YAML: 456 yaml.dump(ledger.to_dict(), f, default_flow_style=False, sort_keys=False) 457 else: 458 json.dump(ledger.to_dict(), f, indent=2) 459 460 def start_session( 461 self, 462 session_id: str, 463 context: str = "sovereign_os", 464 started_at: datetime = None 465 ) -> SessionEconomics: 466 """ 467 Start tracking a session for economic purposes. 468 469 Args: 470 session_id: Unique session identifier 471 context: Type of work ('sovereign_os', 'deep_work', 'meetings', etc.) 472 started_at: Optional explicit start time 473 474 Returns: 475 SessionEconomics object for tracking 476 """ 477 ledger = self.get_today_ledger() 478 479 session = SessionEconomics( 480 session_id=session_id, 481 context=context, 482 started_at=started_at or datetime.now(), 483 ) 484 485 ledger.sessions[session_id] = session 486 self._save_ledger(ledger) 487 488 return session 489 490 def end_session( 491 self, 492 session_id: str, 493 ended_at: datetime = None 494 ) -> Optional[SessionEconomics]: 495 """ 496 End a session and compute final economics. 497 498 Args: 499 session_id: Session to end 500 ended_at: Optional explicit end time 501 502 Returns: 503 Final SessionEconomics or None if session not found 504 """ 505 ledger = self.get_today_ledger() 506 507 session = ledger.sessions.get(session_id) 508 if not session: 509 return None 510 511 session.ended_at = ended_at or datetime.now() 512 513 # Compute duration 514 duration = (session.ended_at - session.started_at).total_seconds() / 3600 515 session.duration_hours = duration 516 517 # Allocate sats based on duration 518 session.sats_allocated = int(duration * ledger.sats_per_hour) 519 520 # Update ledger totals 521 ledger.hours_used = sum( 522 s.duration_hours for s in ledger.sessions.values() 523 if s.ended_at # Only count completed sessions 524 ) 525 ledger.sats_used = sum( 526 s.sats_allocated for s in ledger.sessions.values() 527 ) 528 529 self._save_ledger(ledger) 530 self._update_fo_state(ledger) 531 532 return session 533 534 def record_work_item( 535 self, 536 session_id: str, 537 description: str, 538 v_score: float, 539 duration_minutes: float = 0 540 ) -> None: 541 """ 542 Record a work item with V-score for value attribution. 543 544 Args: 545 session_id: Session this work belongs to 546 description: What was done 547 v_score: Value score (0-1) 548 duration_minutes: Approximate time spent on this item 549 """ 550 ledger = self.get_today_ledger() 551 session = ledger.sessions.get(session_id) 552 553 if not session: 554 return 555 556 work_item = { 557 'description': description, 558 'v_score': v_score, 559 'duration_minutes': duration_minutes, 560 'timestamp': datetime.now().isoformat(), 561 'sats_attributed': int( 562 (duration_minutes / 60) * ledger.sats_per_hour * v_score 563 ) 564 } 565 566 session.work_items.append(work_item) 567 568 # Update weighted value 569 session.weighted_value += v_score * (duration_minutes / 60) 570 571 self._save_ledger(ledger) 572 573 def update_active_session(self, session_id: str) -> Optional[SessionEconomics]: 574 """ 575 Update duration for an active (ongoing) session. 576 577 Call this periodically to keep the ledger current. 578 """ 579 ledger = self.get_today_ledger() 580 session = ledger.sessions.get(session_id) 581 582 if not session or not session.is_active: 583 return None 584 585 # Compute current duration 586 duration = (datetime.now() - session.started_at).total_seconds() / 3600 587 session.duration_hours = duration 588 session.sats_allocated = int(duration * ledger.sats_per_hour) 589 590 # Update ledger totals 591 ledger.hours_used = sum(s.duration_hours for s in ledger.sessions.values()) 592 ledger.sats_used = sum(s.sats_allocated for s in ledger.sessions.values()) 593 594 self._save_ledger(ledger) 595 596 return session 597 598 def get_daily_summary(self) -> Dict[str, Any]: 599 """ 600 Get a summary of today's attention economics. 601 602 Returns dict with budget, usage, sessions, etc. 603 """ 604 ledger = self.get_today_ledger() 605 606 return { 607 'date': ledger.date.isoformat(), 608 'budget': { 609 'hours': ledger.total_hours_budget, 610 'sats': ledger.total_sats_budget, 611 'usd': round(ledger.total_sats_budget / 100_000_000 * 100000, 2), 612 }, 613 'used': { 614 'hours': round(ledger.hours_used, 2), 615 'sats': ledger.sats_used, 616 'usd': round(ledger.sats_used / 100_000_000 * 100000, 2), 617 }, 618 'remaining': { 619 'hours': round(ledger.remaining_hours, 2), 620 'sats': ledger.remaining_sats, 621 'usd': round(ledger.remaining_sats / 100_000_000 * 100000, 2), 622 }, 623 'utilization': round(ledger.utilization, 3), 624 'sessions': { 625 sid: { 626 'context': s.context, 627 'hours': round(s.duration_hours, 2), 628 'sats': s.sats_allocated, 629 'active': s.is_active, 630 'v_weighted': round(s.weighted_value, 2), 631 } 632 for sid, s in ledger.sessions.items() 633 } 634 } 635 636 def get_value_report(self) -> Dict[str, Any]: 637 """ 638 Get value attribution report - what value was created vs cost. 639 640 TWO VALUE CONCEPTS: 641 642 1. V-Score (Attention Efficiency) 643 - What % of your invested attention converted to value 644 - V=0.75 means 75% of potential captured 645 - Formula: V × duration × sats_per_hour 646 647 2. Efficiency Multiplier (Free Energy Gain) 648 - Sovereign OS makes you 45-70% more efficient 649 - This multiplies your effective hourly rate 650 - Formula: base_rate × (1 + efficiency_gain) 651 - NEW VALUE = (effective_rate - base_rate) × hours 652 - This surplus is "ecosystem capital" - value that wouldn't 653 exist without the platform 654 655 Combined: High V-scores on work enabled by efficiency gains = 656 compounding value creation. 657 658 Returns comprehensive value report with work items ranked by value. 659 """ 660 ledger = self.get_today_ledger() 661 btc_price = self._config.get('operator', {}).get('btc_price_usd', 100000) 662 663 def sats_to_usd(sats: int) -> float: 664 return round(sats / 100_000_000 * btc_price, 2) 665 666 # Aggregate all work items across sessions 667 all_work_items = [] 668 total_value_sats = 0 669 total_cost_sats = 0 670 total_work_item_cost_sats = 0 # Cost derived from work item durations 671 672 for session in ledger.sessions.values(): 673 session_cost = session.sats_allocated 674 total_cost_sats += session_cost 675 676 for item in session.work_items: 677 item_value = item.get('sats_attributed', 0) 678 item_duration_hrs = item['duration_minutes'] / 60 679 item_cost = int(item_duration_hrs * ledger.sats_per_hour) 680 total_value_sats += item_value 681 total_work_item_cost_sats += item_cost 682 all_work_items.append({ 683 'session_id': session.session_id, 684 'description': item['description'], 685 'v_score': item['v_score'], 686 'duration_minutes': item['duration_minutes'], 687 'cost_sats': item_cost, 688 'value_sats': item_value, 689 'value_usd': sats_to_usd(item_value), 690 'timestamp': item.get('timestamp', ''), 691 }) 692 693 # Use work item costs if session costs aren't tracked 694 if total_cost_sats == 0 and total_work_item_cost_sats > 0: 695 total_cost_sats = total_work_item_cost_sats 696 697 # Sort by value (highest first) 698 all_work_items.sort(key=lambda x: x['value_sats'], reverse=True) 699 700 # Calculate aggregate metrics 701 avg_v = sum(w['v_score'] for w in all_work_items) / len(all_work_items) if all_work_items else 0 702 total_tracked_minutes = sum(w['duration_minutes'] for w in all_work_items) 703 704 # Untracked value (session time without recorded work items) 705 tracked_hours = total_tracked_minutes / 60 706 untracked_hours = max(0, ledger.hours_used - tracked_hours) 707 untracked_sats = int(untracked_hours * ledger.sats_per_hour) 708 709 return { 710 'date': ledger.date.isoformat(), 711 712 # Investment (what you spent) 713 'investment': { 714 'hours': round(ledger.hours_used, 2), 715 'sats': total_cost_sats, 716 'usd': sats_to_usd(total_cost_sats), 717 }, 718 719 # Value created (what you got) 720 'value_created': { 721 'tracked_sats': total_value_sats, 722 'tracked_usd': sats_to_usd(total_value_sats), 723 'untracked_hours': round(untracked_hours, 2), 724 'untracked_sats': untracked_sats, 725 }, 726 727 # ROI metrics 728 'roi': { 729 'average_v': round(avg_v, 2), 730 'efficiency': round(total_value_sats / total_cost_sats, 2) if total_cost_sats > 0 else 0, 731 'potential_captured': f"{avg_v * 100:.0f}%" if avg_v > 0 else "N/A", 732 }, 733 734 # Efficiency Multiplier (Free Energy Gain) 735 # Sovereign OS makes you 45-70% more efficient 736 # This creates NEW VALUE that wouldn't exist without the platform 737 'ecosystem_surplus': { 738 'efficiency_gain_low': 0.45, # Conservative 739 'efficiency_gain_mid': 0.60, # Middle estimate 740 'efficiency_gain_high': 0.70, # Aggressive 741 'base_cost_sats': total_cost_sats, 742 # NEW VALUE = base_cost × efficiency_gain 743 # (This is value that wouldn't exist without Sovereign OS) 744 'surplus_sats_low': int(total_cost_sats * 0.45), 745 'surplus_sats_mid': int(total_cost_sats * 0.60), 746 'surplus_sats_high': int(total_cost_sats * 0.70), 747 'surplus_usd_mid': sats_to_usd(int(total_cost_sats * 0.60)), 748 # Effective hourly rate with efficiency 749 'effective_rate_usd': round(ledger.sats_per_hour / 100_000_000 * btc_price * 1.60, 2), 750 }, 751 752 # Work items ranked by value 753 'top_value_items': all_work_items[:10], 754 755 # Low value items (might want to deprioritize similar work) 756 'low_value_items': [w for w in all_work_items if w['v_score'] < 0.4][-5:], 757 758 # Tracking coverage 759 'tracking': { 760 'tracked_hours': round(tracked_hours, 2), 761 'total_hours': round(ledger.hours_used, 2), 762 'coverage': round(tracked_hours / ledger.hours_used, 2) if ledger.hours_used > 0 else 0, 763 'work_items_count': len(all_work_items), 764 }, 765 766 # Platform Economics (Bill Gates Principle) 767 'platform_economics': self._compute_platform_economics(total_cost_sats, btc_price), 768 } 769 770 def _compute_platform_economics(self, base_cost_sats: int, btc_price: float) -> Dict[str, Any]: 771 """ 772 Compute platform economics metrics. 773 774 The "Bill Gates Principle": Platform should create more value than it extracts. 775 776 Key metrics: 777 - Ecosystem Surplus: New value created by platform 778 - Capture Rate: What fraction platform takes 779 - Value Creation Ratio (VCR): Total value created / platform cost 780 """ 781 # Efficiency multiplier (how much more productive with Sovereign OS) 782 efficiency_multiplier = 1.60 # 60% more productive 783 784 # Ecosystem surplus = base_cost × (E - 1) 785 surplus_sats = int(base_cost_sats * (efficiency_multiplier - 1)) 786 787 # Platform costs (Claude subscription amortized daily) 788 # $200/month ≈ $6.67/day ≈ 6,667 sats at $100k BTC 789 claude_monthly_usd = 200 790 claude_daily_usd = claude_monthly_usd / 30 791 claude_daily_sats = int(claude_daily_usd / btc_price * 100_000_000) 792 793 # Infrastructure (estimated) 794 infra_sats = 1000 # Minimal for now 795 796 platform_cost_sats = claude_daily_sats + infra_sats 797 798 # Platform capture (default 10% of surplus) 799 capture_rate = 0.10 800 platform_share_sats = int(surplus_sats * capture_rate) 801 operator_share_sats = surplus_sats - platform_share_sats 802 803 # Value Creation Ratio (VCR) = Surplus / Platform Cost 804 # This is the "Bill Gates test" - should be > 5 805 vcr = surplus_sats / platform_cost_sats if platform_cost_sats > 0 else 0 806 807 # Bill Gates test: Does platform create more value than it costs? 808 bill_gates_test = vcr > 5 809 810 def sats_to_usd(sats: int) -> float: 811 return round(sats / 100_000_000 * btc_price, 2) 812 813 return { 814 'efficiency_multiplier': efficiency_multiplier, 815 816 'ecosystem_surplus': { 817 'total_sats': surplus_sats, 818 'total_usd': sats_to_usd(surplus_sats), 819 'operator_share_sats': operator_share_sats, 820 'operator_share_usd': sats_to_usd(operator_share_sats), 821 'platform_share_sats': platform_share_sats, 822 'platform_share_usd': sats_to_usd(platform_share_sats), 823 }, 824 825 'platform_costs': { 826 'claude_sats': claude_daily_sats, 827 'infra_sats': infra_sats, 828 'total_sats': platform_cost_sats, 829 'total_usd': sats_to_usd(platform_cost_sats), 830 }, 831 832 'metrics': { 833 'capture_rate': capture_rate, 834 'value_creation_ratio': round(vcr, 1), 835 'bill_gates_test': bill_gates_test, 836 'test_result': 'PASS' if bill_gates_test else 'FAIL', 837 }, 838 839 'summary': f"Platform creates {vcr:.0f}x the value it costs (target: >5x)", 840 } 841 842 def format_value_report(self) -> str: 843 """ 844 Format value report as human-readable string. 845 846 Suitable for terminal output or inclusion in reports. 847 """ 848 report = self.get_value_report() 849 850 lines = [ 851 "", 852 "╔══════════════════════════════════════════════════════════════════╗", 853 "║ VALUE ATTRIBUTION REPORT ║", 854 "║ V-Score → Sat Value ║", 855 "╚══════════════════════════════════════════════════════════════════╝", 856 "", 857 f"Date: {report['date']}", 858 "", 859 "┌─────────────────────────────────────────────────────────────────┐", 860 "│ INVESTMENT (Attention Spent) │", 861 "├─────────────────────────────────────────────────────────────────┤", 862 f"│ Time: {report['investment']['hours']:.2f} hours", 863 f"│ Cost: {report['investment']['sats']:,} sats (${report['investment']['usd']})", 864 "└─────────────────────────────────────────────────────────────────┘", 865 "", 866 "┌─────────────────────────────────────────────────────────────────┐", 867 "│ VALUE CREATED (V-weighted) │", 868 "├─────────────────────────────────────────────────────────────────┤", 869 f"│ Tracked: {report['value_created']['tracked_sats']:,} sats (${report['value_created']['tracked_usd']})", 870 f"│ Untracked: {report['value_created']['untracked_hours']:.2f}hr = {report['value_created']['untracked_sats']:,} sats (no V-score)", 871 "└─────────────────────────────────────────────────────────────────┘", 872 "", 873 "┌─────────────────────────────────────────────────────────────────┐", 874 "│ ROI METRICS │", 875 "├─────────────────────────────────────────────────────────────────┤", 876 f"│ Average V: {report['roi']['average_v']:.2f}", 877 f"│ Potential: {report['roi']['potential_captured']} of invested attention → value", 878 f"│ Coverage: {report['tracking']['coverage']:.0%} of session time has V-scores", 879 "└─────────────────────────────────────────────────────────────────┘", 880 ] 881 882 # Ecosystem Surplus (Free Energy Gain) 883 surplus = report.get('ecosystem_surplus', {}) 884 if surplus and surplus.get('base_cost_sats', 0) > 0: 885 lines.extend([ 886 "", 887 "┌─────────────────────────────────────────────────────────────────┐", 888 "│ ECOSYSTEM SURPLUS (Free Energy Gain) │", 889 "│ Value that wouldn't exist without Sovereign OS │", 890 "├─────────────────────────────────────────────────────────────────┤", 891 f"│ Efficiency Gain: 45-70% (mid: 60%)", 892 f"│ Base Investment: {surplus['base_cost_sats']:,} sats", 893 f"│ NEW VALUE CREATED: {surplus['surplus_sats_mid']:,} sats (${surplus['surplus_usd_mid']})", 894 f"│ Effective Rate: ${surplus['effective_rate_usd']}/hr (vs $172.11 base)", 895 "├─────────────────────────────────────────────────────────────────┤", 896 "│ This surplus is 'ecosystem capital' - platform value creation │", 897 "│ that compounds across the network (see: investment-thesis.md) │", 898 "└─────────────────────────────────────────────────────────────────┘", 899 ]) 900 901 # Platform Economics (Bill Gates Principle) 902 platform = report.get('platform_economics', {}) 903 if platform and platform.get('metrics', {}).get('value_creation_ratio', 0) > 0: 904 metrics = platform['metrics'] 905 surplus = platform['ecosystem_surplus'] 906 costs = platform['platform_costs'] 907 908 test_icon = "✓" if metrics['bill_gates_test'] else "✗" 909 lines.extend([ 910 "", 911 "┌─────────────────────────────────────────────────────────────────┐", 912 "│ PLATFORM ECONOMICS (Bill Gates Principle) │", 913 "│ 'Platform should create more value than it extracts' │", 914 "├─────────────────────────────────────────────────────────────────┤", 915 f"│ Efficiency: {platform['efficiency_multiplier']:.0%} (you're {platform['efficiency_multiplier']:.2f}x more productive)", 916 f"│ Surplus Created: {surplus['total_sats']:,} sats (${surplus['total_usd']})", 917 f"│ Operator gets: {surplus['operator_share_sats']:,} sats (90%)", 918 f"│ Platform gets: {surplus['platform_share_sats']:,} sats (10%)", 919 f"│ Platform Cost: {costs['total_sats']:,} sats (${costs['total_usd']}/day)", 920 "├─────────────────────────────────────────────────────────────────┤", 921 f"│ {test_icon} BILL GATES TEST: VCR = {metrics['value_creation_ratio']:.0f}x (target: >5x)", 922 f"│ {platform['summary']}", 923 "└─────────────────────────────────────────────────────────────────┘", 924 ]) 925 926 # Top value items 927 if report['top_value_items']: 928 lines.extend([ 929 "", 930 "┌─────────────────────────────────────────────────────────────────┐", 931 "│ TOP VALUE ITEMS │", 932 "├─────────────────────────────────────────────────────────────────┤", 933 ]) 934 for i, item in enumerate(report['top_value_items'][:5], 1): 935 desc = item['description'][:40] + "..." if len(item['description']) > 40 else item['description'] 936 lines.append(f"│ {i}. V={item['v_score']:.2f} | {item['value_sats']:,} sats | {desc}") 937 lines.append("└─────────────────────────────────────────────────────────────────┘") 938 939 # Low value items (warning) 940 if report['low_value_items']: 941 lines.extend([ 942 "", 943 "┌─────────────────────────────────────────────────────────────────┐", 944 "│ ⚠️ LOW VALUE ITEMS (V < 0.40) │", 945 "├─────────────────────────────────────────────────────────────────┤", 946 ]) 947 for item in report['low_value_items']: 948 desc = item['description'][:40] + "..." if len(item['description']) > 40 else item['description'] 949 lines.append(f"│ V={item['v_score']:.2f} | {desc}") 950 lines.append("│ Consider: deprioritize similar work next time │") 951 lines.append("└─────────────────────────────────────────────────────────────────┘") 952 953 lines.append("") 954 return "\n".join(lines) 955 956 def _update_fo_state(self, ledger: DailyAttentionLedger) -> None: 957 """Update FO-STATE.json with attention economics.""" 958 if not self.FO_STATE_PATH.exists(): 959 return 960 961 try: 962 with open(self.FO_STATE_PATH) as f: 963 state = json.load(f) 964 965 state['attention_economics'] = { 966 'date': ledger.date.isoformat(), 967 'budget_sats': ledger.total_sats_budget, 968 'used_sats': ledger.sats_used, 969 'remaining_sats': ledger.remaining_sats, 970 'utilization': round(ledger.utilization, 3), 971 'sessions_count': len(ledger.sessions), 972 'active_sessions': sum(1 for s in ledger.sessions.values() if s.is_active), 973 'updated_at': datetime.now().isoformat(), 974 } 975 976 with open(self.FO_STATE_PATH, 'w') as f: 977 json.dump(state, f, indent=2) 978 979 except Exception as e: 980 # Don't fail if FO-STATE update fails 981 pass 982 983 984 # Integration with CrossSessionTracker 985 def wire_to_cross_session_tracker( 986 economics: AttentionEconomics, 987 tracker: 'CrossSessionTracker' 988 ) -> None: 989 """ 990 Wire AttentionEconomics to an existing CrossSessionTracker. 991 992 This integrates the economic layer with session lifecycle events. 993 """ 994 original_register = tracker.register_session 995 996 def register_with_economics(session_id: str, source: str = "unknown", started_at: datetime = None): 997 # Call original registration 998 session_info = original_register(session_id, source, started_at) 999 1000 # Start economic tracking 1001 economics.start_session( 1002 session_id=session_id, 1003 context=source, 1004 started_at=started_at or datetime.now() 1005 ) 1006 1007 return session_info 1008 1009 tracker.register_session = register_with_economics 1010 1011 1012 if __name__ == "__main__": 1013 print("=== Attention Economics Test ===\n") 1014 1015 economics = AttentionEconomics() 1016 1017 # Simulate a day's sessions 1018 print("Starting session 1 (sovereign_os)...") 1019 s1 = economics.start_session("session_001", context="sovereign_os") 1020 1021 # Record some work 1022 economics.record_work_item( 1023 "session_001", 1024 description="Built attention economics bridge", 1025 v_score=0.75, 1026 duration_minutes=45 1027 ) 1028 1029 economics.record_work_item( 1030 "session_001", 1031 description="Updated CLAUDE.md protocol", 1032 v_score=0.60, 1033 duration_minutes=15 1034 ) 1035 1036 # End session 1037 import time 1038 time.sleep(1) # Simulate some time passing 1039 economics.end_session("session_001") 1040 1041 # Get summary 1042 summary = economics.get_daily_summary() 1043 1044 print(f"\nDaily Summary:") 1045 print(f" Budget: {summary['budget']['sats']:,} sats (${summary['budget']['usd']})") 1046 print(f" Used: {summary['used']['sats']:,} sats (${summary['used']['usd']})") 1047 print(f" Remaining: {summary['remaining']['sats']:,} sats") 1048 print(f" Utilization: {summary['utilization']:.1%}") 1049 1050 print(f"\nSessions:") 1051 for sid, sdata in summary['sessions'].items(): 1052 print(f" {sid}: {sdata['hours']:.2f}hr = {sdata['sats']:,} sats (V={sdata['v_weighted']:.2f})") 1053 1054 print("\n=== Test Complete ===")