← back to Handbag Auth Nextjs
auction-viewer/analytics-engine.py
499 lines
#!/usr/bin/env python3
"""
LUXVAULT Analytics Engine
Advanced statistical analysis and predictive modeling for auction data
"""
import sqlite3
import json
import sys
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
import statistics
import math
class AuctionAnalytics:
"""Advanced analytics engine for auction data"""
def __init__(self, db_path: str):
self.db_path = db_path
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
def __del__(self):
if hasattr(self, 'conn'):
self.conn.close()
def calculate_deal_score(self, auction: Dict) -> float:
"""
Calculate Deal Score (0-100) based on multiple factors:
- Savings percentage (40% weight)
- Price point relative to brand average (25% weight)
- Time urgency (20% weight)
- Historical price performance (15% weight)
"""
score = 0.0
# Factor 1: Savings Percentage (40 points max)
if auction['estimate_low'] > 0 and auction['current_price'] > 0:
savings_pct = ((auction['estimate_low'] - auction['current_price']) / auction['estimate_low']) * 100
# Cap at 100% savings, score linearly up to 40 points
savings_score = min(40, (savings_pct / 100) * 40)
score += max(0, savings_score) # No negative scores
# Factor 2: Price Point Relative to Brand Average (25 points max)
brand_avg = self._get_brand_average_price(auction['search_term'])
if brand_avg > 0 and auction['current_price'] > 0:
price_ratio = auction['current_price'] / brand_avg
# Lower prices relative to average score higher
if price_ratio <= 0.5:
score += 25
elif price_ratio <= 1.0:
score += 25 * (1 - (price_ratio - 0.5) * 2)
# else: 0 points if above average
# Factor 3: Time Urgency (20 points max)
if auction['end_date']:
try:
end_date = datetime.strptime(auction['end_date'][:10], '%Y-%m-%d')
days_remaining = (end_date - datetime.now()).days
if days_remaining < 0:
urgency_score = 0
elif days_remaining <= 1:
urgency_score = 20
elif days_remaining <= 3:
urgency_score = 15
elif days_remaining <= 7:
urgency_score = 10
elif days_remaining <= 14:
urgency_score = 5
else:
urgency_score = 2
score += urgency_score
except:
score += 5 # Default mid-range if date parsing fails
# Factor 4: Historical Price Performance (15 points max)
price_percentile = self._get_price_percentile(auction)
score += price_percentile * 15 / 100
return round(min(100, max(0, score)), 2)
def _get_brand_average_price(self, brand: str) -> float:
"""Get average current price for a brand"""
cursor = self.conn.cursor()
result = cursor.execute(
"SELECT AVG(current_price) as avg FROM auctions WHERE search_term = ? AND current_price > 0",
(brand,)
).fetchone()
return result['avg'] if result['avg'] else 0
def _get_price_percentile(self, auction: Dict) -> float:
"""Calculate what percentile this price is in (lower = better deal)"""
cursor = self.conn.cursor()
# Get all prices for this brand
prices = cursor.execute(
"SELECT current_price FROM auctions WHERE search_term = ? AND current_price > 0 ORDER BY current_price",
(auction['search_term'],)
).fetchall()
if not prices or auction['current_price'] <= 0:
return 50 # Default to median
prices_list = [p['current_price'] for p in prices]
position = sum(1 for p in prices_list if p < auction['current_price'])
percentile = (position / len(prices_list)) * 100
# Invert so lower prices get higher scores
return 100 - percentile
def get_top_deals(self, limit: int = 50) -> List[Dict]:
"""Get top deals by Deal Score"""
cursor = self.conn.cursor()
auctions = cursor.execute("""
SELECT * FROM auctions
WHERE current_price > 0 AND estimate_low > 0
ORDER BY created_at DESC
LIMIT 500
""").fetchall()
deals = []
for auction in auctions:
auction_dict = dict(auction)
deal_score = self.calculate_deal_score(auction_dict)
# Calculate savings
savings_pct = ((auction['estimate_low'] - auction['current_price']) / auction['estimate_low']) * 100
savings_amt = auction['estimate_low'] - auction['current_price']
deals.append({
**auction_dict,
'deal_score': deal_score,
'savings_percent': round(savings_pct, 2),
'savings_amount': round(savings_amt, 2)
})
# Sort by deal score
deals.sort(key=lambda x: x['deal_score'], reverse=True)
return deals[:limit]
def analyze_price_trends(self) -> Dict:
"""Analyze price trends by brand over time"""
cursor = self.conn.cursor()
# Get monthly price trends by brand
trends_query = """
SELECT
search_term,
strftime('%Y-%m', end_date) as month,
AVG(current_price) as avg_price,
MIN(current_price) as min_price,
MAX(current_price) as max_price,
COUNT(*) as auction_count,
AVG(CASE
WHEN estimate_low > 0 AND current_price > 0
THEN ((estimate_low - current_price) / estimate_low * 100)
ELSE 0
END) as avg_savings_pct
FROM auctions
WHERE current_price > 0 AND end_date IS NOT NULL
GROUP BY search_term, month
ORDER BY month DESC, search_term
"""
results = cursor.execute(trends_query).fetchall()
# Organize by brand
trends_by_brand = {}
for row in results:
brand = row['search_term']
if brand not in trends_by_brand:
trends_by_brand[brand] = []
trends_by_brand[brand].append({
'month': row['month'],
'avg_price': round(row['avg_price'], 2),
'min_price': round(row['min_price'], 2),
'max_price': round(row['max_price'], 2),
'auction_count': row['auction_count'],
'avg_savings_pct': round(row['avg_savings_pct'], 2)
})
# Calculate trend direction (last 3 months)
trend_summary = {}
for brand, data in trends_by_brand.items():
if len(data) >= 2:
recent_prices = [d['avg_price'] for d in data[:3]]
if len(recent_prices) >= 2:
trend = 'increasing' if recent_prices[0] > recent_prices[-1] else 'decreasing'
change_pct = ((recent_prices[0] - recent_prices[-1]) / recent_prices[-1]) * 100
else:
trend = 'stable'
change_pct = 0
else:
trend = 'insufficient_data'
change_pct = 0
trend_summary[brand] = {
'trend': trend,
'change_percent': round(change_pct, 2),
'monthly_data': data[:12] # Last 12 months
}
return trend_summary
def detect_price_anomalies(self) -> List[Dict]:
"""Detect price outliers using statistical methods"""
cursor = self.conn.cursor()
# Get all auctions with price data
auctions = cursor.execute("""
SELECT * FROM auctions
WHERE current_price > 0 AND estimate_low > 0
""").fetchall()
# Group by brand and detect outliers
anomalies = []
brands = {}
for auction in auctions:
brand = auction['search_term']
if brand not in brands:
brands[brand] = []
brands[brand].append(dict(auction))
for brand, items in brands.items():
if len(items) < 10: # Need sufficient data
continue
prices = [item['current_price'] for item in items]
mean_price = statistics.mean(prices)
if len(prices) >= 2:
stdev_price = statistics.stdev(prices)
else:
continue
# Detect outliers (more than 2 standard deviations)
for item in items:
z_score = (item['current_price'] - mean_price) / stdev_price if stdev_price > 0 else 0
if abs(z_score) > 2:
anomaly_type = 'unusually_high' if z_score > 0 else 'unusually_low'
anomalies.append({
**item,
'anomaly_type': anomaly_type,
'z_score': round(z_score, 2),
'brand_avg_price': round(mean_price, 2),
'deviation_pct': round(((item['current_price'] - mean_price) / mean_price) * 100, 2)
})
# Sort by absolute z-score
anomalies.sort(key=lambda x: abs(x['z_score']), reverse=True)
return anomalies[:50]
def predict_final_price(self, auction_id: str) -> Dict:
"""Predict final auction price based on historical patterns"""
cursor = self.conn.cursor()
# Get the auction
auction = cursor.execute(
"SELECT * FROM auctions WHERE auction_id = ?", (auction_id,)
).fetchone()
if not auction:
return {'error': 'Auction not found'}
auction_dict = dict(auction)
brand = auction['search_term']
# Get historical data for this brand
historical = cursor.execute("""
SELECT current_price, estimate_low, estimate_high
FROM auctions
WHERE search_term = ? AND current_price > 0 AND estimate_low > 0
""", (brand,)).fetchall()
if len(historical) < 5:
return {
'auction_id': auction_id,
'predicted_price': auction['estimate_low'],
'confidence': 'low',
'method': 'estimate_fallback'
}
# Calculate average ratio of final price to estimate
ratios = []
for hist in historical:
if hist['estimate_low'] > 0:
ratio = hist['current_price'] / hist['estimate_low']
ratios.append(ratio)
if not ratios:
return {
'auction_id': auction_id,
'predicted_price': auction['estimate_low'],
'confidence': 'low',
'method': 'estimate_fallback'
}
avg_ratio = statistics.mean(ratios)
median_ratio = statistics.median(ratios)
# Predict using median ratio (more robust to outliers)
predicted = auction['estimate_low'] * median_ratio
# Calculate confidence interval
if len(ratios) >= 2:
stdev = statistics.stdev(ratios)
confidence_range = predicted * stdev
else:
confidence_range = predicted * 0.2
return {
'auction_id': auction_id,
'current_price': auction['current_price'],
'estimate_low': auction['estimate_low'],
'estimate_high': auction['estimate_high'],
'predicted_final_price': round(predicted, 2),
'prediction_range_low': round(predicted - confidence_range, 2),
'prediction_range_high': round(predicted + confidence_range, 2),
'confidence': 'high' if len(historical) >= 20 else 'medium',
'sample_size': len(historical)
}
def generate_recommendations(self, user_budget: float = None, preferred_brands: List[str] = None) -> List[Dict]:
"""Generate personalized auction recommendations"""
deals = self.get_top_deals(limit=100)
# Filter by budget if provided
if user_budget:
deals = [d for d in deals if d['current_price'] <= user_budget]
# Filter by preferred brands if provided
if preferred_brands:
deals = [d for d in deals if d['search_term'] in preferred_brands]
# Add recommendation score and reasons
for deal in deals:
reasons = []
if deal['deal_score'] >= 80:
reasons.append('Exceptional deal score')
elif deal['deal_score'] >= 60:
reasons.append('Great value opportunity')
if deal['savings_percent'] >= 50:
reasons.append(f"{deal['savings_percent']:.0f}% below estimate")
elif deal['savings_percent'] >= 25:
reasons.append(f"{deal['savings_percent']:.0f}% savings")
# Check time urgency
if deal['end_date']:
try:
end_date = datetime.strptime(deal['end_date'][:10], '%Y-%m-%d')
days_remaining = (end_date - datetime.now()).days
if 0 < days_remaining <= 3:
reasons.append(f'Ends in {days_remaining} days')
except:
pass
deal['recommendation_reasons'] = reasons
return deals[:20]
def calculate_market_insights(self) -> Dict:
"""Calculate comprehensive market insights"""
cursor = self.conn.cursor()
# Overall market statistics
overall_stats = cursor.execute("""
SELECT
COUNT(*) as total_auctions,
AVG(current_price) as avg_price,
AVG(CASE
WHEN estimate_low > 0 AND current_price > 0
THEN ((estimate_low - current_price) / estimate_low * 100)
ELSE 0
END) as avg_savings_pct,
COUNT(DISTINCT search_term) as unique_brands,
COUNT(DISTINCT auction_house) as unique_houses
FROM auctions
WHERE current_price > 0
""").fetchone()
# Price distribution by range
price_distribution = cursor.execute("""
SELECT
CASE
WHEN current_price < 1000 THEN 'Under $1K'
WHEN current_price < 5000 THEN '$1K-$5K'
WHEN current_price < 10000 THEN '$5K-$10K'
WHEN current_price < 20000 THEN '$10K-$20K'
ELSE 'Over $20K'
END as price_range,
COUNT(*) as count
FROM auctions
WHERE current_price > 0
GROUP BY price_range
""").fetchall()
# Top performing brands (best average savings)
top_brands = cursor.execute("""
SELECT
search_term,
COUNT(*) as auction_count,
AVG(current_price) as avg_price,
AVG(CASE
WHEN estimate_low > 0 AND current_price > 0
THEN ((estimate_low - current_price) / estimate_low * 100)
ELSE 0
END) as avg_savings_pct
FROM auctions
WHERE current_price > 0 AND estimate_low > 0
GROUP BY search_term
HAVING COUNT(*) >= 10
ORDER BY avg_savings_pct DESC
LIMIT 10
""").fetchall()
return {
'market_summary': {
'total_auctions': overall_stats['total_auctions'],
'average_price': round(overall_stats['avg_price'], 2),
'average_savings_percent': round(overall_stats['avg_savings_pct'], 2),
'unique_brands': overall_stats['unique_brands'],
'unique_auction_houses': overall_stats['unique_houses']
},
'price_distribution': [
{'range': row['price_range'], 'count': row['count']}
for row in price_distribution
],
'top_value_brands': [
{
'brand': row['search_term'],
'auction_count': row['auction_count'],
'avg_price': round(row['avg_price'], 2),
'avg_savings_percent': round(row['avg_savings_pct'], 2)
}
for row in top_brands
]
}
def main():
"""Main function for CLI usage"""
if len(sys.argv) < 3:
print(json.dumps({'error': 'Usage: analytics-engine.py <db_path> <command> [args]'}))
sys.exit(1)
db_path = sys.argv[1]
command = sys.argv[2]
try:
analytics = AuctionAnalytics(db_path)
if command == 'deal-scores':
limit = int(sys.argv[3]) if len(sys.argv) > 3 else 50
result = analytics.get_top_deals(limit)
print(json.dumps(result, default=str))
elif command == 'trends':
result = analytics.analyze_price_trends()
print(json.dumps(result, default=str))
elif command == 'anomalies':
result = analytics.detect_price_anomalies()
print(json.dumps(result, default=str))
elif command == 'predict':
if len(sys.argv) < 4:
print(json.dumps({'error': 'Auction ID required'}))
sys.exit(1)
auction_id = sys.argv[3]
result = analytics.predict_final_price(auction_id)
print(json.dumps(result, default=str))
elif command == 'recommendations':
budget = float(sys.argv[3]) if len(sys.argv) > 3 else None
result = analytics.generate_recommendations(user_budget=budget)
print(json.dumps(result, default=str))
elif command == 'market-insights':
result = analytics.calculate_market_insights()
print(json.dumps(result, default=str))
else:
print(json.dumps({'error': f'Unknown command: {command}'}))
sys.exit(1)
except Exception as e:
print(json.dumps({'error': str(e)}))
sys.exit(1)
if __name__ == '__main__':
main()