← back to Watches
scripts/rolex-wayback-crawler.py
312 lines
#!/usr/bin/env python3
"""
Rolex Historical Price Crawler via Wayback Machine
Crawls 20 years (2005-2025) of Rolex prices from archived Chrono24 pages.
"""
import json
import sys
import os
import time
import re
from datetime import datetime
from pathlib import Path
try:
import requests
from bs4 import BeautifulSoup
except ImportError:
print("Installing dependencies...", file=sys.stderr)
import subprocess
subprocess.check_call([sys.executable, "-m", "pip", "install", "requests", "beautifulsoup4", "-q"])
import requests
from bs4 import BeautifulSoup
# Configuration
SCRIPT_DIR = Path(__file__).parent.absolute()
PROJECT_DIR = SCRIPT_DIR.parent
DATA_FILE = PROJECT_DIR / "data" / "rolex-historical-prices.json"
LOG_DIR = PROJECT_DIR / "logs"
# Rolex models to crawl
ROLEX_MODELS = [
{"id": "submariner", "name": "Submariner", "chrono24_slug": "rolex/submariner"},
{"id": "submariner-date", "name": "Submariner Date", "chrono24_slug": "rolex/submariner-date"},
{"id": "daytona", "name": "Cosmograph Daytona", "chrono24_slug": "rolex/daytona"},
{"id": "gmt-master-ii", "name": "GMT-Master II", "chrono24_slug": "rolex/gmt-master-ii"},
{"id": "gmt-master", "name": "GMT-Master", "chrono24_slug": "rolex/gmt-master"},
{"id": "datejust", "name": "Datejust", "chrono24_slug": "rolex/datejust"},
{"id": "datejust-36", "name": "Datejust 36", "chrono24_slug": "rolex/datejust-36"},
{"id": "datejust-41", "name": "Datejust 41", "chrono24_slug": "rolex/datejust-41"},
{"id": "day-date", "name": "Day-Date", "chrono24_slug": "rolex/day-date"},
{"id": "day-date-40", "name": "Day-Date 40", "chrono24_slug": "rolex/day-date-40"},
{"id": "explorer", "name": "Explorer", "chrono24_slug": "rolex/explorer"},
{"id": "explorer-ii", "name": "Explorer II", "chrono24_slug": "rolex/explorer-ii"},
{"id": "sea-dweller", "name": "Sea-Dweller", "chrono24_slug": "rolex/sea-dweller"},
{"id": "deepsea", "name": "Deepsea", "chrono24_slug": "rolex/sea-dweller-deepsea"},
{"id": "yacht-master", "name": "Yacht-Master", "chrono24_slug": "rolex/yacht-master"},
{"id": "yacht-master-ii", "name": "Yacht-Master II", "chrono24_slug": "rolex/yacht-master-ii"},
{"id": "sky-dweller", "name": "Sky-Dweller", "chrono24_slug": "rolex/sky-dweller"},
{"id": "air-king", "name": "Air-King", "chrono24_slug": "rolex/air-king"},
{"id": "oyster-perpetual", "name": "Oyster Perpetual", "chrono24_slug": "rolex/oyster-perpetual"},
{"id": "milgauss", "name": "Milgauss", "chrono24_slug": "rolex/milgauss"},
{"id": "cellini", "name": "Cellini", "chrono24_slug": "rolex/cellini"},
]
HEADERS = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
}
REQUEST_DELAY = 2
RETRY_DELAY = 5
MAX_RETRIES = 3
def load_existing_data():
"""Load existing historical data."""
if DATA_FILE.exists():
with open(DATA_FILE) as f:
return json.load(f)
return {
"metadata": {
"source": "Wayback Machine (Chrono24 archives)",
"created": datetime.now().isoformat(),
"lastUpdated": None,
"dateRange": "2005-2025"
},
"models": {},
"crawlStatus": {}
}
def save_data(data):
"""Save historical data."""
data["metadata"]["lastUpdated"] = datetime.now().isoformat()
DATA_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(DATA_FILE, 'w') as f:
json.dump(data, f, indent=2)
def fetch_cdx_snapshots(url, from_year, to_year, limit=100):
"""Fetch CDX index data for a URL from Wayback Machine."""
all_snapshots = []
# Query year by year to avoid timeouts
for year in range(from_year, to_year + 1):
cdx_url = "https://web.archive.org/cdx/search/cdx"
params = {
"url": url,
"matchType": "prefix",
"from": str(year),
"to": str(year),
"output": "json",
"filter": "statuscode:200",
"collapse": "timestamp:6", # One per month
"limit": limit
}
for attempt in range(MAX_RETRIES):
try:
resp = requests.get(cdx_url, params=params, headers=HEADERS, timeout=90)
resp.raise_for_status()
results = resp.json()
if len(results) > 1: # First row is headers
snapshots = results[1:] # Skip header row
all_snapshots.extend(snapshots)
print(f" {year}: {len(snapshots)} snapshots", file=sys.stderr)
else:
print(f" {year}: 0 snapshots", file=sys.stderr)
time.sleep(REQUEST_DELAY)
break
except Exception as e:
if attempt < MAX_RETRIES - 1:
print(f" {year}: Retry {attempt+1} after error: {e}", file=sys.stderr)
time.sleep(RETRY_DELAY)
else:
print(f" {year}: Failed after {MAX_RETRIES} attempts: {e}", file=sys.stderr)
return all_snapshots
def extract_prices_from_page(html_content):
"""Extract prices from archived Chrono24 page."""
prices = []
soup = BeautifulSoup(html_content, 'html.parser')
# Price patterns
price_patterns = [
r'\$\s*([\d,]+)',
r'USD\s*([\d,]+)',
r'([\d,]+)\s*USD',
]
# Find price elements
price_elements = soup.find_all(['span', 'div', 'strong'], class_=lambda x: x and 'price' in str(x).lower())
for elem in price_elements:
text = elem.get_text()
for pattern in price_patterns:
match = re.search(pattern, text)
if match:
try:
price = int(match.group(1).replace(',', ''))
if 500 < price < 500000: # Reasonable Rolex price range
prices.append(price)
except ValueError:
continue
# Also search full page text for prices
page_text = soup.get_text()
for pattern in price_patterns:
matches = re.findall(pattern, page_text)
for match in matches:
try:
price = int(match.replace(',', ''))
if 500 < price < 500000:
prices.append(price)
except ValueError:
continue
return list(set(prices)) # Dedupe
def fetch_archived_page(timestamp, original_url):
"""Fetch a page from Wayback Machine archive."""
archive_url = f"https://web.archive.org/web/{timestamp}/{original_url}"
try:
resp = requests.get(archive_url, headers=HEADERS, timeout=60)
resp.raise_for_status()
return resp.text
except Exception as e:
print(f" Error fetching {archive_url}: {e}", file=sys.stderr)
return None
def crawl_model(model, from_year=2005, to_year=2025):
"""Crawl historical prices for a Rolex model."""
print(f"\nCrawling: {model['name']}", file=sys.stderr)
print(f"Date range: {from_year} - {to_year}", file=sys.stderr)
print("="*50, file=sys.stderr)
prices_by_date = {}
chrono24_url = f"chrono24.com/{model['chrono24_slug']}"
# Get CDX snapshots
snapshots = fetch_cdx_snapshots(chrono24_url, from_year, to_year)
print(f"Found {len(snapshots)} total snapshots", file=sys.stderr)
# Process snapshots
for i, snapshot in enumerate(snapshots[:50]): # Limit to 50 per model
try:
timestamp = snapshot[1]
original_url = snapshot[2]
# Parse date from timestamp (YYYYMMDDHHMMSS)
date = f"{timestamp[:4]}-{timestamp[4:6]}-{timestamp[6:8]}"
print(f" [{i+1}/{min(len(snapshots), 50)}] {date}...", file=sys.stderr, end="")
# Fetch archived page
html = fetch_archived_page(timestamp, f"http://{original_url}")
if html:
prices = extract_prices_from_page(html)
if prices:
prices_by_date[date] = {
"prices": prices,
"min": min(prices),
"max": max(prices),
"avg": sum(prices) // len(prices),
"count": len(prices),
"archiveUrl": f"https://web.archive.org/web/{timestamp}/{original_url}"
}
print(f" Found {len(prices)} prices (${min(prices):,} - ${max(prices):,})", file=sys.stderr)
else:
print(" No prices found", file=sys.stderr)
else:
print(" Failed to fetch", file=sys.stderr)
time.sleep(REQUEST_DELAY)
except Exception as e:
print(f" Error: {e}", file=sys.stderr)
continue
return prices_by_date
def run_full_crawl(from_year=2005, to_year=2025):
"""Run full historical crawl for all Rolex models."""
print(f"\n{'#'*60}", file=sys.stderr)
print(f"# ROLEX HISTORICAL PRICE CRAWL", file=sys.stderr)
print(f"# Date range: {from_year} - {to_year}", file=sys.stderr)
print(f"# Models: {len(ROLEX_MODELS)}", file=sys.stderr)
print(f"{'#'*60}\n", file=sys.stderr)
data = load_existing_data()
for model in ROLEX_MODELS:
# Skip if already crawled recently
if model['id'] in data.get('crawlStatus', {}):
last_crawl = data['crawlStatus'][model['id']].get('lastCrawled')
if last_crawl:
print(f"Skipping {model['name']} (already crawled)", file=sys.stderr)
continue
prices = crawl_model(model, from_year, to_year)
# Store results
data['models'][model['id']] = {
"name": model['name'],
"priceHistory": prices,
"totalDataPoints": len(prices)
}
data['crawlStatus'][model['id']] = {
"lastCrawled": datetime.now().isoformat(),
"dataPoints": len(prices)
}
# Save after each model
save_data(data)
print(f"\nSaved {len(prices)} data points for {model['name']}", file=sys.stderr)
# Summary
total_points = sum(m.get('totalDataPoints', 0) for m in data['models'].values())
print(f"\n{'='*60}", file=sys.stderr)
print(f"CRAWL COMPLETE", file=sys.stderr)
print(f"Models crawled: {len(data['models'])}", file=sys.stderr)
print(f"Total data points: {total_points}", file=sys.stderr)
print(f"{'='*60}\n", file=sys.stderr)
return data
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description='Rolex Historical Price Crawler')
parser.add_argument('--from-year', type=int, default=2005, help='Start year')
parser.add_argument('--to-year', type=int, default=2025, help='End year')
parser.add_argument('--model', type=str, help='Specific model to crawl')
args = parser.parse_args()
if args.model:
model = next((m for m in ROLEX_MODELS if m['id'] == args.model), None)
if model:
data = load_existing_data()
prices = crawl_model(model, args.from_year, args.to_year)
data['models'][model['id']] = {
"name": model['name'],
"priceHistory": prices,
"totalDataPoints": len(prices)
}
save_data(data)
else:
print(f"Model not found: {args.model}", file=sys.stderr)
else:
run_full_crawl(args.from_year, args.to_year)