Omnichannel Marketing Orchestration with Hypergraph: Intelligent Campaign Automation

from dataclasses import dataclass, field
from typing import Dict, List, Set, Optional
from datetime import datetime, timedelta
from enum import Enum

class Channel(Enum):
    EMAIL = "email"
    SMS = "sms"
    PUSH = "push"
    DISPLAY = "display"
    SOCIAL = "social"
    DIRECT_MAIL = "direct_mail"

@dataclass
class CustomerSegment:
    id: str
    name: str
    size: int
    criteria: Dict[str, str]

@dataclass
class CampaignHyperedge:
    """Hyperedge representing a campaign action"""
    id: str
    segment_id: str
    channel: Channel
    content_id: str
    time_window_start: datetime
    time_window_end: datetime
    budget_allocation: float
    priority: int
    frequency_cap: int  # Max touches per customer per week
    objective: str  # conversion, engagement, awareness

@dataclass
class FatigueState:
    """Track customer fatigue across channels"""
    customer_id: str
    channel_touches: Dict[Channel, int]  # Touches per channel this week
    total_touches: int
    last_touch: datetime

class HypergraphOrchestrator:
    """Orchestration engine using hypergraph model"""

    def __init__(self):
        self.hyperedges: List[CampaignHyperedge] = []
        self.segments: Dict[str, CustomerSegment] = {}
        self.fatigue_states: Dict[str, FatigueState] = {}

        # Channel capacity (messages per day)
        self.channel_capacity: Dict[Channel, int] = {
            Channel.EMAIL: 1000000,
            Channel.SMS: 100000,
            Channel.PUSH: 500000,
            Channel.DISPLAY: 10000000,
            Channel.SOCIAL: 5000000,
            Channel.DIRECT_MAIL: 10000,
        }

        # Global frequency caps
        self.global_weekly_cap = 7
        self.channel_weekly_caps: Dict[Channel, int] = {
            Channel.EMAIL: 3,
            Channel.SMS: 2,
            Channel.PUSH: 5,
            Channel.DISPLAY: 20,
            Channel.SOCIAL: 10,
            Channel.DIRECT_MAIL: 1,
        }

    def add_segment(self, segment: CustomerSegment) -> None:
        self.segments[segment.id] = segment

    def add_campaign_hyperedge(self, hyperedge: CampaignHyperedge) -> None:
        self.hyperedges.append(hyperedge)

    def check_eligibility(
        self,
        customer_id: str,
        hyperedge: CampaignHyperedge
    ) -> tuple[bool, str]:
        """Check if customer is eligible for campaign action"""

        # Get or create fatigue state
        if customer_id not in self.fatigue_states:
            self.fatigue_states[customer_id] = FatigueState(
                customer_id=customer_id,
                channel_touches={c: 0 for c in Channel},
                total_touches=0,
                last_touch=datetime.min,
            )

        state = self.fatigue_states[customer_id]

        # Check global fatigue
        if state.total_touches >= self.global_weekly_cap:
            return False, "Global fatigue cap reached"

        # Check channel fatigue
        channel_cap = self.channel_weekly_caps.get(hyperedge.channel, 10)
        if state.channel_touches.get(hyperedge.channel, 0) >= channel_cap:
            return False, f"Channel fatigue cap reached for {hyperedge.channel.value}"

        # Check campaign-specific frequency cap
        # (simplified - would need per-campaign tracking)
        if state.channel_touches.get(hyperedge.channel, 0) >= hyperedge.frequency_cap:
            return False, "Campaign frequency cap reached"

        # Check time window
        now = datetime.utcnow()
        if now < hyperedge.time_window_start or now > hyperedge.time_window_end:
            return False, "Outside campaign time window"

        return True, "Eligible"

    def select_next_best_action(
        self,
        customer_id: str,
        context: Dict[str, str]
    ) -> Optional[CampaignHyperedge]:
        """Select the best campaign action for a customer"""

        eligible_actions = []

        for hyperedge in self.hyperedges:
            is_eligible, reason = self.check_eligibility(customer_id, hyperedge)
            if is_eligible:
                eligible_actions.append(hyperedge)

        if not eligible_actions:
            return None

        # Score and rank eligible actions
        scored_actions = []
        for action in eligible_actions:
            score = self._score_action(customer_id, action, context)
            scored_actions.append((score, action))

        # Sort by score descending
        scored_actions.sort(key=lambda x: x[0], reverse=True)

        return scored_actions[0][1] if scored_actions else None

    def _score_action(
        self,
        customer_id: str,
        action: CampaignHyperedge,
        context: Dict[str, str]
    ) -> float:
        """Score a campaign action for a customer"""
        score = 0.0

        # Priority weight
        score += action.priority * 10

        # Channel affinity (simplified - would use ML model)
        channel_affinity = {
            Channel.EMAIL: 0.7,
            Channel.SMS: 0.5,
            Channel.PUSH: 0.6,
            Channel.DISPLAY: 0.3,
            Channel.SOCIAL: 0.4,
            Channel.DIRECT_MAIL: 0.2,
        }
        score += channel_affinity.get(action.channel, 0.5) * 20

        # Time relevance
        now = datetime.utcnow()
        time_to_end = (action.time_window_end - now).total_seconds() / 3600
        urgency_score = min(1.0, 24 / max(time_to_end, 1)) * 10
        score += urgency_score

        # Budget remaining (simplified)
        score += min(action.budget_allocation / 1000, 10)

        return score

    def record_touch(
        self,
        customer_id: str,
        hyperedge: CampaignHyperedge
    ) -> None:
        """Record that a campaign action was delivered"""

        if customer_id not in self.fatigue_states:
            self.fatigue_states[customer_id] = FatigueState(
                customer_id=customer_id,
                channel_touches={c: 0 for c in Channel},
                total_touches=0,
                last_touch=datetime.min,
            )

        state = self.fatigue_states[customer_id]
        state.channel_touches[hyperedge.channel] = \
            state.channel_touches.get(hyperedge.channel, 0) + 1
        state.total_touches += 1
        state.last_touch = datetime.utcnow()

    def get_orchestration_plan(
        self,
        segment_id: str,
        horizon_days: int = 7
    ) -> List[Dict]:
        """Generate orchestration plan for a segment"""

        segment = self.segments.get(segment_id)
        if not segment:
            return []

        plan = []
        for hyperedge in self.hyperedges:
            if hyperedge.segment_id == segment_id:
                plan.append({
                    "campaign_id": hyperedge.id,
                    "channel": hyperedge.channel.value,
                    "content_id": hyperedge.content_id,
                    "time_window": f"{hyperedge.time_window_start} - {hyperedge.time_window_end}",
                    "budget": hyperedge.budget_allocation,
                    "estimated_reach": int(segment.size * 0.8),  # Simplified
                    "priority": hyperedge.priority,
                })

        return plan


# Example usage
orchestrator = HypergraphOrchestrator()

# Add segments
orchestrator.add_segment(CustomerSegment(
    id="seg_high_value",
    name="High Value Customers",
    size=50000,
    criteria={"ltv": ">1000", "active": "true"}
))

# Add campaign hyperedges
now = datetime.utcnow()
campaigns = [
    CampaignHyperedge(
        id="camp_1",
        segment_id="seg_high_value",
        channel=Channel.EMAIL,
        content_id="promo_winter_sale",
        time_window_start=now,
        time_window_end=now + timedelta(days=7),
        budget_allocation=5000,
        priority=10,
        frequency_cap=2,
        objective="conversion"
    ),
    CampaignHyperedge(
        id="camp_2",
        segment_id="seg_high_value",
        channel=Channel.SMS,
        content_id="flash_sale_alert",
        time_window_start=now + timedelta(days=2),
        time_window_end=now + timedelta(days=3),
        budget_allocation=2000,
        priority=8,
        frequency_cap=1,
        objective="conversion"
    ),
]

for camp in campaigns:
    orchestrator.add_campaign_hyperedge(camp)

# Get next best action for a customer
customer_id = "cust_12345"
context = {"device": "mobile", "location": "US", "time_of_day": "evening"}

action = orchestrator.select_next_best_action(customer_id, context)
if action:
    print(f"Next best action for {customer_id}:")
    print(f"  Campaign: {action.id}")
    print(f"  Channel: {action.channel.value}")
    print(f"  Content: {action.content_id}")

    # Record the touch
    orchestrator.record_touch(customer_id, action)

# Get orchestration plan
plan = orchestrator.get_orchestration_plan("seg_high_value")
print("\nOrchestration Plan:")
for item in plan:
    print(f"  {item['campaign_id']}: {item['channel']} - {item['content_id']}")