# revenue_tracker.py import json import time import os import redis from datetime import datetime, timedelta from typing import Dict, Any from budget_manager import BudgetManager # Import the BudgetManager class RevenueTracker: """ Tracks revenue, calculates ROI, and forecasts future earnings. """ 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 # Inject BudgetManager instance # Load Redis config self.redis_config_path = redis_config_path 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 will be disabled.") self.revenue_target = 1000.0 # Example revenue target per day self.alert_threshold = 0.9 # Trigger alert when revenue is 90% of target def log_revenue_event(self, amount: float, agent: str, task: str, cost: float = 0.0, metadata: Dict[str, Any] = None): """Log a revenue-generating event.""" if not self.redis_client: print("Revenue tracking is disabled (Redis not configured).") return now = datetime.now() timestamp = now.isoformat() revenue_event = { "timestamp": timestamp, "amount": amount, "agent": agent, "task": task, "cost": cost, "profit": amount - cost, "metadata": metadata or {} } # Store the revenue event in Redis self.redis_client.lpush(f"{self.namespace}:events", json.dumps(revenue_event)) self.redis_client.ltrim(f"{self.namespace}:events", 0, 10000) # Keep last 10000 events # Update daily revenue total day_key = f"{self.namespace}:daily_revenue:{now.strftime('%Y-%m-%d')}" self.redis_client.incrbyfloat(day_key, amount) self.redis_client.expire(day_key, 86400 * 2) # Keep for 2 days def calculate_roi(self, agent: str, task: str, start_time: datetime = None, end_time: datetime = None) -> float: """Calculate ROI for a specific agent and task within a time range.""" if not self.redis_client: return 0.0 total_revenue = 0.0 total_cost = 0.0 # Retrieve revenue events (inefficient for large datasets, consider alternative storage/querying) event_strings = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) for event_string in event_strings: event = json.loads(event_string) event_time = datetime.fromisoformat(event["timestamp"]) if event["agent"] == agent and event["task"] == task: if start_time and event_time < start_time: continue if end_time and event_time > end_time: continue total_revenue += event["amount"] total_cost += event["cost"] if total_cost == 0: return float('inf') if total_revenue > 0 else 0.0 # Avoid division by zero else: return (total_revenue - total_cost) / total_cost def identify_highest_value_actions(self, top_n: int = 5) -> list: """Identify the top N highest-value actions based on profit.""" if not self.redis_client: return [] action_profits = {} # Retrieve revenue events (inefficient, consider alternative storage/querying) event_strings = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) for event_string in event_strings: event = json.loads(event_string) action_key = f"{event['agent']}:{event['task']}" profit = event["amount"] - event["cost"] action_profits[action_key] = action_profits.get(action_key, 0) + profit # Sort actions by profit in descending order sorted_actions = sorted(action_profits.items(), key=lambda item: item[1], reverse=True) return sorted_actions[:top_n] def predict_revenue(self, planned_activities: list) -> float: """Predict revenue from a list of planned activities (basic placeholder).""" # This is a very basic placeholder. A real implementation would use a model. predicted_revenue = 0.0 for activity in planned_activities: # Assuming each activity has a 'predicted_value' field predicted_revenue += activity.get("predicted_value", 0.0) return predicted_revenue def generate_revenue_report(self, start_date: datetime = None, end_date: datetime = None) -> dict: """Generate a revenue report for a given time period.""" if not self.redis_client: return {} report = { "total_revenue": 0.0, "total_cost": 0.0, "profit": 0.0, "event_count": 0, "daily_revenue": {} } if not start_date: start_date = datetime.min # Effectively no start date if not end_date: end_date = datetime.max # Retrieve revenue events (inefficient) event_strings = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) for event_string in event_strings: event = json.loads(event_string) event_time = datetime.fromisoformat(event["timestamp"]) if start_date <= event_time <= end_date: report["total_revenue"] += event["amount"] report["total_cost"] += event["cost"] report["profit"] += event["amount"] - event["cost"] report["event_count"] += 1 day_key = event_time.strftime('%Y-%m-%d') if day_key not in report["daily_revenue"]: report["daily_revenue"][day_key] = 0.0 report["daily_revenue"][day_key] += event["amount"] return report def check_revenue_target(self): """Check if the daily revenue target has been met and trigger an alert if not.""" if not self.redis_client: return now = datetime.now() day_key = f"{self.namespace}:daily_revenue:{now.strftime('%Y-%m-%d')}" daily_revenue = float(self.redis_client.get(day_key) or 0.0) if daily_revenue < self.revenue_target * self.alert_threshold: print(f"Alert: Daily revenue is below target! Current revenue: ${daily_revenue}, Target: ${self.revenue_target}") def get_daily_revenue(self) -> float: """Return the total revenue earned today.""" if not self.redis_client: return 0.0 now = datetime.now() day_key = f"{self.namespace}:daily_revenue:{now.strftime('%Y-%m-%d')}" return float(self.redis_client.get(day_key) or 0.0) def integrate_budget(self, amount: float, model: str) -> bool: """Integrate with the BudgetManager to track costs and ensure we're within budget.""" if self.budget_manager.is_within_budget(): self.budget_manager.log_cost(amount, model) return True else: print("Budget exceeded. AI operation blocked.") return False if __name__ == "__main__": # Example Usage budget_manager = BudgetManager() # Initialize BudgetManager revenue_tracker = RevenueTracker(budget_manager) # Pass BudgetManager instance # Log a revenue event revenue_tracker.log_revenue_event(amount=50.0, agent="AgentA", task="Task1", cost=10.0, metadata={"client": "ClientX"}) revenue_tracker.log_revenue_event(amount=75.0, agent="AgentB", task="Task2", cost=15.0, metadata={"client": "ClientY"}) # Calculate ROI for AgentA and Task1 roi = revenue_tracker.calculate_roi(agent="AgentA", task="Task1") print(f"ROI for AgentA, Task1: {roi}") # Identify highest value actions top_actions = revenue_tracker.identify_highest_value_actions(top_n=3) print(f"Top Actions: {top_actions}") # Predict revenue (placeholder) planned_activities = [{"name": "ActivityC", "predicted_value": 100.0}, {"name": "ActivityD", "predicted_value": 150.0}] predicted_revenue = revenue_tracker.predict_revenue(planned_activities) print(f"Predicted Revenue: {predicted_revenue}") # Generate a revenue report report = revenue_tracker.generate_revenue_report() print(f"Revenue Report: {report}") # Check revenue target revenue_tracker.check_revenue_target() # Integrate with budget if revenue_tracker.integrate_budget(amount=2.5, model="GPT-3"): print("AI operation cost logged successfully.") else: print("AI operation blocked due to budget constraints.") print(f"Daily Revenue: ${revenue_tracker.get_daily_revenue()}")