#!/usr/bin/env python3 """ Genesis Comprehensive Cost Tracking System Real-time monitoring with Gemini vs Claude comparison and parallel agent breakdown """ import json import time import sqlite3 from datetime import datetime, timedelta from pathlib import Path from typing import Dict, List, Any import threading import requests class GenesisCostTracker: def __init__(self, db_path="data/genesis-costs.db"): self.db_path = Path(db_path) self.db_path.parent.mkdir(parents=True, exist_ok=True) # Pricing rates (per 1K tokens) self.rates = { 'gemini_2_0_flash': 0.002, # $0.002 per 1K tokens 'claude_sonnet': 0.015, # $15 per 1M = $0.015 per 1K 'claude_haiku': 0.0025, # $0.25 per 1M = $0.00025 per 1K 'claude_opus': 0.075 # $75 per 1M = $0.075 per 1K } # Agent tracking self.agents = { 'genesis-memory-1': {'range': '1-73', 'focus': 'ChromaDB semantic'}, 'genesis-memory-2': {'range': '74-146', 'focus': 'Pinecone vectors'}, 'genesis-memory-3': {'range': '147-219', 'focus': 'Qdrant knowledge'}, 'genesis-memory-4': {'range': '220-292', 'focus': 'Mem0 compression'}, 'genesis-memory-5': {'range': '293-365', 'focus': 'Redis coordination'} } self.init_database() def init_database(self): """Initialize SQLite database for cost tracking""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Create tables cursor.execute(''' CREATE TABLE IF NOT EXISTS cost_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, agent_id TEXT NOT NULL, operation TEXT NOT NULL, input_tokens INTEGER NOT NULL, output_tokens INTEGER NOT NULL, total_tokens INTEGER NOT NULL, gemini_cost REAL NOT NULL, claude_equivalent_cost REAL NOT NULL, savings REAL NOT NULL, model_used TEXT NOT NULL, session_id TEXT ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS hourly_reports ( id INTEGER PRIMARY KEY AUTOINCREMENT, report_timestamp TEXT NOT NULL, total_tokens INTEGER NOT NULL, total_gemini_cost REAL NOT NULL, total_claude_cost REAL NOT NULL, total_savings REAL NOT NULL, agents_active INTEGER NOT NULL, operations_completed INTEGER NOT NULL, average_response_time REAL, report_data TEXT NOT NULL ) ''') conn.commit() conn.close() def log_cost_event(self, agent_id: str, operation: str, input_tokens: int, output_tokens: int, model_used: str = "gemini-2.0-flash-exp", session_id: str = None) -> Dict[str, Any]: """Log a cost event for tracking""" total_tokens = input_tokens + output_tokens gemini_cost = (total_tokens / 1000) * self.rates['gemini_2_0_flash'] claude_cost = (total_tokens / 1000) * self.rates['claude_sonnet'] savings = claude_cost - gemini_cost savings_pct = (savings / claude_cost * 100) if claude_cost > 0 else 0 event = { 'timestamp': datetime.now().isoformat(), 'agent_id': agent_id, 'operation': operation, 'input_tokens': input_tokens, 'output_tokens': output_tokens, 'total_tokens': total_tokens, 'gemini_cost': round(gemini_cost, 6), 'claude_equivalent_cost': round(claude_cost, 4), 'savings': round(savings, 4), 'savings_percentage': round(savings_pct, 2), 'model_used': model_used, 'session_id': session_id or f"session_{int(time.time())}" } # Store in database conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT INTO cost_events (timestamp, agent_id, operation, input_tokens, output_tokens, total_tokens, gemini_cost, claude_equivalent_cost, savings, model_used, session_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', (event['timestamp'], agent_id, operation, input_tokens, output_tokens, total_tokens, gemini_cost, claude_cost, savings, model_used, session_id)) conn.commit() conn.close() # Real-time console output print(f"๐Ÿ’ฐ Cost Event: {agent_id} | {operation}") print(f" Tokens: {total_tokens:,} | Gemini: ${gemini_cost:.4f} | Savings: ${savings:.2f} ({savings_pct:.1f}%)") return event def get_session_summary(self, time_window_hours: int = 24) -> Dict[str, Any]: """Get comprehensive session summary""" since_time = (datetime.now() - timedelta(hours=time_window_hours)).isoformat() conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Get summary data cursor.execute(''' SELECT COUNT(*) as events, SUM(total_tokens) as total_tokens, SUM(gemini_cost) as total_gemini_cost, SUM(claude_equivalent_cost) as total_claude_cost, SUM(savings) as total_savings, COUNT(DISTINCT agent_id) as active_agents, AVG(total_tokens) as avg_tokens_per_event FROM cost_events WHERE timestamp > ? ''', (since_time,)) summary_data = cursor.fetchone() # Get per-agent breakdown cursor.execute(''' SELECT agent_id, COUNT(*) as events, SUM(total_tokens) as tokens, SUM(gemini_cost) as cost, SUM(savings) as savings FROM cost_events WHERE timestamp > ? GROUP BY agent_id ORDER BY tokens DESC ''', (since_time,)) agent_breakdown = cursor.fetchall() conn.close() if summary_data[0] == 0: # No events return { 'time_window_hours': time_window_hours, 'total_events': 0, 'total_tokens': 0, 'total_cost': 0, 'total_savings': 0, 'savings_percentage': 0, 'active_agents': 0, 'agent_breakdown': [] } total_claude_cost = summary_data[3] or 0 total_savings = summary_data[4] or 0 savings_pct = (total_savings / total_claude_cost * 100) if total_claude_cost > 0 else 0 return { 'time_window_hours': time_window_hours, 'total_events': summary_data[0], 'total_tokens': summary_data[1] or 0, 'total_cost': round(summary_data[2] or 0, 4), 'total_claude_equivalent': round(total_claude_cost, 2), 'total_savings': round(total_savings, 2), 'savings_percentage': round(savings_pct, 2), 'active_agents': summary_data[5] or 0, 'avg_tokens_per_event': round(summary_data[6] or 0, 1), 'agent_breakdown': [ { 'agent_id': row[0], 'events': row[1], 'tokens': row[2], 'cost': round(row[3], 4), 'savings': round(row[4], 2), 'focus': self.agents.get(row[0], {}).get('focus', 'Unknown') } for row in agent_breakdown ] } # Real-time monitoring example if __name__ == "__main__": tracker = GenesisCostTracker() # Simulate some events for testing test_events = [ ('genesis-memory-1', 'conversation_extraction', 1500, 800), ('genesis-memory-2', 'vector_optimization', 2000, 1200), ('genesis-memory-3', 'knowledge_synthesis', 1800, 900), ('genesis-memory-4', 'intelligent_compression', 2200, 600), ('genesis-memory-5', 'realtime_coordination', 1000, 400) ] print("๐Ÿงช Testing cost tracking with sample events...") for agent_id, operation, input_tokens, output_tokens in test_events: tracker.log_cost_event(agent_id, operation, input_tokens, output_tokens) time.sleep(0.5) print("\n๐Ÿ“Š Session Summary:") summary = tracker.get_session_summary() print(f"Total Cost: ${summary['total_cost']} (Gemini)") print(f"Claude Equivalent: ${summary['total_claude_equivalent']}") print(f"Savings: ${summary['total_savings']} ({summary['savings_percentage']}%)") print(f"Active Agents: {summary['active_agents']}")