# revenue_tracker.py import json import time import os import redis from datetime import datetime, timedelta from typing import Dict, Any class RevenueTracker: """ Tracks revenue-generating activities, calculates ROI, and provides revenue forecasting. """ def __init__(self, redis_config_path: str = r"E:\genesis-system\genesis_config.json", budget_manager=None): self.namespace = "genesis:revenue" self.redis_client = None self.budget_manager = budget_manager # Load Redis config self.redis_config_path = redis_config_path self._load_redis_config() def _load_redis_config(self): if os.path.exists(self.redis_config_path): with open(self.redis_config_path, 'r') as f: config = json.load(f).get("redis") if config: self.redis_client = redis.Redis( host=config["host"], port=config["port"], password=config["password"], ssl=config.get("ssl", False), decode_responses=True ) else: print(f"Warning: Redis config not found at {self.redis_config_path}. Revenue tracking will be disabled.") def log_revenue_event(self, amount: float, agent: str, task: str, cost: float = 0.0, metadata: Dict[str, Any] = None): """Logs a revenue-generating event with associated agent, task, and cost.""" if not self.redis_client: return now = datetime.now() event_id = f"{self.namespace}:event:{now.strftime('%Y%m%d%H%M%S%f')}" event_data = { "timestamp": now.isoformat(), "amount": amount, "agent": agent, "task": task, "cost": cost, "profit": amount - cost, "metadata": metadata or {} } self.redis_client.set(event_id, json.dumps(event_data)) self.redis_client.lpush(f"{self.namespace}:events", event_id) self.redis_client.ltrim(f"{self.namespace}:events", 0, 10000) # Keep last 10000 events def calculate_roi(self, agent: str = None, task: str = None, start_date: datetime = None, end_date: datetime = None) -> float: """Calculates the ROI for a given agent, task, or overall (if both are None) within a date range.""" if not self.redis_client: return 0.0 total_revenue = 0.0 total_cost = 0.0 event_ids = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) for event_id in event_ids: event_data = json.loads(self.redis_client.get(event_id)) event_time = datetime.fromisoformat(event_data["timestamp"]) if start_date and event_time < start_date: continue if end_date and event_time > end_date: continue if agent and event_data["agent"] != agent: continue if task and event_data["task"] != task: continue total_revenue += event_data["amount"] total_cost += event_data["cost"] if total_cost == 0.0: return 0.0 if total_revenue == 0.0 else float('inf') # Avoid division by zero and handle cases with revenue but no cost. return (total_revenue - total_cost) / total_cost def identify_highest_value_actions(self, top_n: int = 5, start_date: datetime = None, end_date: datetime = None) -> list: """Identifies the top N highest-value actions (task/agent combinations) based on profit.""" if not self.redis_client: return [] action_profits = {} event_ids = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) for event_id in event_ids: event_data = json.loads(self.redis_client.get(event_id)) event_time = datetime.fromisoformat(event_data["timestamp"]) if start_date and event_time < start_date: continue if end_date and event_time > end_date: continue action_key = (event_data["agent"], event_data["task"]) profit = event_data["profit"] if action_key in action_profits: action_profits[action_key] += profit else: action_profits[action_key] = profit sorted_actions = sorted(action_profits.items(), key=lambda item: item[1], reverse=True) return sorted_actions[:top_n] def predict_revenue(self, planned_activities: list, historical_data_days: int = 30) -> float: """Predicts revenue from planned activities based on historical performance.""" if not self.redis_client: return 0.0 total_predicted_revenue = 0.0 end_date = datetime.now() start_date = end_date - timedelta(days=historical_data_days) for activity in planned_activities: agent = activity.get("agent") task = activity.get("task") estimated_cost = activity.get("cost", 0.0) if not agent or not task: print(f"Warning: Agent or task missing for planned activity: {activity}") continue historical_roi = self.calculate_roi(agent=agent, task=task, start_date=start_date, end_date=end_date) predicted_profit = estimated_cost * historical_roi total_predicted_revenue += (predicted_profit + estimated_cost) # Revenue = Profit + Cost return total_predicted_revenue def generate_revenue_report(self, start_date: datetime = None, end_date: datetime = None) -> Dict[str, float]: """Generates a revenue report for a given date range.""" if not self.redis_client: return {} total_revenue = 0.0 total_cost = 0.0 agent_revenue = {} task_revenue = {} event_ids = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) for event_id in event_ids: event_data = json.loads(self.redis_client.get(event_id)) event_time = datetime.fromisoformat(event_data["timestamp"]) if start_date and event_time < start_date: continue if end_date and event_time > end_date: continue revenue = event_data["amount"] cost = event_data["cost"] agent = event_data["agent"] task = event_data["task"] total_revenue += revenue total_cost += cost agent_revenue[agent] = agent_revenue.get(agent, 0.0) + revenue task_revenue[task] = task_revenue.get(task, 0.0) + revenue return { "total_revenue": total_revenue, "total_cost": total_cost, "profit": total_revenue - total_cost, "agent_revenue": agent_revenue, "task_revenue": task_revenue } def check_revenue_targets(self, target_revenue: float, alert_threshold: float = 0.9) -> bool: """Checks if revenue targets are being met and triggers an alert if below threshold.""" if not self.redis_client: return False report = self.generate_revenue_report() current_revenue = report.get("total_revenue", 0.0) if current_revenue < target_revenue * alert_threshold: print(f"Alert: Revenue target not being met! Current revenue: ${current_revenue}, Target: ${target_revenue}") return True # Alert triggered return False def is_within_budget(self, cost: float) -> bool: """Checks if adding the cost to the budget is feasible""" if not self.budget_manager: print("Warning: Budget Manager not integrated. Skipping budget check.") return True if not self.budget_manager.is_within_budget(): print("Budget limit has been reached.") return False return True def log_cost_and_check_budget(self, amount: float, model: str): """Log cost in budget manager if integrated""" if self.budget_manager: self.budget_manager.log_cost(amount, model) if __name__ == "__main__": # Example Usage from budget_manager import BudgetManager # Assuming budget_manager.py is in the same directory # Initialize BudgetManager (optional, for integration) budget_manager = BudgetManager() # Initialize RevenueTracker with BudgetManager revenue_tracker = RevenueTracker(budget_manager=budget_manager) # Log a revenue event revenue_tracker.log_revenue_event(amount=50.0, agent="AgentA", task="Task1", cost=10.0, metadata={"customer_id": "123"}) revenue_tracker.log_revenue_event(amount=75.0, agent="AgentB", task="Task2", cost=15.0, metadata={"customer_id": "456"}) revenue_tracker.log_revenue_event(amount=25.0, agent="AgentA", task="Task2", cost=5.0, metadata={"customer_id": "789"}) # Calculate ROI for AgentA roi_agent_a = revenue_tracker.calculate_roi(agent="AgentA") print(f"ROI for AgentA: {roi_agent_a}") # Identify highest value actions top_actions = revenue_tracker.identify_highest_value_actions(top_n=2) print(f"Top 2 Highest Value Actions: {top_actions}") # Predict revenue from planned activities planned_activities = [ {"agent": "AgentA", "task": "Task1", "cost": 8.0}, {"agent": "AgentB", "task": "Task2", "cost": 12.0} ] predicted_revenue = revenue_tracker.predict_revenue(planned_activities) print(f"Predicted Revenue from Planned Activities: ${predicted_revenue}") # Generate revenue report report = revenue_tracker.generate_revenue_report() print("Revenue Report:") print(report) # Check revenue targets target_revenue = 100.0 revenue_tracker.check_revenue_targets(target_revenue) # Simulate logging costs via budget manager if revenue_tracker.is_within_budget(10): revenue_tracker.log_cost_and_check_budget(10, "GPT-4") else: print("Cost was not logged due to budget limits.")