← back to Watches
analytics/market_analysis.py
445 lines
#!/usr/bin/env python3
"""
Omega Watch Market Analysis
Clustering, segmentation, and market intelligence
"""
import json
import numpy as np
import pandas as pd
from datetime import datetime
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from scipy.cluster.hierarchy import dendrogram, linkage
from collections import defaultdict
import warnings
warnings.filterwarnings('ignore')
class WatchMarketAnalyzer:
"""Advanced market analysis for Omega watches"""
def __init__(self, data_path):
self.data_path = data_path
self.watches = None
self.watch_df = None
def load_data(self):
"""Load watch data"""
with open(self.data_path, 'r') as f:
data = json.load(f)
self.watches = data['watches']
# Create feature dataframe
rows = []
for watch in self.watches:
prices = watch.get('priceHistory', [])
if len(prices) < 2:
continue
prices_sorted = sorted(prices, key=lambda x: x['year'])
first_price = prices_sorted[0]['price']
last_price = prices_sorted[-1]['price']
years_span = prices_sorted[-1]['year'] - prices_sorted[0]['year']
if years_span == 0 or first_price == 0:
continue
cagr = ((last_price / first_price) ** (1 / years_span) - 1) * 100
rows.append({
'watch_id': watch['id'],
'model': watch['model'],
'series': watch['series'],
'year_introduced': watch['yearIntroduced'],
'current_price': last_price,
'first_price': first_price,
'cagr': cagr,
'total_appreciation': ((last_price - first_price) / first_price) * 100,
'years_tracked': years_span,
'data_points': len(prices_sorted),
'case_size': self._extract_numeric(watch['specifications'].get('caseSize', '0mm')),
'water_resistance': self._extract_numeric(watch['specifications'].get('waterResistance', '0m')),
'power_reserve': self._extract_numeric(watch['specifications'].get('powerReserve', '0 hours')),
'is_manual': 1 if 'manual' in watch['specifications'].get('movement', '').lower() else 0,
'has_chronograph': 1 if any('chrono' in f.lower() for f in watch['specifications'].get('features', [])) else 0
})
self.watch_df = pd.DataFrame(rows)
return self.watch_df
def _extract_numeric(self, value):
"""Extract numeric value from string"""
if isinstance(value, (int, float)):
return float(value)
try:
return float(''.join(c for c in str(value) if c.isdigit() or c == '.'))
except:
return 0.0
def perform_clustering(self, n_clusters=4):
"""Perform k-means clustering on watches"""
df = self.watch_df.copy()
# Select features for clustering
features = ['current_price', 'cagr', 'case_size', 'water_resistance',
'power_reserve', 'is_manual', 'has_chronograph', 'year_introduced']
X = df[features].values
# Normalize features
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
# K-means clustering
kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
clusters = kmeans.fit_predict(X_scaled)
df['cluster'] = clusters
# Analyze each cluster
cluster_profiles = []
for i in range(n_clusters):
cluster_watches = df[df['cluster'] == i]
profile = {
'cluster_id': i,
'count': len(cluster_watches),
'avg_current_price': round(cluster_watches['current_price'].mean(), 2),
'avg_cagr': round(cluster_watches['cagr'].mean(), 2),
'avg_case_size': round(cluster_watches['case_size'].mean(), 2),
'avg_water_resistance': round(cluster_watches['water_resistance'].mean(), 2),
'pct_manual': round((cluster_watches['is_manual'].sum() / len(cluster_watches)) * 100, 2),
'pct_chronograph': round((cluster_watches['has_chronograph'].sum() / len(cluster_watches)) * 100, 2),
'series_distribution': cluster_watches['series'].value_counts().to_dict(),
'watches': cluster_watches[['watch_id', 'model', 'current_price', 'cagr']].to_dict('records')
}
# Cluster characterization
if profile['avg_cagr'] > df['cagr'].median() and profile['avg_current_price'] < df['current_price'].median():
profile['characterization'] = 'Growth Opportunities'
elif profile['avg_current_price'] > df['current_price'].median() and profile['avg_cagr'] > df['cagr'].median():
profile['characterization'] = 'Premium High Performers'
elif profile['avg_current_price'] < df['current_price'].median() and profile['avg_cagr'] < df['cagr'].median():
profile['characterization'] = 'Value/Entry Level'
else:
profile['characterization'] = 'Stable Mid-Tier'
cluster_profiles.append(profile)
return {
'cluster_profiles': cluster_profiles,
'inertia': round(kmeans.inertia_, 2),
'silhouette_score': self._calculate_silhouette_score(X_scaled, clusters)
}
def _calculate_silhouette_score(self, X, labels):
"""Calculate silhouette score for clustering quality"""
from sklearn.metrics import silhouette_score
try:
return round(silhouette_score(X, labels), 4)
except:
return None
def identify_value_watches(self):
"""Identify undervalued watches (high growth potential, below median price)"""
df = self.watch_df.copy()
median_price = df['current_price'].median()
median_cagr = df['cagr'].median()
# Value criteria: Above median growth, below median price
value_watches = df[
(df['cagr'] > median_cagr) &
(df['current_price'] < median_price)
].copy()
# Calculate value score (higher CAGR, lower price = better value)
max_cagr = df['cagr'].max()
max_price = df['current_price'].max()
value_watches['value_score'] = (
(value_watches['cagr'] / max_cagr) * 0.6 +
((max_price - value_watches['current_price']) / max_price) * 0.4
) * 100
value_watches_sorted = value_watches.sort_values('value_score', ascending=False)
return value_watches_sorted[[
'watch_id', 'model', 'series', 'current_price', 'cagr',
'total_appreciation', 'value_score'
]].to_dict('records')
def analyze_by_collection(self):
"""Analyze performance by collection/series"""
df = self.watch_df.copy()
series_analysis = []
for series in df['series'].unique():
series_data = df[df['series'] == series]
analysis = {
'series': series,
'watch_count': len(series_data),
'avg_current_price': round(series_data['current_price'].mean(), 2),
'median_current_price': round(series_data['current_price'].median(), 2),
'price_range': {
'min': round(series_data['current_price'].min(), 2),
'max': round(series_data['current_price'].max(), 2)
},
'avg_cagr': round(series_data['cagr'].mean(), 2),
'median_cagr': round(series_data['cagr'].median(), 2),
'avg_total_appreciation': round(series_data['total_appreciation'].mean(), 2),
'volatility': round(series_data['cagr'].std(), 2),
'avg_year_introduced': round(series_data['year_introduced'].mean(), 0),
'top_performer': series_data.nlargest(1, 'cagr')[['model', 'cagr', 'current_price']].to_dict('records')[0] if len(series_data) > 0 else None
}
series_analysis.append(analysis)
# Sort by average CAGR
series_analysis.sort(key=lambda x: x['avg_cagr'], reverse=True)
return series_analysis
def analyze_by_decade(self):
"""Analyze performance by decade introduced"""
df = self.watch_df.copy()
df['decade'] = (df['year_introduced'] // 10) * 10
decade_analysis = []
for decade in sorted(df['decade'].unique()):
decade_data = df[df['decade'] == decade]
analysis = {
'decade': f"{decade}s",
'watch_count': len(decade_data),
'avg_cagr': round(decade_data['cagr'].mean(), 2),
'median_cagr': round(decade_data['cagr'].median(), 2),
'avg_current_price': round(decade_data['current_price'].mean(), 2),
'total_appreciation_avg': round(decade_data['total_appreciation'].mean(), 2),
'best_performing_watch': decade_data.nlargest(1, 'cagr')[['model', 'cagr']].to_dict('records')[0] if len(decade_data) > 0 else None
}
decade_analysis.append(analysis)
return decade_analysis
def calculate_market_volatility(self):
"""Calculate market-wide volatility metrics"""
volatility_metrics = []
for watch in self.watches:
prices = watch.get('priceHistory', [])
if len(prices) < 3:
continue
prices_sorted = sorted(prices, key=lambda x: x['year'])
# Calculate year-over-year returns
returns = []
for i in range(1, len(prices_sorted)):
prev_price = prices_sorted[i-1]['price']
curr_price = prices_sorted[i]['price']
if prev_price > 0:
yoy_return = ((curr_price - prev_price) / prev_price) * 100
returns.append(yoy_return)
if not returns:
continue
volatility_metrics.append({
'watch_id': watch['id'],
'model': watch['model'],
'series': watch['series'],
'volatility': round(np.std(returns), 2),
'avg_return': round(np.mean(returns), 2),
'max_return': round(max(returns), 2),
'min_return': round(min(returns), 2),
'coefficient_of_variation': round((np.std(returns) / abs(np.mean(returns))) if np.mean(returns) != 0 else 0, 2)
})
volatility_df = pd.DataFrame(volatility_metrics)
return {
'individual_volatility': volatility_metrics,
'market_summary': {
'avg_volatility': round(volatility_df['volatility'].mean(), 2),
'median_volatility': round(volatility_df['volatility'].median(), 2),
'most_volatile': volatility_df.nlargest(5, 'volatility')[['model', 'volatility', 'avg_return']].to_dict('records'),
'least_volatile': volatility_df.nsmallest(5, 'volatility')[['model', 'volatility', 'avg_return']].to_dict('records')
}
}
def identify_investment_opportunities(self):
"""Comprehensive investment opportunity analysis"""
df = self.watch_df.copy()
opportunities = []
# Calculate percentiles
cagr_75th = df['cagr'].quantile(0.75)
price_50th = df['current_price'].quantile(0.50)
price_25th = df['current_price'].quantile(0.25)
for _, watch in df.iterrows():
opportunity_score = 0
reasons = []
# High growth rate
if watch['cagr'] > cagr_75th:
opportunity_score += 30
reasons.append(f"Top 25% growth rate ({watch['cagr']:.1f}% CAGR)")
# Below median price
if watch['current_price'] < price_50th:
opportunity_score += 20
reasons.append(f"Below median price (${watch['current_price']:,.0f})")
# Value play: very low price, decent growth
if watch['current_price'] < price_25th and watch['cagr'] > df['cagr'].median():
opportunity_score += 25
reasons.append("Value play: Low entry price with above-average growth")
# Manual movement (collector appeal)
if watch['is_manual'] == 1:
opportunity_score += 15
reasons.append("Manual movement (high collector appeal)")
# Chronograph (complexity premium)
if watch['has_chronograph'] == 1:
opportunity_score += 10
reasons.append("Chronograph complication")
# Vintage (pre-1990)
if watch['year_introduced'] < 1990:
opportunity_score += 15
reasons.append(f"Vintage ({watch['year_introduced']})")
# Long track record
if watch['years_tracked'] > 30:
opportunity_score += 10
reasons.append(f"{watch['years_tracked']} years of price history")
if opportunity_score > 50: # Threshold for opportunity
opportunities.append({
'watch_id': watch['watch_id'],
'model': watch['model'],
'series': watch['series'],
'current_price': watch['current_price'],
'cagr': watch['cagr'],
'opportunity_score': opportunity_score,
'investment_reasons': reasons,
'year_introduced': watch['year_introduced']
})
# Sort by opportunity score
opportunities.sort(key=lambda x: x['opportunity_score'], reverse=True)
return opportunities
def generate_heat_map_data(self):
"""Generate data for heat map visualizations"""
df = self.watch_df.copy()
# Create decade bins
df['decade'] = (df['year_introduced'] // 10) * 10
# Heat map: Series x Decade showing average CAGR
heatmap_data = df.groupby(['series', 'decade']).agg({
'cagr': 'mean',
'current_price': 'mean',
'watch_id': 'count'
}).reset_index()
heatmap_data.columns = ['series', 'decade', 'avg_cagr', 'avg_price', 'count']
return {
'series_by_decade': heatmap_data.to_dict('records'),
'price_vs_appreciation': df[['current_price', 'cagr', 'model', 'series']].to_dict('records')
}
def main():
"""Run comprehensive market analysis"""
print("=" * 80)
print("OMEGA WATCH MARKET ANALYSIS")
print("=" * 80)
analyzer = WatchMarketAnalyzer('/root/Projects/watches/data/watches.json')
# Load data
print("\n[1/8] Loading watch data...")
df = analyzer.load_data()
print(f" Loaded {len(df)} watches with complete data")
# Clustering
print("\n[2/8] Performing k-means clustering...")
clustering_results = analyzer.perform_clustering(n_clusters=4)
print(f" Created {len(clustering_results['cluster_profiles'])} clusters")
print(f" Silhouette score: {clustering_results['silhouette_score']}")
# Value watches
print("\n[3/8] Identifying value watches...")
value_watches = analyzer.identify_value_watches()
print(f" Found {len(value_watches)} value opportunities")
# Collection analysis
print("\n[4/8] Analyzing by collection...")
collection_analysis = analyzer.analyze_by_collection()
print(f" Analyzed {len(collection_analysis)} collections")
# Decade analysis
print("\n[5/8] Analyzing by decade...")
decade_analysis = analyzer.analyze_by_decade()
print(f" Analyzed {len(decade_analysis)} decades")
# Volatility
print("\n[6/8] Calculating market volatility...")
volatility = analyzer.calculate_market_volatility()
print(f" Market avg volatility: {volatility['market_summary']['avg_volatility']}%")
# Investment opportunities
print("\n[7/8] Identifying investment opportunities...")
opportunities = analyzer.identify_investment_opportunities()
print(f" Found {len(opportunities)} high-potential opportunities")
# Heat map data
print("\n[8/8] Generating visualization data...")
heatmap_data = analyzer.generate_heat_map_data()
print(f" Generated {len(heatmap_data['series_by_decade'])} data points for heat maps")
# Compile results
output = {
'generated_at': datetime.now().isoformat(),
'summary': {
'total_watches_analyzed': len(df),
'collections': df['series'].nunique(),
'avg_market_cagr': round(df['cagr'].mean(), 2),
'median_market_price': round(df['current_price'].median(), 2)
},
'clustering': clustering_results,
'value_watches': value_watches[:10], # Top 10
'collection_analysis': collection_analysis,
'decade_analysis': decade_analysis,
'volatility_analysis': volatility,
'investment_opportunities': opportunities[:15], # Top 15
'visualization_data': heatmap_data
}
output_path = '/root/Projects/watches/analytics/market_analysis.json'
with open(output_path, 'w') as f:
json.dump(output, f, indent=2)
print(f"\n" + "=" * 80)
print(f"Market analysis complete! Results saved to: {output_path}")
print("=" * 80)
return output
if __name__ == '__main__':
results = main()