# 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 # Assuming budget_manager.py is in the same directory class RevenueTracker: """ Tracks revenue-generating activities, calculates ROI, and forecasts future revenue. """ def __init__(self, budget_manager: BudgetManager = None, redis_config_path: str = r"E:\genesis-system\genesis_config.json"): self.namespace = "genesis:revenue" self.redis_client = None self.budget_manager = budget_manager or BudgetManager() # Use injected or default BudgetManager self.redis_config_path = redis_config_path # Load Redis config 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 file not found at {self.redis_config_path}. Revenue tracking might be limited.") self.revenue_target = 1000.0 # Example revenue target self.profit_margin_target = 0.2 # Example profit margin target (20%) def log_revenue_event(self, agent: str, task: str, revenue: float, cost: float, details: Dict[str, Any] = None): """Logs a revenue-generating event with associated costs.""" if not self.redis_client: print("Redis client not initialized. Revenue event not logged.") return timestamp = datetime.now().isoformat() event = { "timestamp": timestamp, "agent": agent, "task": task, "revenue": revenue, "cost": cost, "details": details or {} } self.redis_client.lpush(f"{self.namespace}:events", json.dumps(event)) self.redis_client.ltrim(f"{self.namespace}:events", 0, 10000) # Keep the last 10000 events # Update daily revenue day_key = f"{self.namespace}:daily_revenue:{datetime.now().strftime('%Y-%m-%d')}" self.redis_client.incrbyfloat(day_key, revenue) self.redis_client.expire(day_key, 86400 * 2) # Keep for 2 days def calculate_roi(self, agent: str = None, task: str = None, start_date: datetime = None, end_date: datetime = None) -> float: """Calculates ROI based on revenue and cost. Filters by agent/task if provided.""" total_revenue = 0.0 total_cost = 0.0 # Retrieve revenue events from Redis if not self.redis_client: print("Redis client not initialized. Cannot calculate ROI.") return 0.0 events = [] for event_json in self.redis_client.lrange(f"{self.namespace}:events", 0, -1): events.append(json.loads(event_json)) # Filter and aggregate revenue and cost for event in events: event_date = datetime.fromisoformat(event["timestamp"].replace('Z', '+00:00')) #Handle UTC timezone if start_date and event_date < start_date: continue if end_date and event_date > end_date: continue if agent and event["agent"] != agent: continue if task and event["task"] != task: continue total_revenue += event["revenue"] total_cost += event["cost"] if total_cost == 0: return 0.0 # Avoid division by zero return (total_revenue - total_cost) / total_cost def identify_highest_value_actions(self, top_n: int = 5) -> List[Dict[str, Any]]: """Identifies the top N highest-value actions based on (revenue - cost).""" if not self.redis_client: print("Redis client not initialized. Cannot identify highest value actions.") return [] events = [] for event_json in self.redis_client.lrange(f"{self.namespace}:events", 0, -1): events.append(json.loads(event_json)) action_values = {} for event in events: action_key = f"{event['agent']}:{event['task']}" profit = event['revenue'] - event['cost'] if action_key in action_values: action_values[action_key] += profit else: action_values[action_key] = profit sorted_actions = sorted(action_values.items(), key=lambda item: item[1], reverse=True)[:top_n] result = [] for action, value in sorted_actions: agent, task = action.split(":") result.append({"agent": agent, "task": task, "profit": value}) return result def predict_revenue(self, planned_activities: List[Dict[str, Any]]) -> float: """Predicts revenue from planned activities based on historical data.""" # This is a simplified placeholder. A real implementation would use a more sophisticated model. # It averages revenue from similar past tasks. if not self.redis_client: print("Redis client not initialized. Cannot predict revenue.") return 0.0 total_predicted_revenue = 0.0 for activity in planned_activities: agent = activity.get("agent") task = activity.get("task") planned_cost = activity.get("cost", 0.0) # Retrieve past revenue events for the same agent and task events = [] for event_json in self.redis_client.lrange(f"{self.namespace}:events", 0, -1): events.append(json.loads(event_json)) relevant_revenues = [event["revenue"] for event in events if event["agent"] == agent and event["task"] == task] if relevant_revenues: average_revenue = sum(relevant_revenues) / len(relevant_revenues) total_predicted_revenue += average_revenue - planned_cost # Predict profit else: print(f"Warning: No historical data found for agent '{agent}' and task '{task}'. Using a zero revenue prediction.") # If no data, assume zero revenue (or implement a default prediction) return total_predicted_revenue def generate_revenue_report(self, start_date: datetime, end_date: datetime) -> Dict[str, Any]: """Generates a revenue report for a given time period.""" total_revenue = 0.0 total_cost = 0.0 if not self.redis_client: print("Redis client not initialized. Cannot generate report.") return {} events = [] for event_json in self.redis_client.lrange(f"{self.namespace}:events", 0, -1): events.append(json.loads(event_json)) for event in events: event_date = datetime.fromisoformat(event["timestamp"].replace('Z', '+00:00')) #Handle UTC timezone if start_date and event_date < start_date: continue if end_date and event_date > end_date: continue total_revenue += event["revenue"] total_cost += event["cost"] profit = total_revenue - total_cost roi = (profit / total_cost) if total_cost else 0.0 return { "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), "total_revenue": total_revenue, "total_cost": total_cost, "profit": profit, "roi": roi } def check_revenue_targets(self): """Checks if revenue targets are met and sends alerts if not.""" 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) report = self.generate_revenue_report(start_of_month, end_of_month) total_revenue = report.get("total_revenue", 0.0) profit = report.get("profit", 0.0) if total_revenue < self.revenue_target: print(f"Alert: Revenue target not met. Current revenue: ${total_revenue}, Target: ${self.revenue_target}") if total_revenue > 0: profit_margin = profit / total_revenue else: profit_margin = 0 if profit_margin < self.profit_margin_target: print(f"Alert: Profit margin target not met. Current margin: {profit_margin:.2f}, Target: {self.profit_margin_target:.2f}") def integrate_budget_constraints(self, agent: str, task: str, cost: float): """Integrates with the BudgetManager to check budget constraints before allowing an action.""" if self.budget_manager.is_within_budget(): self.budget_manager.log_cost(cost, f"{agent}:{task}") return True else: print(f"Budget exceeded. Action '{task}' by agent '{agent}' blocked.") return False if __name__ == "__main__": # Example Usage redis_config_path = r"E:\genesis-system\genesis_config.json" # Replace with the correct path bm = BudgetManager() rt = RevenueTracker(budget_manager=bm, redis_config_path=redis_config_path) # Log some revenue events rt.log_revenue_event("AgentA", "Task1", 50.0, 10.0, {"customer": "CustomerX"}) rt.log_revenue_event("AgentB", "Task2", 75.0, 15.0, {"customer": "CustomerY"}) rt.log_revenue_event("AgentA", "Task1", 60.0, 12.0, {"customer": "CustomerZ"}) # Calculate ROI roi_agent_a = rt.calculate_roi(agent="AgentA") print(f"ROI for AgentA: {roi_agent_a}") # Identify highest value actions top_actions = rt.identify_highest_value_actions(top_n=2) print(f"Top Actions: {top_actions}") # Predict revenue from planned activities planned_activities = [ {"agent": "AgentA", "task": "Task1", "cost": 11.0}, {"agent": "AgentB", "task": "Task3", "cost": 18.0} # No historical data for Task3 ] predicted_revenue = rt.predict_revenue(planned_activities) print(f"Predicted Revenue: {predicted_revenue}") # Generate a revenue report today = datetime.now() start_date = today - timedelta(days=7) report = rt.generate_revenue_report(start_date, today) print(f"Revenue Report: {report}") # Check revenue targets rt.check_revenue_targets() # Integrate budget constraints if rt.integrate_budget_constraints("AgentC", "Task4", 8.0): print("Action allowed within budget.") else: print("Action blocked due to budget constraints.") if rt.integrate_budget_constraints("AgentC", "Task4", 8.0): print("Action allowed within budget.") else: print("Action blocked due to budget constraints.")