← back to Watches
sdk/omega_api_client.py
489 lines
"""
Omega Watch Price History API - Python Client
Version: 2.0.0
Official Python client library for the Omega Watch Price History API
"""
import requests
import json
from typing import Optional, Dict, List, Any
from urllib.parse import urljoin
class OmegaWatchAPIError(Exception):
"""Custom exception for API errors"""
pass
class OmegaWatchAPI:
"""
Official Python client for the Omega Watch Price History API
Example:
>>> api = OmegaWatchAPI()
>>> watches = api.get_watches(series='Speedmaster', limit=10)
>>> print(f"Found {watches['total']} Speedmaster watches")
"""
def __init__(
self,
base_url: str = "http://45.61.58.125:7600",
timeout: int = 10,
api_key: Optional[str] = None
):
"""
Initialize the API client
Args:
base_url: Base URL for the API (default: http://45.61.58.125:7600)
timeout: Request timeout in seconds (default: 10)
api_key: API key for authenticated requests (optional)
"""
self.base_url = base_url
self.timeout = timeout
self.api_key = api_key
self.session = requests.Session()
if self.api_key:
self.session.headers.update({'X-API-Key': self.api_key})
def _request(
self,
method: str,
path: str,
data: Optional[Dict] = None,
params: Optional[Dict] = None
) -> Any:
"""
Make an HTTP request
Args:
method: HTTP method (GET, POST, etc.)
path: API endpoint path
data: Request body data
params: Query parameters
Returns:
Response data (dict or string)
Raises:
OmegaWatchAPIError: If the request fails
"""
url = urljoin(self.base_url, path)
# Remove None values from params
if params:
params = {k: v for k, v in params.items() if v is not None}
try:
response = self.session.request(
method=method,
url=url,
json=data,
params=params,
timeout=self.timeout
)
response.raise_for_status()
# Return JSON if content type is JSON
content_type = response.headers.get('content-type', '')
if 'application/json' in content_type:
return response.json()
return response.text
except requests.exceptions.Timeout:
raise OmegaWatchAPIError(f"Request timeout after {self.timeout} seconds")
except requests.exceptions.HTTPError as e:
try:
error_data = e.response.json()
error_msg = error_data.get('error', str(e))
except:
error_msg = str(e)
raise OmegaWatchAPIError(f"HTTP {e.response.status_code}: {error_msg}")
except requests.exceptions.RequestException as e:
raise OmegaWatchAPIError(f"Request failed: {str(e)}")
# Health & System
def get_health(self) -> Dict:
"""
Get system health status
Returns:
Health information including uptime, memory, cache stats
"""
return self._request('GET', '/api/health')
# Watches
def get_watches(
self,
series: Optional[str] = None,
min_year: Optional[int] = None,
max_year: Optional[int] = None,
min_price: Optional[int] = None,
max_price: Optional[int] = None,
page: int = 1,
limit: int = 50
) -> Dict:
"""
Get all watches with optional filtering and pagination
Args:
series: Filter by series name
min_year: Minimum year introduced
max_year: Maximum year introduced
min_price: Minimum current price (USD)
max_price: Maximum current price (USD)
page: Page number (default: 1)
limit: Items per page (default: 50)
Returns:
Paginated watch list with metadata
"""
params = {
'series': series,
'minYear': min_year,
'maxYear': max_year,
'minPrice': min_price,
'maxPrice': max_price,
'page': page,
'limit': limit
}
return self._request('GET', '/api/watches', params=params)
def get_watch_by_id(self, watch_id: str) -> Dict:
"""
Get watch by ID with complete price history
Args:
watch_id: Unique watch identifier
Returns:
Watch details and price history
"""
return self._request('GET', f'/api/watches/{watch_id}/history')
def get_trending(self, limit: int = 10) -> Dict:
"""
Get trending watches (most viewed)
Args:
limit: Number of results (default: 10)
Returns:
Trending watches with view counts
"""
return self._request('GET', '/api/watches/trending', params={'limit': limit})
# Search
def search(
self,
query: str,
series: Optional[str] = None,
movement: Optional[str] = None,
min_price: Optional[int] = None,
max_price: Optional[int] = None,
page: int = 1,
limit: int = 20
) -> Dict:
"""
Search watches with advanced filters
Args:
query: Search query string
series: Filter by series
movement: Filter by movement type
min_price: Minimum price
max_price: Maximum price
page: Page number
limit: Items per page
Returns:
Search results with pagination
"""
params = {
'q': query,
'series': series,
'movement': movement,
'minPrice': min_price,
'maxPrice': max_price,
'page': page,
'limit': limit
}
return self._request('GET', '/api/search', params=params)
# Analytics
def get_price_predictions(self, watch_id: str) -> Dict:
"""
Get price predictions for a watch
Args:
watch_id: Watch ID
Returns:
Price predictions for next 3 years with confidence scores
"""
return self._request('GET', f'/api/price-predictions/{watch_id}')
def get_statistics(self) -> Dict:
"""
Get comprehensive market statistics
Returns:
Market statistics including total watches, average appreciation, top performers
"""
return self._request('GET', '/api/statistics')
def get_analytics_predictions(self) -> Dict:
"""
Get ML-generated predictions from analytics engine
Returns:
Advanced prediction analytics
"""
return self._request('GET', '/api/analytics/predictions')
def get_market_analysis(self) -> Dict:
"""
Get comprehensive market analysis
Returns:
Market analysis including investment opportunities
"""
return self._request('GET', '/api/analytics/market')
def get_statistical_insights(self) -> Dict:
"""
Get advanced statistical insights
Returns:
Statistical insights and trends
"""
return self._request('GET', '/api/analytics/statistics')
def get_investment_opportunities(self) -> Dict:
"""
Get top investment opportunities
Returns:
Investment opportunities and value watches
"""
return self._request('GET', '/api/analytics/investment-opportunities')
def get_risk_metrics(self) -> Dict:
"""
Get risk-adjusted performance metrics
Returns:
Risk metrics for all watches
"""
return self._request('GET', '/api/analytics/risk-metrics')
# Collections
def get_collections(self) -> Dict:
"""
Get all collections with statistics
Returns:
Collections data with counts, average prices, appreciation rates
"""
return self._request('GET', '/api/collections')
# Comparison
def compare_watches(self, watch_ids: List[str]) -> Dict:
"""
Compare multiple watches side-by-side
Args:
watch_ids: List of watch IDs to compare (2-10 watches)
Returns:
Comparison results with price history and specifications
"""
return self._request('POST', '/api/compare', data={'watchIds': watch_ids})
# Export
def export_data(self, format: str = 'json') -> Any:
"""
Export data in specified format
Args:
format: Export format ('json' or 'csv')
Returns:
Exported data (dict for JSON, string for CSV)
"""
return self._request('GET', f'/api/export/{format}')
# Watchlist
def add_to_watchlist(self, user_id: str, watch_id: str) -> Dict:
"""
Add watch to user watchlist
Args:
user_id: User identifier
watch_id: Watch identifier
Returns:
Updated watchlist
"""
return self._request(
'POST',
'/api/watchlist',
data={'userId': user_id, 'watchId': watch_id, 'action': 'add'}
)
def remove_from_watchlist(self, user_id: str, watch_id: str) -> Dict:
"""
Remove watch from user watchlist
Args:
user_id: User identifier
watch_id: Watch identifier
Returns:
Updated watchlist
"""
return self._request(
'POST',
'/api/watchlist',
data={'userId': user_id, 'watchId': watch_id, 'action': 'remove'}
)
def get_watchlist(self, user_id: str) -> Dict:
"""
Get user watchlist
Args:
user_id: User identifier
Returns:
User's complete watchlist
"""
return self._request('GET', f'/api/watchlist/{user_id}')
# Admin
def clear_cache(self) -> Dict:
"""
Clear server cache (requires admin access)
Returns:
Success response
"""
return self._request('POST', '/api/admin/clear-cache')
def create_backup(self) -> Dict:
"""
Create database backup (requires admin access)
Returns:
Backup information including path and timestamp
"""
return self._request('GET', '/api/admin/backup')
# Helper Methods
@staticmethod
def calculate_appreciation(watch: Dict) -> float:
"""
Calculate appreciation percentage for a watch
Args:
watch: Watch object with priceHistory
Returns:
Appreciation percentage
"""
price_history = watch.get('priceHistory', [])
if len(price_history) < 2:
return 0.0
first_price = price_history[0]['price']
last_price = price_history[-1]['price']
return ((last_price - first_price) / first_price) * 100
@staticmethod
def get_current_price(watch: Dict) -> float:
"""
Get current price for a watch
Args:
watch: Watch object with priceHistory
Returns:
Current price
"""
price_history = watch.get('priceHistory', [])
if not price_history:
return 0.0
return price_history[-1]['price']
@staticmethod
def format_price(price: float) -> str:
"""
Format price as USD currency
Args:
price: Price value
Returns:
Formatted price string
"""
return f"${price:,.0f}"
def __enter__(self):
"""Context manager entry"""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit - close session"""
self.session.close()
def close(self):
"""Close the HTTP session"""
self.session.close()
# Example usage
if __name__ == '__main__':
# Initialize client
api = OmegaWatchAPI()
# Get health status
health = api.get_health()
print(f"API Status: {health['status']}")
print(f"Total Watches: {health['database']['watches']}")
# Get Speedmaster watches
speedmasters = api.get_watches(series='Speedmaster', limit=5)
print(f"\nFound {speedmasters['total']} Speedmaster watches")
for watch in speedmasters['watches']:
current_price = api.get_current_price(watch)
appreciation = api.calculate_appreciation(watch)
print(f" - {watch['model']}: {api.format_price(current_price)} ({appreciation:.1f}% appreciation)")
# Get trending watches
trending = api.get_trending(limit=3)
print(f"\n Top Trending Watches:")
for watch in trending['trending']:
print(f" - {watch['model']} ({watch['views']} views)")
# Get market statistics
stats = api.get_statistics()
print(f"\nMarket Statistics:")
print(f" Total Watches: {stats['totalWatches']}")
print(f" Average Appreciation: {stats['averageAppreciation']:.2f}%")