# revenue_tracker.py import json import time import os import redis from datetime import datetime, timedelta from typing import Dict, Any, List class RevenueTracker: """ Tracks revenue, calculates ROI, forecasts revenue, and generates reports. """ def __init__(self, budget_manager, redis_config=None, namespace="genesis:revenue"): self.namespace = namespace self.redis_client = None self.budget_manager = budget_manager # Inject BudgetManager # Load Redis config from argument, or if not provided, from environment variable or default path 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.revenue_target = 1000.0 # Example target self.alert_threshold = 0.9 # Alert when 90% of target is reached def log_revenue(self, amount: float, activity: str, agent: str, cost: float = 0.0): """Log a revenue-generating event.""" if not self.redis_client: return now = datetime.now() timestamp = now.isoformat() revenue_event = { "timestamp": timestamp, "amount": amount, "activity": activity, "agent": agent, "cost": cost, "profit": amount - cost } self.redis_client.lpush(f"{self.namespace}:events", json.dumps(revenue_event)) self.redis_client.ltrim(f"{self.namespace}:events", 0, 1000) # Keep last 1000 events # Update total revenue self.redis_client.incrbyfloat(f"{self.namespace}:total", amount) # Agent-specific revenue self.redis_client.incrbyfloat(f"{self.namespace}:agent:{agent}", amount) # Activity-specific revenue self.redis_client.incrbyfloat(f"{self.namespace}:activity:{activity}", amount) # Cost attribution via BudgetManager self.budget_manager.log_cost(cost, "revenue_activity") self._check_target_alert() def calculate_roi(self, agent: str) -> float: """Calculates ROI for a given agent.""" if not self.redis_client: return 0.0 total_revenue = float(self.redis_client.get(f"{self.namespace}:agent:{agent}") or 0.0) # Assuming cost is tracked separately via budget manager # In a real system, this would require more sophisticated cost tracking per agent total_cost = self.budget_manager.get_current_spend() # Approximation if total_cost == 0: return float('inf') if total_revenue > 0 else 0.0 return (total_revenue - total_cost) / total_cost def get_highest_value_actions(self, top_n: int = 5) -> List[Dict[str, Any]]: """Identifies the highest-value revenue-generating actions.""" if not self.redis_client: return [] activity_revenues = {} keys = self.redis_client.keys(f"{self.namespace}:activity:*") for key in keys: activity = key.split(":")[-1] revenue = float(self.redis_client.get(key) or 0.0) activity_revenues[activity] = revenue sorted_activities = sorted(activity_revenues.items(), key=lambda item: item[1], reverse=True) return [{"activity": k, "revenue": v} for k, v in sorted_activities[:top_n]] def forecast_revenue(self, planned_activities: List[Dict[str, Any]]) -> float: """Predicts revenue from planned activities (simplified).""" # This is a placeholder. A real implementation would use a model. total_predicted_revenue = 0.0 for activity in planned_activities: # Simple heuristic: Assume each planned activity generates 10% more than the average # of the top 3 highest value actions. top_actions = self.get_highest_value_actions(top_n=3) if top_actions: avg_revenue = sum(action["revenue"] for action in top_actions) / len(top_actions) predicted_revenue = avg_revenue * 1.10 # 10% increase total_predicted_revenue += predicted_revenue return total_predicted_revenue def generate_revenue_report(self) -> Dict[str, Any]: """Generates a revenue report.""" if not self.redis_client: return {} total_revenue = float(self.redis_client.get(f"{self.namespace}:total") or 0.0) events = [] event_strings = self.redis_client.lrange(f"{self.namespace}:events", 0, -1) for event_str in event_strings: events.append(json.loads(event_str)) report = { "total_revenue": total_revenue, "events": events, "highest_value_actions": self.get_highest_value_actions(), } return report def set_revenue_target(self, target: float): """Sets the revenue target.""" self.revenue_target = target def _check_target_alert(self): """Checks if the revenue target is close to being reached and triggers an alert.""" if not self.redis_client: return total_revenue = float(self.redis_client.get(f"{self.namespace}:total") or 0.0) if total_revenue >= self.revenue_target * self.alert_threshold: print(f"ALERT: Revenue target is {self.alert_threshold*100}% reached! Current revenue: ${total_revenue}, Target: ${self.revenue_target}") def get_total_revenue(self) -> float: """Returns the total revenue.""" if not self.redis_client: return 0.0 return float(self.redis_client.get(f"{self.namespace}:total") or 0.0) if __name__ == "__main__": # Example Usage from budget_manager import BudgetManager # Initialize BudgetManager (ensure Redis is running) bm = BudgetManager() # Use default budget limit # Initialize RevenueTracker, injecting BudgetManager rt = RevenueTracker(budget_manager=bm) # Log some revenue events rt.log_revenue(amount=50.0, activity="Lead Generation", agent="AgentA", cost=5.0) rt.log_revenue(amount=75.0, activity="Sales Conversion", agent="AgentB", cost=10.0) rt.log_revenue(amount=30.0, activity="Customer Support", agent="AgentA", cost=2.0) # Calculate ROI for AgentA roi_agent_a = rt.calculate_roi(agent="AgentA") print(f"ROI for AgentA: {roi_agent_a}") # Get highest value actions highest_value_actions = rt.get_highest_value_actions() print(f"Highest Value Actions: {highest_value_actions}") # Forecast revenue planned_activities = [ {"activity": "Marketing Campaign"}, {"activity": "New Product Launch"} ] forecasted_revenue = rt.forecast_revenue(planned_activities) print(f"Forecasted Revenue: {forecasted_revenue}") # Generate revenue report revenue_report = rt.generate_revenue_report() print(f"Revenue Report: {revenue_report}") # Check total revenue total_revenue = rt.get_total_revenue() print(f"Total Revenue: ${total_revenue}") # Trigger alert by setting target low rt.set_revenue_target(150.0) rt._check_target_alert() # Example of reading Redis config from environment variable # os.environ["REDIS_CONFIG"] = json.dumps({"host": "localhost", "port": 6379, "password": ""}) # rt_env = RevenueTracker(budget_manager=bm, redis_config=json.loads(os.environ["REDIS_CONFIG"]))