← back to Handbag Auth Nextjs
analytics/analytics_api.py
697 lines
#!/usr/bin/env python3
"""
LUXVAULT Analytics API Server
Serves advanced analytics endpoints for auction insights
"""
from flask import Flask, jsonify, request
from flask_cors import CORS
from analytics_engine import AuctionAnalytics
from visualizations import AnalyticsVisualizations
import json
import os
app = Flask(__name__)
CORS(app)
# Initialize analytics engine
analytics = None
visualizations = None
def get_analytics():
"""Get or initialize analytics engine"""
global analytics
if analytics is None:
analytics = AuctionAnalytics()
return analytics
def get_visualizations():
"""Get or initialize visualizations engine"""
global visualizations
if visualizations is None:
visualizations = AnalyticsVisualizations()
return visualizations
# =========================
# PREDICTIVE ANALYTICS
# =========================
@app.route('/api/analytics/predictions', methods=['GET'])
def get_predictions():
"""
Get price predictions for brands
Query params:
- brand: specific brand (optional)
- days_ahead: prediction timeframe (default: 30)
"""
try:
engine = get_analytics()
brand = request.args.get('brand')
days_ahead = int(request.args.get('days_ahead', 30))
if brand:
prediction = engine.predict_price(brand, days_ahead)
return jsonify({
'success': True,
'prediction': prediction
})
else:
# Get predictions for all brands
predictions = []
for b in engine.df['search_term'].unique():
pred = engine.predict_price(b, days_ahead)
if pred:
predictions.append(pred)
predictions.sort(key=lambda x: x['trend_percent_annual'], reverse=True)
return jsonify({
'success': True,
'predictions': predictions
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/undervalued', methods=['GET'])
def get_undervalued():
"""
Get undervalued auctions
Query params:
- threshold: discount threshold percent (default: 20)
- limit: max results (default: 50)
"""
try:
engine = get_analytics()
threshold = float(request.args.get('threshold', 20))
limit = int(request.args.get('limit', 50))
undervalued = engine.identify_undervalued_auctions(threshold_percent=threshold)
return jsonify({
'success': True,
'count': len(undervalued),
'undervalued': undervalued[:limit]
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/optimal-bid', methods=['GET'])
def get_optimal_bid():
"""
Get optimal bid suggestions for a brand
Query params:
- brand: brand name (required)
"""
try:
brand = request.args.get('brand')
if not brand:
return jsonify({'success': False, 'error': 'Brand parameter required'}), 400
engine = get_analytics()
bid_data = engine.predict_optimal_bid(brand)
return jsonify({
'success': True,
'bid_guidance': bid_data
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/roi-potential', methods=['GET'])
def get_roi_potential():
"""
Calculate ROI potential for an auction
Query params:
- auction_id: auction identifier (required)
"""
try:
auction_id = request.args.get('auction_id')
if not auction_id:
return jsonify({'success': False, 'error': 'auction_id parameter required'}), 400
engine = get_analytics()
roi_data = engine.calculate_roi_potential(auction_id)
return jsonify({
'success': True,
'roi_analysis': roi_data
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
# =========================
# TREND ANALYSIS
# =========================
@app.route('/api/analytics/trends', methods=['GET'])
def get_trends():
"""
Get brand trends analysis
Query params:
- days: lookback period (default: 90)
- limit: max results (default: 20)
"""
try:
engine = get_analytics()
days = int(request.args.get('days', 90))
limit = int(request.args.get('limit', 20))
trends = engine.analyze_brand_trends(days=days)
return jsonify({
'success': True,
'period_days': days,
'trends': trends[:limit]
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/price-movement', methods=['GET'])
def get_price_movement():
"""
Get price movement over time
Query params:
- brand: specific brand (optional)
"""
try:
engine = get_analytics()
brand = request.args.get('brand')
movement = engine.price_movement_by_brand(brand=brand)
return jsonify({
'success': True,
'brand': brand if brand else 'all',
'price_movement': movement
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/seasonality', methods=['GET'])
def get_seasonality():
"""
Get seasonal patterns for a brand
Query params:
- brand: brand name (required)
"""
try:
brand = request.args.get('brand')
if not brand:
return jsonify({'success': False, 'error': 'Brand parameter required'}), 400
engine = get_analytics()
patterns = engine.detect_seasonal_patterns(brand)
return jsonify({
'success': True,
'brand': brand,
'seasonal_patterns': patterns
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/auction-houses', methods=['GET'])
def get_auction_houses():
"""Get auction house performance comparison"""
try:
engine = get_analytics()
performance = engine.auction_house_performance()
return jsonify({
'success': True,
'auction_houses': performance
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
# =========================
# VALUE SCORING
# =========================
@app.route('/api/analytics/deal-score', methods=['GET'])
def get_deal_score():
"""
Get composite deal score for an auction
Query params:
- auction_id: auction identifier (required)
"""
try:
auction_id = request.args.get('auction_id')
if not auction_id:
return jsonify({'success': False, 'error': 'auction_id parameter required'}), 400
engine = get_analytics()
score = engine.calculate_deal_score(auction_id)
return jsonify({
'success': True,
'deal_score': score
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/investment-grade', methods=['GET'])
def get_investment_grade():
"""
Get investment grade rating for an auction
Query params:
- auction_id: auction identifier (required)
"""
try:
auction_id = request.args.get('auction_id')
if not auction_id:
return jsonify({'success': False, 'error': 'auction_id parameter required'}), 400
engine = get_analytics()
grade = engine.calculate_investment_grade(auction_id)
return jsonify({
'success': True,
'investment_grade': grade
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/top-deals', methods=['GET'])
def get_top_deals():
"""
Get top-rated deals with comprehensive scoring
Query params:
- limit: max results (default: 20)
- min_score: minimum deal score (default: 50)
"""
try:
engine = get_analytics()
limit = int(request.args.get('limit', 20))
min_score = float(request.args.get('min_score', 50))
# Score all auctions
deals = []
for auction_id in engine.df['auction_id'].unique():
score_data = engine.calculate_deal_score(auction_id)
if score_data['deal_score'] >= min_score:
auction = engine.df[engine.df['auction_id'] == auction_id].iloc[0]
investment = engine.calculate_investment_grade(auction_id)
deals.append({
'auction_id': auction_id,
'title': auction['title'],
'brand': auction['search_term'],
'price': float(auction['current_price']),
'deal_score': score_data['deal_score'],
'grade': score_data['grade'],
'investment_stars': investment['investment_stars'] if investment else 0,
'savings_percent': float(auction['savings_percent']),
'auction_house': auction['auction_house'],
'url': auction['url']
})
deals.sort(key=lambda x: x['deal_score'], reverse=True)
return jsonify({
'success': True,
'count': len(deals),
'deals': deals[:limit]
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
# =========================
# REPORTS & INSIGHTS
# =========================
@app.route('/api/analytics/reports/daily', methods=['GET'])
def get_daily_report():
"""Generate daily insights report"""
try:
engine = get_analytics()
report = engine.generate_daily_report()
return jsonify({
'success': True,
'report': report
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/reports/brand-comparison', methods=['GET'])
def get_brand_comparison():
"""Generate brand comparison report"""
try:
engine = get_analytics()
comparison = engine.generate_brand_comparison()
return jsonify({
'success': True,
'brand_comparison': comparison
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/recommendations', methods=['GET'])
def get_recommendations():
"""
Get personalized deal recommendations
Query params:
- max_budget: maximum price (optional)
- brands: comma-separated brand list (optional)
- min_score: minimum deal score (default: 50)
"""
try:
engine = get_analytics()
max_budget = request.args.get('max_budget')
if max_budget:
max_budget = float(max_budget)
brands = request.args.get('brands')
preferred_brands = brands.split(',') if brands else None
min_score = float(request.args.get('min_score', 50))
recommendations = engine.get_personalized_recommendations(
max_budget=max_budget,
preferred_brands=preferred_brands,
min_deal_score=min_score
)
return jsonify({
'success': True,
'count': len(recommendations),
'recommendations': recommendations,
'filters': {
'max_budget': max_budget,
'brands': preferred_brands,
'min_score': min_score
}
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
# =========================
# SYSTEM
# =========================
@app.route('/api/analytics/status', methods=['GET'])
def get_status():
"""Get analytics system status"""
try:
engine = get_analytics()
return jsonify({
'success': True,
'status': {
'total_auctions': len(engine.df),
'brands': len(engine.df['search_term'].unique()),
'auction_houses': len(engine.df['auction_house'].unique()),
'date_range': {
'start': str(engine.df['end_date'].min().date()),
'end': str(engine.df['end_date'].max().date())
},
'price_range': {
'min': float(engine.df['current_price'].min()),
'max': float(engine.df['current_price'].max()),
'avg': float(engine.df['current_price'].mean())
}
}
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/refresh', methods=['POST'])
def refresh_data():
"""Refresh analytics data from database"""
try:
global analytics
analytics = AuctionAnalytics()
return jsonify({
'success': True,
'message': 'Analytics data refreshed',
'total_auctions': len(analytics.df)
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
# =========================
# VISUALIZATIONS
# =========================
@app.route('/api/analytics/charts/price-trends', methods=['GET'])
def get_price_trends_chart():
"""Get price trend chart data"""
try:
viz = get_visualizations()
brand = request.args.get('brand')
months = int(request.args.get('months', 12))
chart_data = viz.price_trend_chart(brand=brand, months=months)
return jsonify({
'success': True,
'chart_data': chart_data
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/charts/brand-volume', methods=['GET'])
def get_brand_volume_chart():
"""Get brand volume bar chart data"""
try:
viz = get_visualizations()
top_n = int(request.args.get('top_n', 10))
chart_data = viz.brand_volume_chart(top_n=top_n)
return jsonify({
'success': True,
'chart_data': chart_data
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/charts/price-distribution', methods=['GET'])
def get_price_distribution_chart():
"""Get price distribution histogram"""
try:
viz = get_visualizations()
brand = request.args.get('brand')
bins = int(request.args.get('bins', 20))
chart_data = viz.price_distribution_histogram(brand=brand, bins=bins)
return jsonify({
'success': True,
'chart_data': chart_data
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/charts/savings-scatter', methods=['GET'])
def get_savings_scatter_chart():
"""Get savings scatter plot data"""
try:
viz = get_visualizations()
sample_size = int(request.args.get('sample_size', 200))
chart_data = viz.savings_scatter_plot(sample_size=sample_size)
return jsonify({
'success': True,
'chart_data': chart_data
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/charts/brand-radar', methods=['GET'])
def get_brand_radar_chart():
"""Get brand comparison radar chart"""
try:
viz = get_visualizations()
brands = request.args.get('brands')
if brands:
brands = brands.split(',')
chart_data = viz.brand_comparison_radar(brands=brands)
return jsonify({
'success': True,
'chart_data': chart_data
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/charts/auction-house-pie', methods=['GET'])
def get_auction_house_pie():
"""Get auction house pie chart"""
try:
viz = get_visualizations()
chart_data = viz.auction_house_pie()
return jsonify({
'success': True,
'chart_data': chart_data
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/charts/deal-scores', methods=['GET'])
def get_deal_scores_chart():
"""Get deal score distribution chart"""
try:
viz = get_visualizations()
chart_data = viz.deal_score_distribution()
return jsonify({
'success': True,
'chart_data': chart_data
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/charts/roi-projection', methods=['GET'])
def get_roi_projection_chart():
"""Get ROI projection chart for a brand"""
try:
brand = request.args.get('brand')
if not brand:
return jsonify({'success': False, 'error': 'Brand parameter required'}), 400
viz = get_visualizations()
chart_data = viz.roi_projection_chart(brand=brand)
return jsonify({
'success': True,
'chart_data': chart_data
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/charts/trending-timeline', methods=['GET'])
def get_trending_timeline_chart():
"""Get trending brands timeline chart"""
try:
viz = get_visualizations()
days = int(request.args.get('days', 180))
chart_data = viz.trending_brands_timeline(days=days)
return jsonify({
'success': True,
'chart_data': chart_data
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/analytics/charts/all', methods=['GET'])
def get_all_charts():
"""Get all chart data at once"""
try:
viz = get_visualizations()
all_charts = viz.generate_all_charts()
return jsonify({
'success': True,
'charts': all_charts
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/', methods=['GET'])
def index():
"""API documentation"""
endpoints = {
'Predictive Analytics': {
'/api/analytics/predictions': 'Price predictions for brands',
'/api/analytics/undervalued': 'Identify undervalued auctions',
'/api/analytics/optimal-bid': 'Optimal bid guidance',
'/api/analytics/roi-potential': 'ROI potential analysis'
},
'Trend Analysis': {
'/api/analytics/trends': 'Brand trends and momentum',
'/api/analytics/price-movement': 'Price movement over time',
'/api/analytics/seasonality': 'Seasonal patterns',
'/api/analytics/auction-houses': 'Auction house performance'
},
'Value Scoring': {
'/api/analytics/deal-score': 'Composite deal score',
'/api/analytics/investment-grade': 'Investment grade rating',
'/api/analytics/top-deals': 'Top-rated deals'
},
'Reports & Insights': {
'/api/analytics/reports/daily': 'Daily insights report',
'/api/analytics/reports/brand-comparison': 'Brand comparison',
'/api/analytics/recommendations': 'Personalized recommendations'
},
'Visualizations': {
'/api/analytics/charts/price-trends': 'Price trend charts',
'/api/analytics/charts/brand-volume': 'Brand volume bar chart',
'/api/analytics/charts/price-distribution': 'Price distribution histogram',
'/api/analytics/charts/savings-scatter': 'Savings scatter plot',
'/api/analytics/charts/brand-radar': 'Brand comparison radar',
'/api/analytics/charts/auction-house-pie': 'Auction house pie chart',
'/api/analytics/charts/deal-scores': 'Deal score distribution',
'/api/analytics/charts/roi-projection': 'ROI projection chart',
'/api/analytics/charts/trending-timeline': 'Trending timeline',
'/api/analytics/charts/all': 'All charts at once'
},
'System': {
'/api/analytics/status': 'System status',
'/api/analytics/refresh': 'Refresh data (POST)'
}
}
return jsonify({
'name': 'LUXVAULT Analytics API',
'version': '1.0.0',
'endpoints': endpoints
})
def main():
"""Start analytics API server"""
port = int(os.environ.get('PORT', 7501))
print("=" * 60)
print("LUXVAULT Advanced Analytics API")
print("=" * 60)
print(f"Server: http://45.61.58.125:{port}")
print(f"API Docs: http://45.61.58.125:{port}/")
print("=" * 60)
print("\nInitializing analytics engine...")
# Pre-load analytics
get_analytics()
print(f"Loaded {len(analytics.df)} auctions for analysis")
print("\nServer ready!")
app.run(host='0.0.0.0', port=port, debug=False)
if __name__ == '__main__':
main()