rates/backend/jobs/check_rate_parity.py

"""
Rate Parity Check Job

Compares our own hotel's scraped Booking.com rate against the COMPARABLE
Newbook tariff for each date — like-for-like, not headline vs headline.

The scraped Booking.com lead-in rate carries condition flags (breakfast
included, free cancellation, no prepayment); Newbook's tariffs_data holds
every tariff by name with per-day rates (e.g. DIRECT B&B FLEX / PPAY / ADV,
DIRECT DBB ...). We classify tariffs from their names and pick the cheapest
one matching the Booking.com rate's basis: a prepaid B&B rate on Booking.com
is compared against the cheapest prepaid B&B tariff, not the flexi rate.

Fallback tiers when no exact match exists (each alert records which was used):
  matched    — terms (flex/prepaid) and board match
  terms      — terms match, any board except dinner-inclusive
  any        — any non-dinner tariff
  headline   — legacy rate_gross (first tariff) when tariffs_data is absent

Parity is then measured against an expected markup — we deliberately price
Booking.com higher to cover commission: expected = comparable + markup
(% or flat £), alert when the deviation exceeds the tolerance (% or £).
Dates back within tolerance auto-resolve; acknowledged dates stay quiet.

Config (system_config):
  parity_check_enabled     true/false (default true)
  parity_match_mode        'best_available' (default: cheapest bookable
                           non-dinner tariff vs the BC lead-in, which is
                           already BC's best available) or 'match_terms'
                           (flex vs flex, prepaid vs prepaid)
  parity_markup_value      expected Booking.com premium over Newbook (default 0)
  parity_markup_unit       'pct' or 'gbp' (default pct)
  parity_tolerance_value   allowed deviation from expected before alerting (default 2)
  parity_tolerance_unit    'pct' or 'gbp' (default pct)
  (legacy fallbacks: parity_expected_markup_pct, parity_tolerance_pct)

Schedule: daily at 06:45, after the 05:20 Newbook fetch and 05:30 scrape.
"""
import json
import logging
import re
from datetime import date, timedelta

from sqlalchemy import text

from database import SyncSessionLocal

logger = logging.getLogger(__name__)

HORIZON_DAYS = 90

PREPAID_PATTERNS = ('ppay', 'prepay', 'pre-pay', 'advance', 'adv', 'non-ref', 'nonref', 'no refund', 'nrf', 'saver')
DINNER_PATTERNS = ('dbb', 'dinner', 'half board', 'half-board')


def get_parity_config(db) -> dict:
    rows = db.execute(
        text("""SELECT config_key, config_value FROM system_config
                WHERE config_key IN ('parity_check_enabled',
                                     'parity_markup_value', 'parity_markup_unit',
                                     'parity_tolerance_value', 'parity_tolerance_unit',
                                     'parity_expected_markup_pct', 'parity_tolerance_pct')""")
    ).fetchall()
    cfg = {r[0]: r[1] for r in rows}

    def num(key: str, default: float, legacy_key: str = None) -> float:
        raw = cfg.get(key)
        if raw in (None, '') and legacy_key:
            raw = cfg.get(legacy_key)
        try:
            return float(raw if raw not in (None, '') else default)
        except (TypeError, ValueError):
            return default

    def unit(key: str) -> str:
        return 'gbp' if (cfg.get(key) or 'pct').lower() in ('gbp', '£', 'abs') else 'pct'

    enabled_raw = (cfg.get('parity_check_enabled') or 'true').lower()
    return {
        "enabled": enabled_raw in ('true', '1', 'yes', 'enabled'),
        "match_mode": 'match_terms' if (cfg.get('parity_match_mode') or '').lower() == 'match_terms' else 'best_available',
        "markup_value": num('parity_markup_value', 0.0, 'parity_expected_markup_pct'),
        "markup_unit": unit('parity_markup_unit'),
        "tolerance_value": num('parity_tolerance_value', 2.0, 'parity_tolerance_pct'),
        "tolerance_unit": unit('parity_tolerance_unit'),
    }


def expected_booking_rate(newbook_rate: float, cfg: dict) -> float:
    """Expected Booking.com rate: Newbook + markup (% or flat £)."""
    if cfg["markup_unit"] == 'gbp':
        return newbook_rate + cfg["markup_value"]
    return newbook_rate * (1 + cfg["markup_value"] / 100)


def evaluate_parity(booking_rate: float, newbook_rate: float, cfg: dict):
    """Compare actual Booking.com rate against expected.
    Returns (deviation_pct, deviation_gbp, breach) or None if expected invalid.
    The breach test uses the tolerance in its own unit (% or £)."""
    expected = expected_booking_rate(newbook_rate, cfg)
    if expected <= 0:
        return None
    dev_gbp = booking_rate - expected
    dev_pct = dev_gbp / expected * 100
    if cfg["tolerance_unit"] == 'gbp':
        breach = abs(dev_gbp) > cfg["tolerance_value"]
    else:
        breach = abs(dev_pct) > cfg["tolerance_value"]
    return dev_pct, dev_gbp, breach


def _classify_tariff(name: str, description: str = '') -> dict:
    s = f"{name} {description}".lower()
    is_dinner = any(p in s for p in DINNER_PATTERNS)
    is_prepaid = any(re.search(rf'\b{re.escape(p)}\b', s) if p == 'adv' else (p in s)
                     for p in PREPAID_PATTERNS)
    return {"prepaid": is_prepaid, "dinner": is_dinner}


def _candidate_tariffs(tariffs_data, days_ahead: int) -> list:
    """Bookable-tonight tariff options from one category's tariffs_data."""
    if isinstance(tariffs_data, str):
        try:
            tariffs_data = json.loads(tariffs_data)
        except (ValueError, TypeError):
            return []
    out = []
    for t in (tariffs_data or {}).get('tariffs', []):
        rate = t.get('rate')
        if not t.get('success') or not rate or rate <= 0:
            continue
        if (t.get('min_stay') or 1) > 1:
            continue
        if t.get('min_advance_days') and days_ahead < t['min_advance_days']:
            continue
        cls = _classify_tariff(t.get('name', ''), t.get('description', ''))
        out.append({"rate": float(rate), "name": t.get('name', 'Unknown'), **cls})
    return out


def _pick_comparable(candidates: list, booking_flex: bool, booking_breakfast: bool,
                     match_mode: str = 'best_available'):
    """Newbook tariff to compare against the Booking.com lead-in.
    best_available: cheapest bookable non-dinner tariff (the BC lead-in is
    already BC's best available, so this is BAR vs BAR).
    match_terms: cheapest tariff matching the BC rate's flex/prepaid basis,
    falling back to best available. All our tariffs include breakfast, so
    board matching reduces to excluding dinner-inclusive tariffs unless
    nothing else exists.
    Returns (tariff, match_quality) or (None, None)."""
    if not candidates:
        return None, None
    tiers = []
    if match_mode == 'match_terms':
        want_prepaid = not booking_flex
        tiers.append(([c for c in candidates if c["prepaid"] == want_prepaid and not c["dinner"]], 'matched'))
    tiers.append(([c for c in candidates if not c["dinner"]], 'best available'))
    tiers.append((candidates, 'best available'))
    for pool, quality in tiers:
        if pool:
            return min(pool, key=lambda c: c["rate"]), quality
    return None, None


def gather_comparisons(db, start: date, end: date, cfg: dict) -> list:
    """Per-date like-for-like comparison rows for the horizon.
    Each row: {date, booking_rate, booking_room, booking_basis, newbook_rate,
               newbook_tariff, match_quality, dev_pct, dev_gbp, breach}"""
    booking_rows = db.execute(text("""
        SELECT DISTINCT ON (r.rate_date)
            r.rate_date, r.rate_gross, r.room_type,
            r.breakfast_included, r.free_cancellation, r.no_prepayment
        FROM booking_com_rates r
        JOIN booking_com_hotels h ON r.hotel_id = h.id
        WHERE h.tier = 'own'
          AND r.rate_date BETWEEN :fd AND :td
          AND r.rate_gross IS NOT NULL
        ORDER BY r.rate_date, r.scraped_at DESC, r.rate_gross ASC
    """), {"fd": start, "td": end}).mappings().all()

    newbook_rows = db.execute(text("""
        SELECT DISTINCT ON (rate_date, category_id)
            rate_date, category_id, rate_gross, tariffs_data
        FROM newbook_current_rates
        WHERE rate_date BETWEEN :fd AND :td
        ORDER BY rate_date, category_id, valid_from DESC
    """), {"fd": start, "td": end}).mappings().all()

    newbook_by_date: dict = {}
    for r in newbook_rows:
        newbook_by_date.setdefault(r["rate_date"], []).append(r)

    today = date.today()
    out = []
    for b in booking_rows:
        d = b["rate_date"]
        cat_rows = newbook_by_date.get(d)
        if not cat_rows:
            continue

        booking_rate = float(b["rate_gross"])
        booking_flex = bool(b["free_cancellation"] or b["no_prepayment"])
        booking_breakfast = bool(b["breakfast_included"])
        days_ahead = (d - today).days

        candidates = []
        for cat in cat_rows:
            candidates.extend(_candidate_tariffs(cat["tariffs_data"], days_ahead))

        chosen, quality = _pick_comparable(candidates, booking_flex, booking_breakfast, cfg["match_mode"])
        if chosen:
            newbook_rate = chosen["rate"]
            tariff_name = chosen["name"]
        else:
            # No tariff detail — legacy fallback to cheapest headline rate
            headline = [float(c["rate_gross"]) for c in cat_rows if c["rate_gross"] and float(c["rate_gross"]) > 0]
            if not headline:
                continue
            newbook_rate = min(headline)
            tariff_name = 'headline rate'
            quality = 'headline'

        result = evaluate_parity(booking_rate, newbook_rate, cfg)
        if result is None:
            continue
        dev_pct, dev_gbp, breach = result

        basis_bits = [
            'flex' if booking_flex else 'prepaid',
            'B&B' if booking_breakfast else 'room only',
        ]
        out.append({
            "date": d,
            "booking_rate": booking_rate,
            "booking_room": (b["room_type"] or '').split('\n')[0].strip() or None,
            "booking_basis": '/'.join(basis_bits),
            "newbook_rate": newbook_rate,
            "newbook_tariff": tariff_name,
            "match_quality": quality,
            "dev_pct": dev_pct,
            "dev_gbp": dev_gbp,
            "breach": breach,
        })
    return out


def run_parity_check() -> dict:
    db = SyncSessionLocal()
    try:
        cfg = get_parity_config(db)
        if not cfg["enabled"]:
            logger.info("Parity check skipped (disabled)")
            return {"status": "disabled"}

        today = date.today()
        start = today
        end = today + timedelta(days=HORIZON_DAYS)

        # Resolve any lingering active alerts for dates that have already passed
        db.execute(text("""
            UPDATE rate_parity_alerts
            SET alert_status = 'resolved'
            WHERE alert_status = 'active'
              AND rate_date < :today
        """), {"today": today})

        comparisons = gather_comparisons(db, start, end, cfg)

        # Latest alert per date in the horizon
        alert_rows = db.execute(text("""
            SELECT DISTINCT ON (rate_date)
                id, rate_date, alert_status, difference_pct
            FROM rate_parity_alerts
            WHERE rate_date BETWEEN :fd AND :td
            ORDER BY rate_date, created_at DESC
        """), {"fd": start, "td": end}).fetchall()
        latest_alert = {r[1]: {"id": r[0], "status": r[2], "diff": float(r[3] or 0)} for r in alert_rows}

        created = updated = resolved = 0
        for c in comparisons:
            alert = latest_alert.get(c["date"])
            # e.g. "DIRECT B&B PPAY vs BC prepaid/B&B"
            category_label = f"{c['newbook_tariff']} vs BC {c['booking_basis']}"[:255]

            if c["breach"]:
                if alert and alert["status"] == "active":
                    if round(c["dev_pct"], 2) != round(alert["diff"], 2):
                        db.execute(text("""
                            UPDATE rate_parity_alerts
                            SET newbook_rate = :nb, booking_com_rate = :bc,
                                difference_pct = :diff, alert_type = :atype,
                                room_category = :room
                            WHERE id = :id
                        """), {
                            "id": alert["id"], "nb": c["newbook_rate"], "bc": c["booking_rate"],
                            "diff": round(c["dev_pct"], 2),
                            "atype": "higher" if c["dev_pct"] > 0 else "lower",
                            "room": category_label,
                        })
                        updated += 1
                elif alert and alert["status"] == "acknowledged":
                    pass  # user has dealt with this date — don't nag
                else:
                    db.execute(text("""
                        INSERT INTO rate_parity_alerts
                            (rate_date, room_category, newbook_rate, booking_com_rate,
                             difference_pct, alert_type, alert_status)
                        VALUES (:date, :room, :nb, :bc, :diff, :atype, 'active')
                    """), {
                        "date": c["date"], "room": category_label,
                        "nb": c["newbook_rate"], "bc": c["booking_rate"],
                        "diff": round(c["dev_pct"], 2),
                        "atype": "higher" if c["dev_pct"] > 0 else "lower",
                    })
                    created += 1
            else:
                if alert and alert["status"] in ("active", "acknowledged"):
                    db.execute(
                        text("UPDATE rate_parity_alerts SET alert_status = 'resolved' WHERE id = :id"),
                        {"id": alert["id"]}
                    )
                    resolved += 1

        db.commit()
        summary = {
            "status": "ok",
            "match_mode": cfg["match_mode"],
            "dates_compared": len(comparisons),
            "created": created,
            "updated": updated,
            "resolved": resolved,
            "markup_value": cfg["markup_value"],
            "markup_unit": cfg["markup_unit"],
            "tolerance_value": cfg["tolerance_value"],
            "tolerance_unit": cfg["tolerance_unit"],
        }
        logger.info(f"Parity check: {summary}")
        return summary
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()