← back to Watches
analytics/price_prediction.py
435 lines
#!/usr/bin/env python3
"""
Omega Watch Price Prediction Models
Advanced statistical analysis and forecasting for watch price appreciation
"""
import json
import numpy as np
import pandas as pd
from datetime import datetime
from scipy import stats
from sklearn.linear_model import LinearRegression, Ridge, Lasso
from sklearn.preprocessing import PolynomialFeatures
from sklearn.metrics import r2_score, mean_absolute_error, mean_squared_error
from sklearn.model_selection import train_test_split, cross_val_score
import warnings
warnings.filterwarnings('ignore')
class WatchPricePredictor:
"""Advanced price prediction model for Omega watches"""
def __init__(self, data_path):
self.data_path = data_path
self.watches = None
self.price_df = None
self.models = {}
self.predictions = {}
def load_data(self):
"""Load and parse watch data"""
with open(self.data_path, 'r') as f:
data = json.load(f)
self.watches = data['watches']
# Create comprehensive dataframe with all price history
rows = []
for watch in self.watches:
for price_point in watch.get('priceHistory', []):
rows.append({
'watch_id': watch['id'],
'model': watch['model'],
'series': watch['series'],
'reference': watch['reference'],
'year_introduced': watch['yearIntroduced'],
'year': price_point['year'],
'price': price_point['price'],
'condition': price_point.get('condition', 'unknown'),
'age_at_price': price_point['year'] - watch['yearIntroduced'],
'case_size': self._extract_numeric(watch['specifications'].get('caseSize', '0mm')),
'movement_type': 'manual' if 'manual' in watch['specifications'].get('movement', '').lower() else 'automatic',
'water_resistance': self._extract_numeric(watch['specifications'].get('waterResistance', '0m')),
'power_reserve': self._extract_numeric(watch['specifications'].get('powerReserve', '0 hours')),
})
self.price_df = pd.DataFrame(rows)
return self.price_df
def _extract_numeric(self, value):
"""Extract numeric value from string like '42mm' or '300m'"""
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 calculate_appreciation_metrics(self):
"""Calculate comprehensive appreciation metrics for each watch"""
metrics = []
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
# Calculate CAGR (Compound Annual Growth Rate)
cagr = ((last_price / first_price) ** (1 / years_span) - 1) * 100
# Calculate total appreciation
total_appreciation = ((last_price - first_price) / first_price) * 100
# Calculate volatility (standard deviation of year-over-year returns)
yearly_returns = []
for i in range(1, len(prices_sorted)):
if prices_sorted[i-1]['price'] > 0:
ret = (prices_sorted[i]['price'] - prices_sorted[i-1]['price']) / prices_sorted[i-1]['price']
yearly_returns.append(ret)
volatility = np.std(yearly_returns) * 100 if yearly_returns else 0
# Risk-adjusted return (Sharpe-like ratio)
risk_adjusted_return = cagr / volatility if volatility > 0 else 0
metrics.append({
'watch_id': watch['id'],
'model': watch['model'],
'series': watch['series'],
'year_introduced': watch['yearIntroduced'],
'first_price': first_price,
'current_price': last_price,
'years_tracked': years_span,
'cagr': round(cagr, 2),
'total_appreciation_pct': round(total_appreciation, 2),
'volatility': round(volatility, 2),
'risk_adjusted_return': round(risk_adjusted_return, 2),
'data_points': len(prices_sorted),
'avg_price': round(np.mean([p['price'] for p in prices_sorted]), 2)
})
return pd.DataFrame(metrics)
def build_prediction_models(self):
"""Build multiple regression models for price prediction"""
df = self.price_df.copy()
# Prepare features
X = df[['age_at_price', 'year_introduced', 'case_size', 'water_resistance', 'power_reserve']].values
y = df['price'].values
# Log transform prices for better model performance
y_log = np.log1p(y)
# Train-test split
X_train, X_test, y_train, y_test = train_test_split(X, y_log, test_size=0.2, random_state=42)
# Model 1: Linear Regression
lr_model = LinearRegression()
lr_model.fit(X_train, y_train)
lr_pred = lr_model.predict(X_test)
lr_score = r2_score(y_test, lr_pred)
# Model 2: Ridge Regression (handles multicollinearity)
ridge_model = Ridge(alpha=1.0)
ridge_model.fit(X_train, y_train)
ridge_pred = ridge_model.predict(X_test)
ridge_score = r2_score(y_test, ridge_pred)
# Model 3: Polynomial Features
poly = PolynomialFeatures(degree=2)
X_poly_train = poly.fit_transform(X_train)
X_poly_test = poly.transform(X_test)
poly_model = Ridge(alpha=10.0)
poly_model.fit(X_poly_train, y_train)
poly_pred = poly_model.predict(X_poly_test)
poly_score = r2_score(y_test, poly_pred)
self.models = {
'linear': {'model': lr_model, 'score': lr_score, 'poly': None},
'ridge': {'model': ridge_model, 'score': ridge_score, 'poly': None},
'polynomial': {'model': poly_model, 'score': poly_score, 'poly': poly}
}
return {
'linear_r2': round(lr_score, 4),
'ridge_r2': round(ridge_score, 4),
'polynomial_r2': round(poly_score, 4),
'best_model': max([('linear', lr_score), ('ridge', ridge_score), ('polynomial', poly_score)], key=lambda x: x[1])[0]
}
def predict_future_prices(self, watch_id, years_ahead=[5, 10]):
"""Predict future prices for a specific watch"""
watch = next((w for w in self.watches if w['id'] == watch_id), None)
if not watch:
return None
prices = watch.get('priceHistory', [])
if len(prices) < 3:
return None
# Prepare historical data
prices_sorted = sorted(prices, key=lambda x: x['year'])
years = np.array([p['year'] for p in prices_sorted]).reshape(-1, 1)
prices_arr = np.array([p['price'] for p in prices_sorted])
# Log transform
prices_log = np.log1p(prices_arr)
# Fit model
model = LinearRegression()
model.fit(years, prices_log)
# Calculate confidence intervals
predictions_log = model.predict(years)
residuals = prices_log - predictions_log
std_residuals = np.std(residuals)
# Predict future
current_year = datetime.now().year
predictions = {}
for years_out in years_ahead:
future_year = current_year + years_out
future_year_arr = np.array([[future_year]])
pred_log = model.predict(future_year_arr)[0]
pred_price = np.expm1(pred_log)
# 95% confidence interval
ci_lower = np.expm1(pred_log - 1.96 * std_residuals)
ci_upper = np.expm1(pred_log + 1.96 * std_residuals)
predictions[f'{years_out}_year'] = {
'year': future_year,
'predicted_price': round(pred_price, 2),
'confidence_interval_lower': round(ci_lower, 2),
'confidence_interval_upper': round(ci_upper, 2),
'confidence_level': '95%'
}
# Calculate expected annual appreciation
if len(prices_sorted) >= 2:
years_span = prices_sorted[-1]['year'] - prices_sorted[0]['year']
cagr = ((prices_sorted[-1]['price'] / prices_sorted[0]['price']) ** (1 / years_span) - 1) * 100
else:
cagr = 0
return {
'watch_id': watch_id,
'model': watch['model'],
'current_price': prices_sorted[-1]['price'],
'current_year': prices_sorted[-1]['year'],
'historical_cagr': round(cagr, 2),
'model_r2_score': round(model.score(years, prices_log), 4),
'predictions': predictions
}
def analyze_price_anomalies(self):
"""Detect price anomalies and outliers using statistical methods"""
anomalies = []
for watch in self.watches:
prices = watch.get('priceHistory', [])
if len(prices) < 4:
continue
prices_sorted = sorted(prices, key=lambda x: x['year'])
price_values = [p['price'] for p in prices_sorted]
# Calculate z-scores
mean_price = np.mean(price_values)
std_price = np.std(price_values)
if std_price == 0:
continue
for i, price_point in enumerate(prices_sorted):
z_score = (price_point['price'] - mean_price) / std_price
# Flag outliers (|z| > 2)
if abs(z_score) > 2:
anomalies.append({
'watch_id': watch['id'],
'model': watch['model'],
'year': price_point['year'],
'price': price_point['price'],
'z_score': round(z_score, 2),
'mean_price': round(mean_price, 2),
'std_dev': round(std_price, 2),
'anomaly_type': 'high_outlier' if z_score > 2 else 'low_outlier',
'deviation_pct': round(((price_point['price'] - mean_price) / mean_price) * 100, 2)
})
return anomalies
def calculate_correlation_matrix(self):
"""Calculate correlations between specifications and price appreciation"""
df = self.price_df.copy()
# Group by watch to get final appreciation
watch_stats = []
for watch_id in df['watch_id'].unique():
watch_data = df[df['watch_id'] == watch_id].sort_values('year')
if len(watch_data) < 2:
continue
first_price = watch_data.iloc[0]['price']
last_price = watch_data.iloc[-1]['price']
years_span = watch_data.iloc[-1]['year'] - watch_data.iloc[0]['year']
if years_span > 0 and first_price > 0:
cagr = ((last_price / first_price) ** (1 / years_span) - 1) * 100
watch_stats.append({
'watch_id': watch_id,
'case_size': watch_data.iloc[0]['case_size'],
'water_resistance': watch_data.iloc[0]['water_resistance'],
'power_reserve': watch_data.iloc[0]['power_reserve'],
'age_at_last_price': watch_data.iloc[-1]['age_at_price'],
'is_manual': 1 if watch_data.iloc[0]['movement_type'] == 'manual' else 0,
'cagr': cagr,
'total_appreciation': ((last_price - first_price) / first_price) * 100
})
stats_df = pd.DataFrame(watch_stats)
# Calculate correlations
correlation_matrix = stats_df[[
'case_size', 'water_resistance', 'power_reserve',
'age_at_last_price', 'is_manual', 'cagr', 'total_appreciation'
]].corr()
# Extract key correlations with appreciation
cagr_correlations = correlation_matrix['cagr'].drop('cagr').to_dict()
appreciation_correlations = correlation_matrix['total_appreciation'].drop('total_appreciation').to_dict()
return {
'cagr_correlations': {k: round(v, 4) for k, v in cagr_correlations.items()},
'total_appreciation_correlations': {k: round(v, 4) for k, v in appreciation_correlations.items()},
'correlation_matrix': correlation_matrix.to_dict()
}
def generate_investment_insights(self):
"""Generate actionable investment insights"""
metrics_df = self.calculate_appreciation_metrics()
# Sort by different criteria
top_cagr = metrics_df.nlargest(5, 'cagr')[['model', 'series', 'cagr', 'current_price']].to_dict('records')
top_total_return = metrics_df.nlargest(5, 'total_appreciation_pct')[['model', 'series', 'total_appreciation_pct', 'current_price']].to_dict('records')
best_risk_adjusted = metrics_df.nlargest(5, 'risk_adjusted_return')[['model', 'series', 'risk_adjusted_return', 'volatility', 'cagr']].to_dict('records')
# Low volatility growth (stable investments)
stable_growth = metrics_df[(metrics_df['cagr'] > 10) & (metrics_df['volatility'] < metrics_df['volatility'].median())]
stable_investments = stable_growth.nlargest(5, 'cagr')[['model', 'series', 'cagr', 'volatility']].to_dict('records')
# High growth potential (high CAGR, relatively underpriced)
median_price = metrics_df['current_price'].median()
growth_opportunities = metrics_df[(metrics_df['cagr'] > 15) & (metrics_df['current_price'] < median_price)]
opportunities = growth_opportunities.nlargest(5, 'cagr')[['model', 'series', 'cagr', 'current_price']].to_dict('records')
return {
'top_performers_by_cagr': top_cagr,
'top_total_returns': top_total_return,
'best_risk_adjusted_returns': best_risk_adjusted,
'stable_growth_investments': stable_investments,
'growth_opportunities': opportunities,
'market_summary': {
'median_cagr': round(metrics_df['cagr'].median(), 2),
'mean_cagr': round(metrics_df['cagr'].mean(), 2),
'median_volatility': round(metrics_df['volatility'].median(), 2),
'total_watches_analyzed': len(metrics_df)
}
}
def main():
"""Run comprehensive price prediction analysis"""
print("=" * 80)
print("OMEGA WATCH PRICE PREDICTION ANALYSIS")
print("=" * 80)
predictor = WatchPricePredictor('/root/Projects/watches/data/watches.json')
# Load data
print("\n[1/7] Loading watch data...")
df = predictor.load_data()
print(f" Loaded {len(df)} price points from {len(predictor.watches)} watches")
# Calculate appreciation metrics
print("\n[2/7] Calculating appreciation metrics...")
metrics_df = predictor.calculate_appreciation_metrics()
print(f" Analyzed {len(metrics_df)} watches with complete price history")
# Build prediction models
print("\n[3/7] Building prediction models...")
model_scores = predictor.build_prediction_models()
print(f" Linear R2: {model_scores['linear_r2']}")
print(f" Ridge R2: {model_scores['ridge_r2']}")
print(f" Polynomial R2: {model_scores['polynomial_r2']}")
print(f" Best model: {model_scores['best_model']}")
# Predict future prices for all watches
print("\n[4/7] Generating future price predictions...")
all_predictions = []
for watch in predictor.watches:
pred = predictor.predict_future_prices(watch['id'], years_ahead=[5, 10, 20])
if pred:
all_predictions.append(pred)
print(f" Generated predictions for {len(all_predictions)} watches")
# Detect anomalies
print("\n[5/7] Detecting price anomalies...")
anomalies = predictor.analyze_price_anomalies()
print(f" Found {len(anomalies)} price anomalies")
# Calculate correlations
print("\n[6/7] Calculating feature correlations...")
correlations = predictor.calculate_correlation_matrix()
print(" CAGR Correlations:")
for feature, corr in correlations['cagr_correlations'].items():
print(f" {feature}: {corr}")
# Generate investment insights
print("\n[7/7] Generating investment insights...")
insights = predictor.generate_investment_insights()
# Save all results
output = {
'generated_at': datetime.now().isoformat(),
'summary': {
'total_watches': len(predictor.watches),
'total_price_points': len(df),
'watches_with_predictions': len(all_predictions),
'anomalies_detected': len(anomalies)
},
'model_performance': model_scores,
'appreciation_metrics': metrics_df.to_dict('records'),
'future_predictions': all_predictions,
'price_anomalies': anomalies,
'correlations': correlations,
'investment_insights': insights
}
output_path = '/root/Projects/watches/analytics/price_predictions.json'
with open(output_path, 'w') as f:
json.dump(output, f, indent=2)
print(f"\n" + "=" * 80)
print(f"Analysis complete! Results saved to: {output_path}")
print("=" * 80)
return output
if __name__ == '__main__':
results = main()