# revenue_tracker.py import json import time import os import redis from datetime import datetime, timedelta from typing import Dict, List, Tuple class RevenueTracker: """ Tracks revenue, calculates ROI, forecasts revenue, and generates reports. """ def __init__(self, budget_manager=None): self.namespace = "genesis:revenue" self.redis_client = None self.budget_manager = budget_manager # Inject BudgetManager # Load Redis config 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 ) def log_revenue(self, amount: float, source: str, agent: str = "unknown", task: str = "unknown", metadata: Dict = None): """Log a revenue-generating event.""" if not self.redis_client: return now = datetime.now() revenue_entry = { "timestamp": now.isoformat(), "amount": amount, "source": source, "agent": agent, "task": task, "metadata": metadata if metadata else {} } self.redis_client.lpush(f"{self.namespace}:revenue_logs", json.dumps(revenue_entry)) self.redis_client.ltrim(f"{self.namespace}:revenue_logs", 0, 5000) # Keep last 5000 entries # Increment total revenue (daily and overall) 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 self.redis_client.incrbyfloat(f"{self.namespace}:total_revenue", amount) def calculate_roi(self, agent: str, task: str) -> float: """Calculate ROI for a specific agent/task.""" if not self.redis_client: return 0.0 total_revenue = 0.0 total_cost = 0.0 # Aggregate Revenue revenue_logs = self.get_revenue_logs() for log in revenue_logs: if log["agent"] == agent and log["task"] == task: total_revenue += log["amount"] # Aggregate Costs (using BudgetManager, if available) if self.budget_manager: #This is a simplified example. A more robust implementation would track costs per agent/task directly. #For example, you could add agent/task to the budget_manager.log_cost method. total_cost = self.budget_manager.get_current_spend() # This is a VERY rough estimate. 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 identify_highest_value_actions(self, top_n: int = 5) -> List[Tuple[str, float]]: """Identify the highest-value actions based on revenue generated.""" if not self.redis_client: return [] revenue_logs = self.get_revenue_logs() action_revenue = {} # {(agent, task): revenue} for log in revenue_logs: key = (log["agent"], log["task"]) if key in action_revenue: action_revenue[key] += log["amount"] else: action_revenue[key] = log["amount"] sorted_actions = sorted(action_revenue.items(), key=lambda item: item[1], reverse=True) return sorted_actions[:top_n] def forecast_revenue(self, planned_activities: List[Dict], conversion_rate: float = 0.1, avg_revenue_per_conversion: float = 5.0) -> float: """ Forecast revenue from planned activities. Args: planned_activities: A list of dictionaries, where each dictionary represents a planned activity. Each activity should have a 'reach' key indicating the potential audience size. conversion_rate: The estimated conversion rate (e.g., 0.1 for 10%). avg_revenue_per_conversion: The average revenue generated per conversion. Returns: The forecasted revenue. """ if not self.redis_client: return 0.0 total_forecasted_revenue = 0.0 for activity in planned_activities: reach = activity.get("reach", 0) forecasted_conversions = reach * conversion_rate forecasted_revenue = forecasted_conversions * avg_revenue_per_conversion total_forecasted_revenue += forecasted_revenue return total_forecasted_revenue def generate_revenue_report(self, start_date: datetime, end_date: datetime) -> Dict: """Generate a revenue report for a given date range.""" if not self.redis_client: return {} total_revenue = 0.0 daily_revenue = {} revenue_logs = self.get_revenue_logs() for log in revenue_logs: log_date = datetime.fromisoformat(log["timestamp"]).date() if start_date.date() <= log_date <= end_date.date(): total_revenue += log["amount"] date_str = log_date.strftime("%Y-%m-%d") if date_str in daily_revenue: daily_revenue[date_str] += log["amount"] else: daily_revenue[date_str] = log["amount"] return { "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), "total_revenue": total_revenue, "daily_revenue": daily_revenue } def check_revenue_targets(self, target_revenue: float, alert_threshold: float = 0.9) -> bool: """Check if revenue targets are being met and trigger alerts if necessary.""" if not self.redis_client: return False total_revenue = float(self.redis_client.get(f"{self.namespace}:total_revenue") or 0.0) if total_revenue >= target_revenue * alert_threshold: print(f"Alert: Revenue is at {total_revenue:.2f}, exceeding {alert_threshold*100}% of the target ({target_revenue:.2f})!") return True return False def get_revenue_logs(self) -> List[Dict]: """Retrieve revenue logs from Redis.""" if not self.redis_client: return [] logs = self.redis_client.lrange(f"{self.namespace}:revenue_logs", 0, -1) return [json.loads(log) for log in logs] def integrate_budget(self, amount: float, model: str): """Integrate with budget constraints (using BudgetManager).""" if self.budget_manager: if self.budget_manager.is_within_budget(): self.budget_manager.log_cost(amount, model) else: print("Budget exceeded! Operation blocked.") return False return True if __name__ == "__main__": from budget_manager import BudgetManager # Import here to avoid circular dependency # Instantiate BudgetManager budget_manager = BudgetManager() # Instantiate RevenueTracker with BudgetManager revenue_tracker = RevenueTracker(budget_manager=budget_manager) # Log some revenue events revenue_tracker.log_revenue(amount=5.0, source="Ad Campaign A", agent="Agent1", task="Lead Generation", metadata={"campaign_id": "123"}) revenue_tracker.log_revenue(amount=12.0, source="Direct Sales", agent="Agent2", task="Customer Acquisition") revenue_tracker.log_revenue(amount=3.0, source="Ad Campaign B", agent="Agent1", task="Lead Generation") # Calculate ROI for Agent1/Lead Generation roi = revenue_tracker.calculate_roi(agent="Agent1", task="Lead Generation") print(f"ROI for Agent1/Lead Generation: {roi}") # Identify highest-value actions highest_value_actions = revenue_tracker.identify_highest_value_actions(top_n=2) print(f"Highest Value Actions: {highest_value_actions}") # Forecast revenue from planned activities planned_activities = [ {"reach": 1000}, {"reach": 500} ] forecasted_revenue = revenue_tracker.forecast_revenue(planned_activities) print(f"Forecasted Revenue: {forecasted_revenue}") # Generate revenue report start_date = datetime(2023, 1, 1) end_date = datetime.now() revenue_report = revenue_tracker.generate_revenue_report(start_date, end_date) print(f"Revenue Report: {revenue_report}") # Check revenue targets revenue_tracker.check_revenue_targets(target_revenue=50.0) # Integrate with budget constraints revenue_tracker.integrate_budget(amount=2.5, model="GPT-3")