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