import asyncio from datetime import datetime, timedelta from zoneinfo import ZoneInfo import httpx from app.configs.settings import settings EXCLUDED_COMPETITORS = { "https://www.plazabeachresorts.com/", "https://www.baypalmsresort.com/", "https://www.thebayviewplaza.com/", } ALLOWED_COMPETITORS_PROPERTIES_TOKEN = { "ChgI3vKYl56ruN6AARoLL2cvMXRobWc1aGwQAQ", # The Beachcomber "ChkIxIOjhofkq5uOARoML2cvMTFyOTI3ZGc2EAE", # RumFish Beach at TradeWinds "ChgIlOjmr4votcb-ARoLL2cvMXR5a3c1YmoQAQ", # Sirata Beach Resort St Pete Beach "ChcIlpeClLW624ciGgsvZy8xdzExMnpxchAB", # Bilmar Beach Resort "ChcI9s-yofmNuolrGgsvZy8xdGs4Y3htbRAB", #St. Pete Beach Suites } async def get_competitors( client: httpx.AsyncClient, check_in_date: str, check_out_date: str, ): response = await client.get( "https://serpapi.com/search.json", params={ "engine": "google_hotels", "q": "st pete beach", "check_in_date": check_in_date, "check_out_date": check_out_date, "adults": 2, "children": 0, "gl": "us", "hl": "en", "google_domain": "google.com", "currency": "USD", "no_cache": "true", "api_key": settings.SERP_API_KEY, }, ) response.raise_for_status() properties = response.json().get("properties", []) return [ { "name": competitor.get("name"), "price": competitor.get("rate_per_night", {}).get("extracted_before_taxes_fees") or competitor.get("total_rate", {}).get("extracted_before_taxes_fees"), } for competitor in properties if competitor.get("link") not in EXCLUDED_COMPETITORS and competitor.get("property_token") in ALLOWED_COMPETITORS_PROPERTIES_TOKEN ] async def get_competitors_prices(days: int = 2): """ Fetch competitor prices for the next `days` check-in dates. Example: days=2 -> Today, Tomorrow days=30 -> Today through next 29 days """ tz = ZoneInfo("America/New_York") today = datetime.now(tz).date() date_pairs = [ ( (today + timedelta(days=i)).strftime("%Y-%m-%d"), (today + timedelta(days=i + 1)).strftime("%Y-%m-%d"), ) for i in range(days) ] async with httpx.AsyncClient(timeout=60.0) as client: tasks = [ get_competitors(client, check_in, check_out) for check_in, check_out in date_pairs ] results = await asyncio.gather(*tasks, return_exceptions=True) final_data = {} for (check_in, _), result in zip(date_pairs, results): if isinstance(result, Exception): print(f"Failed to fetch competitors for {check_in}: {result}") final_data[check_in] = [] else: final_data[check_in] = result return final_data from copy import deepcopy async def adjust_predictions_with_competitors( results: dict, competitors: dict, agree_weight: float = 0.15, conflict_weight: float = 0.30, floor_pct: float = 0.85, ): """ Adjust predicted prices using competitor market trends. Instead of comparing raw competitor prices against our hotel's actual price (which is flawed — different hotels have different price tiers), this uses the competitor price *trend* relative to their own baseline to determine whether the market is moving up or down. Formula: 1. Compute competitor avg per date and a baseline (mean of all dates). 2. Market direction = competitor_avg[date] vs baseline_avg. 3. Model direction = predicted vs actual (our current price). 4. When both signals agree → light blend (agree_weight). When they disagree → heavier blend (conflict_weight) pulls the model toward market-adjusted price. 5. Market-adjusted price = model_price * competitor_trend_ratio (competitor_avg[date] / baseline_avg). Args: results: {room_id: [{date, actual, predicted, ...}, ...]} competitors: {date_str: [{name, price}, ...]} agree_weight: Blend weight when model and market agree (default 0.15). conflict_weight: Blend weight when they disagree (default 0.30). floor_pct: Minimum ratio — adjusted price never drops below floor_pct * model_price (default 0.85 = 85%). Returns: Same structure as results with updated predicted prices. """ output = deepcopy(results) # ------------------------------- # Calculate average competitor price per date # ------------------------------- competitor_avg: dict[str, float] = {} for date, hotels in competitors.items(): prices = [] for h in hotels: price = h.get("price") if price is None: continue try: price = float(price) except Exception: continue if price <= 0: continue prices.append(price) if prices: competitor_avg[date] = sum(prices) / len(prices) if not competitor_avg: return output # Baseline = mean of all competitor date averages baseline_avg = sum(competitor_avg.values()) / len(competitor_avg) # --------------------------------- # Update predictions # --------------------------------- for room_id, rows in output.items(): if not isinstance(rows, list): continue for row in rows: date = row.get("date") if date not in competitor_avg: continue if row.get("predicted") is None: continue model = float(row["predicted"]) market = competitor_avg[date] # Competitor trend ratio: is the market above or below its own average? trend_ratio = market / baseline_avg # e.g. 1.05 = market 5% above baseline # Market-adjusted price: apply the same trend to our model prediction market_adjusted = model * trend_ratio # Determine directions market_up = trend_ratio >= 1.0 model_up = True # default if no actual if row.get("actual") is not None: current = float(row["actual"]) model_up = model >= current # Choose blend weight if model_up == market_up: weight = agree_weight # signals agree — nudge lightly else: weight = conflict_weight # signals disagree — lean toward market # Blend: weighted average of model price and market-adjusted price adjusted = (1 - weight) * model + weight * market_adjusted # Floor guardrail: never let competitors drag the price too low adjusted = max(adjusted, model * floor_pct) row["predicted"] = round(adjusted, 2) # Debug / transparency info row["competitor_avg"] = round(market, 2) row["competitor_baseline"] = round(baseline_avg, 2) row["competitor_trend"] = round(trend_ratio, 4) row["competitor_weight"] = weight return output