#!/usr/bin/env python3 """ Genesis Revenue Attribution Engine =================================== Attributes revenue to AIVA actions with multi-touch support. PM-039: Revenue Attribution Engine - Supports: direct, assisted, influenced attribution - Generates attribution reports - Tracks attribution chains """ import os import json import logging from typing import Dict, Any, List, Optional from datetime import datetime, timedelta from dataclasses import dataclass, asdict from collections import defaultdict from pathlib import Path logger = logging.getLogger("AttributionEngine") logging.basicConfig(level=logging.INFO) @dataclass class AttributionRecord: """Represents an attribution record linking revenue to AIVA action.""" attribution_id: str revenue_event_id: str aiva_action_id: str attribution_type: str # direct, assisted, influenced attribution_weight: float # 0.0-1.0 attributed_amount: float timestamp: str metadata: Dict[str, Any] def to_dict(self) -> Dict[str, Any]: return asdict(self) class AttributionEngine: """ Attributes revenue to AIVA actions using multi-touch attribution models. Supports direct, assisted, and influenced attribution types. """ # Attribution weights for different types ATTRIBUTION_WEIGHTS = { "direct": 1.0, # 100% credit to single action "assisted": 0.5, # 50% credit to assisting actions "influenced": 0.25 # 25% credit to influencing actions } def __init__(self, storage_path: str = None, revenue_tracker=None): """ Initialize attribution engine. Args: storage_path: Path for attribution records storage revenue_tracker: Reference to RevenueTracker instance """ self.storage_path = Path(storage_path or "data/attribution") self.storage_path.mkdir(parents=True, exist_ok=True) self.revenue_tracker = revenue_tracker # Attribution records cache self.attributions: List[AttributionRecord] = [] # Action to attribution mapping self.action_attributions: Dict[str, List[str]] = defaultdict(list) # Load existing records self._load_attributions() logger.info(f"Attribution Engine initialized (storage: {self.storage_path})") def _load_attributions(self) -> None: """Load attribution records from storage.""" records_file = self.storage_path / "attributions.jsonl" if records_file.exists(): try: with open(records_file, 'r') as f: for line in f: if line.strip(): data = json.loads(line) record = AttributionRecord(**data) self.attributions.append(record) self.action_attributions[record.aiva_action_id].append( record.attribution_id ) logger.info(f"Loaded {len(self.attributions)} attribution records") except Exception as e: logger.error(f"Failed to load attributions: {e}") def _save_attribution(self, record: AttributionRecord) -> None: """Persist attribution record.""" records_file = self.storage_path / "attributions.jsonl" try: with open(records_file, 'a') as f: f.write(json.dumps(record.to_dict()) + "\n") except Exception as e: logger.error(f"Failed to save attribution: {e}") def _generate_attribution_id(self) -> str: """Generate unique attribution ID.""" import uuid return f"attr_{uuid.uuid4().hex[:12]}" def attribute(self, revenue_event_id: str, aiva_action_id: str, attribution_type: str = "direct", amount: float = None, metadata: Dict[str, Any] = None) -> AttributionRecord: """ Attribute revenue to an AIVA action. Args: revenue_event_id: ID of the revenue event aiva_action_id: ID of the AIVA action responsible attribution_type: Type (direct, assisted, influenced) amount: Revenue amount (optional, can be fetched from tracker) metadata: Additional attribution metadata Returns: Created AttributionRecord """ if attribution_type not in self.ATTRIBUTION_WEIGHTS: raise ValueError(f"Invalid attribution type: {attribution_type}") weight = self.ATTRIBUTION_WEIGHTS[attribution_type] attributed_amount = (amount or 0) * weight record = AttributionRecord( attribution_id=self._generate_attribution_id(), revenue_event_id=revenue_event_id, aiva_action_id=aiva_action_id, attribution_type=attribution_type, attribution_weight=weight, attributed_amount=attributed_amount, timestamp=datetime.utcnow().isoformat(), metadata=metadata or {} ) # Store record self.attributions.append(record) self.action_attributions[aiva_action_id].append(record.attribution_id) # Persist self._save_attribution(record) logger.info( f"Attribution created: {attribution_type} - " f"Action {aiva_action_id} -> Revenue {revenue_event_id} " f"(${attributed_amount:.2f})" ) return record def attribute_multi_touch(self, revenue_event_id: str, action_chain: List[Dict[str, str]], total_amount: float) -> List[AttributionRecord]: """ Apply multi-touch attribution across a chain of actions. Args: revenue_event_id: ID of the revenue event action_chain: List of dicts with action_id and attribution_type total_amount: Total revenue amount to attribute Returns: List of created AttributionRecords """ records = [] # Calculate total weight for normalization total_weight = sum( self.ATTRIBUTION_WEIGHTS.get(a.get("type", "influenced"), 0.25) for a in action_chain ) for action in action_chain: action_id = action.get("action_id") attr_type = action.get("type", "influenced") if not action_id: continue weight = self.ATTRIBUTION_WEIGHTS.get(attr_type, 0.25) # Normalize weights so they sum to 1.0 normalized_weight = weight / total_weight attributed_amount = total_amount * normalized_weight record = self.attribute( revenue_event_id=revenue_event_id, aiva_action_id=action_id, attribution_type=attr_type, amount=attributed_amount / weight, # Pre-adjust for weight metadata={"multi_touch": True, "chain_position": action_chain.index(action)} ) records.append(record) return records def get_action_revenue(self, aiva_action_id: str) -> Dict[str, Any]: """ Get total attributed revenue for an AIVA action. Args: aiva_action_id: ID of the AIVA action Returns: Dict with total and breakdown by attribution type """ attribution_ids = self.action_attributions.get(aiva_action_id, []) records = [a for a in self.attributions if a.attribution_id in attribution_ids] total = sum(r.attributed_amount for r in records) by_type = defaultdict(float) for record in records: by_type[record.attribution_type] += record.attributed_amount return { "action_id": aiva_action_id, "total_attributed": total, "by_type": dict(by_type), "attribution_count": len(records) } def get_attributions_for_revenue(self, revenue_event_id: str) -> List[Dict[str, Any]]: """Get all attributions for a revenue event.""" records = [a for a in self.attributions if a.revenue_event_id == revenue_event_id] return [r.to_dict() for r in records] def generate_report(self, start_date: str = None, end_date: str = None) -> Dict[str, Any]: """ Generate attribution report for a date range. Args: start_date: Start date (YYYY-MM-DD) end_date: End date (YYYY-MM-DD) Returns: Comprehensive attribution report """ # Default to last 30 days if not end_date: end_date = datetime.utcnow().strftime("%Y-%m-%d") if not start_date: start_date = (datetime.utcnow() - timedelta(days=30)).strftime("%Y-%m-%d") # Filter attributions by date filtered = [ a for a in self.attributions if start_date <= a.timestamp[:10] <= end_date ] # Calculate totals total_attributed = sum(a.attributed_amount for a in filtered) # By type breakdown by_type = defaultdict(lambda: {"count": 0, "amount": 0.0}) for attr in filtered: by_type[attr.attribution_type]["count"] += 1 by_type[attr.attribution_type]["amount"] += attr.attributed_amount # Top actions action_totals = defaultdict(float) for attr in filtered: action_totals[attr.aiva_action_id] += attr.attributed_amount top_actions = sorted( action_totals.items(), key=lambda x: x[1], reverse=True )[:10] # Daily breakdown daily = defaultdict(float) for attr in filtered: date_key = attr.timestamp[:10] daily[date_key] += attr.attributed_amount return { "report_period": { "start": start_date, "end": end_date }, "summary": { "total_attributed_revenue": total_attributed, "total_attributions": len(filtered), "unique_actions": len(set(a.aiva_action_id for a in filtered)), "unique_revenue_events": len(set(a.revenue_event_id for a in filtered)) }, "by_attribution_type": dict(by_type), "top_actions": [ {"action_id": action_id, "attributed_revenue": amount} for action_id, amount in top_actions ], "daily_breakdown": dict(sorted(daily.items())), "generated_at": datetime.utcnow().isoformat() } def get_attribution_funnel(self, revenue_event_id: str) -> Dict[str, Any]: """ Get the attribution funnel for a revenue event. Shows all actions that contributed to the revenue. Args: revenue_event_id: Revenue event ID Returns: Attribution funnel with all contributing actions """ attributions = self.get_attributions_for_revenue(revenue_event_id) funnel = { "revenue_event_id": revenue_event_id, "total_attributions": len(attributions), "direct": [], "assisted": [], "influenced": [] } for attr in attributions: attr_type = attr.get("attribution_type", "influenced") funnel[attr_type].append({ "action_id": attr.get("aiva_action_id"), "amount": attr.get("attributed_amount"), "timestamp": attr.get("timestamp") }) return funnel def calculate_roi_by_action_type(self) -> Dict[str, Dict[str, float]]: """ Calculate ROI metrics grouped by action type patterns. Returns: ROI metrics by action type """ # Group by action patterns (extracted from metadata) action_types = defaultdict(lambda: {"revenue": 0.0, "count": 0}) for attr in self.attributions: # Try to extract action type from metadata or action_id action_type = attr.metadata.get("action_type", "unknown") if action_type == "unknown" and attr.aiva_action_id: # Extract type from action_id pattern if "workflow" in attr.aiva_action_id.lower(): action_type = "workflow" elif "funnel" in attr.aiva_action_id.lower(): action_type = "funnel" elif "voice" in attr.aiva_action_id.lower(): action_type = "voice_ai" else: action_type = "other" action_types[action_type]["revenue"] += attr.attributed_amount action_types[action_type]["count"] += 1 return dict(action_types) def get_metrics(self) -> Dict[str, Any]: """Get comprehensive attribution metrics.""" return { "total_attributions": len(self.attributions), "total_attributed_revenue": sum(a.attributed_amount for a in self.attributions), "unique_actions": len(self.action_attributions), "by_type": { "direct": sum(1 for a in self.attributions if a.attribution_type == "direct"), "assisted": sum(1 for a in self.attributions if a.attribution_type == "assisted"), "influenced": sum(1 for a in self.attributions if a.attribution_type == "influenced") } } # Global instance _engine: Optional[AttributionEngine] = None def get_engine() -> AttributionEngine: """Get or create global attribution engine instance.""" global _engine if _engine is None: _engine = AttributionEngine() return _engine if __name__ == "__main__": # Self-test engine = AttributionEngine(storage_path="/tmp/genesis_attribution_test") # Test direct attribution record1 = engine.attribute( revenue_event_id="rev_123", aiva_action_id="aiva_workflow_001", attribution_type="direct", amount=500.00, metadata={"action_type": "workflow"} ) print(f"Direct Attribution: {record1.to_dict()}") # Test assisted attribution record2 = engine.attribute( revenue_event_id="rev_123", aiva_action_id="aiva_email_002", attribution_type="assisted", amount=500.00, metadata={"action_type": "email"} ) print(f"Assisted Attribution: {record2.to_dict()}") # Test multi-touch attribution records = engine.attribute_multi_touch( revenue_event_id="rev_456", action_chain=[ {"action_id": "aiva_ad_001", "type": "influenced"}, {"action_id": "aiva_email_003", "type": "assisted"}, {"action_id": "aiva_call_001", "type": "direct"} ], total_amount=1000.00 ) print(f"\nMulti-touch Attribution ({len(records)} records):") for r in records: print(f" - {r.attribution_type}: ${r.attributed_amount:.2f}") # Generate report print("\n=== Attribution Report ===") report = engine.generate_report() print(json.dumps(report, indent=2)) # Action revenue print(f"\nAction Revenue: {engine.get_action_revenue('aiva_workflow_001')}")