← back to Watches
analytics/report_generator.py
362 lines
#!/usr/bin/env python3
"""
Comprehensive Analytics Report Generator
Combines all analysis modules into executive summary
"""
import json
from datetime import datetime
from pathlib import Path
class AnalyticsReportGenerator:
"""Generate comprehensive analytics reports"""
def __init__(self, analytics_dir='/root/Projects/watches/analytics'):
self.analytics_dir = Path(analytics_dir)
self.reports = {}
def load_all_analyses(self):
"""Load all generated analysis files"""
files = {
'predictions': 'price_predictions.json',
'market': 'market_analysis.json',
'statistics': 'statistical_insights.json'
}
for key, filename in files.items():
filepath = self.analytics_dir / filename
if filepath.exists():
with open(filepath, 'r') as f:
self.reports[key] = json.load(f)
return self.reports
def generate_executive_summary(self):
"""Generate executive summary of all findings"""
summary = {
'report_generated': datetime.now().isoformat(),
'report_type': 'Omega Watch Investment Analysis - Executive Summary'
}
# Market overview
if 'market' in self.reports:
market = self.reports['market']
summary['market_overview'] = {
'total_watches_analyzed': market['summary']['total_watches_analyzed'],
'collections': market['summary']['collections'],
'average_cagr': market['summary']['avg_market_cagr'],
'median_price': market['summary']['median_market_price']
}
# Key findings from predictions
if 'predictions' in self.reports:
pred = self.reports['predictions']
summary['prediction_highlights'] = {
'model_performance': pred['model_performance']['best_model'],
'model_r2_score': pred['model_performance'].get(f"{pred['model_performance']['best_model']}_r2", 0),
'anomalies_detected': pred['summary']['anomalies_detected'],
'watches_with_predictions': pred['summary']['watches_with_predictions']
}
# Statistical insights
if 'statistics' in self.reports:
stats = self.reports['statistics']
summary['statistical_summary'] = stats.get('summary_statistics', {})
return summary
def generate_top_recommendations(self):
"""Generate top investment recommendations"""
recommendations = []
if 'market' in self.reports:
market = self.reports['market']
# Top opportunities
opportunities = market.get('investment_opportunities', [])[:5]
for opp in opportunities:
rec = {
'watch_id': opp['watch_id'],
'model': opp['model'],
'series': opp['series'],
'current_price': opp['current_price'],
'cagr': opp['cagr'],
'opportunity_score': opp['opportunity_score'],
'recommendation_type': 'High Growth Opportunity',
'key_reasons': opp['investment_reasons'][:3] # Top 3 reasons
}
# Add prediction if available
if 'predictions' in self.reports:
pred_data = self.reports['predictions']['future_predictions']
watch_pred = next((p for p in pred_data if p['watch_id'] == opp['watch_id']), None)
if watch_pred and '5_year' in watch_pred['predictions']:
rec['5_year_forecast'] = watch_pred['predictions']['5_year']['predicted_price']
recommendations.append(rec)
return recommendations
def generate_risk_analysis_summary(self):
"""Generate risk analysis summary"""
if 'statistics' not in self.reports:
return {}
risk_data = self.reports['statistics'].get('risk_metrics', [])
# Top risk-adjusted performers
top_sharpe = sorted(risk_data, key=lambda x: x['risk_adjusted_ranking'], reverse=True)[:5]
# High risk watches
high_risk = sorted(risk_data, key=lambda x: x['risk_metrics']['volatility'], reverse=True)[:5]
# Best max drawdown
best_drawdown = sorted(risk_data, key=lambda x: abs(x['risk_metrics']['max_drawdown']))[:5]
return {
'best_risk_adjusted': [{
'model': w['model'],
'sharpe_ratio': w['risk_metrics']['sharpe_ratio'],
'mean_return': w['risk_metrics']['mean_return'],
'volatility': w['risk_metrics']['volatility']
} for w in top_sharpe],
'highest_risk': [{
'model': w['model'],
'volatility': w['risk_metrics']['volatility'],
'max_drawdown': w['risk_metrics']['max_drawdown']
} for w in high_risk],
'most_stable': [{
'model': w['model'],
'max_drawdown': w['risk_metrics']['max_drawdown'],
'volatility': w['risk_metrics']['volatility']
} for w in best_drawdown]
}
def generate_collection_comparison(self):
"""Generate collection performance comparison"""
if 'market' not in self.reports:
return {}
collections = self.reports['market'].get('collection_analysis', [])
comparison = {
'top_performing_collections': sorted(collections, key=lambda x: x['avg_cagr'], reverse=True)[:3],
'most_expensive_collections': sorted(collections, key=lambda x: x['avg_current_price'], reverse=True)[:3],
'most_volatile_collections': sorted(collections, key=lambda x: x.get('volatility', 0), reverse=True)[:3],
'collection_count': len(collections)
}
return comparison
def generate_decade_insights(self):
"""Generate insights by decade"""
if 'market' not in self.reports:
return {}
decades = self.reports['market'].get('decade_analysis', [])
return {
'best_performing_decade': max(decades, key=lambda x: x['avg_cagr']) if decades else None,
'worst_performing_decade': min(decades, key=lambda x: x['avg_cagr']) if decades else None,
'all_decades': decades
}
def generate_value_opportunities(self):
"""Generate value investment opportunities"""
if 'market' not in self.reports:
return []
value_watches = self.reports['market'].get('value_watches', [])
opportunities = []
for watch in value_watches[:10]:
opp = {
'watch_id': watch['watch_id'],
'model': watch['model'],
'series': watch['series'],
'current_price': watch['current_price'],
'cagr': watch['cagr'],
'value_score': watch['value_score'],
'investment_thesis': f"High growth ({watch['cagr']:.1f}% CAGR) at below-median price (${watch['current_price']:,.0f})"
}
opportunities.append(opp)
return opportunities
def generate_statistical_highlights(self):
"""Generate key statistical highlights"""
if 'statistics' not in self.reports:
return {}
stats = self.reports['statistics']
highlights = {
'market_summary': stats.get('summary_statistics', {}),
'significant_tests': []
}
# Find significant hypothesis test results
tests = stats.get('hypothesis_tests', [])
for test in tests:
if test.get('significant'):
highlights['significant_tests'].append({
'comparison': test['comparison'],
'result': test['interpretation'],
'p_value': test['p_value'],
'effect_size': test['cohens_d']
})
return highlights
def generate_markdown_report(self):
"""Generate formatted markdown report"""
md = []
md.append("# Omega Watch Investment Analysis Report")
md.append(f"\nGenerated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
md.append("---\n")
# Executive Summary
summary = self.generate_executive_summary()
md.append("## Executive Summary\n")
if 'market_overview' in summary:
mo = summary['market_overview']
md.append(f"- **Total Watches Analyzed**: {mo['total_watches_analyzed']}")
md.append(f"- **Collections**: {mo['collections']}")
md.append(f"- **Market Average CAGR**: {mo['average_cagr']}%")
md.append(f"- **Median Market Price**: ${mo['median_price']:,.2f}\n")
# Top Recommendations
md.append("\n## Top Investment Recommendations\n")
recommendations = self.generate_top_recommendations()
for i, rec in enumerate(recommendations[:5], 1):
md.append(f"\n### {i}. {rec['model']}")
md.append(f"- **Series**: {rec['series']}")
md.append(f"- **Current Price**: ${rec['current_price']:,.2f}")
md.append(f"- **Historical CAGR**: {rec['cagr']:.2f}%")
md.append(f"- **Opportunity Score**: {rec['opportunity_score']}/100")
if '5_year_forecast' in rec:
md.append(f"- **5-Year Forecast**: ${rec['5_year_forecast']:,.2f}")
md.append(f"\n**Key Reasons**:")
for reason in rec['key_reasons']:
md.append(f" - {reason}")
# Risk Analysis
md.append("\n\n## Risk Analysis\n")
risk_summary = self.generate_risk_analysis_summary()
if 'best_risk_adjusted' in risk_summary:
md.append("\n### Best Risk-Adjusted Returns (Top 3)\n")
for i, watch in enumerate(risk_summary['best_risk_adjusted'][:3], 1):
md.append(f"{i}. **{watch['model']}**: Sharpe Ratio {watch['sharpe_ratio']:.4f}, "
f"Return {watch['mean_return']:.2f}%, Volatility {watch['volatility']:.2f}%")
# Collection Performance
md.append("\n\n## Collection Performance\n")
collections = self.generate_collection_comparison()
if 'top_performing_collections' in collections:
md.append("\n### Top Performing Collections\n")
for coll in collections['top_performing_collections']:
md.append(f"\n**{coll['series']}**")
md.append(f"- Average CAGR: {coll['avg_cagr']}%")
md.append(f"- Average Price: ${coll['avg_current_price']:,.2f}")
md.append(f"- Watch Count: {coll['watch_count']}")
# Decade Analysis
md.append("\n\n## Historical Performance by Decade\n")
decades = self.generate_decade_insights()
if decades.get('best_performing_decade'):
best = decades['best_performing_decade']
md.append(f"\n**Best Decade**: {best['decade']} - {best['avg_cagr']:.2f}% avg annual return")
if decades.get('worst_performing_decade'):
worst = decades['worst_performing_decade']
md.append(f"\n**Weakest Decade**: {worst['decade']} - {worst['avg_cagr']:.2f}% avg annual return")
# Value Opportunities
md.append("\n\n## Value Investment Opportunities\n")
md.append("\nWatches with strong growth potential at below-median prices:\n")
value_opps = self.generate_value_opportunities()
for i, opp in enumerate(value_opps[:5], 1):
md.append(f"\n{i}. **{opp['model']}** ({opp['series']})")
md.append(f" - Price: ${opp['current_price']:,.2f}")
md.append(f" - CAGR: {opp['cagr']:.2f}%")
md.append(f" - Value Score: {opp['value_score']:.1f}/100")
# Statistical Highlights
md.append("\n\n## Statistical Highlights\n")
stat_highlights = self.generate_statistical_highlights()
if 'market_summary' in stat_highlights and 'cagr' in stat_highlights['market_summary']:
cagr_stats = stat_highlights['market_summary']['cagr']
md.append(f"\n**Market CAGR Distribution**:")
md.append(f"- Mean: {cagr_stats['mean']}%")
md.append(f"- Median: {cagr_stats['median']}%")
md.append(f"- Range: {cagr_stats['min']}% to {cagr_stats['max']}%")
md.append(f"- 90th Percentile: {cagr_stats['percentiles']['90th']}%")
md.append("\n\n---")
md.append("\n*This report is generated from statistical analysis of historical Omega watch prices.*")
md.append("\n*Past performance does not guarantee future results.*")
return '\n'.join(md)
def generate_full_report(self):
"""Generate complete comprehensive report"""
self.load_all_analyses()
report = {
'generated_at': datetime.now().isoformat(),
'executive_summary': self.generate_executive_summary(),
'top_recommendations': self.generate_top_recommendations(),
'risk_analysis': self.generate_risk_analysis_summary(),
'collection_comparison': self.generate_collection_comparison(),
'decade_insights': self.generate_decade_insights(),
'value_opportunities': self.generate_value_opportunities(),
'statistical_highlights': self.generate_statistical_highlights()
}
# Save JSON report
json_path = self.analytics_dir / 'comprehensive_report.json'
with open(json_path, 'w') as f:
json.dump(report, f, indent=2)
# Save Markdown report
markdown_report = self.generate_markdown_report()
md_path = self.analytics_dir / 'INVESTMENT_REPORT.md'
with open(md_path, 'w') as f:
f.write(markdown_report)
return {
'json_report': str(json_path),
'markdown_report': str(md_path),
'report_data': report
}
def main():
"""Generate comprehensive report"""
print("=" * 80)
print("GENERATING COMPREHENSIVE ANALYTICS REPORT")
print("=" * 80)
generator = AnalyticsReportGenerator()
print("\n[1/2] Loading all analysis results...")
generator.load_all_analyses()
print(f" Loaded {len(generator.reports)} analysis reports")
print("\n[2/2] Generating comprehensive report...")
result = generator.generate_full_report()
print(f"\n" + "=" * 80)
print("Report generation complete!")
print(f"JSON Report: {result['json_report']}")
print(f"Markdown Report: {result['markdown_report']}")
print("=" * 80)
return result
if __name__ == '__main__':
result = main()