AI Agent for Hospitality: Automate Revenue Management, Guest Experience & Hotel Operations

The average 300-room hotel generates over 2.6 million data points per year — reservation patterns, guest preferences, competitor rates, weather forecasts, event calendars, housekeeping logs, F&B consumption, and OTA performance metrics. A revenue manager manually updating spreadsheets captures maybe 5% of that signal. An AI agent captures all of it, in real time, and acts on it autonomously.

Hotels running AI agents in 2026 are seeing 8-15% RevPAR increases, 20-30% reductions in F&B waste, and 25-40% improvements in housekeeping efficiency. This is not incremental optimization. This is a structural advantage that compounds every day your competitors operate manually.

This guide covers six operational domains where AI agents deliver measurable ROI for hotel properties, with production-ready Python code for each. Every example is designed for integration with standard PMS, CRS, and POS systems.

1. Revenue Management & Dynamic Pricing

Revenue management is where AI agents deliver the fastest, largest return. The core problem is deceptively simple: set the right price for the right room on the right channel at the right time. In practice, this means simultaneously optimizing across room types (standard, deluxe, suite, accessible), rate plans (BAR, AAA, corporate, package), channels (direct, Booking.com, Expedia, GDS), and booking windows (0-365 days out) — a combinatorial space that grows exponentially with property size.

Traditional RMS tools like IDeaS G3 or Duetto GameChanger handle demand forecasting and rate recommendations well. An AI agent goes further: it acts autonomously, pushing rate changes across all distribution channels, monitoring competitive responses, adjusting overbooking limits based on real-time no-show probabilities, and evaluating group pricing requests against transient displacement.

Demand Forecasting & Dynamic Rate Optimization

import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass

@dataclass
class RateRecommendation:
    room_type: str
    channel: str
    recommended_rate: float
    current_rate: float
    confidence: float
    demand_score: float
    reasoning: str

class RevenueManagementAgent:
    """AI agent for hotel revenue management and dynamic pricing."""

    def __init__(self, pms, channel_manager, comp_intel, weather_api, event_api):
        self.pms = pms                    # Property Management System
        self.cm = channel_manager          # Channel manager (SiteMinder, etc.)
        self.comp = comp_intel             # Competitor rate scraper
        self.weather = weather_api
        self.events = event_api
        self.min_rate_floor = {}           # Minimum rates by room type
        self.max_rate_ceiling = {}         # Maximum rates by room type

    def generate_daily_pricing(self, hotel_id: str, horizon_days: int = 90) -> list:
        """Generate optimal rates for all room types across all channels."""
        recommendations = []
        hotel = self.pms.get_hotel(hotel_id)

        for target_date in self._date_range(horizon_days):
            # Build demand signal from multiple sources
            demand = self._forecast_demand(hotel_id, target_date)

            # Get current booking pace vs. historical
            pace = self._calculate_booking_pace(hotel_id, target_date)

            # Competitor rate intelligence
            comp_rates = self.comp.get_rates(
                hotel['comp_set_ids'], target_date
            )

            for room_type in hotel['room_types']:
                # Current inventory position
                inventory = self.pms.get_availability(
                    hotel_id, room_type['code'], target_date
                )
                total_rooms = room_type['count']
                sold = total_rooms - inventory['available']
                occupancy_pct = sold / total_rooms

                # Days until arrival (booking window position)
                days_out = (target_date - datetime.now().date()).days

                # Calculate optimal rate
                optimal = self._optimize_rate(
                    room_type=room_type,
                    demand_score=demand['score'],
                    occupancy_pct=occupancy_pct,
                    pace_vs_historical=pace['ratio'],
                    comp_median=comp_rates.get(room_type['comp_category'], {}).get('median'),
                    days_out=days_out,
                    day_of_week=target_date.weekday(),
                    event_impact=demand['event_multiplier'],
                    weather_impact=demand['weather_multiplier']
                )

                # Generate per-channel rates (maintain parity rules)
                for channel in hotel['active_channels']:
                    channel_rate = self._apply_channel_strategy(
                        optimal['rate'], channel, room_type
                    )

                    recommendations.append(RateRecommendation(
                        room_type=room_type['code'],
                        channel=channel['name'],
                        recommended_rate=channel_rate,
                        current_rate=self.cm.get_current_rate(
                            hotel_id, room_type['code'],
                            channel['id'], target_date
                        ),
                        confidence=optimal['confidence'],
                        demand_score=demand['score'],
                        reasoning=optimal['reasoning']
                    ))

        return recommendations

    def _forecast_demand(self, hotel_id: str, target_date) -> dict:
        """Multi-signal demand forecast combining 7 data sources."""
        signals = {}

        # 1. Historical occupancy (same DOW, same week-of-year, 3yr avg)
        historical = self.pms.get_historical_occupancy(
            hotel_id, target_date, lookback_years=3
        )
        signals['historical_occ'] = np.mean(historical)

        # 2. Booking pace (reservations on-the-books vs same point last year)
        pace = self._calculate_booking_pace(hotel_id, target_date)
        signals['pace_ratio'] = pace['ratio']

        # 3. Local events (conferences, concerts, sports, holidays)
        events = self.events.get_events(
            hotel_id, target_date,
            radius_km=25, min_attendance=500
        )
        event_multiplier = 1.0
        for event in events:
            impact = min(event['expected_attendance'] / 5000, 0.4)
            event_multiplier += impact
        signals['event_multiplier'] = min(event_multiplier, 1.8)

        # 4. Weather forecast (beach resorts, ski lodges, outdoor venues)
        weather = self.weather.get_forecast(hotel_id, target_date)
        weather_mult = self._weather_demand_impact(weather)
        signals['weather_multiplier'] = weather_mult

        # 5. Competitor occupancy signals (rate increases = market demand)
        comp_rate_change = self.comp.get_rate_trend(hotel_id, target_date, days=7)
        signals['comp_trend'] = comp_rate_change

        # 6. Day-of-week seasonality
        dow_factor = self._get_dow_factor(hotel_id, target_date.weekday())
        signals['dow_factor'] = dow_factor

        # 7. Flight search volume to destination (forward-looking indicator)
        flight_demand = self._get_flight_search_index(hotel_id, target_date)
        signals['flight_demand'] = flight_demand

        # Weighted composite score (0.0 = dead, 1.0 = average, 2.0 = peak)
        score = (
            signals['historical_occ'] * 0.25 +
            signals['pace_ratio'] * 0.20 +
            signals['event_multiplier'] * 0.20 +
            signals['dow_factor'] * 0.15 +
            signals['comp_trend'] * 0.10 +
            signals['weather_multiplier'] * 0.05 +
            signals['flight_demand'] * 0.05
        )

        return {
            'score': score,
            'signals': signals,
            'event_multiplier': signals['event_multiplier'],
            'weather_multiplier': signals['weather_multiplier']
        }

    def _optimize_rate(self, room_type, demand_score, occupancy_pct,
                       pace_vs_historical, comp_median, days_out,
                       day_of_week, event_impact, weather_impact) -> dict:
        """Calculate optimal rate using constrained revenue maximization."""
        base_rate = room_type['base_rate']
        floor = self.min_rate_floor.get(room_type['code'], base_rate * 0.7)
        ceiling = self.max_rate_ceiling.get(room_type['code'], base_rate * 2.5)

        # Demand multiplier (exponential curve, aggressive near sellout)
        if occupancy_pct > 0.92:
            demand_mult = 1.4 + (occupancy_pct - 0.92) * 6.0  # Steep curve
        elif occupancy_pct > 0.80:
            demand_mult = 1.1 + (occupancy_pct - 0.80) * 2.5
        else:
            demand_mult = 0.85 + occupancy_pct * 0.3

        # Booking window adjustment (last-minute vs. advance)
        if days_out <= 1:
            window_mult = 1.15  # Last-minute premium
        elif days_out <= 7:
            window_mult = 1.05 + (7 - days_out) * 0.015
        elif days_out > 60:
            window_mult = 0.95  # Early-bird slight discount
        else:
            window_mult = 1.0

        # Pace adjustment (ahead of pace = raise, behind = lower)
        pace_mult = 1.0 + (pace_vs_historical - 1.0) * 0.3

        # Competitor positioning (stay within 10% of comp median)
        if comp_median:
            comp_ratio = (base_rate * demand_mult) / comp_median
            if comp_ratio > 1.15:
                demand_mult *= 0.92  # Pull back if too far above market
            elif comp_ratio < 0.85:
                demand_mult *= 1.05  # Raise if leaving money on table

        # Length-of-stay pricing incentive
        los_discount = 0.0  # Applied separately per LOS threshold

        optimal_rate = base_rate * demand_mult * window_mult * pace_mult
        optimal_rate = max(floor, min(ceiling, optimal_rate))

        reasoning = (
            f"Demand={demand_score:.2f}, Occ={occupancy_pct:.0%}, "
            f"Pace={pace_vs_historical:.2f}x, Events={event_impact:.1f}x, "
            f"DaysOut={days_out}, CompMedian=${comp_median or 'N/A'}"
        )

        return {
            'rate': round(optimal_rate, 2),
            'confidence': min(0.95, 0.6 + occupancy_pct * 0.3),
            'reasoning': reasoning
        }

    def optimize_overbooking(self, hotel_id: str, date) -> dict:
        """Calculate optimal overbooking level per room type."""
        results = {}
        hotel = self.pms.get_hotel(hotel_id)

        for room_type in hotel['room_types']:
            # Historical no-show rate by DOW, season, channel mix
            noshow_rate = self._predict_noshow_rate(
                hotel_id, room_type['code'], date
            )
            cancellation_rate = self._predict_cancellation_rate(
                hotel_id, room_type['code'], date
            )

            total_rooms = room_type['count']
            adr = self.pms.get_adr(hotel_id, room_type['code'], date)

            # Walk cost (compensation + reputation damage)
            walk_cost = adr * 2.5 + 150  # Room at competitor + transport + goodwill

            # Revenue from overbooking (expected gain from selling extra rooms)
            # Optimal overbook = rooms * (noshow_rate + cancel_rate) * confidence
            expected_noshows = total_rooms * (noshow_rate + cancellation_rate * 0.3)

            # Risk-adjusted overbooking (conservative: 70-80% of expected)
            overbook_rooms = int(expected_noshows * 0.75)

            expected_revenue_gain = overbook_rooms * adr * 0.85
            expected_walk_cost = overbook_rooms * 0.15 * walk_cost

            results[room_type['code']] = {
                'overbook_rooms': overbook_rooms,
                'noshow_rate': f"{noshow_rate:.1%}",
                'expected_revenue_gain': round(expected_revenue_gain),
                'expected_walk_cost': round(expected_walk_cost),
                'net_expected_value': round(expected_revenue_gain - expected_walk_cost)
            }

        return results

2. Guest Experience Personalization

Guest personalization in hospitality goes far beyond "Welcome back, Mr. Smith." The AI agent builds a living guest profile that accumulates across every touchpoint: booking channel preferences, room temperature settings, minibar consumption, spa booking patterns, restaurant choices, complaint history, loyalty tier, total lifetime value, and even sentiment extracted from post-stay surveys.

The best hotel chains treat this as a competitive moat. A guest who receives a personalized experience — preferred pillow type already on the bed, dietary restrictions pre-communicated to the restaurant, proactive upgrade when availability allows — has a 40% higher rebooking rate and generates 2.3x more ancillary revenue than a generic guest.

Pre-Arrival Profiling & Smart Room Assignment

from enum import Enum
from typing import Optional

class LoyaltyTier(Enum):
    NEW = "new"
    SILVER = "silver"
    GOLD = "gold"
    PLATINUM = "platinum"
    AMBASSADOR = "ambassador"

class GuestExperienceAgent:
    """AI agent for end-to-end guest experience personalization."""

    def __init__(self, pms, crm, iot_controller, llm, feedback_system):
        self.pms = pms
        self.crm = crm
        self.iot = iot_controller       # Smart room controls (thermostats, lighting)
        self.llm = llm
        self.feedback = feedback_system

    def pre_arrival_workflow(self, reservation_id: str) -> dict:
        """Execute complete pre-arrival personalization pipeline."""
        reservation = self.pms.get_reservation(reservation_id)
        guest = self.crm.get_profile(reservation['guest_id'])
        past_stays = self.crm.get_stay_history(guest['id'])

        # Step 1: Build comprehensive guest profile
        profile = self._build_arrival_profile(guest, past_stays, reservation)

        # Step 2: Optimal room assignment (preference matching + upgrade logic)
        room = self._assign_optimal_room(profile, reservation)

        # Step 3: Pre-configure room IoT settings
        if room.get('smart_enabled'):
            self._preconfigure_room(room['number'], profile)

        # Step 4: Generate staff briefing
        briefing = self._generate_staff_brief(profile, room, reservation)

        # Step 5: Identify upsell opportunities
        upsells = self._identify_upsells(profile, reservation)

        return {
            'guest': guest['name'],
            'room_assigned': room['number'],
            'upgrade_applied': room.get('upgraded', False),
            'iot_preconfigured': room.get('smart_enabled', False),
            'staff_briefing': briefing,
            'upsell_opportunities': upsells
        }

    def _assign_optimal_room(self, profile: dict, reservation: dict) -> dict:
        """Assign best room considering preferences, loyalty, and revenue."""
        available = self.pms.get_available_rooms(
            reservation['hotel_id'],
            reservation['room_type'],
            reservation['check_in'],
            reservation['check_out']
        )

        scored_rooms = []
        for room in available:
            score = 0

            # Floor preference (some guests always want high floor)
            if profile.get('preferred_floor') == 'high' and room['floor'] >= 8:
                score += 30
            elif profile.get('preferred_floor') == 'low' and room['floor'] <= 3:
                score += 30

            # View preference
            if profile.get('preferred_view') == room.get('view_type'):
                score += 25

            # Bed configuration
            if profile.get('preferred_bed') == room.get('bed_config'):
                score += 20

            # Quiet room (away from elevator, ice machine)
            if profile.get('noise_sensitive') and room.get('quiet_zone'):
                score += 20

            # Accessibility
            if profile.get('accessibility_needs') and room.get('accessible'):
                score += 50

            # Proximity to amenities (gym, pool, restaurant)
            if profile.get('frequent_gym') and room.get('near_gym'):
                score += 10

            # Previous room (some guests want the same room)
            if room['number'] == profile.get('last_room_number'):
                score += 15

            scored_rooms.append({**room, 'preference_score': score})

        # Sort by score and pick best match
        scored_rooms.sort(key=lambda r: -r['preference_score'])
        best_room = scored_rooms[0]

        # Upgrade logic: loyalty tier + lifetime value + availability
        upgrade = self._evaluate_upgrade(profile, reservation, available)
        if upgrade:
            best_room = upgrade
            best_room['upgraded'] = True

        return best_room

    def _evaluate_upgrade(self, profile: dict, reservation: dict,
                          available: list) -> Optional[dict]:
        """Decide whether to offer a complimentary upgrade."""
        tier = profile.get('loyalty_tier', LoyaltyTier.NEW)
        ltv = profile.get('lifetime_value', 0)
        total_stays = profile.get('total_stays', 0)

        # Upgrade probability based on loyalty + value
        upgrade_score = 0
        if tier == LoyaltyTier.AMBASSADOR:
            upgrade_score = 90
        elif tier == LoyaltyTier.PLATINUM:
            upgrade_score = 70
        elif tier == LoyaltyTier.GOLD:
            upgrade_score = 40
        elif ltv > 25000:  # High-value non-loyalty guest
            upgrade_score = 50

        # Reduce if hotel is >85% occupied (upgrades cost displacement)
        occupancy = self.pms.get_occupancy(
            reservation['hotel_id'], reservation['check_in']
        )
        if occupancy > 0.85:
            upgrade_score *= 0.5
        elif occupancy < 0.60:
            upgrade_score *= 1.3  # More generous when lots of availability

        if upgrade_score >= 50:
            # Find next-tier room with availability
            next_tier = self.pms.get_next_room_tier(reservation['room_type'])
            if next_tier and self._has_availability(next_tier, reservation):
                return self._get_best_room(next_tier, available)

        return None

    def _preconfigure_room(self, room_number: str, profile: dict):
        """Set IoT room controls based on guest preferences."""
        settings = {}

        if profile.get('preferred_temp'):
            settings['thermostat'] = profile['preferred_temp']
        else:
            settings['thermostat'] = 72  # Default Fahrenheit

        if profile.get('preferred_lighting'):
            settings['lighting_scene'] = profile['preferred_lighting']
        else:
            settings['lighting_scene'] = 'warm_welcome'

        # Curtain position
        if profile.get('early_riser'):
            settings['curtain_mode'] = 'auto_open_sunrise'
        else:
            settings['curtain_mode'] = 'closed'

        # TV welcome screen language
        settings['tv_language'] = profile.get('language', 'en')

        self.iot.apply_settings(room_number, settings)

    def analyze_post_stay_feedback(self, stay_id: str) -> dict:
        """Analyze post-stay survey and trigger service recovery if needed."""
        survey = self.feedback.get_survey(stay_id)
        stay = self.pms.get_stay(stay_id)

        # NPS prediction from survey responses
        nps_prediction = self._predict_nps(survey)

        # Extract specific issues
        issues = self.llm.extract_issues(survey.get('comments', ''))

        # Service recovery triggers
        recovery_actions = []
        if nps_prediction['score'] < 7:
            # Detractor: immediate service recovery
            recovery_actions.append({
                'action': 'personal_outreach',
                'channel': 'email_from_gm',
                'timing': 'within_24h',
                'offer': self._calibrate_recovery_offer(
                    nps_prediction['score'],
                    stay['total_revenue'],
                    stay['guest_ltv']
                )
            })

        if nps_prediction['score'] >= 9:
            # Promoter: request public review
            recovery_actions.append({
                'action': 'review_request',
                'platforms': ['google', 'tripadvisor'],
                'timing': '48h_post_checkout',
                'personalized_message': True
            })

        # Update guest profile with new preferences learned
        learned = self._extract_preferences(survey, stay)
        self.crm.update_preferences(stay['guest_id'], learned)

        return {
            'nps_predicted': nps_prediction['score'],
            'nps_confidence': nps_prediction['confidence'],
            'issues_found': issues,
            'recovery_actions': recovery_actions,
            'preferences_learned': learned
        }

3. Housekeeping & Maintenance Optimization

Housekeeping is the single largest labor cost in hotel operations, typically accounting for 25-35% of total labor expense. A 300-room hotel employs 40-60 housekeepers and spends $2-3M annually on room cleaning alone. The inefficiency comes from static scheduling: every room gets the same cleaning protocol regardless of guest type, stay length, or actual condition.

An AI agent transforms housekeeping from a batch process into a dynamic, priority-driven operation. It factors in checkout times from the PMS, VIP arrivals with specific ETAs, special cleaning requirements (allergen-free, pet rooms, connecting rooms), staff skill levels, floor clustering to minimize transit time, and real-time DND status from IoT door sensors.

Intelligent Room Prioritization & Predictive Maintenance

from datetime import datetime, time
from typing import List, Dict
import heapq

class HousekeepingMaintenanceAgent:
    """AI agent for housekeeping scheduling and predictive maintenance."""

    def __init__(self, pms, iot_sensors, work_order_system, staff_system):
        self.pms = pms
        self.sensors = iot_sensors
        self.work_orders = work_order_system
        self.staff = staff_system
        self.clean_times = {
            'checkout_standard': 35,    # minutes
            'checkout_suite': 55,
            'checkout_vip': 45,
            'stayover_standard': 20,
            'stayover_vip': 30,
            'deep_clean': 90,
            'pet_room': 50,
            'allergen_free': 60
        }

    def generate_daily_schedule(self, hotel_id: str, date: str) -> dict:
        """Generate optimized housekeeping schedule with priority queuing."""
        rooms = self.pms.get_room_status(hotel_id, date)
        staff_available = self.staff.get_available(hotel_id, date, 'housekeeping')

        # Build priority queue (min-heap by priority, then deadline)
        priority_queue = []

        for room in rooms:
            priority, deadline, clean_type = self._classify_room(room)

            # Check IoT signals for real-time adjustments
            iot_status = self.sensors.get_room_status(room['number'])
            if iot_status.get('dnd_active'):
                continue  # Skip DND rooms, re-check hourly
            if iot_status.get('guest_departed') and room['status'] == 'stayover':
                # Guest checked out early, reprioritize
                priority = 2
                clean_type = 'checkout_standard'

            estimated_minutes = self.clean_times.get(clean_type, 35)

            # Special request modifiers
            if room.get('extra_bed_requested'):
                estimated_minutes += 10
            if room.get('crib_requested'):
                estimated_minutes += 8
            if room.get('connecting_room'):
                estimated_minutes += 5

            heapq.heappush(priority_queue, (priority, deadline, {
                'room': room['number'],
                'floor': room['floor'],
                'type': clean_type,
                'estimated_minutes': estimated_minutes,
                'special_requests': room.get('special_requests', []),
                'vip': room.get('vip', False),
                'guest_eta': room.get('eta')
            }))

        # Assign rooms to staff with floor clustering
        assignments = self._assign_with_clustering(
            priority_queue, staff_available
        )

        # Calculate linen and amenity requirements
        supply_forecast = self._forecast_supplies(priority_queue)

        return {
            'schedule': assignments,
            'total_rooms': len(priority_queue),
            'staff_utilized': len(staff_available),
            'estimated_completion': self._estimate_completion(assignments),
            'supply_requirements': supply_forecast,
            'efficiency_score': self._calculate_efficiency(assignments)
        }

    def _classify_room(self, room: dict) -> tuple:
        """Classify room into priority tier, deadline, and clean type."""
        # Priority 1: VIP arrivals (must be immaculate before ETA)
        if room.get('vip') and room['status'] == 'arriving':
            eta = room.get('eta', '14:00')
            clean_type = 'checkout_vip' if room.get('suite') else 'checkout_standard'
            return (1, eta, clean_type)

        # Priority 2: Early check-in requests
        if room['status'] == 'arriving' and room.get('early_checkin'):
            return (2, room.get('eta', '12:00'), 'checkout_standard')

        # Priority 3: Pet rooms and allergen-free (extra time needed)
        if room.get('pet_room'):
            return (3, '15:00', 'pet_room')
        if room.get('allergen_free'):
            return (3, '15:00', 'allergen_free')

        # Priority 4: Regular checkouts
        if room['status'] in ('checkout', 'departed'):
            clean_type = 'checkout_suite' if room.get('suite') else 'checkout_standard'
            return (4, '15:00', clean_type)

        # Priority 5: Stayovers (most flexible)
        if room['status'] == 'stayover':
            clean_type = 'stayover_vip' if room.get('vip') else 'stayover_standard'
            preferred = room.get('preferred_clean_time', '14:00')
            return (5, preferred, clean_type)

        # Priority 6: Deep cleans (scheduled)
        if room.get('deep_clean_due'):
            return (6, '17:00', 'deep_clean')

        return (7, '18:00', 'stayover_standard')

    def _assign_with_clustering(self, priority_queue: list,
                                 staff: list) -> List[Dict]:
        """Assign rooms to staff optimizing for floor proximity."""
        # Group by floor first
        floors = {}
        tasks = []
        while priority_queue:
            priority, deadline, task = heapq.heappop(priority_queue)
            tasks.append((priority, deadline, task))
            floor = task['floor']
            if floor not in floors:
                floors[floor] = []
            floors[floor].append((priority, deadline, task))

        assignments = []
        staff_idx = 0

        # Assign high-priority tasks first, then cluster by floor
        high_priority = [(p, d, t) for p, d, t in tasks if p <= 2]
        normal_priority = [(p, d, t) for p, d, t in tasks if p > 2]

        # Distribute high-priority across available staff
        for i, (_, _, task) in enumerate(high_priority):
            staff_member = staff[i % len(staff)]
            if not any(a['staff_id'] == staff_member['id'] for a in assignments):
                assignments.append({
                    'staff_id': staff_member['id'],
                    'staff_name': staff_member['name'],
                    'rooms': [task],
                    'total_minutes': task['estimated_minutes']
                })
            else:
                existing = next(a for a in assignments if a['staff_id'] == staff_member['id'])
                existing['rooms'].append(task)
                existing['total_minutes'] += task['estimated_minutes']

        # Cluster remaining rooms by floor and distribute evenly
        for floor, floor_tasks in sorted(floors.items()):
            normal_floor = [t for _, _, t in floor_tasks if (_, _, t) not in [(p, d, t) for p, d, t in high_priority]]
            for task in normal_floor:
                # Find staff with lowest load on this floor (or nearby)
                best_staff = min(
                    assignments if assignments else [{'staff_id': s['id'], 'staff_name': s['name'], 'rooms': [], 'total_minutes': 0} for s in staff[:1]],
                    key=lambda a: a['total_minutes']
                )
                best_staff['rooms'].append(task)
                best_staff['total_minutes'] += task['estimated_minutes']

        return assignments

    def run_predictive_maintenance(self, hotel_id: str) -> dict:
        """Analyze IoT sensor data for predictive maintenance alerts."""
        assets = self.sensors.get_monitored_assets(hotel_id)
        alerts = []

        for asset in assets:
            readings = self.sensors.get_readings(asset['id'], hours=72)

            if asset['type'] == 'hvac':
                alert = self._analyze_hvac(asset, readings)
            elif asset['type'] == 'elevator':
                alert = self._analyze_elevator(asset, readings)
            elif asset['type'] == 'plumbing':
                alert = self._analyze_plumbing(asset, readings)
            elif asset['type'] == 'electrical':
                alert = self._analyze_electrical(asset, readings)
            else:
                continue

            if alert:
                # Calculate guest impact
                affected_rooms = self._get_affected_rooms(asset)
                occupied_affected = [
                    r for r in affected_rooms
                    if self.pms.is_occupied(r, datetime.now().date())
                ]

                alert['guest_impact'] = len(occupied_affected)
                alert['affected_rooms'] = affected_rooms
                alert['preventive_cost'] = asset.get('avg_preventive_cost', 500)
                alert['emergency_cost'] = asset.get('avg_emergency_cost', 2500)
                alert['savings'] = alert['emergency_cost'] - alert['preventive_cost']

                alerts.append(alert)

                # Auto-create work order for critical/high priority
                if alert['severity'] in ('critical', 'high'):
                    self.work_orders.create(
                        asset_id=asset['id'],
                        description=f"Predictive: {alert['issue']}. "
                                   f"Confidence: {alert['confidence']:.0%}. "
                                   f"Est. failure: {alert['est_failure_days']} days.",
                        priority=alert['severity'],
                        affected_rooms=affected_rooms
                    )

        return {
            'assets_scanned': len(assets),
            'alerts_generated': len(alerts),
            'critical': len([a for a in alerts if a['severity'] == 'critical']),
            'potential_savings': sum(a['savings'] for a in alerts),
            'alerts': sorted(alerts, key=lambda a: a['severity'] == 'critical', reverse=True)
        }

    def _analyze_hvac(self, asset: dict, readings: list) -> dict:
        """Detect HVAC anomalies from sensor patterns."""
        temps = [r['supply_temp'] for r in readings]
        power = [r['power_draw'] for r in readings]
        vibration = [r.get('vibration', 0) for r in readings]

        issues = []

        # Compressor degradation: rising power draw for same output
        if len(power) > 24:
            power_trend = np.polyfit(range(len(power)), power, 1)[0]
            if power_trend > 0.05:  # kW/hour increasing
                issues.append('compressor_degradation')

        # Refrigerant leak: supply temp not reaching setpoint
        setpoint_diff = [abs(t - asset.get('setpoint', 72)) for t in temps[-12:]]
        if np.mean(setpoint_diff) > 4.0:
            issues.append('refrigerant_low_or_leak')

        # Bearing wear: increasing vibration signature
        if np.mean(vibration[-12:]) > asset.get('vibration_baseline', 0.5) * 1.5:
            issues.append('bearing_wear')

        if issues:
            return {
                'asset': asset['name'],
                'location': asset['location'],
                'type': 'hvac',
                'issue': ', '.join(issues),
                'severity': 'critical' if 'compressor_degradation' in issues else 'high',
                'confidence': 0.85,
                'est_failure_days': 7 if 'compressor_degradation' in issues else 21
            }

        return None

4. Food & Beverage Intelligence

F&B operations are the most complex and highest-waste area of hotel management. The average hotel restaurant wastes 15-25% of food purchased, worth $200K-500K annually for a full-service 300-room property. The problem is compounding variability: breakfast demand depends on occupancy and guest mix. Lunch depends on meeting room bookings and pool weather. Dinner depends on local events, group dining, and restaurant reputation. Banquet depends on event bookings that were confirmed months ago.

An AI agent brings demand forecasting, menu engineering, and inventory optimization together into a single decision loop that updates daily.

Demand Forecasting & Menu Engineering

from dataclasses import dataclass, field
from typing import List, Tuple

@dataclass
class MenuItemAnalysis:
    name: str
    category: str
    food_cost: float
    selling_price: float
    contribution_margin: float
    popularity_index: float
    classification: str  # star, plowhorse, puzzle, dog

class FBIntelligenceAgent:
    """AI agent for hotel food & beverage optimization."""

    def __init__(self, pms, pos_system, inventory_system, procurement):
        self.pms = pms
        self.pos = pos_system
        self.inventory = inventory_system
        self.procurement = procurement

    def forecast_daily_demand(self, hotel_id: str, date: str,
                               outlet: str) -> dict:
        """Forecast covers and revenue by meal period."""
        # Gather all demand signals
        occupancy = self.pms.get_forecasted_occupancy(hotel_id, date)
        guest_mix = self.pms.get_guest_mix(hotel_id, date)  # business/leisure ratio
        groups = self.pms.get_group_blocks(hotel_id, date)
        events = self.pms.get_banquet_events(hotel_id, date)
        day_of_week = self._get_dow(date)
        weather = self._get_weather_forecast(hotel_id, date)
        local_events = self._get_local_events(hotel_id, date)

        # Historical capture rates by meal period
        historical = self.pos.get_historical_capture_rates(
            hotel_id, outlet, lookback_days=90
        )

        forecasts = {}
        for meal_period in ['breakfast', 'lunch', 'dinner']:
            # Base capture rate from history
            base_rate = historical[meal_period][day_of_week]

            # Adjustments
            adjustments = 1.0

            # Group impact (groups with meal packages = guaranteed covers)
            group_covers = sum(
                g['pax'] for g in groups
                if meal_period in g.get('meal_package', [])
            )

            # Business travelers eat breakfast in-house more (85% vs 60% leisure)
            if meal_period == 'breakfast':
                biz_ratio = guest_mix.get('business', 0.5)
                adjustments *= (0.60 + biz_ratio * 0.35)

            # Weather impact on lunch (nice weather = fewer in-house lunches)
            if meal_period == 'lunch' and weather.get('clear_sky'):
                if outlet == 'main_restaurant':
                    adjustments *= 0.80  # Guests go out
                elif outlet == 'pool_bar':
                    adjustments *= 1.40  # Pool bar booms

            # Local events impact dinner
            if meal_period == 'dinner' and local_events:
                for event in local_events:
                    if event['type'] in ('concert', 'sports'):
                        adjustments *= 0.85  # Guests eat out
                    elif event['type'] == 'conference':
                        adjustments *= 1.15  # Conference overflow

            # Calculate forecasted covers
            total_guests = occupancy * self.pms.get_total_rooms(hotel_id)
            transient_covers = int(total_guests * base_rate * adjustments)
            total_covers = transient_covers + group_covers

            # Revenue forecast
            avg_check = self.pos.get_avg_check(
                hotel_id, outlet, meal_period, day_of_week
            )

            forecasts[meal_period] = {
                'total_covers': total_covers,
                'group_covers': group_covers,
                'transient_covers': transient_covers,
                'capture_rate': round(base_rate * adjustments, 3),
                'avg_check': avg_check,
                'forecasted_revenue': round(total_covers * avg_check, 2),
                'confidence': 0.85 if occupancy > 0.5 else 0.70
            }

        return forecasts

    def run_menu_engineering(self, hotel_id: str, outlet: str,
                             period_days: int = 30) -> dict:
        """Analyze menu using contribution margin and popularity matrix."""
        items = self.pos.get_item_sales(hotel_id, outlet, days=period_days)
        total_items_sold = sum(item['quantity'] for item in items)
        num_items = len(items)

        analyzed = []
        for item in items:
            cm = item['selling_price'] - item['food_cost']
            popularity = item['quantity'] / total_items_sold

            analyzed.append(MenuItemAnalysis(
                name=item['name'],
                category=item['category'],
                food_cost=item['food_cost'],
                selling_price=item['selling_price'],
                contribution_margin=cm,
                popularity_index=popularity,
                classification=''  # Assigned below
            ))

        # Calculate thresholds
        avg_cm = np.mean([a.contribution_margin for a in analyzed])
        avg_popularity = 1.0 / num_items * 0.7  # 70% rule for popularity threshold

        # Classify each item
        for item in analyzed:
            high_cm = item.contribution_margin >= avg_cm
            high_pop = item.popularity_index >= avg_popularity

            if high_cm and high_pop:
                item.classification = 'star'        # Keep, promote, protect price
            elif not high_cm and high_pop:
                item.classification = 'ploworse'   # Increase price or reduce cost
            elif high_cm and not high_pop:
                item.classification = 'puzzle'      # Promote, reposition, rename
            else:
                item.classification = 'dog'         # Remove or redesign completely

        # Generate actionable recommendations
        recommendations = []
        for item in analyzed:
            if item.classification == 'plowhorse':
                # Calculate price increase needed to reach avg CM
                price_increase = avg_cm - item.contribution_margin
                recommendations.append({
                    'item': item.name,
                    'action': f'Increase price by ${price_increase:.2f} or reduce '
                             f'food cost by {price_increase/item.food_cost:.0%}',
                    'impact': 'high'
                })
            elif item.classification == 'puzzle':
                recommendations.append({
                    'item': item.name,
                    'action': 'Reposition on menu (eye magnets, box highlight), '
                             'retrain servers to recommend, consider renaming',
                    'impact': 'medium'
                })
            elif item.classification == 'dog':
                if item.popularity_index < avg_popularity * 0.3:
                    recommendations.append({
                        'item': item.name,
                        'action': 'Remove from menu. Replace with new item targeting '
                                 'star quadrant (high CM, high appeal).',
                        'impact': 'medium'
                    })

        return {
            'items_analyzed': len(analyzed),
            'stars': [a.name for a in analyzed if a.classification == 'star'],
            'plowhorse': [a.name for a in analyzed if a.classification == 'plowhorse'],
            'puzzles': [a.name for a in analyzed if a.classification == 'puzzle'],
            'dogs': [a.name for a in analyzed if a.classification == 'dog'],
            'avg_contribution_margin': round(avg_cm, 2),
            'recommendations': recommendations,
            'total_food_cost_pct': round(
                sum(a.food_cost * a.popularity_index for a in analyzed) /
                sum(a.selling_price * a.popularity_index for a in analyzed) * 100, 1
            )
        }

    def optimize_inventory(self, hotel_id: str, forecast_days: int = 7) -> dict:
        """Optimize perishable inventory based on demand forecasts."""
        outlets = self.pos.get_outlets(hotel_id)
        all_orders = []

        for day_offset in range(forecast_days):
            date = self._offset_date(day_offset)
            for outlet in outlets:
                forecast = self.forecast_daily_demand(hotel_id, date, outlet['id'])

                # Convert covers to ingredient requirements
                for meal_period, data in forecast.items():
                    menu_mix = self.pos.get_menu_mix(
                        hotel_id, outlet['id'], meal_period
                    )
                    for item_name, pct in menu_mix.items():
                        expected_orders = int(data['total_covers'] * pct)
                        recipe = self.inventory.get_recipe(item_name)
                        for ingredient, qty_per in recipe.items():
                            all_orders.append({
                                'ingredient': ingredient,
                                'quantity_needed': expected_orders * qty_per,
                                'date_needed': date,
                                'outlet': outlet['id']
                            })

        # Aggregate by ingredient and apply safety stock
        aggregated = {}
        for order in all_orders:
            key = order['ingredient']
            if key not in aggregated:
                aggregated[key] = {
                    'total_needed': 0,
                    'dates': [],
                    'perishable': self.inventory.is_perishable(key),
                    'shelf_life_days': self.inventory.get_shelf_life(key),
                    'current_stock': self.inventory.get_current_stock(hotel_id, key),
                    'unit_cost': self.inventory.get_unit_cost(key)
                }
            aggregated[key]['total_needed'] += order['quantity_needed']
            aggregated[key]['dates'].append(order['date_needed'])

        # Generate purchase orders
        purchase_orders = []
        for ingredient, data in aggregated.items():
            deficit = data['total_needed'] - data['current_stock']
            safety_stock = data['total_needed'] * 0.15  # 15% buffer

            if deficit + safety_stock > 0:
                order_qty = deficit + safety_stock

                # For perishables: split deliveries to minimize waste
                if data['perishable'] and data['shelf_life_days'] < forecast_days:
                    deliveries = self._split_perishable_orders(
                        ingredient, order_qty, data['dates'],
                        data['shelf_life_days']
                    )
                    purchase_orders.extend(deliveries)
                else:
                    purchase_orders.append({
                        'ingredient': ingredient,
                        'quantity': round(order_qty, 1),
                        'delivery_date': min(data['dates']),
                        'estimated_cost': round(order_qty * data['unit_cost'], 2)
                    })

        waste_reduction = self._estimate_waste_reduction(aggregated)

        return {
            'purchase_orders': purchase_orders,
            'total_cost': round(sum(po['estimated_cost'] for po in purchase_orders), 2),
            'estimated_waste_reduction': waste_reduction,
            'items_optimized': len(aggregated)
        }

5. Distribution & Channel Management

Distribution costs are the second largest expense after labor for most hotels. OTA commissions run 15-25% per booking (Booking.com 15-18%, Expedia 18-25%), while direct bookings cost 3-5% in marketing and technology. A 300-room hotel paying $3.5M in OTA commissions could save $1.5-2M annually by shifting just 15-20% of OTA bookings to direct channels.

The distribution agent manages rate parity across all channels, scores channel performance on a commission-adjusted basis, optimizes metasearch bidding, and drives direct booking conversion through intelligent website personalization and retargeting.

Rate Parity Monitoring & Channel Optimization

from datetime import datetime
from typing import Dict, List, Optional

class DistributionAgent:
    """AI agent for hotel distribution and channel management."""

    def __init__(self, channel_manager, rate_shopper, booking_engine, meta_api):
        self.cm = channel_manager        # SiteMinder, D-Edge, etc.
        self.shopper = rate_shopper       # OTA Insight, RateGain
        self.engine = booking_engine      # Direct booking engine
        self.meta = meta_api              # Google Hotel Ads, TripAdvisor, trivago

    def monitor_rate_parity(self, hotel_id: str) -> dict:
        """Real-time rate parity check across all distribution channels."""
        channels = ['booking_com', 'expedia', 'hotels_com', 'agoda',
                    'direct_web', 'direct_mobile', 'gds_amadeus',
                    'gds_sabre', 'gds_travelport']

        check_dates = self._get_check_dates(days_ahead=30, sample_size=10)
        violations = []
        parity_score = {'total_checks': 0, 'in_parity': 0}

        for date in check_dates:
            for room_type in self.cm.get_room_types(hotel_id):
                rates = {}
                for channel in channels:
                    rate = self.shopper.get_displayed_rate(
                        hotel_id, channel, room_type, date
                    )
                    if rate:
                        rates[channel] = rate

                if not rates:
                    continue

                parity_score['total_checks'] += 1
                direct_rate = rates.get('direct_web')
                min_ota_rate = min(
                    (r for c, r in rates.items() if c not in ('direct_web', 'direct_mobile')),
                    default=None
                )

                # Check for parity violations
                if direct_rate and min_ota_rate:
                    # Direct should never be more than 2% above lowest OTA
                    if direct_rate > min_ota_rate * 1.02:
                        violations.append({
                            'date': date,
                            'room_type': room_type,
                            'direct_rate': direct_rate,
                            'lowest_ota': min_ota_rate,
                            'lowest_channel': min(
                                ((c, r) for c, r in rates.items()
                                 if c not in ('direct_web', 'direct_mobile')),
                                key=lambda x: x[1]
                            )[0],
                            'difference_pct': round(
                                (direct_rate - min_ota_rate) / min_ota_rate * 100, 1
                            ),
                            'severity': 'high' if direct_rate > min_ota_rate * 1.05 else 'medium'
                        })
                    else:
                        parity_score['in_parity'] += 1

                # Check OTA-to-OTA parity (some OTAs undercut via packages)
                ota_rates = {c: r for c, r in rates.items()
                            if c not in ('direct_web', 'direct_mobile')}
                if ota_rates:
                    max_ota = max(ota_rates.values())
                    min_ota = min(ota_rates.values())
                    if max_ota > min_ota * 1.05:
                        violations.append({
                            'date': date,
                            'room_type': room_type,
                            'type': 'ota_disparity',
                            'max_channel': max(ota_rates, key=ota_rates.get),
                            'max_rate': max_ota,
                            'min_channel': min(ota_rates, key=ota_rates.get),
                            'min_rate': min_ota,
                            'severity': 'low'
                        })

        return {
            'parity_score': round(
                parity_score['in_parity'] / max(parity_score['total_checks'], 1) * 100, 1
            ),
            'violations': violations,
            'critical_violations': len([v for v in violations if v.get('severity') == 'high']),
            'auto_corrections': self._auto_correct_violations(violations, hotel_id)
        }

    def score_channel_performance(self, hotel_id: str,
                                   period_days: int = 30) -> dict:
        """Score channels by commission-adjusted RevPAR (Net RevPAR)."""
        channels = self.cm.get_active_channels(hotel_id)
        channel_scores = []

        for channel in channels:
            bookings = self.cm.get_bookings(hotel_id, channel['id'], days=period_days)

            if not bookings:
                continue

            total_revenue = sum(b['total_revenue'] for b in bookings)
            total_room_nights = sum(b['nights'] for b in bookings)
            commission_pct = channel.get('commission_pct', 0)
            total_commission = total_revenue * (commission_pct / 100)
            net_revenue = total_revenue - total_commission

            # Cancellation rate by channel
            cancellations = len([b for b in bookings if b.get('cancelled')])
            cancel_rate = cancellations / len(bookings) if bookings else 0

            # Average booking window (advance purchase days)
            avg_lead_time = np.mean([
                (b['check_in'] - b['booking_date']).days for b in bookings
                if not b.get('cancelled')
            ])

            # Length of stay
            avg_los = np.mean([b['nights'] for b in bookings if not b.get('cancelled')])

            # Ancillary revenue correlation
            ancillary = np.mean([
                b.get('ancillary_revenue', 0) for b in bookings
                if not b.get('cancelled')
            ])

            # Net RevPAR contribution
            net_revpar = net_revenue / total_room_nights if total_room_nights else 0

            # Composite channel value score
            score = (
                net_revpar * 0.35 +
                (1 - cancel_rate) * 100 * 0.20 +
                avg_los * 20 * 0.15 +
                avg_lead_time * 0.5 * 0.15 +
                ancillary * 0.15
            )

            channel_scores.append({
                'channel': channel['name'],
                'gross_revenue': round(total_revenue),
                'commission': round(total_commission),
                'net_revenue': round(net_revenue),
                'room_nights': total_room_nights,
                'gross_adr': round(total_revenue / total_room_nights, 2),
                'net_adr': round(net_revenue / total_room_nights, 2),
                'net_revpar_contribution': round(net_revpar, 2),
                'cancel_rate': f"{cancel_rate:.1%}",
                'avg_lead_time': round(avg_lead_time, 1),
                'avg_los': round(avg_los, 1),
                'ancillary_per_stay': round(ancillary, 2),
                'value_score': round(score, 1)
            })

        channel_scores.sort(key=lambda c: -c['value_score'])

        return {
            'channel_rankings': channel_scores,
            'direct_share': self._calculate_direct_share(channel_scores),
            'total_commission_paid': sum(c['commission'] for c in channel_scores),
            'recommendations': self._generate_channel_recommendations(channel_scores)
        }

    def optimize_metasearch_bids(self, hotel_id: str) -> dict:
        """Optimize bids for Google Hotel Ads, TripAdvisor, trivago."""
        platforms = ['google_hotel_ads', 'tripadvisor', 'trivago']
        optimizations = []

        for platform in platforms:
            # Get historical performance
            perf = self.meta.get_performance(hotel_id, platform, days=30)

            if not perf:
                continue

            current_bid = perf['avg_cpc']
            impressions = perf['impressions']
            clicks = perf['clicks']
            bookings = perf['bookings']
            revenue = perf['booking_revenue']
            spend = perf['total_spend']

            ctr = clicks / impressions if impressions else 0
            conversion_rate = bookings / clicks if clicks else 0
            roas = revenue / spend if spend else 0
            cpa = spend / bookings if bookings else float('inf')

            # Target ROAS: 8x for Google, 6x for TripAdvisor, 5x for trivago
            target_roas = {'google_hotel_ads': 8, 'tripadvisor': 6, 'trivago': 5}
            target = target_roas[platform]

            # Bid adjustment
            if roas > target * 1.2:
                # Performing well: increase bid to capture more volume
                new_bid = current_bid * 1.15
                action = 'increase'
            elif roas < target * 0.8:
                # Underperforming: decrease bid
                new_bid = current_bid * 0.85
                action = 'decrease'
            elif roas < target * 0.5:
                # Severely underperforming: pause or drastically reduce
                new_bid = current_bid * 0.50
                action = 'reduce_significantly'
            else:
                new_bid = current_bid
                action = 'maintain'

            optimizations.append({
                'platform': platform,
                'current_cpc': round(current_bid, 2),
                'recommended_cpc': round(new_bid, 2),
                'action': action,
                'current_roas': round(roas, 1),
                'target_roas': target,
                'monthly_spend': round(spend),
                'monthly_revenue': round(revenue),
                'cpa': round(cpa, 2),
                'conversion_rate': f"{conversion_rate:.2%}"
            })

        return {
            'optimizations': optimizations,
            'total_meta_spend': sum(o['monthly_spend'] for o in optimizations),
            'total_meta_revenue': sum(o['monthly_revenue'] for o in optimizations),
            'blended_roas': round(
                sum(o['monthly_revenue'] for o in optimizations) /
                max(sum(o['monthly_spend'] for o in optimizations), 1), 1
            )
        }

6. ROI Analysis: 300-Room Hotel Portfolio

Let's put concrete numbers behind every agent for a realistic 300-room full-service hotel with a $45M total revenue (rooms $32M, F&B $8M, other $5M), running at 76% average occupancy with a $155 ADR and a $117.80 RevPAR.

Implementation Cost vs. Return

The key metrics to track across your AI agent deployment:

• RevPAR (Revenue Per Available Room) — Primary rooms revenue metric. Target: +8-15% within 6 months.
• TRevPAR (Total Revenue Per Available Room) — Includes F&B, spa, parking. Captures cross-selling impact. Target: +6-10%.
• GOPPAR (Gross Operating Profit Per Available Room) — The ultimate profitability metric. Factors in both revenue gains and cost savings. Target: +12-20%.
• ADR (Average Daily Rate) — Monitor alongside occupancy to ensure you are optimizing revenue, not just filling rooms.
• Guest Satisfaction Score (GSS) — Must not decrease as you automate. Target: +5-15 points.
• Direct booking percentage — Tracks channel shift away from high-commission OTAs. Target: +10-15 percentage points.

The hotels seeing the best results in 2026 are not choosing between revenue management and guest experience or between operational efficiency and personalization. They are deploying AI agents across all six operational domains simultaneously, because the compounding effects are where the real competitive advantage emerges. A better-personalized guest books direct next time (lower commission), stays longer (higher ADR), spends more at the restaurant (higher TRevPAR), and leaves a better review (more organic bookings). Each agent amplifies the others.