Rate parity: expected-markup config, alert writer job, alerts UI + direct-link UI

Parity was read-only — the alerts table had no producer, so the Market
View badge could never fire. Now:
- jobs/check_rate_parity.py: daily 06:45 job comparing own Booking.com
  lead-in rate vs cheapest Newbook rate per date, measured against an
  EXPECTED markup (we deliberately price Booking.com higher to cover
  commission): alert when deviation from newbook*(1+markup%) exceeds the
  tolerance. Creates/updates active alerts, auto-resolves dates back in
  line, leaves acknowledged dates alone.
- config keys: parity_check_enabled, parity_expected_markup_pct,
  parity_tolerance_pct (system_config)
- POST /competitors/parity/check manual trigger; GET /parity now uses the
  same markup/tolerance and cheapest-across-categories Newbook rate
- Settings -> Rate Parity tab: markup %, tolerance %, enable toggle,
  run-now with result summary
- Market View -> Parity Alerts tab: status-filtered list w/ acknowledge
- Market View -> Hotels: direct-link dropdown per competitor (new PUT
  /competitors/hotels/{id}/direct-link) — closes the never-written
  direct_hotel_id gap so the matrix direct-rates sub-row can populate

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

backend/api/competitors.py

@@ -551,6 +551,39 @@ async def update_hotel_notes(
     return {"status": "success"}
 
 
+class DirectLinkUpdate(BaseModel):
+    direct_hotel_id: Optional[int] = None
+
+
+@router.put("/hotels/{hotel_id}/direct-link")
+async def update_hotel_direct_link(
+    hotel_id: int,
+    payload: DirectLinkUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user)
+):
+    """Link (or unlink with null) a Booking.com hotel to a direct booking
+    engine competitor, enabling the direct-rates sub-row in the rate matrix."""
+    if payload.direct_hotel_id is not None:
+        exists = await db.execute(
+            text("SELECT 1 FROM direct_competitor_hotels WHERE id = :id"),
+            {'id': payload.direct_hotel_id}
+        )
+        if not exists.fetchone():
+            raise HTTPException(status_code=404, detail="Direct hotel not found")
+
+    result = await db.execute(
+        text("UPDATE booking_com_hotels SET direct_hotel_id = :did WHERE id = :hotel_id RETURNING id"),
+        {'hotel_id': hotel_id, 'did': payload.direct_hotel_id}
+    )
+
+    if not result.fetchone():
+        raise HTTPException(status_code=404, detail="Hotel not found")
+
+    await db.commit()
+    return {"status": "success"}
+
+
 # ============================================
 # COMPETITOR RATES MATRIX
 # ============================================
@@ -711,21 +744,32 @@ async def get_rate_parity(
     )
     booking_rates = {row.rate_date: dict(row._mapping) for row in booking_rates_result.fetchall()}
 
-    # Get Newbook rates (best rate per date across categories)
+    # Get Newbook rates (cheapest per date across latest category snapshots)
     newbook_rates_result = await db.execute(
         text("""
-            SELECT DISTINCT ON (rate_date)
-                rate_date,
-                rate_gross as newbook_rate,
-                category_id
-            FROM newbook_current_rates
-            WHERE rate_date >= :from_date AND rate_date <= :to_date
-            ORDER BY rate_date, valid_from DESC
+            SELECT rate_date, MIN(rate_gross) AS newbook_rate FROM (
+                SELECT DISTINCT ON (rate_date, category_id)
+                    rate_date, rate_gross
+                FROM newbook_current_rates
+                WHERE rate_date >= :from_date AND rate_date <= :to_date
+                  AND rate_gross IS NOT NULL AND rate_gross > 0
+                ORDER BY rate_date, category_id, valid_from DESC
+            ) t GROUP BY rate_date
         """),
         {'from_date': start, 'to_date': end}
     )
     newbook_rates = {row.rate_date: dict(row._mapping) for row in newbook_rates_result.fetchall()}
 
+    # Expected-markup config: we deliberately price Booking.com higher to
+    # cover commission, so parity is measured against newbook × (1 + markup%)
+    from jobs.check_rate_parity import get_parity_config, deviation_from_expected
+    from database import SyncSessionLocal
+    cfg_db = SyncSessionLocal()
+    try:
+        cfg = get_parity_config(cfg_db)
+    finally:
+        cfg_db.close()
+
     # Compare rates
     parity_issues = []
     all_dates = set(booking_rates.keys()) | set(newbook_rates.keys())
@@ -743,13 +787,17 @@ async def get_rate_parity(
         if not booking_rate or not newbook_rate:
             continue
 
-        diff_pct = ((float(booking_rate) - float(newbook_rate)) / float(newbook_rate)) * 100
+        expected_rate = float(newbook_rate) * (1 + cfg["markup_pct"] / 100)
+        diff_pct = deviation_from_expected(float(booking_rate), float(newbook_rate), cfg["markup_pct"])
+        if diff_pct is None:
+            continue
 
-        if abs(diff_pct) > 1:  # More than 1% difference
+        if abs(diff_pct) > cfg["tolerance_pct"]:
             parity_issues.append({
                 'rate_date': rate_date.isoformat(),
                 'booking_rate': float(booking_rate),
                 'newbook_rate': float(newbook_rate),
+                'expected_rate': round(expected_rate, 2),
                 'difference_pct': round(diff_pct, 2),
                 'alert_type': 'higher' if diff_pct > 0 else 'lower',
                 'booking_room_type': booking.get('booking_room_type'),
@@ -759,11 +807,24 @@ async def get_rate_parity(
     return {
         'from_date': start.isoformat(),
         'to_date': end.isoformat(),
+        'expected_markup_pct': cfg["markup_pct"],
+        'tolerance_pct': cfg["tolerance_pct"],
         'issues_count': len(parity_issues),
         'issues': parity_issues
     }
 
 
+@router.post("/parity/check")
+async def trigger_parity_check(
+    current_user: dict = Depends(get_current_user)
+):
+    """Run the parity check now (same logic as the daily 06:45 job)."""
+    import asyncio
+    from jobs.check_rate_parity import run_parity_check
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(None, run_parity_check)
+
+
 # ============================================
 # PARITY ALERTS
 # ============================================