Restore guestline-monitor features: room stock/occupancy, rate history, room stats

- dates endpoint now returns total_avail, benchmark rate (tier-resolved),
  min-stay nights, max_rooms (summed per-room stock) and friendly labels
- rooms endpoint groups by room with stock, occupancy, best/bench rate and
  nested rate plans; injects known-but-absent rooms; honours room_order
- new /history/{stay_date} endpoint: per room/rate series, benchmark mode,
  or overall availability + cheapest rate per scrape run
- new /stats endpoint: per-room stock + windowed avg rate / current occ /
  est. final occ (look-back window so past dates give completed bookings)
- DirectRates page: occupancy pills, avail x/stock, Room Stats tab,
  history chart modal (plotly), expandable rate plans with history links
- fix discovery trigger crashing (run_until_complete inside running loop)
- fix migration script reading wrong column name (engine_profile)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/api/direct.py

@@ -1,7 +1,6 @@
 """
 Direct booking engine API — hotel management, discovery, scrape control, rate data.
 """
-import asyncio
 import json
 import logging
 from datetime import date, timedelta
@@ -11,7 +10,7 @@ from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks
 from pydantic import BaseModel
 from sqlalchemy import text
 
-from database import AsyncSessionLocal, SyncSessionLocal
+from database import AsyncSessionLocal
 from auth import get_current_user, require_cap
 from services.direct_profiles import PROFILES, detect_profile, get_profile
 from services.direct_scraper import run_discovery, get_discovery_status, run_scrape
@@ -45,6 +44,37 @@ class DetectRequest(BaseModel):
     url: str
 
 
+# ─── Benchmark helpers (ported from guestline-monitor) ──────────────────────
+
+def _resolve_bench(prices: dict, bench_room: str, offsets: dict):
+    """
+    Derive the benchmark room's price from whatever room prices are available.
+    `prices` maps room_id -> price on the benchmark rate plan.
+    All offsets are relative to tier_base_room; bench = room_price - room_offset + bench_offset.
+    Returns (price, is_calculated).
+    """
+    if prices.get(bench_room) is not None:
+        return float(prices[bench_room]), False
+
+    bench_offset = offsets.get(bench_room, 0)
+    for room_id, room_offset in offsets.items():
+        if room_id == bench_room:
+            continue
+        if prices.get(room_id) is not None:
+            return float(prices[room_id]) - room_offset + bench_offset, True
+
+    for v in prices.values():
+        if v is not None:
+            return float(v), True
+
+    return None, False
+
+
+def _room_stock(rows) -> dict:
+    """room_id -> MAX(availability) ever seen = estimated stock."""
+    return {r["room_id"]: r["stock"] for r in rows}
+
+
 # ─── Profiles ────────────────────────────────────────────────────────────────
 
 @router.get("/profiles")
@@ -161,10 +191,7 @@ async def trigger_discovery(hotel_id: int, background_tasks: BackgroundTasks, us
     if not hotel:
         raise HTTPException(status_code=404, detail="Hotel not found")
 
-    background_tasks.add_task(
-        asyncio.get_event_loop().run_until_complete,
-        run_discovery(hotel_id, hotel["profile_name"], hotel["params"])
-    )
+    background_tasks.add_task(run_discovery, hotel_id, hotel["profile_name"], hotel["params"])
     return {"status": "discovery started", "hotel_id": hotel_id}
 
 
@@ -220,39 +247,81 @@ async def hotel_dates(
         if not hotel:
             raise HTTPException(status_code=404, detail="Hotel not found")
 
-        # Latest snapshot per date: cheapest available price_incl
+        # Latest snapshot per (date, room, rate)
         rates_result = await db.execute(
             text("""
-                SELECT
-                    r.stay_date,
-                    MIN(r.price_incl) FILTER (WHERE r.availability > 0) AS cheapest_rate,
-                    BOOL_OR(r.availability > 0)                          AS has_availability,
-                    BOOL_OR(r.min_stay_nights IS NOT NULL
-                            AND r.min_stay_nights > 1)                   AS has_min_stay,
-                    MAX(r.scraped_at)                                     AS scraped_at
-                FROM (
-                    SELECT DISTINCT ON (room_id, rate_id)
-                        stay_date, room_id, rate_id, availability,
-                        price_incl, min_stay_nights, scraped_at
-                    FROM direct_rates
-                    WHERE hotel_id = :hid
-                      AND stay_date BETWEEN :fd AND :td
-                    ORDER BY room_id, rate_id, scraped_at DESC
-                ) r
-                GROUP BY r.stay_date
-                ORDER BY r.stay_date
+                SELECT DISTINCT ON (stay_date, room_id, rate_id)
+                    stay_date, room_id, rate_id, availability,
+                    price_incl, min_stay_nights, scraped_at
+                FROM direct_rates
+                WHERE hotel_id = :hid
+                  AND stay_date BETWEEN :fd AND :td
+                ORDER BY stay_date, room_id, rate_id, scraped_at DESC
             """),
             {"hid": hotel_id, "fd": from_date, "td": to_date}
         )
-        dates = [dict(r) for r in rates_result.mappings().all()]
+        rows = rates_result.mappings().all()
 
-        return {
-            "hotel_id": hotel_id,
-            "hotel_name": hotel["name"],
-            "from_date": str(from_date),
-            "to_date": str(to_date),
-            "dates": dates,
-        }
+        # Stock = max availability ever seen per room (across all history)
+        stock_result = await db.execute(
+            text("""SELECT room_id, MAX(availability) AS stock
+                    FROM direct_rates WHERE hotel_id = :hid GROUP BY room_id"""),
+            {"hid": hotel_id}
+        )
+        room_stock = {r["room_id"]: r["stock"] for r in stock_result.mappings().all()}
+
+    bench_room = hotel["benchmark_room"] or ""
+    bench_rate = hotel["benchmark_rate"] or ""
+    offsets = hotel["tier_offsets"] or {}
+    max_rooms = sum(v for v in room_stock.values() if v) or None
+
+    # Group by date; availability counted once per room (max across its rate plans)
+    by_date: dict = {}
+    for r in rows:
+        d = by_date.setdefault(r["stay_date"], {
+            "room_avail": {}, "bench_prices": {}, "prices": [],
+            "min_stay": None, "scraped_at": None,
+        })
+        room_avail = d["room_avail"]
+        room_avail[r["room_id"]] = max(room_avail.get(r["room_id"], 0), r["availability"] or 0)
+        if r["price_incl"] is not None and (r["availability"] or 0) > 0:
+            d["prices"].append(float(r["price_incl"]))
+        if r["rate_id"] == bench_rate and r["price_incl"] is not None:
+            d["bench_prices"][r["room_id"]] = r["price_incl"]
+        if r["min_stay_nights"] and (d["min_stay"] is None or r["min_stay_nights"] > d["min_stay"]):
+            d["min_stay"] = r["min_stay_nights"]
+        if d["scraped_at"] is None or r["scraped_at"] > d["scraped_at"]:
+            d["scraped_at"] = r["scraped_at"]
+
+    dates = []
+    for stay_date in sorted(by_date):
+        d = by_date[stay_date]
+        total_avail = sum(d["room_avail"].values())
+        bench_price, bench_calc = _resolve_bench(d["bench_prices"], bench_room, offsets)
+        dates.append({
+            "stay_date": stay_date,
+            "total_avail": total_avail,
+            "cheapest_rate": min(d["prices"]) if d["prices"] else None,
+            "bench_rate": round(bench_price, 2) if bench_price is not None else None,
+            "bench_calculated": bench_calc,
+            "has_availability": total_avail > 0,
+            "min_stay_nights": d["min_stay"],
+            "has_min_stay": bool(d["min_stay"] and d["min_stay"] > 1),
+            "scraped_at": d["scraped_at"],
+        })
+
+    return {
+        "hotel_id": hotel_id,
+        "hotel_name": hotel["name"],
+        "from_date": str(from_date),
+        "to_date": str(to_date),
+        "max_rooms": max_rooms,
+        "room_labels": hotel["room_labels"] or {},
+        "rate_labels": hotel["rate_labels"] or {},
+        "benchmark_room": bench_room or None,
+        "benchmark_rate": bench_rate or None,
+        "dates": dates,
+    }
 
 
 @router.get("/hotels/{hotel_id}/date/{rate_date}/rooms")
@@ -264,7 +333,7 @@ async def hotel_date_rooms(
     require_cap(user, "view_direct_rates")
     async with AsyncSessionLocal() as db:
         cfg_row = await db.execute(
-            text("SELECT room_labels, rate_labels, tier_offsets, tier_base_room, benchmark_room, benchmark_rate FROM direct_competitor_hotels WHERE id = :id"),
+            text("SELECT room_labels, rate_labels, room_order, tier_offsets, tier_base_room, benchmark_room, benchmark_rate FROM direct_competitor_hotels WHERE id = :id"),
             {"id": hotel_id}
         )
         hotel = cfg_row.mappings().fetchone()
@@ -283,55 +352,280 @@ async def hotel_date_rooms(
             """),
             {"hid": hotel_id, "sd": rate_date}
         )
-        rooms = [dict(r) for r in rooms_result.mappings().all()]
+        rows = [dict(r) for r in rooms_result.mappings().all()]
 
-        # Resolve tier-normalised benchmark rates
-        room_labels = hotel["room_labels"] or {}
-        rate_labels = hotel["rate_labels"] or {}
-        tier_offsets = hotel["tier_offsets"] or {}
-        tier_base_room = hotel["tier_base_room"]
-        benchmark_room = hotel["benchmark_room"]
-        benchmark_rate = hotel["benchmark_rate"]
+        # Stock = max availability ever seen per room
+        stock_result = await db.execute(
+            text("""SELECT room_id, MAX(availability) AS stock
+                    FROM direct_rates WHERE hotel_id = :hid GROUP BY room_id"""),
+            {"hid": hotel_id}
+        )
+        room_stock = {r["room_id"]: r["stock"] for r in stock_result.mappings().all()}
 
-        # Find benchmark price
-        bench_price = None
-        if benchmark_room and benchmark_rate:
-            bench_match = next(
-                (r for r in rooms
-                 if r["room_id"] == benchmark_room and r["rate_id"] == benchmark_rate
-                 and r["availability"] > 0 and r["price_incl"]),
-                None
-            )
-            if bench_match:
-                bench_price = float(bench_match["price_incl"])
-            elif tier_base_room and tier_offsets:
-                base_match = next(
-                    (r for r in rooms
-                     if r["room_id"] == tier_base_room and r["availability"] > 0 and r["price_incl"]),
-                    None
-                )
-                if base_match:
-                    base_price = float(base_match["price_incl"])
-                    bench_offset = tier_offsets.get(benchmark_room, 0)
-                    base_offset = tier_offsets.get(tier_base_room, 0)
-                    bench_price = base_price - base_offset + bench_offset
+    room_labels = hotel["room_labels"] or {}
+    rate_labels = hotel["rate_labels"] or {}
+    tier_offsets = hotel["tier_offsets"] or {}
+    benchmark_room = hotel["benchmark_room"] or ""
+    benchmark_rate = hotel["benchmark_rate"] or ""
+    bench_offset = tier_offsets.get(benchmark_room, 0)
 
-        enriched = []
-        for r in rooms:
-            r["room_label"] = room_labels.get(r["room_id"], r["room_id"])
-            r["rate_label"] = rate_labels.get(r["rate_id"], r["rate_id"])
-            # Derive bench_rate for this room from tier offsets
-            r["bench_rate"] = None
-            if bench_price is not None and tier_offsets and tier_base_room:
-                room_offset = tier_offsets.get(r["room_id"])
-                bench_offset = tier_offsets.get(benchmark_room, 0)
-                if room_offset is not None:
-                    r["bench_rate"] = round(bench_price + (room_offset - bench_offset), 2)
-            enriched.append(r)
+    # Group flat room×rate rows into one entry per room
+    rooms: dict = {}
+    for r in rows:
+        rid = r["room_id"]
+        room = rooms.setdefault(rid, {
+            "room_id": rid,
+            "room_label": room_labels.get(rid, rid),
+            "availability": r["availability"],
+            "stock": room_stock.get(rid),
+            "best_rate": None,
+            "bench_rate": None,
+            "bench_calculated": False,
+            "min_stay_nights": r["min_stay_nights"],
+            "unavailable": False,
+            "rates": [],
+        })
+        room["availability"] = max(room["availability"] or 0, r["availability"] or 0)
+        if r["price_incl"] is not None:
+            p = float(r["price_incl"])
+            if room["best_rate"] is None or p < room["best_rate"]:
+                room["best_rate"] = p
+        if r["min_stay_nights"] and not room["min_stay_nights"]:
+            room["min_stay_nights"] = r["min_stay_nights"]
+        room["rates"].append({
+            "rate_id": r["rate_id"],
+            "rate_label": rate_labels.get(r["rate_id"], r["rate_id"]),
+            "price_incl": r["price_incl"],
+            "price_excl": r["price_excl"],
+            "min_stay_nights": r["min_stay_nights"],
+        })
 
-        return {
-            "hotel_id": hotel_id,
-            "date": str(rate_date),
-            "bench_price": bench_price,
-            "rooms": enriched,
+    # Resolve the benchmark room's price today from whatever rooms are present
+    bench_prices = {}
+    for rid, room in rooms.items():
+        p = next((rt["price_incl"] for rt in room["rates"] if rt["rate_id"] == benchmark_rate), None)
+        if p is not None:
+            bench_prices[rid] = p
+    bench_price, _ = _resolve_bench(bench_prices, benchmark_room, tier_offsets)
+
+    # Per-room benchmark: direct quote if present, else estimate via tier offsets
+    for rid, room in rooms.items():
+        direct = bench_prices.get(rid)
+        room_offset = tier_offsets.get(rid)
+        if direct is not None:
+            room["bench_rate"] = round(float(direct), 2)
+        elif bench_price is not None and room_offset is not None:
+            room["bench_rate"] = round(bench_price + (room_offset - bench_offset), 2)
+            room["bench_calculated"] = True
+
+    # Inject known rooms absent from this date's snapshot (no availability at all)
+    for rid in set(room_stock) - set(rooms):
+        room_offset = tier_offsets.get(rid)
+        derived = round(bench_price + (room_offset - bench_offset), 2) \
+            if bench_price is not None and room_offset is not None else None
+        rooms[rid] = {
+            "room_id": rid,
+            "room_label": room_labels.get(rid, rid),
+            "availability": None,
+            "stock": room_stock.get(rid),
+            "best_rate": None,
+            "bench_rate": derived,
+            "bench_calculated": derived is not None,
+            "min_stay_nights": None,
+            "unavailable": True,
+            "rates": [],
         }
+
+    # Sort by configured room_order, then by stock (largest first)
+    room_order = hotel["room_order"] or []
+
+    def sort_key(r):
+        try:
+            return (0, room_order.index(r["room_id"]), 0)
+        except ValueError:
+            return (1, -(r["stock"] or 0), 0)
+
+    return {
+        "hotel_id": hotel_id,
+        "date": str(rate_date),
+        "bench_price": bench_price,
+        "rooms": sorted(rooms.values(), key=sort_key),
+    }
+
+
+# ─── Rate history ─────────────────────────────────────────────────────────────
+
+@router.get("/hotels/{hotel_id}/history/{stay_date}")
+async def rate_history(
+    hotel_id: int,
+    stay_date: date,
+    room_id: Optional[str] = None,
+    rate_id: Optional[str] = None,
+    mode: Optional[str] = None,
+    user=Depends(get_current_user)
+):
+    """
+    Time series of tracked rates for one stay date, across scrape runs.
+    - room_id + rate_id: history for that exact room/rate combo
+    - mode=benchmark: benchmark-room price history (tier-resolved)
+    - default: total availability + cheapest rate per scrape
+    """
+    require_cap(user, "view_direct_rates")
+    async with AsyncSessionLocal() as db:
+        cfg_row = await db.execute(
+            text("SELECT benchmark_room, benchmark_rate, tier_offsets FROM direct_competitor_hotels WHERE id = :id"),
+            {"id": hotel_id}
+        )
+        hotel = cfg_row.mappings().fetchone()
+        if not hotel:
+            raise HTTPException(status_code=404, detail="Hotel not found")
+
+        if room_id and rate_id:
+            result = await db.execute(
+                text("""SELECT scraped_at, availability, price_incl
+                        FROM direct_rates
+                        WHERE hotel_id = :hid AND stay_date = :sd
+                          AND room_id = :room AND rate_id = :rate
+                        ORDER BY scraped_at"""),
+                {"hid": hotel_id, "sd": stay_date, "room": room_id, "rate": rate_id}
+            )
+            return [dict(r) for r in result.mappings().all()]
+
+        result = await db.execute(
+            text("""SELECT scraped_at, room_id, rate_id, availability, price_incl
+                    FROM direct_rates
+                    WHERE hotel_id = :hid AND stay_date = :sd
+                    ORDER BY scraped_at"""),
+            {"hid": hotel_id, "sd": stay_date}
+        )
+        rows = result.mappings().all()
+
+    bench_room = hotel["benchmark_room"] or ""
+    bench_rate = hotel["benchmark_rate"] or ""
+    offsets = hotel["tier_offsets"] or {}
+
+    # Group by scrape timestamp
+    by_scrape: dict = {}
+    for r in rows:
+        s = by_scrape.setdefault(r["scraped_at"], {"room_avail": {}, "bench_prices": {}, "prices": []})
+        s["room_avail"][r["room_id"]] = max(s["room_avail"].get(r["room_id"], 0), r["availability"] or 0)
+        if r["price_incl"] is not None:
+            s["prices"].append(float(r["price_incl"]))
+            if r["rate_id"] == bench_rate:
+                s["bench_prices"][r["room_id"]] = r["price_incl"]
+
+    series = []
+    for scraped_at in sorted(by_scrape):
+        s = by_scrape[scraped_at]
+        if mode == "benchmark":
+            price, calc = _resolve_bench(s["bench_prices"], bench_room, offsets)
+            series.append({
+                "scraped_at": scraped_at,
+                "price_incl": round(price, 2) if price is not None else None,
+                "calculated": calc,
+            })
+        else:
+            series.append({
+                "scraped_at": scraped_at,
+                "availability": sum(s["room_avail"].values()),
+                "price_incl": min(s["prices"]) if s["prices"] else None,
+            })
+    return series
+
+
+# ─── Per-room stats ───────────────────────────────────────────────────────────
+
+STAT_WINDOWS = {
+    "this_week": 7,
+    "this_month": 30,
+    "next_6mo": 182,
+    "next_12mo": 365,
+}
+
+
+@router.get("/hotels/{hotel_id}/stats")
+async def hotel_stats(hotel_id: int, user=Depends(get_current_user)):
+    """
+    Per-room stats on the benchmark rate across time windows: stock (max
+    availability ever seen), avg rate, current occupancy vs stock, and
+    estimated final occupancy from the equivalent look-back window (past
+    dates no longer change, so they reflect completed bookings).
+    """
+    require_cap(user, "view_direct_rates")
+    today = date.today()
+
+    async with AsyncSessionLocal() as db:
+        cfg_row = await db.execute(
+            text("SELECT room_labels, benchmark_rate FROM direct_competitor_hotels WHERE id = :id"),
+            {"id": hotel_id}
+        )
+        hotel = cfg_row.mappings().fetchone()
+        if not hotel:
+            raise HTTPException(status_code=404, detail="Hotel not found")
+
+        stock_result = await db.execute(
+            text("""SELECT room_id, MAX(availability) AS stock
+                    FROM direct_rates WHERE hotel_id = :hid GROUP BY room_id"""),
+            {"hid": hotel_id}
+        )
+        stock = {r["room_id"]: r["stock"] for r in stock_result.mappings().all()}
+
+        bench_rate = hotel["benchmark_rate"] or ""
+        result = await db.execute(
+            text("""SELECT DISTINCT ON (stay_date, room_id)
+                        stay_date, room_id, availability, price_incl
+                    FROM direct_rates
+                    WHERE hotel_id = :hid AND rate_id = :rate
+                      AND stay_date BETWEEN :fd AND :td
+                    ORDER BY stay_date, room_id, scraped_at DESC"""),
+            {"hid": hotel_id, "rate": bench_rate,
+             "fd": today - timedelta(days=365), "td": today + timedelta(days=365)}
+        )
+        rows = result.mappings().all()
+
+    room_labels = hotel["room_labels"] or {}
+    by_room: dict = {}
+    for r in rows:
+        by_room.setdefault(r["room_id"], []).append(r)
+
+    result_out = {}
+    for room_id, room_stock_val in stock.items():
+        room_rows = by_room.get(room_id, [])
+        room_stock_safe = room_stock_val or 1
+        windows = {}
+        for wname, days in STAT_WINDOWS.items():
+            future = [r for r in room_rows if today <= r["stay_date"] < today + timedelta(days=days)]
+            past = [r for r in room_rows if today - timedelta(days=days) <= r["stay_date"] < today]
+
+            if not future and not past:
+                windows[wname] = None
+                continue
+
+            prices = [float(r["price_incl"]) for r in future if r["price_incl"] is not None]
+            avg_price = round(sum(prices) / len(prices), 2) if prices else None
+
+            avg_occ = None
+            if future:
+                avg_avail = sum(r["availability"] or 0 for r in future) / len(future)
+                avg_occ = round((1 - avg_avail / room_stock_safe) * 100, 1)
+
+            est_final_occ = None
+            if past:
+                past_sold = sum(room_stock_safe - (r["availability"] or 0) for r in past)
+                est_final_occ = round(past_sold / (len(past) * room_stock_safe) * 100, 1)
+
+            windows[wname] = {
+                "dates": len(future),
+                "avg_price": avg_price,
+                "avg_occ": avg_occ,
+                "past_dates": len(past),
+                "est_final_occ": est_final_occ,
+            }
+
+        result_out[room_id] = {
+            "stock": room_stock_val,
+            "room_label": room_labels.get(room_id, room_id),
+            "windows": windows,
+        }
+
+    return result_out

backend/migrate_direct_data.py

@@ -10,6 +10,7 @@ import sys
 from datetime import datetime, timezone
 
 import psycopg2
+import psycopg2.extras
 
 ARCHIVE_DIR = os.environ.get(
     "GUESTLINE_ARCHIVE",
@@ -37,7 +38,7 @@ def migrate():
     src = sqlite3.connect(CONFIG_DB)
     src.row_factory = dict_row
     hotels = src.execute(
-        """SELECT id, name, profile, params, room_labels, rate_labels,
+        """SELECT id, name, engine_profile, params, room_labels, rate_labels,
                   room_order, benchmark_room, benchmark_rate,
                   tier_base_room, tier_offsets, scrape_enabled, last_scraped_at
            FROM hotels ORDER BY id"""
@@ -65,7 +66,7 @@ def migrate():
                    RETURNING id""",
                 (
                     h["name"],
-                    h["profile"],
+                    h["engine_profile"],
                     params,
                     h["room_labels"] or "{}",
                     h["rate_labels"] or "{}",
@@ -163,5 +164,4 @@ def migrate():
 
 
 if __name__ == "__main__":
-    import psycopg2.extras
     migrate()