from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
import json
from decimal import Decimal
dictionary=json.loads((ROOT/"data"/"dictionary.json").read_text(encoding="utf-8"))
observation_end=pd.Timestamp(dictionary["observation_end"])
# 教学时间戳精确到秒；合同保证覆盖截至该日末尾，而非以最后一条事件推断覆盖。
cutoff=observation_end.normalize()
observation_exclusive=cutoff+pd.Timedelta(days=1)
assert observation_end == observation_exclusive-pd.Timedelta(seconds=1)
users=load("users").copy()
orders=load("orders").copy()
events=load("events").copy()
assigned=load("assignments").copy()
assert users["user_id"].is_unique and users["user_id"].notna().all()
assert orders["order_id"].is_unique and events["event_id"].is_unique
assert assigned["user_id"].is_unique
users["signup_date"]=pd.to_datetime(users["signup_date"]).dt.normalize()
orders["order_date"]=pd.to_datetime(orders["order_date"])
events["event_time"]=pd.to_datetime(events["event_time"])
assigned["assigned_at"]=pd.to_datetime(assigned["assigned_at"])
users=users.loc[users["signup_date"].le(cutoff)].copy()
orders=orders.loc[orders["order_date"].lt(observation_exclusive)].copy()
events=events.loc[events["event_time"].lt(observation_exclusive)].copy()
assigned=assigned.loc[assigned["assigned_at"].lt(observation_exclusive)].copy()
paid=orders.loc[orders["status"].isin(["paid","refunded"])].copy()
paid["net_cents"]=[int((Decimal(str(a))-Decimal(str(r)))*100) for a,r in zip(paid["amount"],paid["refund_amount"])]
known=paid.loc[paid["user_id"].notna()].copy()
summary=known.groupby("user_id").agg(paid_orders=("order_id","size"),net_cents=("net_cents","sum")).reset_index()
per_user=users.merge(summary,on="user_id",how="left",validate="one_to_one")
per_user[["paid_orders","net_cents"]]=per_user[["paid_orders","net_cents"]].fillna(0).astype("int64")
assert len(per_user)==len(users)
assert per_user["net_cents"].sum()==known["net_cents"].sum()
daily=paid.assign(day=paid["order_date"].dt.normalize()).groupby("day").agg(paid_orders=("order_id","size"),net_cents=("net_cents","sum"))
assert daily["net_cents"].sum()==paid["net_cents"].sum()
# 次日留存：注册后第一个自然日有任何事件；成熟分母使用权威完整观察截止。
active=events.assign(day=events["event_time"].dt.normalize())[["user_id","day"]].drop_duplicates()
eligible=users.loc[users["signup_date"]+pd.Timedelta(days=1)<=cutoff].copy()
eligible["target_day"]=eligible["signup_date"]+pd.Timedelta(days=1)
retained=eligible.merge(active,left_on=["user_id","target_day"],right_on=["user_id","day"],how="left",validate="one_to_one",indicator=True)
retained["retained_d1"]=retained["_merge"].eq("both")
retention=retained.groupby("signup_date").agg(eligible_users=("user_id","size"),retained_users=("retained_d1","sum"))
retention["rate"]=retention["retained_users"]/retention["eligible_users"]
assert retention["rate"].between(0,1).all()
# 会话漏斗：按事件时间与事件编号排序，用状态机保证顺序。
stages=[set(),set(),set()]
ordered=events.sort_values(["user_id","session_id","event_time","event_id"])
for (uid,sid),g in ordered.groupby(["user_id","session_id"]):
    state=0
    for kind in g["event_type"]:
        if state==0 and kind=="view":state=1
        elif state==1 and kind=="cart":state=2
        elif state==2 and kind=="purchase":state=3
    for i in range(state):stages[i].add(uid)
funnel=pd.DataFrame({"stage":["view","ordered_cart","ordered_purchase"],"users":[len(s) for s in stages]})
assert stages[2] <= stages[1] <= stages[0]
# 固定七天转化：仅纳入七天观察窗完整的分配用户；这些用户无活动也保留在分母。
mature_7d=assigned["assigned_at"]+pd.Timedelta(days=7)<=observation_exclusive
pending_assigned=assigned.loc[~mature_7d].copy()
assigned=assigned.loc[mature_7d].copy()
candidate=events.merge(assigned,on="user_id",how="inner",validate="many_to_one")
converted=candidate.loc[candidate["event_type"].eq("purchase") & candidate["event_time"].ge(candidate["assigned_at"]) & candidate["event_time"].lt(candidate["assigned_at"]+pd.Timedelta(days=7)),"user_id"].unique()
ab=assigned.copy()
ab["converted_7d"]=ab["user_id"].isin(converted)
ab_result=ab.groupby("variant").agg(assigned_users=("user_id","size"),conversions=("converted_7d","sum"))
ab_result["rate"]=ab_result["conversions"]/ab_result["assigned_users"]
assert ab_result["assigned_users"].sum()==len(assigned)
for name,frame in [("users",per_user),("daily",daily.reset_index()),("retention",retention.reset_index()),("funnel",funnel),("experiment",ab_result.reset_index())]:
    frame.to_csv(OUT/("project_business_"+name+".csv"),index=False)
print("注册用户 / 付费用户:",len(users),int(per_user["paid_orders"].gt(0).sum()))
print("已支付净额分 / 匿名已支付净额分:",int(paid["net_cents"].sum()),int(paid.loc[paid["user_id"].isna(),"net_cents"].sum()))
print("次日留存:");print(retention.to_string())
print("有序会话漏斗:");print(funnel.to_string(index=False))
print("七日行为转化（描述性统计）:");print(ab_result.to_string())
print("权威完整观察截止:",observation_end)
print("七日观察成熟 / 待观察分配用户:",len(assigned),len(pending_assigned))