# 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, and provides forecasting. """ def __init__(self, budget_manager, redis_config: Optional[Dict] = None): self.namespace = "genesis:revenue" self.redis_client = None # Load Redis 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 ) 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.budget_manager = budget_manager def log_revenue_event(self, agent_id: str, task_id: str, revenue: float, cost: float, description: str = ""): """Logs a revenue-generating event with associated cost.""" if not self.redis_client: return timestamp = datetime.now().isoformat() event = { "timestamp": timestamp, "agent_id": agent_id, "task_id": task_id, "revenue": revenue, "cost": cost, "description": description } self.redis_client.lpush(f"{self.namespace}:events", json.dumps(event)) self.redis_client.ltrim(f"{self.namespace}:events", 0, 1000) # Keep last 1000 events def calculate_roi(self, agent_id: str) -> float: """Calculates the ROI for a given agent.""" if not self.redis_client: return 0.0 total_revenue = 0.0 total_cost = 0.0 events = self.get_agent_revenue_events(agent_id) for event_str in events: event = json.loads(event_str) 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 get_agent_revenue_events(self, agent_id: str) -> List[str]: """Retrieves revenue events for a specific agent.""" if not self.redis_client: return [] all_events = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) agent_events = [event for event in all_events if json.loads(event)["agent_id"] == agent_id] return agent_events def identify_highest_value_actions(self, top_n: int = 5) -> List[Dict]: """Identifies the highest-value actions based on revenue generated.""" if not self.redis_client: return [] all_events = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) task_revenues = {} for event_str in all_events: event = json.loads(event_str) task_id = event["task_id"] revenue = event["revenue"] cost = event["cost"] if task_id not in task_revenues: task_revenues[task_id] = {"revenue": 0.0, "cost": 0.0, "profit": 0.0} task_revenues[task_id]["revenue"] += revenue task_revenues[task_id]["cost"] += cost task_revenues[task_id]["profit"] = task_revenues[task_id]["revenue"] - task_revenues[task_id]["cost"] sorted_tasks = sorted(task_revenues.items(), key=lambda item: item[1]["profit"], reverse=True) top_tasks = [{"task_id": task_id, **data} for task_id, data in sorted_tasks[:top_n]] return top_tasks def predict_revenue(self, planned_activity: Dict) -> float: """Predicts revenue from a planned activity based on historical data.""" # This is a placeholder for a more sophisticated prediction model. # In a real-world scenario, you would use machine learning models # to predict revenue based on various features of the planned activity. # For now, we'll just return a simple estimate. # Example: Assume revenue is correlated with cost. estimated_revenue = planned_activity.get("estimated_cost", 0.0) * 2 # Example multiplier return estimated_revenue def generate_revenue_report(self, start_date: datetime, end_date: datetime) -> List[Dict]: """Generates a revenue report for a given date range.""" if not self.redis_client: return [] report = [] all_events = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) for event_str in all_events: event = json.loads(event_str) event_date = datetime.fromisoformat(event["timestamp"].replace('Z', '+00:00')) # Handle Z timezone if start_date <= event_date <= end_date: report.append(event) return report def check_budget_and_alert(self, cost: float) -> bool: """Checks if the cost exceeds the budget and triggers an alert.""" if not self.budget_manager.is_within_budget(): print("ALERT! Budget exceeded. Halting operation.") return False # Indicate that the operation should be halted return True # Indicate that the operation can proceed def get_profit_margin(self, revenue: float, cost: float) -> float: """Calculates the profit margin.""" if cost == 0: return 0.0 if revenue == 0 else float('inf') return (revenue - cost) / revenue if revenue else 0.0 if __name__ == "__main__": from budget_manager import BudgetManager # Example Usage budget_manager = BudgetManager(daily_limit=100.0) # Higher limit for testing revenue_tracker = RevenueTracker(budget_manager) # Log some revenue events revenue_tracker.log_revenue_event(agent_id="agent1", task_id="task1", revenue=50.0, cost=20.0, description="Sold product A") revenue_tracker.log_revenue_event(agent_id="agent1", task_id="task2", revenue=30.0, cost=15.0, description="Sold product B") revenue_tracker.log_revenue_event(agent_id="agent2", task_id="task3", revenue=70.0, cost=30.0, description="Service C rendered") # Calculate ROI for agent1 roi_agent1 = revenue_tracker.calculate_roi("agent1") print(f"ROI for agent1: {roi_agent1}") # Identify highest value actions top_actions = revenue_tracker.identify_highest_value_actions(top_n=2) print(f"Top actions: {top_actions}") # Predict revenue for a planned activity planned_activity = {"estimated_cost": 25.0} predicted_revenue = revenue_tracker.predict_revenue(planned_activity) print(f"Predicted revenue: {predicted_revenue}") # Generate a revenue report start_date = datetime.now() - timedelta(days=1) end_date = datetime.now() report = revenue_tracker.generate_revenue_report(start_date, end_date) print(f"Revenue report: {report}") # Check budget and alert cost_of_operation = 10.0 if revenue_tracker.check_budget_and_alert(cost_of_operation): print("Operation can proceed.") else: print("Operation halted due to budget.") # Calculate profit margin revenue = 100.0 cost = 60.0 profit_margin = revenue_tracker.get_profit_margin(revenue, cost) print(f"Profit margin: {profit_margin}")