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