ADR Decline by Hotel Category

Tactical Triggers

Revenue Flow

The hotel only pays when the AI demonstrably increases revenue. Zero risk, pure upside alignment.

Three-Layer Architecture

15+ Real-Time Data Sources

Signal Processing Pipeline

Signal Processor

class SignalProcessor:
    """Process raw data signals with anomaly detection and smoothing."""

    def __init__(self, z_threshold=3.0, alpha=0.3):
        self.z_threshold = z_threshold
        self.alpha = alpha  # Exponential smoothing factor
        self.history = defaultdict(list)

    def process(self, source: str, value: float) -> dict:
        self.history[source].append(value)
        window = self.history[source][-100:]

        # Anomaly detection via Z-score
        mean = np.mean(window)
        std = np.std(window) or 1e-6
        z_score = abs(value - mean) / std
        is_anomaly = z_score > self.z_threshold

        # Exponential smoothing
        if len(window) < 2:
            smoothed = value
        else:
            smoothed = self.alpha * value + (1 - self.alpha) * window[-2]

        return {
            'source': source,
            'raw': value,
            'smoothed': round(smoothed, 4),
            'z_score': round(z_score, 2),
            'is_anomaly': is_anomaly,
            'action': 'FILTERED' if is_anomaly else 'PASSED'
        }

Smart OTA Collection (Anti-Detection)

class OTAScraper:
    """Anti-detection OTA price collector using undetected-chromedriver."""

    def __init__(self):
        self.options = uc.ChromeOptions()
        self.options.add_argument('--headless=new')
        self.user_agents = self._load_user_agents()
        self.proxy_pool = ProxyRotator()

    async def collect_price(self, source: str, hotel_id: str):
        driver = uc.Chrome(options=self.options)
        driver.execute_cdp_cmd('Network.setUserAgentOverride', {
            'userAgent': random.choice(self.user_agents)
        })
        await asyncio.sleep(random.uniform(2, 5))  # Human-like delay
        # ... price extraction logic
        return {'source': source, 'price': extracted_price}

OTA Channel Inventory Parity

Real-time room inventory status across major OTA channels. When channels show "tight supply," consumer panic-buying psychology enables higher ADR.

Ensemble Model Architecture

Ensemble Model Service

class EnsembleModelService:
    """Weighted ensemble combining XGBoost and LSTM predictions."""

    def __init__(self, xgb_weight=0.6, lstm_weight=0.4):
        self.xgb_model = XGBRegressor(n_estimators=500, max_depth=8)
        self.lstm_model = LSTMPricePredictor(
            input_dim=20, hidden_dims=[128, 64, 32],
            attention=True, dropout=0.3
        )
        self.weights = {'xgb': xgb_weight, 'lstm': lstm_weight}
        self.scaler = RobustScaler()
        self.drift_detector = DDM()  # Drift Detection Method

    def predict(self, features: dict, sequence: np.ndarray) -> dict:
        xgb_input = self.scaler.transform(pd.DataFrame([features]))
        xgb_pred = self.xgb_model.predict(xgb_input)[0]
        lstm_input = torch.FloatTensor(sequence).unsqueeze(0)
        lstm_pred = self.lstm_model(lstm_input).item()
        ensemble_pred = (
            self.weights['xgb'] * xgb_pred +
            self.weights['lstm'] * lstm_pred
        )
        ci_low, ci_high = self._bootstrap_ci(features, sequence)
        return {
            'price': round(ensemble_pred, 2),
            'confidence_interval': (ci_low, ci_high),
            'xgb_contribution': xgb_pred,
            'lstm_contribution': lstm_pred,
            'drift_score': self.drift_detector.score
        }

Regional Supply Intelligence (v8)

Supply-side features integrated into the ensemble model to predict scarcity-driven pricing opportunities before occupancy thresholds are reached.

Price Elasticity Matrix

Optimal Price Finder

Anomaly Detection

Weighted Competitor Position Analysis

Weather & Event Impact Model

class AdvancedWeatherModel:
    TYPHOON_MULTIPLIERS = {1: 1.05, 3: 1.20, 8: 1.40, 10: 1.60}
    HEATWAVE_THRESHOLD = 35  # Celsius
    RAIN_THRESHOLD = 5       # mm/hr

    def calculate_premium(self, weather: dict) -> float:
        premium = 1.0
        if signal := weather.get('typhoon_signal'):
            premium *= self.TYPHOON_MULTIPLIERS.get(signal, 1.0)
            if weather.get('hours_to_landfall', 999) < 6:
                premium *= 1.30  # Stranded traveler premium
        if weather.get('temperature', 0) > self.HEATWAVE_THRESHOLD:
            premium *= 1.25
        if weather.get('rainfall_mm', 0) > self.RAIN_THRESHOLD:
            premium *= 1.15
        return round(premium, 3)

Price Sensitivity Map — Macau Districts

Quick Presets

Cross-Border Price Barrier (Zhuhai Linkage)

Zhuhai port-area hotels act as virtual auxiliary supply. When Zhuhai is also full or border friction is high, Macau gains pricing sovereignty.

React Dashboard Code

function YieldCommandCenter() {
  const { data: signals } = useQuery('signals', fetchSignals);
  const { data: metrics } = useQuery('metrics', fetchMetrics);

  return (
    <DashboardLayout>
      <KPIRow>
        <KPICard label="Success Fee Pool" value={metrics?.feePool} format="currency" trend={metrics?.feeTrend} variant="success" />
        <KPICard label="Avg ADR Lift" value={metrics?.adrLift} format="percentage" trend={metrics?.liftTrend} />
      </KPIRow>
      <ProfitChart data={metrics?.dailyProfit} />
      <SignalPanel signals={signals} />
    </DashboardLayout>
  );
}

RAM/IAM Access Levels

Deployment Checklist

Circuit Breaker States

CI/CD Pipeline

name: Deploy Pricing Engine
on:
  push:
    branches: [main]
jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: docker build -t macau-engine .
      - run: docker push registry/macau-engine:latest
      - run: kubectl apply -f k8s/deployment.yaml
      - name: Smoke Test
        run: |
          curl -sf https://api.macau-engine.com/healthz
          curl -sf https://api.macau-engine.com/predict/test

Infrastructure Architecture

WeChat Work Interactive Card Payload

async def trigger_wechat_webhook(alert_payload):
    """
    Sends a structured card to WeChat Work with Approval/Veto buttons.
    """
    wechat_card = {
        "msgtype": "template_card",
        "template_card": {
            "card_type": "button_interaction",
            "main_title": {"title": alert_payload["message"]},
            "sub_title_text": alert_payload["details"],
            "button_list": [
                {
                    "text": "Approve Increase",
                    "style": 1,  # Green button
                    "key": f"APPROVE_{alert_payload['opportunity_id']}"
                },
                {
                    "text": "Veto (Keep Base)",
                    "style": 2,  # Red button
                    "key": f"VETO_{alert_payload['opportunity_id']}"
                }
            ]
        }
    }
    async with aiohttp.ClientSession() as session:
        await session.post(WECHAT_WORK_WEBHOOK, json=wechat_card)

Veto Response Handler (FastAPI)

Receives management decisions from WeChat Work buttons and updates the Pricing Engine in real-time.

@app.post("/api/v1/veto-handler")
async def handle_management_decision(payload: dict):
    """
    Receives button interaction from WeChat Work
    and updates the Pricing Engine accordingly.
    """
    decision_key = payload["button_key"]  # e.g., "VETO_12345"
    action, opp_id = decision_key.split("_")

    if action == "VETO":
        # Lock price to MOP 898 baseline in the Feature Store
        await feature_store.store_feature("price_lock", opp_id, 898.0)
        # Log veto event for the Shadow Report audit
        logger.warning(f"Management VETO for Opportunity {opp_id} recorded.")

    else:
        # Authorize the AI to push the high-yield price
        await feature_store.store_feature(
            "price_lock", opp_id, payload["recommended_price"]
        )
        logger.info(f"Management APPROVED increase for Opportunity {opp_id}.")

    return {"status": "success", "action": action}

Border-Adjusted Pricing Engine (v8)

def calculate_border_adjusted_price(
    macau_base: float,
    macau_occupancy: float,
    zhuhai_supply: float,
    clearance_time: int,
    local_ceiling: float = 973
) -> dict:
    """v8: Cross-border price barrier with Zhuhai linkage."""

    # 1. Macau local scarcity score
    macau_scarcity = 1.15 if macau_occupancy > 0.85 else 1.0

    # 2. Zhuhai price anchor factor
    #    If Zhuhai also full (<10% vacancy) → release pricing power
    #    If Zhuhai has excess supply (>40%) → suppress Macau markup
    if zhuhai_supply < 0.10:
        zhuhai_anchor = 1.25  # Zhuhai sold out → Macau sovereignty
    elif zhuhai_supply > 0.40:
        zhuhai_anchor = 0.85  # Zhuhai plenty → constrain pricing
    else:
        zhuhai_anchor = 1.0   # Neutral zone

    # 3. Border friction factor (clearance time as "cutoff" index)
    #    >60min wait → travelers won't cross back to Zhuhai
    friction_factor = 1.12 if clearance_time > 60 else 1.0

    # 4. Supply attrition composite
    supply_factor = (1 / max(zhuhai_supply, 0.05)) ** 0.3

    # 5. Final adjusted price (capped at dynamic ceiling)
    adjusted = macau_base * macau_scarcity * zhuhai_anchor * friction_factor
    return {
        "adjusted_price": min(round(adjusted, 2), local_ceiling),
        "factors": {
            "macau_scarcity": macau_scarcity,
            "zhuhai_anchor": zhuhai_anchor,
            "friction_factor": friction_factor,
            "supply_attrition": round(supply_factor, 3)
        },
        "ceiling_hit": adjusted >= local_ceiling
    }

Access Code Validation Middleware

from fastapi import Header, HTTPException

async def verify_subscription_access(access_code: str = Header(...)):
    # Query your Supabase 'subscriptions' table
    sub_data = supabase.table("subscriptions").select("*").eq(
        "code", access_code
    ).single().execute()

    if not sub_data.data or not sub_data.data['is_active']:
        raise HTTPException(
            status_code=403,
            detail="Access withdrawn. Please contact finance."
        )

    return sub_data.data

Network Security Layers

Model Drift Detection

Live API Rate Limits

Advanced Metrics Dashboard

Alert Configuration

Performance Dashboard (Live)

Signal Type Action Matrix

Real-time signal classification drives automated pricing actions with UI warnings for management oversight.

Cross-Category Spillover Monitor

When upper-tier hotels (4-5 star) approach capacity, demand cascades downward to 3-star properties, granting temporary pricing power.

Dynamic Supply Attrition Index

Port congestion and ferry delays reduce the effective pool of travelers who can leave Macau, creating locked-in demand for overnight stays.

Security Monitoring

Access Audit Log

Key Rotation Schedule

Shadow Price Log (Live)

Every pricing cycle, shadow v9 generates a recommendation that is logged but never applied. Production continues using v8.

30-Day Milestone Timeline

Shadow vs Production Price Comparison

Hourly Revenue Leakage Heatmap

Each cell shows estimated revenue leakage per hour. Red intensity = higher missed yield. Peak window (19:00–21:00) is highlighted.

7-Day Peak Window Leakage Trend

Peak Window Mitigation Actions

Training Completion Status

Module 1: Supply-Side Indicator Reference

Module 2: Reading Supply Signals on the Dashboard

Module 3: Approve/Veto Decision Framework

Module 4: Managing the 19:00–21:00 Peak

The peak leakage window requires heightened attention. During this 2-hour period, walk-in demand spikes and pricing sensitivity drops.

AI-Enhanced Revenue vs Static Baseline — 12 Months

ROI Attribution — Shapley Values

class ROIAttributionModel:
    def __init__(self, lookback_hours=24, confidence=0.85):
        self.lookback = lookback_hours
        self.confidence_threshold = confidence

    def shapley_attribution(self, revenue_data: pd.DataFrame) -> dict:
        """Multi-touch attribution using Shapley values."""
        channels = ['ai_pricing', 'events', 'weather', 'seasonal']
        shapley_values = {}
        for channel in channels:
            marginal = self._marginal_contribution(revenue_data, channel)
            shapley_values[channel] = marginal
        total = sum(shapley_values.values())
        return {k: round(v/total * 100, 1) for k, v in shapley_values.items()}

7-Day Shadow Report

Competitor Displacement Analysis

Automated Report Templates

Go-Live Readiness Checklist

Code Generator

All Access Codes (4)

Code Statistics

Admin Audit Log