← 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()