← back to Handbag Auth Nextjs
analytics/analytics_engine.py
634 lines
#!/usr/bin/env python3
"""
LUXVAULT Advanced Analytics Engine
Provides predictive analytics, trend analysis, value scoring, and insights
"""
import sqlite3
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from scipy import stats
from sklearn.linear_model import LinearRegression, Ridge
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import json
import os
# Configuration
PROJECT_DIR = '/root/Projects/handbag-auth-nextjs'
DB_PATH = os.path.join(PROJECT_DIR, 'data', 'auction-history', 'auctions.db')
class AuctionAnalytics:
"""Advanced analytics for luxury handbag auctions"""
def __init__(self, db_path=DB_PATH):
self.db_path = db_path
self.df = None
self.load_data()
def load_data(self):
"""Load auction data into pandas DataFrame"""
conn = sqlite3.connect(self.db_path)
query = """
SELECT
id,
auction_id,
title,
auction_house,
current_price,
estimate_low,
estimate_high,
end_date,
lot_number,
url,
search_term,
created_at,
CASE
WHEN estimate_low > 0 AND current_price > 0
THEN ((estimate_low - current_price) / estimate_low * 100)
ELSE 0
END as savings_percent,
CASE
WHEN estimate_low > 0 AND current_price > 0
THEN (estimate_low - current_price)
ELSE 0
END as savings_amount
FROM auctions
WHERE current_price > 0
"""
self.df = pd.read_sql_query(query, conn)
self.df['end_date'] = pd.to_datetime(self.df['end_date'])
self.df['created_at'] = pd.to_datetime(self.df['created_at'])
conn.close()
# =========================
# PREDICTIVE ANALYTICS
# =========================
def predict_price(self, brand, days_ahead=30):
"""
Predict future price for a brand using time series analysis
Returns: predicted price, confidence interval, trend
"""
brand_data = self.df[self.df['search_term'] == brand].copy()
if len(brand_data) < 10:
return None
# Sort by date
brand_data = brand_data.sort_values('end_date')
# Create time features
brand_data['days_since_start'] = (brand_data['end_date'] - brand_data['end_date'].min()).dt.days
# Prepare data for regression
X = brand_data[['days_since_start']].values
y = brand_data['current_price'].values
# Train linear regression model
model = LinearRegression()
model.fit(X, y)
# Predict future price
max_days = brand_data['days_since_start'].max()
future_days = max_days + days_ahead
predicted_price = model.predict([[future_days]])[0]
# Calculate confidence interval
y_pred = model.predict(X)
residuals = y - y_pred
std_error = np.std(residuals)
confidence_interval = 1.96 * std_error
# Calculate trend (slope)
trend_percent = (model.coef_[0] / brand_data['current_price'].mean()) * 365 * 100
return {
'brand': brand,
'predicted_price': round(predicted_price, 2),
'confidence_lower': round(predicted_price - confidence_interval, 2),
'confidence_upper': round(predicted_price + confidence_interval, 2),
'trend_percent_annual': round(trend_percent, 2),
'sample_size': len(brand_data),
'current_avg_price': round(brand_data['current_price'].mean(), 2)
}
def identify_undervalued_auctions(self, threshold_percent=20):
"""
Identify auctions selling significantly below historical average
Returns: list of undervalued items with expected value
"""
undervalued = []
for brand in self.df['search_term'].unique():
brand_data = self.df[self.df['search_term'] == brand]
avg_price = brand_data['current_price'].mean()
std_price = brand_data['current_price'].std()
# Find items selling below average - threshold%
threshold_price = avg_price * (1 - threshold_percent / 100)
for _, row in brand_data.iterrows():
if row['current_price'] < threshold_price:
# Calculate z-score
z_score = (row['current_price'] - avg_price) / std_price if std_price > 0 else 0
undervalued.append({
'auction_id': row['auction_id'],
'title': row['title'],
'brand': brand,
'current_price': row['current_price'],
'historical_avg': round(avg_price, 2),
'discount_percent': round(((avg_price - row['current_price']) / avg_price * 100), 2),
'discount_amount': round(avg_price - row['current_price'], 2),
'z_score': round(z_score, 2),
'auction_house': row['auction_house'],
'url': row['url']
})
# Sort by discount amount
undervalued.sort(key=lambda x: x['discount_amount'], reverse=True)
return undervalued
def predict_optimal_bid(self, brand, percentile=75):
"""
Predict optimal bid amount based on historical data
Returns: suggested bid at different confidence levels
"""
brand_data = self.df[self.df['search_term'] == brand]
if len(brand_data) < 5:
return None
prices = brand_data['current_price'].values
return {
'brand': brand,
'sample_size': len(brand_data),
'conservative_bid': round(np.percentile(prices, 25), 2), # 25th percentile
'median_bid': round(np.percentile(prices, 50), 2), # 50th percentile
'competitive_bid': round(np.percentile(prices, 75), 2), # 75th percentile
'aggressive_bid': round(np.percentile(prices, 90), 2), # 90th percentile
'mean_price': round(np.mean(prices), 2),
'std_dev': round(np.std(prices), 2)
}
def calculate_roi_potential(self, auction_id):
"""
Calculate potential ROI for an investment piece
Based on historical appreciation trends
"""
auction = self.df[self.df['auction_id'] == auction_id].iloc[0]
brand = auction['search_term']
brand_data = self.df[self.df['search_term'] == brand].copy()
# Calculate historical trend
brand_data = brand_data.sort_values('end_date')
if len(brand_data) < 10:
return None
# Calculate annual appreciation
brand_data['days_since_start'] = (brand_data['end_date'] - brand_data['end_date'].min()).dt.days
X = brand_data[['days_since_start']].values
y = brand_data['current_price'].values
model = LinearRegression()
model.fit(X, y)
# Annual appreciation rate
annual_appreciation = (model.coef_[0] / brand_data['current_price'].mean()) * 365 * 100
# Project future value
current_price = auction['current_price']
projections = {
'1_year': round(current_price * (1 + annual_appreciation/100), 2),
'3_year': round(current_price * ((1 + annual_appreciation/100) ** 3), 2),
'5_year': round(current_price * ((1 + annual_appreciation/100) ** 5), 2)
}
return {
'auction_id': auction_id,
'brand': brand,
'current_price': current_price,
'annual_appreciation_percent': round(annual_appreciation, 2),
'projected_value_1yr': projections['1_year'],
'projected_value_3yr': projections['3_year'],
'projected_value_5yr': projections['5_year'],
'roi_1yr_percent': round(((projections['1_year'] - current_price) / current_price * 100), 2),
'roi_3yr_percent': round(((projections['3_year'] - current_price) / current_price * 100), 2),
'roi_5yr_percent': round(((projections['5_year'] - current_price) / current_price * 100), 2)
}
# =========================
# TREND ANALYSIS
# =========================
def analyze_brand_trends(self, days=90):
"""
Identify trending brands based on auction frequency and price changes
"""
cutoff_date = datetime.now() - timedelta(days=days)
recent_data = self.df[self.df['end_date'] >= cutoff_date]
older_data = self.df[self.df['end_date'] < cutoff_date]
trends = []
for brand in self.df['search_term'].unique():
recent_brand = recent_data[recent_data['search_term'] == brand]
older_brand = older_data[older_data['search_term'] == brand]
if len(recent_brand) == 0:
continue
# Calculate frequency change
recent_count = len(recent_brand)
older_count = len(older_brand)
avg_older_count = older_count / (len(self.df) - len(recent_data)) * len(recent_data) if len(older_data) > 0 else 0
frequency_change = ((recent_count - avg_older_count) / avg_older_count * 100) if avg_older_count > 0 else 0
# Calculate price change
recent_avg_price = recent_brand['current_price'].mean()
older_avg_price = older_brand['current_price'].mean() if len(older_brand) > 0 else recent_avg_price
price_change = ((recent_avg_price - older_avg_price) / older_avg_price * 100) if older_avg_price > 0 else 0
# Calculate momentum score
momentum_score = (frequency_change * 0.4 + price_change * 0.6)
trends.append({
'brand': brand,
'recent_auctions': recent_count,
'frequency_change_percent': round(frequency_change, 2),
'recent_avg_price': round(recent_avg_price, 2),
'price_change_percent': round(price_change, 2),
'momentum_score': round(momentum_score, 2),
'trend': 'Hot' if momentum_score > 10 else 'Cooling' if momentum_score < -10 else 'Stable'
})
# Sort by momentum
trends.sort(key=lambda x: x['momentum_score'], reverse=True)
return trends
def price_movement_by_brand(self, brand=None):
"""
Track price movements over time for a brand or all brands
"""
if brand:
data = self.df[self.df['search_term'] == brand].copy()
else:
data = self.df.copy()
# Group by month
data['month'] = data['end_date'].dt.to_period('M')
monthly_prices = data.groupby(['month', 'search_term']).agg({
'current_price': ['mean', 'median', 'std', 'count']
}).reset_index()
monthly_prices.columns = ['month', 'brand', 'avg_price', 'median_price', 'std_price', 'count']
monthly_prices['month'] = monthly_prices['month'].astype(str)
return monthly_prices.to_dict('records')
def detect_seasonal_patterns(self, brand):
"""
Detect seasonal patterns in auction prices and frequency
"""
brand_data = self.df[self.df['search_term'] == brand].copy()
if len(brand_data) < 50:
return None
# Group by month
brand_data['month'] = brand_data['end_date'].dt.month
monthly_stats = brand_data.groupby('month').agg({
'current_price': ['mean', 'count'],
'savings_percent': 'mean'
}).reset_index()
monthly_stats.columns = ['month', 'avg_price', 'auction_count', 'avg_savings']
# Calculate seasonality index (normalized to 1.0 = average)
overall_avg_price = brand_data['current_price'].mean()
monthly_stats['price_seasonality'] = monthly_stats['avg_price'] / overall_avg_price
overall_avg_count = brand_data.groupby('month').size().mean()
monthly_stats['volume_seasonality'] = monthly_stats['auction_count'] / overall_avg_count
return monthly_stats.to_dict('records')
def auction_house_performance(self):
"""
Compare auction houses by pricing and deal quality
"""
house_stats = []
for house in self.df['auction_house'].unique():
house_data = self.df[self.df['auction_house'] == house]
house_stats.append({
'auction_house': house,
'total_auctions': len(house_data),
'avg_price': round(house_data['current_price'].mean(), 2),
'median_price': round(house_data['current_price'].median(), 2),
'avg_savings_percent': round(house_data['savings_percent'].mean(), 2),
'avg_savings_amount': round(house_data['savings_amount'].mean(), 2),
'price_range_low': round(house_data['current_price'].min(), 2),
'price_range_high': round(house_data['current_price'].max(), 2),
'deals_count': len(house_data[house_data['savings_percent'] > 20])
})
house_stats.sort(key=lambda x: x['avg_savings_percent'], reverse=True)
return house_stats
# =========================
# VALUE SCORING
# =========================
def calculate_deal_score(self, auction_id):
"""
Create composite deal score combining multiple factors
Scale: 0-100 (higher is better)
"""
auction = self.df[self.df['auction_id'] == auction_id].iloc[0]
brand = auction['search_term']
brand_data = self.df[self.df['search_term'] == brand]
# Factor 1: Savings vs estimate (0-30 points)
savings_score = min(auction['savings_percent'] / 50 * 30, 30)
# Factor 2: Price vs historical average (0-25 points)
avg_price = brand_data['current_price'].mean()
price_discount = ((avg_price - auction['current_price']) / avg_price * 100) if avg_price > 0 else 0
price_score = min(max(price_discount / 30 * 25, 0), 25)
# Factor 3: Brand trend (0-20 points)
trends = self.analyze_brand_trends(days=90)
brand_trend = next((t for t in trends if t['brand'] == brand), None)
trend_score = 0
if brand_trend:
# Positive trend adds value
if brand_trend['trend'] == 'Hot':
trend_score = 20
elif brand_trend['trend'] == 'Stable':
trend_score = 10
# Factor 4: Rarity (0-15 points)
rarity_score = self.calculate_rarity_score(brand)
# Factor 5: Auction house reputation (0-10 points)
prestigious_houses = ["Christie's", "Sotheby's"]
house_score = 10 if auction['auction_house'] in prestigious_houses else 5
total_score = savings_score + price_score + trend_score + rarity_score + house_score
return {
'auction_id': auction_id,
'deal_score': round(total_score, 1),
'grade': self.score_to_grade(total_score),
'breakdown': {
'savings_score': round(savings_score, 1),
'price_score': round(price_score, 1),
'trend_score': round(trend_score, 1),
'rarity_score': round(rarity_score, 1),
'house_score': round(house_score, 1)
}
}
def calculate_rarity_score(self, brand):
"""Calculate rarity score based on auction frequency (0-15 points)"""
brand_count = len(self.df[self.df['search_term'] == brand])
total_count = len(self.df)
frequency = brand_count / total_count
# Less frequent = more rare = higher score
if frequency < 0.02: # Less than 2%
return 15
elif frequency < 0.05: # Less than 5%
return 10
elif frequency < 0.10: # Less than 10%
return 5
else:
return 2
def score_to_grade(self, score):
"""Convert numeric score to letter grade"""
if score >= 80:
return 'A+'
elif score >= 70:
return 'A'
elif score >= 60:
return 'B+'
elif score >= 50:
return 'B'
elif score >= 40:
return 'C+'
elif score >= 30:
return 'C'
else:
return 'D'
def calculate_investment_grade(self, auction_id):
"""
Calculate investment-grade rating for long-term value
Returns: rating from 1-5 stars
"""
roi_data = self.calculate_roi_potential(auction_id)
if not roi_data:
return None
# Criteria for investment grade
annual_appreciation = roi_data['annual_appreciation_percent']
deal_score = self.calculate_deal_score(auction_id)['deal_score']
# Calculate stars (1-5)
stars = 0
# Strong appreciation potential
if annual_appreciation > 10:
stars += 2
elif annual_appreciation > 5:
stars += 1
# Good current deal
if deal_score > 70:
stars += 2
elif deal_score > 50:
stars += 1
# Brand prestige
auction = self.df[self.df['auction_id'] == auction_id].iloc[0]
prestigious_brands = ['Hermes Birkin', 'Hermes Kelly', 'Chanel Classic Flap']
if auction['search_term'] in prestigious_brands:
stars += 1
stars = min(stars, 5)
return {
'auction_id': auction_id,
'investment_stars': stars,
'investment_grade': 'Excellent' if stars >= 4 else 'Good' if stars >= 3 else 'Fair' if stars >= 2 else 'Speculative',
'annual_appreciation': roi_data['annual_appreciation_percent'],
'deal_score': deal_score,
'recommendation': self.get_investment_recommendation(stars)
}
def get_investment_recommendation(self, stars):
"""Get recommendation based on star rating"""
if stars >= 4:
return "Strong Buy - Excellent investment potential"
elif stars >= 3:
return "Buy - Good long-term value"
elif stars >= 2:
return "Consider - Fair investment opportunity"
else:
return "Pass - Better opportunities available"
# =========================
# INSIGHTS & REPORTS
# =========================
def generate_daily_report(self):
"""Generate daily insights report"""
today = datetime.now().date()
yesterday = today - timedelta(days=1)
recent_auctions = self.df[self.df['end_date'].dt.date >= yesterday]
# Top deals
top_deals = recent_auctions.nlargest(10, 'savings_percent')
# Trending brands
trends = self.analyze_brand_trends(days=7)[:5]
# Best auction houses
house_perf = self.auction_house_performance()[:3]
report = {
'date': str(today),
'summary': {
'total_auctions_analyzed': len(self.df),
'recent_auctions': len(recent_auctions),
'avg_savings_percent': round(recent_auctions['savings_percent'].mean(), 2) if len(recent_auctions) > 0 else 0
},
'top_deals': [
{
'auction_id': row['auction_id'],
'title': row['title'],
'brand': row['search_term'],
'price': row['current_price'],
'savings_percent': round(row['savings_percent'], 2),
'auction_house': row['auction_house']
}
for _, row in top_deals.iterrows()
],
'trending_brands': trends,
'best_auction_houses': house_perf
}
return report
def generate_brand_comparison(self):
"""Compare all brands across key metrics"""
comparisons = []
for brand in self.df['search_term'].unique():
brand_data = self.df[self.df['search_term'] == brand]
# Get prediction
prediction = self.predict_price(brand, days_ahead=30)
comparisons.append({
'brand': brand,
'total_auctions': len(brand_data),
'avg_price': round(brand_data['current_price'].mean(), 2),
'median_price': round(brand_data['current_price'].median(), 2),
'avg_savings': round(brand_data['savings_percent'].mean(), 2),
'price_range': {
'min': round(brand_data['current_price'].min(), 2),
'max': round(brand_data['current_price'].max(), 2)
},
'predicted_trend': prediction['trend_percent_annual'] if prediction else 0,
'rarity_score': self.calculate_rarity_score(brand)
})
comparisons.sort(key=lambda x: x['avg_price'], reverse=True)
return comparisons
def get_personalized_recommendations(self, max_budget=None, preferred_brands=None, min_deal_score=50):
"""
Generate personalized deal recommendations
"""
recommendations = []
# Filter by budget
filtered_df = self.df
if max_budget:
filtered_df = filtered_df[filtered_df['current_price'] <= max_budget]
# Filter by brands
if preferred_brands:
filtered_df = filtered_df[filtered_df['search_term'].isin(preferred_brands)]
# Calculate scores for all
for _, auction in filtered_df.iterrows():
deal_score_data = self.calculate_deal_score(auction['auction_id'])
if deal_score_data['deal_score'] >= min_deal_score:
investment_grade = self.calculate_investment_grade(auction['auction_id'])
recommendations.append({
'auction_id': auction['auction_id'],
'title': auction['title'],
'brand': auction['search_term'],
'price': auction['current_price'],
'deal_score': deal_score_data['deal_score'],
'grade': deal_score_data['grade'],
'investment_stars': investment_grade['investment_stars'] if investment_grade else 0,
'investment_grade': investment_grade['investment_grade'] if investment_grade else 'N/A',
'savings_percent': round(auction['savings_percent'], 2),
'auction_house': auction['auction_house'],
'url': auction['url']
})
# Sort by deal score
recommendations.sort(key=lambda x: x['deal_score'], reverse=True)
return recommendations[:20] # Top 20 recommendations
def main():
"""Test analytics engine"""
print("Initializing LUXVAULT Analytics Engine...")
analytics = AuctionAnalytics()
print(f"\nLoaded {len(analytics.df)} auctions")
# Test predictions
print("\n=== PRICE PREDICTIONS ===")
for brand in analytics.df['search_term'].unique()[:3]:
prediction = analytics.predict_price(brand)
if prediction:
print(f"{brand}: ${prediction['predicted_price']} (trend: {prediction['trend_percent_annual']}% annually)")
# Test undervalued
print("\n=== UNDERVALUED AUCTIONS ===")
undervalued = analytics.identify_undervalued_auctions(threshold_percent=15)
for item in undervalued[:5]:
print(f"{item['brand']}: ${item['current_price']} (${item['discount_amount']} below avg)")
# Test trends
print("\n=== TRENDING BRANDS ===")
trends = analytics.analyze_brand_trends(days=90)
for trend in trends[:5]:
print(f"{trend['brand']}: {trend['trend']} (momentum: {trend['momentum_score']})")
print("\nAnalytics engine test complete!")
if __name__ == "__main__":
main()