# revenue_tracker.py import json import time import os import redis from datetime import datetime, timedelta from typing import Dict, Any, List from budget_manager import BudgetManager # Import the BudgetManager class RevenueTracker: """ Tracks revenue, calculates ROI, identifies high-value actions, forecasts revenue, and generates reports. """ def __init__(self, budget_manager: BudgetManager, redis_config_path: str = r"E:\genesis-system\genesis_config.json"): self.namespace = "genesis:revenue" self.redis_client = None self.budget_manager = budget_manager # Integrate BudgetManager # Load Redis config if os.path.exists(redis_config_path): with open(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("Warning: Redis config file not found. Revenue tracking will be limited.") self.revenue_target = 1000.0 # Example revenue target self.alert_threshold = 0.9 # Alert when 90% of target is reached 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 = { "timestamp": now.isoformat(), "amount": amount, "agent": agent, "task": task, "metadata": metadata or {} } self.redis_client.lpush(f"{self.namespace}:events", json.dumps(event)) self.redis_client.ltrim(f"{self.namespace}:events", 0, 1000) # Keep last 1000 events # Track total revenue self.redis_client.incrbyfloat(f"{self.namespace}:total", amount) def get_total_revenue(self) -> float: """Returns the total revenue.""" if not self.redis_client: return 0.0 return float(self.redis_client.get(f"{self.namespace}:total") or 0.0) def calculate_roi(self, agent: str, task: str) -> float: """Calculates ROI for a specific agent and task.""" if not self.redis_client: return 0.0 total_revenue = 0.0 total_cost = 0.0 # Retrieve all revenue events and calculate total revenue for the specified agent/task events = self.get_all_revenue_events() for event_str in events: event = json.loads(event_str) if event["agent"] == agent and event["task"] == task: total_revenue += event["amount"] # Get the cost from the budget_manager for the agent/task (assuming it's logged there) # This is a simplification. More sophisticated cost attribution might be needed. # You might need to modify budget_manager to directly associate costs with agents/tasks. total_cost = self.get_cost_for_agent_task(agent, task) if total_cost == 0: return float('inf') if total_revenue > 0 else 0.0 # Avoid division by zero return (total_revenue - total_cost) / total_cost def get_cost_for_agent_task(self, agent: str, task: str) -> float: """Placeholder: Retrieve cost for agent/task. Needs actual implementation.""" # This is a placeholder. In a real system, you'd need a way to query the # BudgetManager or other cost tracking system to get the cost associated with # a specific agent and task. This might involve adding new methods to BudgetManager. # For now, just return a dummy value. return 5.0 # Example cost def identify_highest_value_actions(self, top_n: int = 5) -> List[Dict[str, Any]]: """Identifies the highest-value actions based on ROI.""" if not self.redis_client: return [] roi_data = [] seen_pairs = set() events = self.get_all_revenue_events() for event_str in events: event = json.loads(event_str) agent = event["agent"] task = event["task"] if (agent, task) not in seen_pairs: roi = self.calculate_roi(agent, task) roi_data.append({"agent": agent, "task": task, "roi": roi}) seen_pairs.add((agent, task)) sorted_actions = sorted(roi_data, key=lambda x: x["roi"], reverse=True) return sorted_actions[:top_n] def forecast_revenue(self, planned_activities: List[Dict[str, Any]]) -> float: """Forecasts revenue based on planned activities. Simple estimate.""" # This is a simplified forecasting method. A real system would use more # sophisticated techniques like time series analysis. total_forecasted_revenue = 0.0 for activity in planned_activities: # Assuming each activity has an estimated revenue field total_forecasted_revenue += activity.get("estimated_revenue", 0.0) return total_forecasted_revenue def generate_revenue_report(self) -> Dict[str, Any]: """Generates a revenue report.""" total_revenue = self.get_total_revenue() highest_value_actions = self.identify_highest_value_actions() report = { "total_revenue": total_revenue, "highest_value_actions": highest_value_actions, "timestamp": datetime.now().isoformat() } return report def check_revenue_target(self): """Checks if the revenue target is reached and sends an alert.""" total_revenue = self.get_total_revenue() if total_revenue >= self.revenue_target * self.alert_threshold: self.send_alert(f"Revenue target approaching! Current revenue: ${total_revenue}") def send_alert(self, message: str): """Placeholder: Sends an alert. Needs actual implementation (e.g., email, Slack).""" print(f"ALERT: {message}") def get_all_revenue_events(self) -> List[str]: """Retrieves all revenue events from Redis.""" if not self.redis_client: return [] return self.redis_client.lrange(f"{self.namespace}:events", 0, -1) def clear_revenue_data(self): """Clears all revenue data (for testing/resetting).""" if not self.redis_client: return self.redis_client.delete(f"{self.namespace}:events") self.redis_client.delete(f"{self.namespace}:total") if __name__ == "__main__": # Example Usage budget_manager = BudgetManager() revenue_tracker = RevenueTracker(budget_manager) # Log some revenue events revenue_tracker.log_revenue_event(amount=50.0, agent="AgentA", task="Task1", metadata={"customer_id": "123"}) revenue_tracker.log_revenue_event(amount=75.0, agent="AgentB", task="Task2", metadata={"customer_id": "456"}) revenue_tracker.log_revenue_event(amount=100.0, agent="AgentA", task="Task1", metadata={"customer_id": "789"}) # Get total revenue total_revenue = revenue_tracker.get_total_revenue() print(f"Total Revenue: ${total_revenue}") # Calculate ROI for AgentA/Task1 roi = revenue_tracker.calculate_roi(agent="AgentA", task="Task1") print(f"ROI for AgentA/Task1: {roi}") # Identify highest value actions highest_value_actions = revenue_tracker.identify_highest_value_actions() print(f"Highest Value Actions: {highest_value_actions}") # Forecast revenue planned_activities = [ {"agent": "AgentC", "task": "Task3", "estimated_revenue": 150.0}, {"agent": "AgentD", "task": "Task4", "estimated_revenue": 200.0} ] forecasted_revenue = revenue_tracker.forecast_revenue(planned_activities) print(f"Forecasted Revenue: ${forecasted_revenue}") # Generate revenue report report = revenue_tracker.generate_revenue_report() print(f"Revenue Report: {report}") # Check revenue target revenue_tracker.check_revenue_target() # Clear the data (optional, for testing) # revenue_tracker.clear_revenue_data()