"""Deposit-flight playbook, step 1: land the bank's deposit book.

The book is formula-generated with sha256 jitter (no RNG, no clock), so
every run lands the same book: 1,200 accounts over 160 days. Steady accounts
keep their deposits, card purchases and logins coming, with a small transfer
now and then. Leaving accounts behave exactly like them, on the same cadence
and with the same history length, then stop all activity and, about six
weeks later, move most of their balance out in one transfer. Nothing in a
leaving account's events before it goes quiet tells it apart: the tell is
the silence.

The book arrives the way a core-banking export does: one typed Parquet file,
uploaded, mapped and synced. Typed columns keep numbers numbers, which the
value-aware label of step 3 compares.
"""

import hashlib
import io
import json
import sys
from collections import Counter
from datetime import UTC, datetime, timedelta
from pathlib import Path

import httpx
import pyarrow as pa
import pyarrow.parquet as pq

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # playbooks/playbook_api.py
from playbook_api import Api, State

api, state = Api(), State("deposit-flight")

START = datetime(2026, 3, 2, tzinfo=UTC)
DAYS = 160
# The datasets of step 5 read the book up to day 120: every leaving account has
# gone quiet by then and none has moved its balance yet.
WINDOW_DAYS = 120
# The book is fixed, but the as-of anchors the observation-time build draws
# (step 5) are seeded by that build's own fingerprint, which carries its freshly
# minted label and snapshot ids: every run trains the with-token model on a
# different draw (by design; the flag-off twin trains identically every run).
# On 400 accounts, with re-scores from day 100, some draws trained a model whose
# risk barely moved with silence in that range: a flat curve at step 9 in about
# 1 run in 10, and a G2 calibration error up to 0.040 against the 0.05 bar at
# step 7. 1,200 accounts and re-scores from day 93 (below) keep every draw far
# from both edges; the measurements are in the change that set these values.
ACCOUNTS = 1200
CYCLES = 8
# Step 9 re-scores weekly from day 93, the day after the last leaving account
# went quiet (its last cycle ends on day 81-92): the first re-score sees the
# leaving accounts in an ordinary pause, the last after three more weeks of
# silence, and none has moved its balance yet (the earliest flight is day 121).
RESCORE_FROM_DAY = 93
RESCORES = 4
CHANNELS = ("branch", "direct_deposit", "mobile", "wire")


def unit(*parts: object) -> float:
    """A deterministic draw in [0, 1) from the parts."""
    digest = hashlib.sha256("|".join(map(str, parts)).encode("utf-8")).digest()
    return int.from_bytes(digest[:8], "big") / 2**64


def account_events(index: int) -> tuple[str, bool, list[dict]]:
    """One account: eight deposit cycles inside the datasets' window, then its fate.

    Every account has the same cycle (a deposit, then a card purchase or a
    login) on gaps of 4-8 days, so every account carries exactly 16 events in
    the window and the same history length. A steady account's last cycle in
    the window ends 6-18 days before it closes (a pause) and it carries on
    afterwards, now and then with a small transfer. A leaving account's last
    cycle ends on day 81-92; 40-44 days later it moves 85-100% of its balance
    out.
    """
    account = f"acct-{index:04d}"
    leaving = index % 2 == 1
    gaps = [4.0 + 4.0 * unit(account, cycle, "gap") for cycle in range(CYCLES)]
    if leaving:
        last = 81.0 + 11.0 * unit(account, "stops")
    else:
        last = WINDOW_DAYS - 6.0 - 12.0 * unit(account, "last")
    first = last - 0.6 * gaps[-1] - sum(gaps[:-1])
    events: list[dict] = []

    def add(day: float, event_type: str, **attributes: object) -> None:
        events.append(
            {
                "event_id": f"evt-{account}-{len(events):04d}",
                "account_id": account,
                "event_type": event_type,
                "occurred_at": (START + timedelta(days=day)).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
                **attributes,
            }
        )

    def cycle(day: float, gap: float, key: object) -> None:
        add(
            day,
            "deposit.posted",
            amount_minor=int(20_000 + 380_000 * unit(account, key, "amount")),
            currency="USD",
            channel=CHANNELS[int(4 * unit(account, key, "channel"))],
        )
        if unit(account, key, "kind") < 0.6:
            add(
                day + 0.6 * gap,
                "card.purchase",
                amount_minor=int(300 + 24_700 * unit(account, key, "card")),
                currency="USD",
            )
        else:
            add(day + 0.6 * gap, "app.login", is_new_device=unit(account, key, "device") < 0.02)

    day = first
    for number, gap in enumerate(gaps):
        cycle(day, gap, number)
        day += gap
    if leaving:
        flight = last + 40.0 + 4.0 * unit(account, "flight")
        share = 0.85 + 0.15 * unit(account, "flight-share")
        add(
            flight,
            "transfer.out",
            amount_minor=int(3_000_000 * share),
            currency="USD",
            balance_share=round(share, 4),
            is_external=True,
        )
        return account, leaving, events
    day, number = WINDOW_DAYS + 1.0 + 3.0 * unit(account, "after"), CYCLES
    while day < DAYS - 1.0:
        gap = 4.0 + 4.0 * unit(account, number, "gap")
        cycle(day, gap, number)
        if unit(account, number, "move") < 0.1:
            share = 0.02 + 0.18 * unit(account, number, "share")  # a small move is not flight
            add(
                day + 0.8 * gap,
                "transfer.out",
                amount_minor=int(2_000_000 * share),
                currency="USD",
                balance_share=round(share, 4),
                is_external=False,
            )
        day += gap
        number += 1
    return account, leaving, events


COLUMNS: dict[str, pa.DataType] = {
    "amount_minor": pa.int64(),
    "currency": pa.string(),
    "channel": pa.string(),
    "is_new_device": pa.bool_(),
    "balance_share": pa.float64(),
    "is_external": pa.bool_(),
}
leavers: list[str] = []
book: list[dict] = []
for index in range(ACCOUNTS):
    account, leaving, events = account_events(index)
    book.extend(events)
    if leaving:
        leavers.append(account)
rows = [
    {
        name: event.get(name)
        for name in ("event_id", "account_id", "event_type", "occurred_at", *COLUMNS)
    }
    for event in book
]
schema = pa.schema(
    [
        ("event_id", pa.string()),
        ("account_id", pa.string()),
        ("event_type", pa.string()),
        ("occurred_at", pa.string()),
        *COLUMNS.items(),
    ]
)
buffer = io.BytesIO()
pq.write_table(pa.Table.from_pylist(rows, schema=schema), buffer)

session = api.post(
    "/uploads",
    {"filename": "deposit_book.parquet", "format": "parquet"},
    key="deposit-flight-upload",
)
httpx.put(session["upload_url"], content=buffer.getvalue(), timeout=300.0).raise_for_status()
api.post(f"/uploads/{session['upload_id']}:complete")
upload = api.get(f"/uploads/{session['upload_id']}")
api.wait(
    f"/syncs/{upload['sync_id']}",
    lambda item: item.get("status") in ("succeeded", "failed"),
    what="the landing sync",
    polls=300,
)
mapping = {
    "version": 1,
    "entity": {"type": {"constant": "account"}, "id_column": "account_id"},
    "event_type": {"from_column": "event_type"},
    "timestamp": {"column": "occurred_at"},
    "event_id_column": "event_id",
    "attributes": [{"column": name} for name in COLUMNS],
}
api.put(f"/sources/{upload['source_id']}/mapping", mapping)
sync = api.post(
    f"/sources/{upload['source_id']}/syncs", {"mode": "full"}, key="deposit-flight-sync"
)
sync = api.wait(
    f"/syncs/{sync['sync_id']}",
    lambda item: item.get("status") in ("succeeded", "failed"),
    what="the book sync",
    polls=300,
)
print(f"book sync: {sync['status']}, rows_rejected: {sync.get('rows_rejected')}")
print(f"events landed: {sync.get('events_normalized') == len(rows)}")
if sync["status"] != "succeeded" or sync.get("rows_rejected"):
    sys.exit(json.dumps(sync)[:3000])

counts = Counter(event["event_type"] for event in book)
for event_type in sorted(counts):
    print(f"  {event_type}: {counts[event_type]}")
print(f"accounts: {ACCOUNTS}, quiet leavers: {len(leavers) > 0}")
end = START + timedelta(days=DAYS)
state.update(
    start=START.strftime("%Y-%m-%dT%H:%M:%SZ"),
    window_end=(START + timedelta(days=WINDOW_DAYS)).strftime("%Y-%m-%dT%H:%M:%SZ"),
    # Step 9 re-scores weekly from day 93: every leaving account is quiet, none has left.
    rescore_from=(START + timedelta(days=RESCORE_FROM_DAY)).strftime("%Y-%m-%dT%H:%M:%SZ"),
    rescores=RESCORES,
    end=end.strftime("%Y-%m-%dT%H:%M:%SZ"),
    leavers=leavers,
    accounts=ACCOUNTS,
)