# 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, calculates ROI, and forecasts future earnings. """ def __init__(self, budget_manager, redis_config=None): self.namespace = "genesis:revenue" self.redis_client = None self.budget_manager = budget_manager # Inject BudgetManager # Load Redis config if provided, otherwise use BudgetManager's config 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 ) elif budget_manager.redis_client: self.redis_client = budget_manager.redis_client else: 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 target self.alert_threshold = 0.9 # Alert when 90% of target is reached def log_revenue_event(self, amount: float, agent: str, task: str, cost: float = 0.0): """Logs a revenue-generating event with associated agent, task, and cost.""" if not self.redis_client: print("Redis client not initialized.") return now = datetime.now() timestamp = now.isoformat() event_data = { "timestamp": timestamp, "amount": amount, "agent": agent, "task": task, "cost": cost, "profit": amount - cost, } self.redis_client.lpush(f"{self.namespace}:events", json.dumps(event_data)) self.redis_client.ltrim(f"{self.namespace}:events", 0, 1000) # Aggregate revenue data self._aggregate_revenue(amount, agent, task, cost) self._check_revenue_target() def _aggregate_revenue(self, amount: float, agent: str, task: str, cost: float): """Aggregates revenue data for ROI calculations.""" if not self.redis_client: return today = datetime.now().strftime("%Y-%m-%d") agent_key = f"{self.namespace}:agent:{agent}:{today}" task_key = f"{self.namespace}:task:{task}:{today}" # Agent revenue self.redis_client.incrbyfloat(f"{agent_key}:revenue", amount) self.redis_client.incrbyfloat(f"{agent_key}:cost", cost) self.redis_client.expire(agent_key, 86400 * 2) # Task revenue self.redis_client.incrbyfloat(f"{task_key}:revenue", amount) self.redis_client.incrbyfloat(f"{task_key}:cost", cost) self.redis_client.expire(task_key, 86400 * 2) def calculate_roi(self, agent: str, task: str) -> float: """Calculates ROI for a given agent or task.""" if not self.redis_client: return 0.0 today = datetime.now().strftime("%Y-%m-%d") agent_key = f"{self.namespace}:agent:{agent}:{today}" task_key = f"{self.namespace}:task:{task}:{today}" agent_revenue = float(self.redis_client.get(f"{agent_key}:revenue") or 0.0) agent_cost = float(self.redis_client.get(f"{agent_key}:cost") or 0.0) task_revenue = float(self.redis_client.get(f"{task_key}:revenue") or 0.0) task_cost = float(self.redis_client.get(f"{task_key}:cost") or 0.0) agent_roi = (agent_revenue - agent_cost) / agent_cost if agent_cost else 0.0 task_roi = (task_revenue - task_cost) / task_cost if task_cost else 0.0 return {"agent_roi": agent_roi, "task_roi": task_roi} def identify_highest_value_actions(self, top_n: int = 5) -> list: """Identifies the top N highest-value actions based on profit.""" if not self.redis_client: return [] all_events = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) events = [json.loads(event) for event in all_events] # Sort events by profit (amount - cost) sorted_events = sorted(events, key=lambda x: x["profit"], reverse=True) return sorted_events[:top_n] def forecast_revenue(self, planned_activities: list) -> float: """Forecasts revenue from planned activities (requires activity-specific revenue estimates).""" total_forecasted_revenue = sum(activity["estimated_revenue"] for activity in planned_activities) return total_forecasted_revenue def generate_revenue_report(self, start_date: str, end_date: str) -> Dict[str, Any]: """Generates a revenue report for a given date range.""" if not self.redis_client: return {} # Basic implementation: aggregates total revenue total_revenue = 0.0 all_events = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) events = [json.loads(event) for event in all_events] for event in events: event_date = datetime.fromisoformat(event["timestamp"]).strftime("%Y-%m-%d") if start_date <= event_date <= end_date: total_revenue += event["amount"] return { "start_date": start_date, "end_date": end_date, "total_revenue": total_revenue, "num_events": len(events), } def _check_revenue_target(self): """Checks if the revenue target is reached and sends an alert.""" if not self.redis_client: return today = datetime.now().strftime("%Y-%m-%d") total_revenue = 0.0 all_events = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) events = [json.loads(event) for event in all_events] for event in events: event_date = datetime.fromisoformat(event["timestamp"]).strftime("%Y-%m-%d") if today == event_date: total_revenue += event["amount"] if total_revenue >= self.revenue_target * self.alert_threshold: print(f"Alert! Revenue target approaching: {total_revenue}/{self.revenue_target}") def is_within_budget(self, cost: float) -> bool: """Checks if the cost is within the budget using the injected BudgetManager.""" if self.budget_manager.is_within_budget(): return True else: print("Warning: Budget exceeded!") return False def log_cost_and_check_budget(self, amount: float, model: str) -> bool: """Logs cost to budget manager and checks if within budget.""" if self.budget_manager.is_within_budget(): self.budget_manager.log_cost(amount, model) return True else: print("Warning: Budget exceeded. Cost not logged.") return False if __name__ == "__main__": # Example Usage from budget_manager import BudgetManager # Initialize BudgetManager budget_manager = BudgetManager() # Initialize RevenueTracker, injecting the BudgetManager revenue_tracker = RevenueTracker(budget_manager) # Log a revenue event revenue_tracker.log_revenue_event(amount=50.0, agent="AgentA", task="Task1", cost=10.0) # Calculate ROI roi = revenue_tracker.calculate_roi(agent="AgentA", task="Task1") print(f"ROI: {roi}") # Identify highest value actions top_actions = revenue_tracker.identify_highest_value_actions(top_n=3) print(f"Top Actions: {top_actions}") # Forecast revenue planned_activities = [ {"activity": "ActivityX", "estimated_revenue": 75.0}, {"activity": "ActivityY", "estimated_revenue": 120.0}, ] forecasted_revenue = revenue_tracker.forecast_revenue(planned_activities) print(f"Forecasted Revenue: {forecasted_revenue}") # Generate revenue report report = revenue_tracker.generate_revenue_report(start_date="2023-01-01", end_date="2024-01-01") print(f"Revenue Report: {report}") # Check budget if revenue_tracker.is_within_budget(cost=5.0): print("Within budget.") else: print("Exceeded budget.") # Log cost and check budget if revenue_tracker.log_cost_and_check_budget(amount=2.0, model="GPT-3"): print("Cost logged and within budget.") else: print("Cost not logged due to budget constraints.")