import datetime import uuid import random import pandas as pd import numpy as np class RevenueAnalyticsEngine: """ A comprehensive Revenue Analytics Engine for tracking patent monetization. """ def __init__(self, db_engine): """ Initializes the RevenueAnalyticsEngine. Args: db_engine: A database engine object (e.g., SQLAlchemy engine) to interact with the database. For this example, we'll simulate a database with Pandas DataFrames. """ self.db_engine = db_engine self.usage_events = pd.DataFrame(columns=['event_id', 'timestamp', 'patent_id', 'customer_id', 'api_endpoint', 'processing_time_ms', 'billed', 'validation_type']) self.customers = pd.DataFrame(columns=['customer_id', 'name', 'contract_type', 'volume_discount', 'enterprise_discount', 'signup_date', 'churned']) self.invoices = pd.DataFrame(columns=['invoice_id', 'customer_id', 'issue_date', 'due_date', 'total_amount', 'paid']) self.metrics_aggregations = pd.DataFrame(columns=['date', 'metric_type', 'value']) self.pricing = {'validation_base': 1.0, 'validation_premium': 5.0} # Example pricing def track_usage(self, patent_id, customer_id, api_endpoint, processing_time_ms, validation_type): """ Tracks API usage events. Args: patent_id: The ID of the patent used. customer_id: The ID of the customer making the API call. api_endpoint: The API endpoint called. processing_time_ms: The processing time in milliseconds. validation_type: Type of validation (e.g., 'base', 'premium') """ event_id = uuid.uuid4() timestamp = datetime.datetime.now() new_event = pd.DataFrame([{ 'event_id': event_id, 'timestamp': timestamp, 'patent_id': patent_id, 'customer_id': customer_id, 'api_endpoint': api_endpoint, 'processing_time_ms': processing_time_ms, 'billed': False, 'validation_type': validation_type }]) self.usage_events = pd.concat([self.usage_events, new_event], ignore_index=True) return event_id def calculate_bill(self, customer_id, validation_type): """ Calculates the bill for a single validation event. Args: customer_id: The ID of the customer. validation_type: The type of validation. Returns: The bill amount. """ customer = self.customers[self.customers['customer_id'] == customer_id] if customer.empty: return 0 # Customer not found contract_type = customer['contract_type'].values[0] volume_discount = customer['volume_discount'].values[0] enterprise_discount = customer['enterprise_discount'].values[0] base_price = self.pricing.get(f'validation_{validation_type}', self.pricing['validation_base']) # Apply discounts price = base_price * (1 - volume_discount) * (1 - enterprise_discount) return price def generate_invoice(self, customer_id, start_date, end_date): """ Generates an invoice for a customer for a given period. Args: customer_id: The ID of the customer. start_date: The start date of the billing period. end_date: The end date of the billing period. Returns: The invoice ID. """ usage_events = self.usage_events[ (self.usage_events['customer_id'] == customer_id) & (self.usage_events['timestamp'] >= start_date) & (self.usage_events['timestamp'] <= end_date) & (self.usage_events['billed'] == False) ] if usage_events.empty: return None # No usage for the period total_amount = 0 for index, row in usage_events.iterrows(): validation_type = row['validation_type'] bill_amount = self.calculate_bill(customer_id, validation_type) total_amount += bill_amount self.usage_events.loc[index, 'billed'] = True invoice_id = uuid.uuid4() issue_date = datetime.datetime.now() due_date = issue_date + datetime.timedelta(days=30) new_invoice = pd.DataFrame([{ 'invoice_id': invoice_id, 'customer_id': customer_id, 'issue_date': issue_date, 'due_date': due_date, 'total_amount': total_amount, 'paid': False }]) self.invoices = pd.concat([self.invoices, new_invoice], ignore_index=True) return invoice_id def revenue_by_patent(self, start_date=None, end_date=None): """ Calculates revenue generated by each patent. Args: start_date: Optional start date for the analysis. end_date: Optional end date for the analysis. Returns: A Pandas DataFrame with revenue by patent. """ filtered_invoices = self.invoices if start_date: filtered_invoices = filtered_invoices[filtered_invoices['issue_date'] >= start_date] if end_date: filtered_invoices = filtered_invoices[filtered_invoices['issue_date'] <= end_date] if filtered_invoices.empty: return pd.DataFrame(columns=['patent_id', 'revenue']) patent_revenue = {} for index, row in self.usage_events.iterrows(): if row['billed']: invoice_id = self.invoices[(self.invoices['customer_id'] == row['customer_id']) & (self.invoices['issue_date'] >= start_date) & (self.invoices['issue_date'] <= end_date)]['invoice_id'].values if len(invoice_id) > 0: customer_id = row['customer_id'] validation_type = row['validation_type'] bill_amount = self.calculate_bill(customer_id, validation_type) patent_id = row['patent_id'] if patent_id in patent_revenue: patent_revenue[patent_id] += bill_amount else: patent_revenue[patent_id] = bill_amount patent_revenue_df = pd.DataFrame(list(patent_revenue.items()), columns=['patent_id', 'revenue']) return patent_revenue_df def revenue_by_customer(self, start_date=None, end_date=None): """ Calculates revenue generated by each customer. Args: start_date: Optional start date for the analysis. end_date: Optional end date for the analysis. Returns: A Pandas DataFrame with revenue by customer. """ filtered_invoices = self.invoices if start_date: filtered_invoices = filtered_invoices[filtered_invoices['issue_date'] >= start_date] if end_date: filtered_invoices = filtered_invoices[filtered_invoices['issue_date'] <= end_date] customer_revenue = filtered_invoices.groupby('customer_id')['total_amount'].sum().reset_index() return customer_revenue def daily_revenue_trend(self, start_date=None, end_date=None): """ Calculates the daily revenue trend. Args: start_date: Optional start date for the analysis. end_date: Optional end date for the analysis. Returns: A Pandas DataFrame with daily revenue trend. """ filtered_invoices = self.invoices if start_date: filtered_invoices = filtered_invoices[filtered_invoices['issue_date'] >= start_date] if end_date: filtered_invoices = filtered_invoices[filtered_invoices['issue_date'] <= end_date] daily_revenue = filtered_invoices.groupby(filtered_invoices['issue_date'].dt.date)['total_amount'].sum().reset_index() daily_revenue.rename(columns={'issue_date': 'date', 'total_amount': 'revenue'}, inplace=True) return daily_revenue def cohort_analysis(self, cohort_period='monthly'): """ Performs cohort analysis based on signup date. Args: cohort_period: 'monthly' or 'weekly' Returns: A Pandas DataFrame with cohort analysis results. """ if self.customers.empty: return pd.DataFrame() # Create cohort groups based on signup date if cohort_period == 'monthly': self.customers['cohort_group'] = self.customers['signup_date'].dt.to_period('M') elif cohort_period == 'weekly': self.customers['cohort_group'] = self.customers['signup_date'].dt.to_period('W') else: raise ValueError("Invalid cohort period. Choose 'monthly' or 'weekly'.") # Calculate retention rate cohort_data = self.customers.groupby('cohort_group')['customer_id'].nunique().reset_index() cohort_data.rename(columns={'customer_id': 'total_customers'}, inplace=True) # Calculate active customers each period retention = self.customers.groupby(['cohort_group', 'churned'])['customer_id'].nunique().unstack(fill_value=0) retention['retained'] = retention[False] retention = retention[['retained']] retention.reset_index(inplace=True) cohort_analysis = pd.merge(cohort_data, retention, on='cohort_group', how='left') cohort_analysis['retention_rate'] = cohort_analysis['retained'] / cohort_analysis['total_customers'] return cohort_analysis def calculate_cost_per_validation(self, patent_id=None): """ Calculates the cost per validation for a given patent or overall. Args: patent_id: Optional patent ID to filter by. Returns: The cost per validation. """ # This would require integration with a cost tracking system. # For now, we'll simulate it with a fixed cost per validation. fixed_cost_per_validation = 0.1 # Example: $0.1 per validation if patent_id: validation_count = len(self.usage_events[self.usage_events['patent_id'] == patent_id]) else: validation_count = len(self.usage_events) if validation_count == 0: return 0 return fixed_cost_per_validation def calculate_profit_margin(self, patent_id=None): """ Calculates the profit margin for a given patent or overall. Args: patent_id: Optional patent ID to filter by. Returns: The profit margin. """ revenue = 0 if patent_id: revenue_data = self.revenue_by_patent(start_date=datetime.datetime(2023, 1, 1), end_date=datetime.datetime.now()) if not revenue_data.empty: revenue = revenue_data[revenue_data['patent_id']==patent_id]['revenue'].values[0] else: revenue = self.invoices['total_amount'].sum() cost = self.calculate_cost_per_validation(patent_id) * len(self.usage_events) if revenue == 0: return 0 profit_margin = (revenue - cost) / revenue return profit_margin def calculate_customer_lifetime_value(self, customer_id): """ Calculates the customer lifetime value (CLTV). This is a simplified CLTV calculation. Args: customer_id: The ID of the customer. Returns: The estimated CLTV. """ customer_revenue = self.revenue_by_customer(start_date=self.customers[self.customers['customer_id'] == customer_id]['signup_date'].values[0], end_date=datetime.datetime.now()) if customer_revenue.empty: return 0 avg_monthly_revenue = customer_revenue['total_amount'].mean() customer_lifespan_months = 36 # Assuming a 3-year lifespan churn_rate = 0.1 # Assuming a 10% churn rate cltv = (avg_monthly_revenue * customer_lifespan_months) / (1 + churn_rate) return cltv def predict_churn(self, customer_id): """ Predicts whether a customer is likely to churn. (This is a placeholder for a real churn prediction model). Args: customer_id: The ID of the customer. Returns: True if the customer is predicted to churn, False otherwise. """ # Simulate churn prediction based on random chance. # A real implementation would use machine learning models. return random.random() < 0.2 # 20% chance of churn prediction def generate_sample_data(self, num_customers=5, num_events=20): """ Generates sample data for testing purposes. """ self.customers = pd.DataFrame({ 'customer_id': [uuid.uuid4() for _ in range(num_customers)], 'name': [f'Customer {i}' for i in range(num_customers)], 'contract_type': ['standard'] * num_customers, 'volume_discount': [0.05] * num_customers, 'enterprise_discount': [0.0] * num_customers, 'signup_date': [datetime.datetime.now() - datetime.timedelta(days=random.randint(30, 365)) for _ in range(num_customers)], 'churned': [False] * num_customers }) patents = ['patent1', 'patent2', 'patent3'] api_endpoints = ['/validate', '/analyze'] validation_types = ['base', 'premium'] for _ in range(num_events): customer_id = random.choice(self.customers['customer_id'].tolist()) patent_id = random.choice(patents) api_endpoint = random.choice(api_endpoints) processing_time_ms = random.randint(100, 500) validation_type = random.choice(validation_types) self.track_usage(patent_id, customer_id, api_endpoint, processing_time_ms, validation_type) def run_simulation(self): """ Runs a simulation to demonstrate the engine's capabilities. """ self.generate_sample_data() # Generate invoices for all customers for customer_id in self.customers['customer_id']: invoice_id = self.generate_invoice(customer_id, datetime.datetime(2023, 1, 1), datetime.datetime.now()) if invoice_id: print(f"Generated invoice {invoice_id} for customer {customer_id}") # Perform analytics queries print("\nRevenue by Patent:") print(self.revenue_by_patent()) print("\nRevenue by Customer:") print(self.revenue_by_customer()) print("\nDaily Revenue Trend:") print(self.daily_revenue_trend()) print("\nCohort Analysis:") print(self.cohort_analysis()) # Calculate optimization metrics print("\nCost per Validation:") print(self.calculate_cost_per_validation()) print("\nProfit Margin:") print(self.calculate_profit_margin()) customer_id = self.customers['customer_id'].iloc[0] #Grab first customer print(f"\nCustomer Lifetime Value for {customer_id}:") print(self.calculate_customer_lifetime_value(customer_id)) print(f"\nChurn Prediction for {customer_id}:") print(self.predict_churn(customer_id)) # Example Usage (Simulated Database) if __name__ == '__main__': # For demonstration, we'll simulate a database with Pandas DataFrames. # In a real application, replace this with a database engine (e.g., SQLAlchemy). class MockDBEngine: pass # Replace with your actual database engine db_engine = MockDBEngine() engine = RevenueAnalyticsEngine(db_engine) engine.run_simulation()