# revenue_tracker.py import json import time import os import redis from datetime import datetime, timedelta from typing import Dict, Any, List class RevenueTracker: """ Tracks revenue, calculates ROI, and provides forecasting. """ def __init__(self, budget_manager, redis_config=None): self.namespace = "genesis:revenue" self.redis_client = None self.budget_manager = budget_manager # Inject BudgetManager # Load Redis config if not provided if redis_config is None: config_path = r"E:\genesis-system\genesis_config.json" if os.path.exists(config_path): with open(config_path, 'r') as f: redis_config = json.load(f).get("redis") if redis_config: self.redis_client = redis.Redis( host=redis_config["host"], port=redis_config["port"], password=redis_config["password"], ssl=redis_config.get("ssl", False), decode_responses=True ) def log_revenue_event(self, amount: float, agent: str, task: str, metadata: Dict[str, Any] = None): """Logs a revenue-generating event.""" if not self.redis_client: return now = datetime.now() event_id = f"{now.timestamp()}" event_key = f"{self.namespace}:events:{event_id}" event_data = { "timestamp": now.isoformat(), "amount": amount, "agent": agent, "task": task, "metadata": metadata or {} } self.redis_client.set(event_key, json.dumps(event_data)) self.redis_client.lpush(f"{self.namespace}:event_list", event_id) # Keep a list of event IDs self.redis_client.ltrim(f"{self.namespace}:event_list", 0, 1000) # Limit to 1000 events def attribute_cost(self, agent: str, task: str, cost: float): """Associates a cost with a specific agent and task.""" if not self.redis_client: return key = f"{self.namespace}:cost:{agent}:{task}" self.redis_client.incrbyfloat(key, cost) def get_cost(self, agent: str, task: str) -> float: """Retrieves the cost associated with a specific agent and task.""" if not self.redis_client: return 0.0 key = f"{self.namespace}:cost:{agent}:{task}" return float(self.redis_client.get(key) or 0.0) def calculate_roi(self, agent: str, task: str) -> float: """Calculates the ROI for a given agent and task.""" revenue = self.get_revenue(agent, task) cost = self.get_cost(agent, task) if cost == 0: return float('inf') if revenue > 0 else 0.0 # Avoid division by zero return (revenue - cost) / cost def get_revenue(self, agent: str, task: str) -> float: """Calculates total revenue for a specific agent and task.""" if not self.redis_client: return 0.0 total_revenue = 0.0 event_ids = self.redis_client.lrange(f"{self.namespace}:event_list", 0, -1) for event_id in event_ids: event_key = f"{self.namespace}:events:{event_id}" event_data_str = self.redis_client.get(event_key) if event_data_str: event_data = json.loads(event_data_str) if event_data["agent"] == agent and event_data["task"] == task: total_revenue += event_data["amount"] return total_revenue def identify_highest_value_actions(self, top_n: int = 5) -> List[Dict[str, Any]]: """Identifies the top N highest-ROI actions.""" if not self.redis_client: return [] roi_data = [] # This is inefficient; consider a better storage mechanism for ROI calculation event_ids = self.redis_client.lrange(f"{self.namespace}:event_list", 0, -1) agents_tasks = set() for event_id in event_ids: event_key = f"{self.namespace}:events:{event_id}" event_data_str = self.redis_client.get(event_key) if event_data_str: event_data = json.loads(event_data_str) agents_tasks.add((event_data["agent"], event_data["task"])) for agent, task in agents_tasks: roi = self.calculate_roi(agent, task) revenue = self.get_revenue(agent, task) cost = self.get_cost(agent, task) roi_data.append({"agent": agent, "task": task, "roi": roi, "revenue": revenue, "cost": cost}) sorted_roi_data = sorted(roi_data, key=lambda x: x["roi"], reverse=True) return sorted_roi_data[:top_n] def predict_revenue(self, agent: str, task: str, num_iterations: int = 10) -> float: """Predicts revenue based on historical data.""" if not self.redis_client: return 0.0 # Simple average revenue per iteration as a baseline event_ids = self.redis_client.lrange(f"{self.namespace}:event_list", 0, -1) relevant_events = [] for event_id in event_ids: event_key = f"{self.namespace}:events:{event_id}" event_data_str = self.redis_client.get(event_key) if event_data_str: event_data = json.loads(event_data_str) if event_data["agent"] == agent and event_data["task"] == task: relevant_events.append(event_data) if not relevant_events: return 0.0 total_revenue = sum(event["amount"] for event in relevant_events) average_revenue = total_revenue / len(relevant_events) return average_revenue * num_iterations def generate_revenue_report(self, start_date: datetime, end_date: datetime) -> List[Dict[str, Any]]: """Generates a revenue report for a given date range.""" if not self.redis_client: return [] report = [] event_ids = self.redis_client.lrange(f"{self.namespace}:event_list", 0, -1) for event_id in event_ids: event_key = f"{self.namespace}:events:{event_id}" event_data_str = self.redis_client.get(event_key) if event_data_str: event_data = json.loads(event_data_str) event_time = datetime.fromisoformat(event_data["timestamp"]) if start_date <= event_time <= end_date: report.append(event_data) return report def check_budget_impact(self, predicted_cost: float) -> bool: """Checks if a planned activity exceeds the budget.""" return self.budget_manager.is_within_budget() and (self.budget_manager.get_current_spend() + predicted_cost <= self.budget_manager.daily_limit) def send_alert(self, message: str): """Sends an alert (e.g., if revenue targets are not met).""" print(f"ALERT: {message}") # Replace with actual alerting mechanism def monitor_targets(self, agent: str, task: str, target_revenue: float): """Monitors revenue targets and sends alerts if not met.""" current_revenue = self.get_revenue(agent, task) if current_revenue < target_revenue: self.send_alert(f"Revenue target not met for {agent}/{task}. Current revenue: {current_revenue}, Target: {target_revenue}") if __name__ == "__main__": from budget_manager import BudgetManager # Assuming budget_manager.py is in the same directory # Initialize Budget Manager budget_manager = BudgetManager(daily_limit=100.0) # Initialize Revenue Tracker revenue_tracker = RevenueTracker(budget_manager) # Log some revenue events revenue_tracker.log_revenue_event(amount=25.0, agent="AgentA", task="Task1", metadata={"customer_id": "123"}) revenue_tracker.log_revenue_event(amount=30.0, agent="AgentB", task="Task2", metadata={"customer_id": "456"}) revenue_tracker.log_revenue_event(amount=15.0, agent="AgentA", task="Task1", metadata={"customer_id": "789"}) # Attribute some costs revenue_tracker.attribute_cost(agent="AgentA", task="Task1", cost=10.0) revenue_tracker.attribute_cost(agent="AgentB", task="Task2", cost=15.0) # Calculate ROI roi_agent_a = revenue_tracker.calculate_roi(agent="AgentA", task="Task1") print(f"ROI for AgentA/Task1: {roi_agent_a}") # Identify highest value actions top_actions = revenue_tracker.identify_highest_value_actions(top_n=3) print(f"Top Actions: {top_actions}") # Predict revenue predicted_revenue = revenue_tracker.predict_revenue(agent="AgentA", task="Task1", num_iterations=5) print(f"Predicted Revenue for AgentA/Task1: {predicted_revenue}") # Generate revenue report start_date = datetime.now() - timedelta(days=1) end_date = datetime.now() report = revenue_tracker.generate_revenue_report(start_date, end_date) print(f"Revenue Report: {report}") # Check budget impact predicted_cost = 20.0 within_budget = revenue_tracker.check_budget_impact(predicted_cost) print(f"Within Budget for predicted cost {predicted_cost}: {within_budget}") # Monitor targets revenue_tracker.monitor_targets(agent="AgentA", task="Task1", target_revenue=50.0)