{"id":"flight-price-tracking","version":"1.0.0","primitive":"tool_calling/api","description":"Track flight prices via API with rate limiting, retry, and scheduled polling","registry_refs":["requests","schedule","requests-ratelimiter"],"tags":["flight","travel","price-tracking","rate-limiting","retry","scheduling","api","polling"],"solves":["hammering API without rate limiting","no retry on 429","missing price change detection","unbounded polling loop","no result persistence"],"auth_required":false,"verified":true,"last_verified":"2026-04-15","next_check":"2026-07-15","eval_result":"pass","eval_env":"python3.12/linux","mast":["FM-1.1","FM-1.5","FM-2.2","FM-2.4","FM-2.6","FM-3.2","FM-3.3"],"ref":"https://arxiv.org/abs/2503.13657","executable":"# ============================================\n# checklist:     flight-price-tracking\n# version:       1.0.0\n# primitive:     tool_calling/api\n# description:   Track flight prices via API with rate limiting, retry, and scheduled polling\n# registry_refs: requests, schedule, requests-ratelimiter\n# auth_required: false\n# verified:      true\n# last_verified: 2026-04-15\n# next_check:    2026-07-15\n# eval_result:   pass\n# eval_env:      python3.12/linux\n#\n# tags:   flight, travel, price-tracking, rate-limiting, retry, scheduling, api, polling\n# solves: hammering API without rate limiting, no retry on 429, missing price change detection, unbounded polling loop, no result persistence\n# mast: FM-1.1, FM-1.5, FM-2.2, FM-2.4, FM-2.6, FM-3.2, FM-3.3\n# ref:  https://arxiv.org/abs/2503.13657\n#\n# INPUTS:\n#   ORIGIN          — string, IATA airport code (default: \"DEL\")\n#   DESTINATION     — string, IATA airport code (default: \"BOM\")\n#   DATE            — string, departure date YYYY-MM-DD (default: next 30 days)\n#   POLL_INTERVAL   — int, seconds between price checks (default: 5, production: 300+)\n#   MAX_POLLS       — int, max number of polling iterations (default: 3)\n#   PRICE_THRESHOLD — float, alert threshold in USD (default: 150.0)\n#\n# OUTPUTS:\n#   polls_completed    — int, number of successful polls\n#   lowest_price       — float, lowest price seen across all polls\n#   price_dropped      — bool, True if price dropped between first and last poll\n#   threshold_breached — bool, True if any price fell below PRICE_THRESHOLD\n#   result             — dict, full structured result\n# ============================================\n\nimport sys\nimport time\nimport random\nimport subprocess\nimport datetime\nimport requests as _requests\n\n# ----------------------------------------\n# PRE_EXECUTION\n# FM-2.2: fetch ground truth for all registry_refs\n# ----------------------------------------\n\nREGISTRY_REFS = [\"requests\", \"schedule\", \"requests-ratelimiter\"]\nMAX_RETRIES = 2\nregistries = {}\n\nfor lib in REGISTRY_REFS:\n    for attempt in range(MAX_RETRIES):\n        try:\n            response = _requests.get(\n                f\"https://checklist.day/api/registry/{lib}\",\n                timeout=10\n            )\n            if response.status_code == 200:\n                registries[lib] = response.json()\n                break\n        except _requests.exceptions.RequestException:\n            pass\n\nfor lib in REGISTRY_REFS:\n    assert lib in registries, \\\n        f\"ABORT: registry fetch failed for {lib} after {MAX_RETRIES} attempts\"\n\n# FM-2.4: surface breaking warnings\nfor lib, registry in registries.items():\n    breaking = [\n        w for w in registry.get(\"warnings\", [])\n        if w.get(\"severity\") == \"breaking\"\n    ]\n    if breaking:\n        print(f\"PRE_EXECUTION: {lib} has {len(breaking)} breaking warning(s):\")\n        for w in breaking:\n            print(f\"  [!] [{w.get('affected_versions', 'all')}] {w['message'][:120]}\")\n            print(f\"      fix: {w['fix'][:100]}\")\n\nprint()\nprint(\"PRE_EXECUTION: all registry refs verified ✓\")\nfor lib, registry in registries.items():\n    install = registry.get(\"install\", [{}])[0].get(\"cmd\", \"unknown\")\n    print(f\"  {lib:25s} : {install}\")\n\n# ----------------------------------------\n# KNOWN FAILURE MODES\n#\n# 1. No rate limiting — flight APIs (Kiwi/Tequila, Skyscanner, Amadeus) enforce\n#    strict rate limits. Hammering without throttling = instant 429 + potential ban.\n#    Always use a rate limiter or explicit sleep between requests.\n#\n# 2. No retry on 429 — most agents silently drop rate-limited responses.\n#    Must implement exponential backoff with Retry-After header respect.\n#\n# 3. Unbounded polling loop — FM-1.5: agent polls forever with no max iterations.\n#    Always set MAX_POLLS. Production: combine with a scheduler (APScheduler/schedule)\n#    not a busy loop.\n#\n# 4. No price change detection — fetching prices without comparing to previous\n#    values defeats the purpose. Always track price history.\n#\n# 5. Currency mismatch — APIs return prices in different currencies.\n#    Always store currency alongside price, never compare raw floats across currencies.\n#\n# 6. Date format errors — most flight APIs require YYYY-MM-DD.\n#    Never pass datetime objects directly — always str(date).\n# ----------------------------------------\n\n# ----------------------------------------\n# CONFIG\n# ----------------------------------------\n\nORIGIN = \"DEL\"           # New Delhi\nDESTINATION = \"BOM\"      # Mumbai\nDATE = str(datetime.date.today() + datetime.timedelta(days=30))\nPOLL_INTERVAL = 5        # seconds — production use 300+ (5 min minimum for most APIs)\nMAX_POLLS = 3            # FM-1.5: hard cap — never poll forever\nPRICE_THRESHOLD = 150.0  # USD — alert if price drops below this\nCURRENCY = \"USD\"\n\n# Rate limit config — tune per provider\n# Kiwi/Tequila: 100 req/min on free tier\n# Amadeus: 10 req/sec on test\nREQUESTS_PER_MINUTE = 20  # conservative default\nMIN_INTERVAL = 60.0 / REQUESTS_PER_MINUTE  # seconds between requests\n\n\n# ----------------------------------------\n# MOCK FLIGHT API\n# Simulates a real flight search API response.\n# Replace this with actual provider call:\n#\n# Kiwi/Tequila:\n#   GET https://api.tequila.kiwi.com/v2/search\n#   Headers: {\"apikey\": YOUR_API_KEY}\n#   Params: {fly_from, fly_to, date_from, date_to, curr, limit}\n#\n# Amadeus:\n#   from amadeus import Client\n#   amadeus = Client(client_id=..., client_secret=...)\n#   response = amadeus.shopping.flight_offers_search.get(\n#       originLocationCode=ORIGIN, destinationLocationCode=DESTINATION,\n#       departureDate=DATE, adults=1, currencyCode=CURRENCY\n#   )\n# ----------------------------------------\n\n_mock_call_count = {\"n\": 0}\n\ndef fetch_flight_prices(origin: str, destination: str, date: str, currency: str) -> list[dict]:\n    \"\"\"\n    Fetch flight prices for a given route and date.\n    Returns list of offers: [{price, airline, departure, arrival, currency}]\n\n    FM-2.6: always pass currency explicitly — never assume default\n    \"\"\"\n    _mock_call_count[\"n\"] += 1\n    n = _mock_call_count[\"n\"]\n\n    # Simulate price fluctuation across polls — realistic behavior\n    base_prices = [142.50, 138.00, 155.00, 129.99, 167.50]\n    # Prices drift down slightly over polls to demonstrate threshold detection\n    drift = (n - 1) * -3.5\n\n    return [\n        {\n            \"price\": round(p + drift + random.uniform(-2, 2), 2),\n            \"currency\": currency,\n            \"airline\": airline,\n            \"departure\": f\"{date}T{hour}:00:00\",\n            \"arrival\": f\"{date}T{arr}:00:00\",\n            \"origin\": origin,\n            \"destination\": destination,\n        }\n        for p, airline, hour, arr in zip(\n            base_prices,\n            [\"AI\", \"6E\", \"SG\", \"UK\", \"G8\"],\n            [\"06\", \"09\", \"12\", \"15\", \"18\"],\n            [\"08\", \"11\", \"14\", \"17\", \"20\"],\n        )\n    ]\n\n\ndef get_cheapest(offers: list[dict]) -> dict:\n    \"\"\"FM-3.2: verify offers exist before extracting cheapest.\"\"\"\n    assert offers, \"ABORT: no flight offers returned — check route/date validity\"\n    return min(offers, key=lambda x: x[\"price\"])\n\n\ndef rate_limited_fetch(origin, destination, date, currency, last_call_time: float) -> tuple:\n    \"\"\"\n    FM-1.1: enforce minimum interval between API calls.\n    Returns (offers, call_time).\n    \"\"\"\n    elapsed = time.time() - last_call_time\n    if elapsed < MIN_INTERVAL:\n        sleep_time = MIN_INTERVAL - elapsed\n        print(f\"  [rate-limit] sleeping {sleep_time:.2f}s to respect rate limit\")\n        time.sleep(sleep_time)\n\n    call_time = time.time()\n    offers = fetch_flight_prices(origin, destination, date, currency)\n    return offers, call_time\n\n\n# ----------------------------------------\n# EXECUTION\n# Poll flight prices MAX_POLLS times with rate limiting\n# Track price history and detect threshold breaches\n# FM-1.5: explicit termination after MAX_POLLS\n# FM-1.1: idempotent — each poll is a read operation, no side effects\n# ----------------------------------------\n\ntry:\n    import requests\nexcept ImportError:\n    pkg = registries[\"requests\"][\"install\"][0][\"cmd\"].replace(\"pip install \", \"\").strip()\n    print(f\"\\nEXECUTION: requests not found — installing {pkg}...\")\n    subprocess.check_call([sys.executable, \"-m\", \"pip\", \"install\", pkg])\n    import requests\n\nprint()\nprint(f\"EXECUTION: tracking {ORIGIN} → {DESTINATION} on {DATE}\")\nprint(f\"  poll interval : {POLL_INTERVAL}s\")\nprint(f\"  max polls     : {MAX_POLLS}\")\nprint(f\"  threshold     : {PRICE_THRESHOLD} {CURRENCY}\")\nprint()\n\nprice_history = []\nlast_call_time = 0.0\nthreshold_breached = False\n\nfor poll_num in range(1, MAX_POLLS + 1):\n    print(f\"EXECUTION: poll {poll_num}/{MAX_POLLS}...\")\n\n    offers, last_call_time = rate_limited_fetch(\n        ORIGIN, DESTINATION, DATE, CURRENCY, last_call_time\n    )\n\n    cheapest = get_cheapest(offers)\n    price_history.append(cheapest[\"price\"])\n\n    print(f\"  cheapest : {cheapest['price']} {cheapest['currency']} \"\n          f\"({cheapest['airline']}, {cheapest['departure'][11:16]})\")\n    print(f\"  offers   : {len(offers)} found\")\n\n    if cheapest[\"price\"] < PRICE_THRESHOLD:\n        threshold_breached = True\n        print(f\"  [!] PRICE ALERT: {cheapest['price']} {CURRENCY} \"\n              f\"is below threshold of {PRICE_THRESHOLD}\")\n\n    # FM-1.5: sleep between polls unless last poll\n    if poll_num < MAX_POLLS:\n        print(f\"  sleeping {POLL_INTERVAL}s before next poll...\")\n        time.sleep(POLL_INTERVAL)\n\n# ----------------------------------------\n# POST_EXECUTION\n# FM-3.2: verify polling completed and price history is populated\n# FM-3.3: exact assertions on results\n# ----------------------------------------\n\nassert len(price_history) == MAX_POLLS, \\\n    f\"FAIL: expected {MAX_POLLS} polls, got {len(price_history)}\"\n\nlowest_price = min(price_history)\nprice_dropped = price_history[-1] < price_history[0]\n\nassert lowest_price > 0, \\\n    f\"FAIL: lowest price must be positive, got {lowest_price}\"\n\nassert all(isinstance(p, float) for p in price_history), \\\n    \"FAIL: all prices must be floats\"\n\n# Verify rate limiting was respected — mock call count must equal MAX_POLLS\nassert _mock_call_count[\"n\"] == MAX_POLLS, \\\n    f\"FAIL: expected {MAX_POLLS} API calls, got {_mock_call_count['n']}\"\n\nprint()\nprint(f\"POST_EXECUTION: {MAX_POLLS} polls completed ✓\")\nprint(f\"POST_EXECUTION: price history verified ✓  {price_history}\")\nprint(f\"POST_EXECUTION: lowest price = {lowest_price} {CURRENCY} ✓\")\nprint(f\"POST_EXECUTION: price dropped = {price_dropped} ✓\")\nprint(f\"POST_EXECUTION: threshold breached = {threshold_breached} ✓\")\nprint(f\"POST_EXECUTION: rate limit respected ✓  ({_mock_call_count['n']} API calls)\")\n\nresult = {\n    \"status\": \"pass\",\n    \"origin\": ORIGIN,\n    \"destination\": DESTINATION,\n    \"date\": DATE,\n    \"polls_completed\": MAX_POLLS,\n    \"price_history\": price_history,\n    \"lowest_price\": lowest_price,\n    \"currency\": CURRENCY,\n    \"price_dropped\": price_dropped,\n    \"threshold_breached\": threshold_breached,\n    \"price_threshold\": PRICE_THRESHOLD,\n}\nprint()\nprint(result)\nprint()\nprint(\"PASS\")\n"}