# revenue_tracker.py import json import time import os import redis from datetime import datetime, timedelta from typing import Dict, List, Optional class RevenueTracker: """ Tracks revenue-generating activities, calculates ROI, forecasts revenue, and generates reports. Integrates with BudgetManager for cost attribution. """ def __init__(self, budget_manager=None): self.namespace = "genesis:revenue" self.redis_client = None self.budget_manager = budget_manager # Load Redis config (same as BudgetManager for consistency) 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 = 100.0 # Example revenue target self.alert_threshold = 0.8 # Alert when revenue is 80% of target def log_revenue_event(self, amount: float, agent: str, task: str, details: Optional[Dict] = None): """Logs a revenue-generating event.""" if not self.redis_client: return now = datetime.now() event_id = f"{now.timestamp()}" # unique id for each event revenue_event = { "timestamp": now.isoformat(), "amount": amount, "agent": agent, "task": task, "details": details or {}, "event_id": event_id } self.redis_client.lpush(f"{self.namespace}:events", json.dumps(revenue_event)) self.redis_client.ltrim(f"{self.namespace}:events", 0, 1000) # Keep last 1000 events # Store event details by ID for easy retrieval self.redis_client.set(f"{self.namespace}:event:{event_id}", json.dumps(revenue_event)) def calculate_roi(self, agent: str, task: str, start_date: datetime, end_date: datetime) -> float: """Calculates ROI for a given agent and task within a date range.""" if not self.redis_client: return 0.0 total_revenue = 0.0 events = self.get_revenue_events(start_date, end_date, agent, task) for event in events: total_revenue += event["amount"] total_cost = self.get_cost_for_agent_task(agent, task, start_date, end_date) # Assumes BudgetManager integration 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, start_date: datetime, end_date: datetime) -> float: """ Retrieves the total cost associated with a specific agent and task within a given date range. This function assumes the budget manager logs costs with enough detail to filter by agent and task. Since the budget_manager.py only logs cost by model and time, we will return a simple estimate. In a real-world scenario, you'd need to modify the BudgetManager to include agent and task details in the cost logs. """ if not self.budget_manager: return 0.0 # No budget manager, no cost data # This is a simplified estimate. A real implementation would need detailed cost logging. # Assume each revenue event corresponds to some cost. This is a very rough estimate. events = self.get_revenue_events(start_date, end_date, agent, task) estimated_cost_per_event = 0.1 # Adjust this based on your AI model costs return len(events) * estimated_cost_per_event def identify_highest_value_actions(self, start_date: datetime, end_date: datetime, top_n: int = 5) -> List[Dict]: """Identifies the highest-value actions based on revenue generated.""" if not self.redis_client: return [] revenue_by_action = {} events = self.get_revenue_events(start_date, end_date) # Get all events for event in events: action_key = (event["agent"], event["task"]) if action_key not in revenue_by_action: revenue_by_action[action_key] = 0.0 revenue_by_action[action_key] += event["amount"] # Sort by revenue in descending order sorted_actions = sorted(revenue_by_action.items(), key=lambda item: item[1], reverse=True) top_actions = [] for (agent, task), revenue in sorted_actions[:top_n]: top_actions.append({"agent": agent, "task": task, "revenue": revenue}) return top_actions def predict_revenue(self, planned_activities: List[Dict], historical_data_start: datetime, historical_data_end: datetime) -> float: """ Predicts revenue from planned activities based on historical performance. Args: planned_activities: A list of dictionaries, each representing a planned activity with keys like "agent", "task", and "expected_frequency". historical_data_start: The start date for historical data to use for prediction. historical_data_end: The end date for historical data to use for prediction. Returns: The predicted revenue for the planned activities. """ if not self.redis_client: return 0.0 total_predicted_revenue = 0.0 for activity in planned_activities: agent = activity["agent"] task = activity["task"] expected_frequency = activity.get("expected_frequency", 1) # How many times per day/week/month # Get historical revenue per action historical_roi = self.calculate_roi(agent, task, historical_data_start, historical_data_end) # Make prediction based on frequency and historical ROI (very simplified) # Adjust this formula as needed predicted_revenue = historical_roi * expected_frequency total_predicted_revenue += predicted_revenue return total_predicted_revenue def generate_revenue_report(self, start_date: datetime, end_date: datetime) -> Dict: """Generates a revenue report for a given date range.""" if not self.redis_client: return {} total_revenue = 0.0 events = self.get_revenue_events(start_date, end_date) for event in events: total_revenue += event["amount"] highest_value_actions = self.identify_highest_value_actions(start_date, end_date) report = { "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), "total_revenue": total_revenue, "highest_value_actions": highest_value_actions } return report def check_revenue_targets(self): """Checks if revenue targets are being met and triggers alerts if necessary.""" today = datetime.now().date() start_of_month = datetime(today.year, today.month, 1) end_of_month = datetime(today.year, today.month + 1, 1) - timedelta(days=1) if today.month < 12 else datetime(today.year + 1, 1, 1) - timedelta(days=1) report = self.generate_revenue_report(start_of_month, end_of_month) current_revenue = report["total_revenue"] if current_revenue < self.revenue_target * self.alert_threshold: self.trigger_alert(f"Revenue is below target! Current revenue: {current_revenue}, Target: {self.revenue_target}") def trigger_alert(self, message: str): """Sends an alert (e.g., to a logging system or notification service).""" print(f"ALERT: {message}") # Replace with a proper logging/notification mechanism def get_revenue_events(self, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None, agent: Optional[str] = None, task: Optional[str] = None) -> List[Dict]: """ Retrieves revenue events from Redis, optionally filtered by date range, agent, and task. """ if not self.redis_client: return [] events = [] event_jsons = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) for event_json in event_jsons: try: event = json.loads(event_json) event_datetime = datetime.fromisoformat(event["timestamp"]) # Apply filters if start_date and event_datetime < start_date: continue if end_date and event_datetime > end_date: continue if agent and event["agent"] != agent: continue if task and event["task"] != task: continue events.append(event) except json.JSONDecodeError: print(f"Error decoding JSON: {event_json}") except ValueError as e: print(f"Error processing timestamp: {e}") return events if __name__ == "__main__": from budget_manager import BudgetManager # Import BudgetManager # Initialize BudgetManager and RevenueTracker budget_manager = BudgetManager() revenue_tracker = RevenueTracker(budget_manager=budget_manager) # Pass BudgetManager instance # Log some revenue events revenue_tracker.log_revenue_event(amount=25.0, agent="AgentA", task="Task1", details={"customer": "CustomerX"}) revenue_tracker.log_revenue_event(amount=30.0, agent="AgentB", task="Task2", details={"customer": "CustomerY"}) revenue_tracker.log_revenue_event(amount=15.0, agent="AgentA", task="Task1", details={"customer": "CustomerZ"}) # Calculate ROI today = datetime.now() start_date = today - timedelta(days=7) # Last 7 days roi_agent_a_task_1 = revenue_tracker.calculate_roi("AgentA", "Task1", start_date, today) print(f"ROI for AgentA, Task1 (last 7 days): {roi_agent_a_task_1}") # Identify highest value actions highest_value_actions = revenue_tracker.identify_highest_value_actions(start_date, today) print(f"Highest Value Actions (last 7 days): {highest_value_actions}") # Generate a revenue report report = revenue_tracker.generate_revenue_report(start_date, today) print(f"Revenue Report (last 7 days): {report}") # Predict revenue planned_activities = [ {"agent": "AgentA", "task": "Task1", "expected_frequency": 5}, {"agent": "AgentB", "task": "Task2", "expected_frequency": 3} ] predicted_revenue = revenue_tracker.predict_revenue(planned_activities, start_date - timedelta(days=30), start_date) print(f"Predicted Revenue: {predicted_revenue}") # Check if revenue targets are being met (example) revenue_tracker.check_revenue_targets()