"""Receivables: ageing (FIFO, Corrected), DSO, credit vs cash split, Monday collection list with P1-P4 priority and plain reasons."""
import numpy as np, pandas as pd, pickle
from pcommon import *

A = pd.read_parquet(f'{SIM}/sim_receivables_customer_ageing.parquet'); O = pd.read_parquet(f'{SIM}/sim_receivables_open_items.parquet')
C = pd.read_parquet(f'{CLEAN}/dim_customer.parquet'); S = pd.read_parquet(f'{WORK}/sales.parquet'); R = pd.read_parquet(f'{WORK}/returns.parquet')
ch = pickle.load(open(f'{WORK}/churn.pkl', 'rb'))['now'].set_index('outlet_id')
for d in (A, O, C):
    d['outlet_id'] = outlet_id(d.dist_cd, d.cust_cd)
A = A.set_index('outlet_id'); C = C.set_index('outlet_id')
B = S[S.sellable]
last = B.groupby('outlet_id').date.max()
s90 = B[B.date > DATA_END - pd.Timedelta(days=90)].groupby('outlet_id').net_amt.sum()
s4 = B[B.date > DATA_END - pd.Timedelta(days=28)].groupby('outlet_id').net_amt.sum()
s_prev12 = B[(B.date <= DATA_END - pd.Timedelta(days=28)) & (B.date > DATA_END - pd.Timedelta(days=112))].groupby('outlet_id').net_amt.sum()
r13 = R[R.date > DATA_END - pd.Timedelta(days=91)].groupby('outlet_id').net_amt.sum(); s13 = B[B.date > DATA_END - pd.Timedelta(days=91)].groupby('outlet_id').net_amt.sum()
mainvan = B[B.date > DATA_END - pd.Timedelta(days=91)].groupby('outlet_id').van.agg(lambda x: x.value_counts().index[0])
lastvan = B.sort_values('date').groupby('outlet_id').van.last()

L = A.copy()
L['past_due_sar'] = L[['open_1-30', 'open_31-60', 'open_61-90', 'open_90+']].sum(axis=1)
L['past_due_over_60_sar'] = L[['open_61-90', 'open_90+']].sum(axis=1)
L['unmatched_sar'] = L.unexplained_by_sfa_invoices_sar
mx = O[O.days_past_due > 0].groupby('outlet_id').days_past_due.max()
L['max_days_past_due'] = L.index.to_series().map(mx).fillna(0).astype(int).values
L['monthly_buying_sar'] = L.index.to_series().map(s90).fillna(0).values / 3
L['days_since_last_purchase'] = (DATA_END - L.index.to_series().map(last)).dt.days.values
L['sales_last_4w_sar'] = L.index.to_series().map(s4).fillna(0).values
L['avg_4w_sales_prior_12w_sar'] = L.index.to_series().map(s_prev12).fillna(0).values / 3
L['return_rate_13w'] = L.index.to_series().map(r13).fillna(0).values / L.index.to_series().map(s13).fillna(0).clip(lower=1).values
L['over_limit'] = (~L.credit_limit_is_placeholder) & (L.outstanding_bal_sar > L.credit_limit_sar)
L['churn_risk_8w'] = L.index.to_series().map(ch.p).values
L['van'] = L.index.to_series().map(mainvan).fillna(L.index.to_series().map(lastvan)).values
reasons, score = [], []
for o, r in L.iterrows():
    rs, sc = [], 0
    slowing = r.avg_4w_sales_prior_12w_sar > 0 and r.sales_last_4w_sar < 0.5 * r.avg_4w_sales_prior_12w_sar
    if r.monthly_buying_sar > 0 and r.outstanding_bal_sar > 2 * r.monthly_buying_sar and slowing:
        rs.append('Balance > 2x monthly buying, orders slowing'); sc += 3
    elif r.monthly_buying_sar > 0 and r.outstanding_bal_sar > 2 * r.monthly_buying_sar:
        x = r.outstanding_bal_sar / r.monthly_buying_sar
        rs.append(f'Balance = {x:.0f}x monthly buying (SAR {r.outstanding_bal_sar:,.0f} vs SAR {r.monthly_buying_sar:,.0f} a month)' if x >= 10 else f'Balance = {x:.1f}x monthly buying'); sc += 1
    if r.days_since_last_purchase != r.days_since_last_purchase:  # blank: nothing bought since the sales data starts in March
        rs.append('No purchase since March (start of the sales data), balance open'); sc += 3
    elif r.days_since_last_purchase >= 45:
        rs.append(f'No purchase in {int(r.days_since_last_purchase)} days, balance open'); sc += 3
    if r.past_due_over_60_sar > 0:
        rs.append(f'SAR {r.past_due_over_60_sar:,.0f} more than 60 days past due'); sc += 2
    elif r.past_due_sar > 0:
        rs.append(f'SAR {r.past_due_sar:,.0f} past due ({r.max_days_past_due} days)'); sc += 1
    if r.unmatched_sar > 0:
        rs.append(f'SAR {r.unmatched_sar:,.0f} not matched to invoices since March (older debt?)'); sc += 2
    if r.return_rate_13w > 0.10 and r.outstanding_bal_sar > r.monthly_buying_sar:
        rs.append(f'High return rate ({r.return_rate_13w:.0%}), balance above a month of buying'); sc += 1
    if r.over_limit:
        rs.append('Over credit limit'); sc += 2
    if r.churn_risk_8w == r.churn_risk_8w and r.churn_risk_8w >= 0.3:
        rs.append(f'Outlet likely to go quiet ({r.churn_risk_8w:.0%})'); sc += 2
    reasons.append('; '.join(rs) if rs else 'No risk flags'); score.append(sc)
L['risk_points'] = score; L['why_flagged'] = reasons
L['exposure_sar'] = L.past_due_sar + L.unmatched_sar
hi_exp = L.exposure_sar >= max(1000, L.exposure_sar[L.exposure_sar > 0].quantile(0.75))
L['priority'] = np.select([(L.risk_points >= 5) & hi_exp, (L.risk_points >= 3) & (L.exposure_sar > 0), (L.risk_points >= 1) | (L.past_due_sar > 0)], ['P1', 'P2', 'P3'], 'P4')
cc = C.reindex(L.index)
L['outlet_name'] = cc.cust_name.values; L['outlet_name_ar'] = cc.cust_name2.values
credit_sales_90 = B[(B.date > DATA_END - pd.Timedelta(days=90)) & B.outlet_id.isin(C.index[C.cust_type == 'Q'])].net_amt.sum()
Q = C[C.cust_type == 'Q']
kpi = {'credit_book_gross_sar': float(Q.outstanding_bal[Q.outstanding_bal > 0].sum()), 'credit_balances_negative_sar': float(Q.outstanding_bal[Q.outstanding_bal < 0].sum()),
       'customers_with_balance': int((Q.outstanding_bal > 0).sum()), 'past_due_sar': float(L.past_due_sar.sum()), 'unmatched_sar': float(L.unmatched_sar.sum()),
       'not_due_sar': float(L.open_not_due.sum()), 'dso_days': float(Q.outstanding_bal[Q.outstanding_bal > 0].sum() / (credit_sales_90 / 90)),
       'p1_accounts': int((L.priority == 'P1').sum()), 'p1_exposure_sar': float(L.loc[L.priority == 'P1', 'exposure_sar'].sum()),
       'p1_balance_sar': float(L.loc[L.priority == 'P1', 'outstanding_bal_sar'].sum()), 'p2_accounts': int((L.priority == 'P2').sum()),
       'p3_accounts': int((L.priority == 'P3').sum()), 'p4_accounts': int((L.priority == 'P4').sum()), 'placeholder_limits': int(C.flag_credit_limit_sentinel.sum())}
kpi['past_due_share_of_balance'] = kpi['past_due_sar'] / kpi['credit_book_gross_sar']
print({k: round(v, 3) if isinstance(v, float) else v for k, v in kpi.items()})
buckets = pd.DataFrame({'bucket': ['Not yet due', '1-30 days', '31-60 days', '61-90 days', 'Over 90 days', 'Not matched to invoices since March'],
                        'bucket_order': range(1, 7), 'amount_sar': [L.open_not_due.sum(), L['open_1-30'].sum(), L['open_31-60'].sum(), L['open_61-90'].sum(), L['open_90+'].sum(), L.unmatched_sar.sum()],
                        'customers': [(L.open_not_due > 0).sum(), (L['open_1-30'] > 0).sum(), (L['open_31-60'] > 0).sum(), (L['open_61-90'] > 0).sum(), (L['open_90+'] > 0).sum(), (L.unmatched_sar > 0).sum()]})
buckets['share_of_book'] = buckets.amount_sar / buckets.amount_sar.sum()
C2 = C.copy(); C2['term'] = C2.invterm_cd.fillna('cash (no term)'); C2['bought_in_window'] = C2.index.isin(B.outlet_id)
terms = C2.groupby(['cust_type', 'term']).agg(customers=('cust_cd', 'size'), buying_customers=('bought_in_window', 'sum'),
                                              customers_with_balance=('outstanding_bal', lambda x: int((x > 0).sum())), balance_sar=('outstanding_bal', lambda x: float(x[x > 0].sum())),
                                              placeholder_limits=('flag_credit_limit_sentinel', 'sum'), real_limits=('flag_credit_limit_sentinel', lambda x: int((~x.astype(bool)).sum()))).reset_index()
terms['real_limits'] = np.where(terms.cust_type == 'Q', terms.real_limits, 0)
O['due_week_end'] = week_end(pd.to_datetime(O.due_date))
due = O.assign(when=np.where(pd.to_datetime(O.due_date) <= DATA_END, 'overdue now', O.due_week_end.dt.strftime('%Y-%m-%d'))).groupby('when').open_amount_sar.sum().reset_index()
pickle.dump({'L': L, 'kpi': kpi, 'buckets': buckets, 'terms': terms, 'due': due}, open(f'{WORK}/receivables.pkl', 'wb'))
print(L.priority.value_counts().to_dict()); print(buckets.round(0).to_string()); print(due.to_string())