"""Reproducible UCI Online Retail revenue and cohort portfolio studies.
Data: Daqing Chen (2015), DOI 10.24432/C5BW33, CC BY 4.0.
Run: python analyze.py (download source ZIP to data/ first).
"""
from pathlib import Path
import zipfile, json, sqlite3
import pandas as pd
import numpy as np

ROOT=Path(__file__).parent
OUT=ROOT/'output'; OUT.mkdir(exist_ok=True)
with zipfile.ZipFile(ROOT/'data/online-retail.zip') as z:
    with z.open('Online Retail.xlsx') as f: raw=pd.read_excel(f)
raw.columns=['invoice','sku','description','quantity','date','unit_price','customer','country']
quality={'raw_rows':len(raw),'exact_duplicate_rows':int(raw.duplicated().sum()),'missing_customer_rows':int(raw.customer.isna().sum()),'cancelled_rows':int(raw.invoice.astype(str).str.upper().str.startswith('C').sum()),'nonpositive_price_rows':int((raw.unit_price<=0).sum())}
df=raw.drop_duplicates().copy()
df['invoice']=df.invoice.astype(str)
df['value']=df.quantity*df.unit_price
# Complete calendar months only. Positive invoice value is not net revenue/profit.
period=(df.date>='2010-12-01')&(df.date<'2011-12-01')
positive=(df.quantity>0)&(df.unit_price>0)&~df.invoice.str.upper().str.startswith('C')
sales=df[period&positive].copy()
sales['month']=sales.date.dt.strftime('%Y-%m')
sales['customer']=sales.customer.astype('Int64').astype('string')
connection=sqlite3.connect(ROOT/'data/retail.sqlite')
sales[['invoice','sku','quantity','unit_price','value','month','customer','country']].to_sql('sales',connection,if_exists='replace',index=False)
query_file=ROOT/'queries.sql'
if not query_file.exists(): query_file=ROOT/'queries.sql.txt'
sql=query_file.read_text()
monthly=pd.read_sql_query(sql.split('-- COHORT')[0],connection)
known=sales[sales.customer.notna()].copy()
first=known.groupby('customer').month.min()
known['cohort']=known.customer.map(first)
known['age']=(pd.to_datetime(known.month).dt.year-pd.to_datetime(known.cohort).dt.year)*12+(pd.to_datetime(known.month).dt.month-pd.to_datetime(known.cohort).dt.month)
cohort=known.groupby(['cohort','age']).customer.nunique().rename('active_customers').reset_index()
sizes=first.value_counts()
cohort['cohort_size']=cohort.cohort.map(sizes)
cohort['retention_pct']=100*cohort.active_customers/cohort.cohort_size
sql_cohort=pd.read_sql_query(sql.split('-- COHORT')[1],connection)
check=cohort.merge(sql_cohort,on=['cohort','age'],suffixes=('_python','_sql'))
assert len(check)==len(cohort)==len(sql_cohort)
assert (check.active_customers_python==check.active_customers_sql).all()
assert np.isclose(monthly.positive_value.sum(),sales.value.sum())
assert np.isclose(cohort[cohort.age==0].retention_pct,100).all()
assert (cohort.retention_pct.between(0,100)).all()
country=sales.groupby('country').agg(positive_value=('value','sum'),orders=('invoice','nunique')).sort_values('positive_value',ascending=False).reset_index()
order_counts=known.groupby('customer').invoice.nunique()
eligible_m1=cohort[cohort.age==1]
m1=100*eligible_m1.active_customers.sum()/eligible_m1.cohort_size.sum()
missing_share=100*sales[sales.customer.isna()].value.sum()/sales.value.sum()
duplicate_value=float(raw[(raw.date>='2010-12-01')&(raw.date<'2011-12-01')&(raw.quantity>0)&(raw.unit_price>0)&~raw.invoice.astype(str).str.upper().str.startswith('C')].eval('quantity*unit_price').sum()-sales.value.sum())
summary={'period':'Dec 2010–Nov 2011','quality':quality,'positive_invoice_value':round(float(sales.value.sum()),2),'orders':int(sales.invoice.nunique()),'identified_customers':len(order_counts),'clean_positive_lines':len(sales),'aov':round(float(sales.value.sum()/sales.invoice.nunique()),2),'repeat_customer_pct':round(100*float((order_counts>1).mean()),2),'month1_weighted_retention_pct':round(float(m1),2),'unidentified_value_pct':round(float(missing_share),2),'duplicate_policy_value_difference':round(duplicate_value,2),'top_country':str(country.iloc[0].country),'top_country_value_share':round(100*float(country.iloc[0].positive_value/sales.value.sum()),2),'checks':{'sql_python_monthly_total':True,'sql_python_cohorts':True,'cohort_zero_100pct':True,'retention_bounds':True}}
for name,table in [('monthly',monthly),('cohorts',cohort),('countries',country)]:
    table.to_csv(OUT/(name+'.csv'),index=False)
payload={'summary':summary,'monthly':json.loads(monthly.round(2).to_json(orient='records')),'cohorts':json.loads(cohort.round(2).to_json(orient='records')),'countries':json.loads(country.round(2).to_json(orient='records'))}
(OUT/'retail-results.json').write_text(json.dumps(payload,indent=2))
print(json.dumps(summary,indent=2))