# 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, costs, and ROI for AI-driven activities. """ def __init__(self, budget_manager, redis_config: Dict[str, Any] = None): self.namespace = "genesis:revenue" self.redis_client = None self.budget_manager = budget_manager # Inject BudgetManager 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 ) else: # Load Redis config from file (fallback if not provided) config_path = r"E:\genesis-system\genesis_config.json" if os.path.exists(config_path): with open(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 ) self.revenue_target = 1000.0 # Example revenue target def log_revenue_event(self, amount: float, activity: str, agent: str, cost: float = 0.0): """Log a revenue-generating event.""" if not self.redis_client: return now = datetime.now() event_data = { "timestamp": now.isoformat(), "amount": amount, "activity": activity, "agent": agent, "cost": cost } self.redis_client.lpush(f"{self.namespace}:events", json.dumps(event_data)) self.redis_client.ltrim(f"{self.namespace}:events", 0, 10000) # Keep last 10000 events def calculate_roi(self, agent: str) -> float: """Calculate the ROI for a specific agent.""" if not self.redis_client: return 0.0 total_revenue = 0.0 total_cost = 0.0 events = self.get_all_events() for event in events: if event["agent"] == agent: total_revenue += event["amount"] total_cost += event["cost"] if total_cost == 0: return 0.0 if total_revenue == 0 else float('inf') # Avoid division by zero return (total_revenue - total_cost) / total_cost if total_cost > 0 else 0.0 def get_highest_value_actions(self, top_n: int = 5) -> List[Dict[str, Any]]: """Identify the highest-value actions based on revenue generated.""" if not self.redis_client: return [] activity_revenue = {} events = self.get_all_events() for event in events: activity = event["activity"] revenue = event["amount"] activity_revenue[activity] = activity_revenue.get(activity, 0.0) + revenue sorted_activities = sorted(activity_revenue.items(), key=lambda item: item[1], reverse=True) return [{"activity": activity, "revenue": revenue} for activity, revenue in sorted_activities[:top_n]] def predict_revenue(self, planned_activities: List[Dict[str, Any]]) -> float: """Predict revenue from planned activities (simple estimation).""" # This is a placeholder. A real implementation would use a model. total_predicted_revenue = 0.0 for activity in planned_activities: # Assume each activity generates a fixed amount of revenue for now total_predicted_revenue += activity.get("estimated_revenue", 10.0) # Default to $10 return total_predicted_revenue def generate_revenue_report(self) -> Dict[str, Any]: """Generate a revenue report.""" if not self.redis_client: return {} total_revenue = 0.0 agent_revenue = {} events = self.get_all_events() for event in events: total_revenue += event["amount"] agent = event["agent"] agent_revenue[agent] = agent_revenue.get(agent, 0.0) + event["amount"] return { "total_revenue": total_revenue, "agent_revenue": agent_revenue, "roi_per_agent": {agent: self.calculate_roi(agent) for agent in agent_revenue}, "highest_value_actions": self.get_highest_value_actions() } def check_revenue_target(self): """Check if revenue target is met and trigger alerts.""" report = self.generate_revenue_report() total_revenue = report.get("total_revenue", 0.0) if total_revenue >= self.revenue_target: self.trigger_alert(f"Revenue target of ${self.revenue_target} reached! Total revenue: ${total_revenue}") else: print(f"Revenue target not yet reached. Current revenue: ${total_revenue}, Target: ${self.revenue_target}") def trigger_alert(self, message: str): """Placeholder for alert system integration (e.g., email, Slack).""" print(f"ALERT: {message}") def get_all_events(self) -> List[Dict[str, Any]]: """Retrieve all revenue events from Redis.""" if not self.redis_client: return [] events = [] event_strings = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) for event_string in event_strings: try: events.append(json.loads(event_string)) except json.JSONDecodeError: print(f"Error decoding JSON: {event_string}") return events def is_within_budget(self, cost: float) -> bool: """Check if adding the cost exceeds the budget.""" return self.budget_manager.is_within_budget() def log_cost_and_check_budget(self, amount: float, model: str, activity: str, agent: str, revenue: float): """Logs cost, checks budget, and logs revenue event.""" if self.is_within_budget(amount): self.budget_manager.log_cost(amount, model) self.log_revenue_event(revenue, activity, agent, amount) else: print("Budget exceeded. Revenue event not logged.") if __name__ == "__main__": # Example usage from budget_manager import BudgetManager # Import BudgetManager budget_manager = BudgetManager() revenue_tracker = RevenueTracker(budget_manager) # Example revenue events revenue_tracker.log_revenue_event(amount=50.0, activity="Lead Generation", agent="AgentA", cost=2.5) revenue_tracker.log_revenue_event(amount=75.0, activity="Sales Conversion", agent="AgentB", cost=5.0) revenue_tracker.log_revenue_event(amount=25.0, activity="Lead Generation", agent="AgentA", cost=1.0) # Example planned activities planned_activities = [ {"activity": "Marketing Campaign", "estimated_revenue": 100.0}, {"activity": "Product Launch", "estimated_revenue": 200.0} ] # Calculate ROI for AgentA roi_agent_a = revenue_tracker.calculate_roi("AgentA") print(f"ROI for AgentA: {roi_agent_a}") # Get highest value actions highest_value_actions = revenue_tracker.get_highest_value_actions() print(f"Highest Value Actions: {highest_value_actions}") # Predict revenue from planned activities predicted_revenue = revenue_tracker.predict_revenue(planned_activities) print(f"Predicted Revenue: {predicted_revenue}") # Generate revenue report report = revenue_tracker.generate_revenue_report() print(f"Revenue Report: {report}") # Check revenue target revenue_tracker.check_revenue_target() # Example of logging cost and checking the budget revenue_tracker.log_cost_and_check_budget(amount=3.0, model="GPT-4", activity="Content Creation", agent="AgentC", revenue=60.0)