"""Calibration, exact expected-F0.5 selection per S1, one-owner arbitration, and the macro-F0.5 metric."""
from multiprocessing import Pool

import numpy as np
import pandas as pd
from sklearn.isotonic import IsotonicRegression


# ----------------------------------------------------------------------------------------------- metric

MULT = np.int64(1 << 32)
_GT_CACHE = {}


def in_sorted(x, s):
    """Membership of x in sorted array s by binary search (no per-call hashing of s)."""
    if len(s) == 0:
        return np.zeros(len(x), dtype=bool)
    pos = np.clip(np.searchsorted(s, x), 0, len(s) - 1)
    return s[pos] == x


def gt_keys(gt_q, gt_c):
    k = (id(gt_q), len(gt_q))
    if k not in _GT_CACHE:
        _GT_CACHE[k] = np.sort(gt_q.astype(np.int64) * MULT + gt_c)
    return _GT_CACHE[k]


def macro_f05(pred, q_eval, n_true, gt_q, gt_c):
    """pred: DataFrame(q, c) of predicted matches. Every q in q_eval counts, singletons included."""
    q_eval = np.asarray(q_eval)
    pred = pred[in_sorted(pred.q.values.astype(np.int64), np.sort(q_eval).astype(np.int64))]
    k = pred.q.values.astype(np.int64) * MULT + pred.c.values
    tp = pd.Series(in_sorted(k, gt_keys(gt_q, gt_c)).astype(np.int64)).groupby(pred.q.values).sum()
    npred = pred.groupby("q").size()
    s = pd.DataFrame({"q": q_eval})
    s["tp"] = s.q.map(tp).fillna(0).values
    s["np"] = s.q.map(npred).fillna(0).values
    s["nt"] = n_true[q_eval]
    p = np.where(s.np > 0, s.tp / s.np.clip(lower=1), 0)
    r = np.where(s.nt > 0, s.tp / s.nt.clip(lower=1), 0)
    f = np.where(s.tp > 0, 1.25 * p * r / (0.25 * p + r + 1e-12), 0.0)
    f = np.where((s.np == 0) & (s.nt == 0), 1.0, f)
    s["f"] = f
    return float(f.mean()), s


# ----------------------------------------------------------------------------------------------- calibration

def fit_isotonic(score, y):
    ir = IsotonicRegression(out_of_bounds="clip", y_min=0.0, y_max=1.0)
    ir.fit(score, y)
    return ir


def crossfit_isotonic(score, y, q, n_folds=5):
    out = np.empty_like(score, dtype=np.float64)
    fold = (q.astype(np.int64) * 2654435761 % 2 ** 32) % n_folds
    for k in range(n_folds):
        m = fold == k
        out[m] = fit_isotonic(score[~m], y[~m]).predict(score[m])
    return out


# ----------------------------------------------------------------------------------------------- expected F0.5

def best_k(p):
    """p sorted descending. Exact E[F0.5] of taking the top-k for every k via Poisson-binomial DPs."""
    n = len(p)
    if n == 0:
        return 0, 1.0
    suf = [None] * (n + 1)
    d = np.array([1.0])
    suf[n] = d
    for i in range(n - 1, -1, -1):
        d = np.convolve(d, [1 - p[i], p[i]])
        suf[i] = d
    best, best_v = 0, suf[0][0]              # k = 0 scores 1 only if there is no true match
    tp = np.array([1.0])
    for k in range(1, n + 1):
        tp = np.convolve(tp, [1 - p[k - 1], p[k - 1]])
        fn = suf[k]
        t = np.arange(k + 1)[:, None]
        f = np.arange(len(fn))[None, :]
        F = 1.25 * t / (0.25 * (t + f) + k)
        v = float((tp[:, None] * fn[None, :] * F).sum())
        if v > best_v:
            best, best_v = k, v
    return best, best_v


def _choose_group(args):
    qs, probs_list, family, thr = args
    out = []
    for q, p in zip(qs, probs_list):
        if family == "threshold":
            k = int((p >= thr).sum())
        else:
            k, _ = best_k(p)
        out.append(k)
    return out


def choose(df, family, thr=None, pool=None, banned=None):
    """df(q, c, p) -> selected rows. Candidates sorted by p desc inside each q."""
    d = df.sort_values(["q", "p"], ascending=[True, False], kind="stable")
    if banned is not None and len(banned):
        mult = np.int64(1 << 32)
        d = d[~np.isin(d.q.values.astype(np.int64) * mult + d.c.values, banned)]
    d = d[d.p >= 1e-4]
    if d.empty:
        return d
    qv = d.q.values
    starts = np.flatnonzero(np.r_[True, qv[1:] != qv[:-1]])
    ends = np.r_[starts[1:], len(d)]
    pv = d.p.values
    qs = qv[starts]
    probs = [pv[a:e] for a, e in zip(starts, ends)]
    step = 20_000
    args = [(qs[i:i + step], probs[i:i + step], family, thr) for i in range(0, len(qs), step)]
    ks = [k for part in (pool.map(_choose_group, args) if pool else map(_choose_group, args)) for k in part]
    take = np.zeros(len(d), dtype=bool)
    for a, k in zip(starts, ks):
        take[a:a + k] = True
    return d[take]


def decide(df, family, thr=None, max_iter=30, pool=None):
    """Per-S1 choice, then one-owner arbitration iterated to a fixpoint: the higher-probability claim keeps
    a contested S2/S3 record and the loser re-decides without it."""
    mult = np.int64(1 << 32)
    banned = np.empty(0, dtype=np.int64)
    sel = choose(df, family, thr, pool)
    for _ in range(max_iter):
        dup = sel[sel.duplicated("c", keep=False)]
        if dup.empty:
            break
        win = dup.sort_values(["c", "p"], ascending=[True, False]).drop_duplicates("c")
        lose = dup.merge(win[["c", "q"]], on="c", suffixes=("", "_w"))
        lose = lose[lose.q != lose.q_w]
        banned = np.union1d(banned, lose.q.values.astype(np.int64) * mult + lose.c.values)
        redo = np.unique(lose.q.values)
        keep = sel[~np.isin(sel.q.values, redo)]
        again = choose(df[np.isin(df.q.values, redo)], family, thr, pool, banned)
        sel = pd.concat([keep, again], ignore_index=True)
    return sel.sort_values(["q", "p"], ascending=[True, False]).reset_index(drop=True)