# 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, and provides revenue forecasting. """ def __init__(self): self.namespace = "genesis:revenue" self.redis_client = None # 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 ) self.profit_margin = 0.2 # Default profit margin def log_revenue(self, amount: float, source: str, agent: str = "N/A", task: str = "N/A"): """Log a revenue-generating event.""" if not self.redis_client: return now = datetime.now() timestamp = now.isoformat() revenue_entry = { "timestamp": timestamp, "amount": amount, "source": source, "agent": agent, "task": task } # Store revenue event self.redis_client.lpush(f"{self.namespace}:events", json.dumps(revenue_entry)) self.redis_client.ltrim(f"{self.namespace}:events", 0, 10000) # Limit log size # Update total revenue self.redis_client.incrbyfloat(f"{self.namespace}:total", amount) # Update agent-specific revenue self.redis_client.incrbyfloat(f"{self.namespace}:agent:{agent}", amount) # Update task-specific revenue self.redis_client.incrbyfloat(f"{self.namespace}:task:{task}", amount) def attribute_cost(self, agent: str, task: str, cost: float): """Attribute cost to a specific agent/task for ROI calculation.""" if not self.redis_client: return self.redis_client.incrbyfloat(f"{self.namespace}:cost:agent:{agent}", cost) self.redis_client.incrbyfloat(f"{self.namespace}:cost:task:{task}", cost) def calculate_roi(self, agent: str = None, task: str = None) -> float: """Calculate ROI for a specific agent or task.""" if not self.redis_client: return 0.0 if agent: revenue = float(self.redis_client.get(f"{self.namespace}:agent:{agent}") or 0.0) cost = float(self.redis_client.get(f"{self.namespace}:cost:agent:{agent}") or 0.0) elif task: revenue = float(self.redis_client.get(f"{self.namespace}:task:{task}") or 0.0) cost = float(self.redis_client.get(f"{self.namespace}:cost:task:{task}") or 0.0) else: return 0.0 # Need to specify agent or task if cost == 0: return float('inf') if revenue > 0 else 0.0 # Avoid division by zero return (revenue - cost) / cost def identify_highest_value_actions(self, top_n: int = 5) -> List[Tuple[str, float]]: """Identify the top N highest-value tasks based on ROI.""" if not self.redis_client: return [] task_keys = [key.split(":")[-1] for key in self.redis_client.keys(f"{self.namespace}:task:*") if "cost" not in key] task_rois = [] for task in task_keys: roi = self.calculate_roi(task=task) task_rois.append((task, roi)) task_rois.sort(key=lambda item: item[1], reverse=True) return task_rois[:top_n] def predict_revenue(self, planned_activities: List[Dict], conversion_rate: float) -> float: """Predict revenue from planned activities based on conversion rate.""" # planned_activities: [{"activity": "email_campaign", "target_audience": 1000}, ...] predicted_revenue = 0.0 for activity in planned_activities: # Placeholder: Assume each converted member generates $X revenue revenue_per_conversion = 10.0 # Example value predicted_conversions = activity["target_audience"] * conversion_rate predicted_revenue += predicted_conversions * revenue_per_conversion return predicted_revenue def generate_revenue_report(self) -> Dict: """Generate a summary revenue report.""" if not self.redis_client: return {} total_revenue = float(self.redis_client.get(f"{self.namespace}:total") or 0.0) agent_revenues = {} task_revenues = {} for key in self.redis_client.keys(f"{self.namespace}:agent:*"): if "cost" not in key: agent = key.split(":")[-1] agent_revenues[agent] = float(self.redis_client.get(key) or 0.0) for key in self.redis_client.keys(f"{self.namespace}:task:*"): if "cost" not in key: task = key.split(":")[-1] task_revenues[task] = float(self.redis_client.get(key) or 0.0) return { "total_revenue": total_revenue, "agent_revenues": agent_revenues, "task_revenues": task_revenues } def set_profit_margin(self, margin: float): """Set the desired profit margin.""" if 0 <= margin <= 1: self.profit_margin = margin else: raise ValueError("Profit margin must be between 0 and 1.") def calculate_profit(self, revenue: float) -> float: """Calculate profit based on the set profit margin.""" return revenue * self.profit_margin def check_revenue_target(self, target_revenue: float) -> bool: """Check if a specific revenue target has been achieved.""" if not self.redis_client: return False total_revenue = float(self.redis_client.get(f"{self.namespace}:total") or 0.0) return total_revenue >= target_revenue def get_total_revenue(self) -> float: """Get the total revenue.""" if not self.redis_client: return 0.0 return float(self.redis_client.get(f"{self.namespace}:total") or 0.0) def get_revenue_events(self, count: int = 10) -> List[Dict]: """Retrieve the most recent revenue events.""" if not self.redis_client: return [] events = [] for event_json in self.redis_client.lrange(f"{self.namespace}:events", 0, count - 1): events.append(json.loads(event_json)) return events if __name__ == "__main__": rt = RevenueTracker() # Log some revenue events rt.log_revenue(100, "Website", agent="Agent1", task="Sales") rt.log_revenue(50, "Referral", agent="Agent2", task="Marketing") rt.log_revenue(75, "Website", agent="Agent1", task="Sales") # Attribute some costs rt.attribute_cost("Agent1", "Sales", 25) rt.attribute_cost("Agent2", "Marketing", 15) # Calculate ROI roi_agent1 = rt.calculate_roi(agent="Agent1") roi_marketing = rt.calculate_roi(task="Marketing") print(f"ROI for Agent1: {roi_agent1}") print(f"ROI for Marketing task: {roi_marketing}") # Identify highest value actions top_actions = rt.identify_highest_value_actions(top_n=3) print(f"Top actions: {top_actions}") # Predict revenue planned_activities = [ {"activity": "email_campaign", "target_audience": 1000}, {"activity": "social_media_ad", "target_audience": 500} ] predicted_revenue = rt.predict_revenue(planned_activities, conversion_rate=0.01) print(f"Predicted revenue: {predicted_revenue}") # Generate revenue report report = rt.generate_revenue_report() print(f"Revenue report: {report}") # Check revenue target target_achieved = rt.check_revenue_target(target_revenue=200) print(f"Target achieved: {target_achieved}") # Get the last 5 revenue events recent_events = rt.get_revenue_events(count=5) print(f"Recent revenue events: {recent_events}")