154 lines
6.8 KiB
Python
154 lines
6.8 KiB
Python
"""
|
|||
|
|
exporters/excel_workbook.py
|
||
|
|
Generates an executive Excel audit workbook containing:
|
||
|
|
1. Sheet 1: 'Executive_Summary_&_Governance' (Methodology, sources, zero-imputation policy).
|
||
|
|
2. Sheet 2: 'Commercial_Sales_Catalog' (Verified CSPC metrics, EAS units, provenance URLs).
|
||
|
|
3. Sheet 3: 'Simulated_Rerankings' (Side-by-side Published vs Simulated ranks and deltas).
|
||
|
|
4. Sheet 4: 'Sensitivity_Analysis' (Rankings across 0%, 5%, 10%, 15%, 20% weights).
|
||
|
|
"""
|
||
|
|
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
import io
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Optional
|
||
|
|
import openpyxl
|
||
|
|
from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
|
||
|
|
import pandas as pd
|
||
|
|
|
||
|
|
from core.cpi_engine import CommercialPowerEngine
|
||
|
|
|
||
|
|
|
||
|
|
def generate_executive_sales_workbook(
|
||
|
|
df: pd.DataFrame,
|
||
|
|
commercial_weight: float = 0.15,
|
||
|
|
mode: str = "matched_only",
|
||
|
|
) -> bytes:
|
||
|
|
"""Generates an in-memory, styled multi-tab Excel workbook."""
|
||
|
|
engine = CommercialPowerEngine()
|
||
|
|
|
||
|
|
df_copy = df.copy()
|
||
|
|
if "commercial_power_index" not in df_copy.columns:
|
||
|
|
df_copy["commercial_power_index"] = engine.calculate_cpi(df_copy)
|
||
|
|
|
||
|
|
sim_df = engine.simulate_ranking(df_copy, commercial_weight=commercial_weight, mode=mode)
|
||
|
|
sensitivity_df = engine.run_sensitivity_matrix(df_copy, mode=mode)
|
||
|
|
metrics = engine.compute_simulation_metrics(sim_df)
|
||
|
|
|
||
|
|
output = io.BytesIO()
|
||
|
|
with pd.ExcelWriter(output, engine="openpyxl") as writer:
|
||
|
|
meta_data = {
|
||
|
|
"Parameter / Metric": [
|
||
|
|
"Report Title",
|
||
|
|
"Generated At (UTC)",
|
||
|
|
"Telemetry Source",
|
||
|
|
"Source Methodology",
|
||
|
|
"Unit of Measurement",
|
||
|
|
"Zero-Imputation Policy",
|
||
|
|
"Total Monitored Cohort",
|
||
|
|
"Verified Cataloged Artists",
|
||
|
|
"Coverage Ratio",
|
||
|
|
"Simulation Mode",
|
||
|
|
"Commercial Signal Weight",
|
||
|
|
"Top Gainer (Rank Delta)",
|
||
|
|
"Top Loser (Rank Delta)",
|
||
|
|
"Mean Absolute Rank Shift",
|
||
|
|
],
|
||
|
|
"Value": [
|
||
|
|
"HIFFI Commercial Record Sales & Simulation Audit",
|
||
|
|
datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
||
|
|
"ChartMasters (Best Selling Artists of All Time)",
|
||
|
|
"CSPC (Commensurate Sales to Popularity Concept)",
|
||
|
|
"EAS (Equivalent Album Sales)",
|
||
|
|
"Enforced (Strict NaN retention; 0 synthetic zeroes)",
|
||
|
|
len(df),
|
||
|
|
metrics["matched_coverage"],
|
||
|
|
f"{metrics['matched_coverage'] / len(df):.1%}",
|
||
|
|
mode.replace("_", " ").title(),
|
||
|
|
f"{int(commercial_weight * 100)}% Commercial / {int((1 - commercial_weight) * 100)}% Baseline",
|
||
|
|
f"{metrics['top_gainer_name']} (▲ +{metrics['top_gainer_delta']})",
|
||
|
|
f"{metrics['top_loser_name']} (▼ {metrics['top_loser_delta']})",
|
||
|
|
f"± {metrics['mean_abs_delta']} positions",
|
||
|
|
],
|
||
|
|
}
|
||
|
|
pd.DataFrame(meta_data).to_excel(writer, sheet_name="Executive_Summary", index=False)
|
||
|
|
|
||
|
|
sales_cols = {
|
||
|
|
"cohort_sales_rank": "Cohort Sales Rank",
|
||
|
|
"chartmasters_sales_rank": "ChartMasters Global Rank",
|
||
|
|
"name": "Artist Name",
|
||
|
|
"total_eas_cspc": "Total Equivalent Album Sales (EAS)",
|
||
|
|
"pure_album_sales": "Traditional Album Sales",
|
||
|
|
"physical_singles_sales": "Physical Singles Sales",
|
||
|
|
"digital_singles_sales": "Digital Track Downloads",
|
||
|
|
"streaming_eas": "Streaming-Equivalent Album Sales (EAS)",
|
||
|
|
"commercial_power_index": "Commercial Power Index (0-100)",
|
||
|
|
"sales_data_status": "Provenance Status",
|
||
|
|
"sales_collected_at": "Collected Timestamp",
|
||
|
|
"sales_source_url": "Source Audit URL",
|
||
|
|
}
|
||
|
|
matched_catalog = (
|
||
|
|
df_copy[df_copy["sales_data_status"] == "VERIFIED_CSPC"]
|
||
|
|
.sort_values(by="total_eas_cspc", ascending=False)
|
||
|
|
.copy()
|
||
|
|
)
|
||
|
|
export_catalog = matched_catalog[[c for c in sales_cols.keys() if c in matched_catalog.columns]].rename(
|
||
|
|
columns=sales_cols
|
||
|
|
)
|
||
|
|
export_catalog.to_excel(writer, sheet_name="Commercial_Sales_Catalog", index=False)
|
||
|
|
|
||
|
|
sim_cols = {
|
||
|
|
"simulated_rank": "Simulated Rank",
|
||
|
|
("baseline_matched_rank" if mode == "matched_only" else "hiffi_xp_published_rank"): "Published / Baseline Rank",
|
||
|
|
"simulation_rank_delta": "Simulated Rank Delta (Signed)",
|
||
|
|
"name": "Artist Name",
|
||
|
|
"simulated_hiffi_score": "Simulated Composite Score",
|
||
|
|
"hiffi_xp_score": "Published HIFFI-XP Score",
|
||
|
|
"commercial_power_index": "Commercial Power Index",
|
||
|
|
"total_eas_cspc": "Total Equivalent Album Sales",
|
||
|
|
"sales_data_status": "Status",
|
||
|
|
}
|
||
|
|
export_sim = sim_df[[c for c in sim_cols.keys() if c in sim_df.columns]].rename(columns=sim_cols)
|
||
|
|
export_sim.to_excel(writer, sheet_name="Simulated_Rerankings", index=False)
|
||
|
|
|
||
|
|
sensitivity_df.to_excel(writer, sheet_name="Sensitivity_Analysis", index=False)
|
||
|
|
|
||
|
|
wb = openpyxl.load_workbook(output)
|
||
|
|
header_fill = PatternFill(start_color="111827", end_color="111827", fill_type="solid")
|
||
|
|
header_font = Font(name="Calibri", size=11, bold=True, color="FFFFFF")
|
||
|
|
thin_border = Border(
|
||
|
|
left=Side(style="thin", color="E5E7EB"),
|
||
|
|
right=Side(style="thin", color="E5E7EB"),
|
||
|
|
top=Side(style="thin", color="E5E7EB"),
|
||
|
|
bottom=Side(style="thin", color="E5E7EB"),
|
||
|
|
)
|
||
|
|
|
||
|
|
for sheet in wb.worksheets:
|
||
|
|
sheet.views.sheetView[0].showGridLines = True
|
||
|
|
for col_idx, col_cells in enumerate(sheet.columns, 1):
|
||
|
|
max_len = max(len(str(cell.value or "")) for cell in col_cells)
|
||
|
|
col_letter = openpyxl.utils.get_column_letter(col_idx)
|
||
|
|
sheet.column_dimensions[col_letter].width = max(max_len + 4, 14)
|
||
|
|
|
||
|
|
header_cell = col_cells[0]
|
||
|
|
header_cell.fill = header_fill
|
||
|
|
header_cell.font = header_font
|
||
|
|
header_cell.alignment = Alignment(horizontal="center", vertical="center")
|
||
|
|
|
||
|
|
for cell in col_cells[1:]:
|
||
|
|
cell.border = thin_border
|
||
|
|
if isinstance(cell.value, float):
|
||
|
|
cell.number_format = "#,##0.00"
|
||
|
|
elif isinstance(cell.value, int) and cell.value > 999:
|
||
|
|
cell.number_format = "#,##0"
|
||
|
|
|
||
|
|
styled_output = io.BytesIO()
|
||
|
|
wb.save(styled_output)
|
||
|
|
return styled_output.getvalue()
|
||
|
|
|
||
|
|
|
||
|
|
def write_disk_workbook(output_path: Path, df: pd.DataFrame) -> None:
|
||
|
|
"""Utility to generate the workbook to local disk for repository artifacts."""
|
||
|
|
content = generate_executive_sales_workbook(df)
|
||
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
output_path.write_bytes(content)
|
||
|
|
print(f"[Excel] Generated verified executive workbook at {output_path}")
|