Scoring method

The exact engine

The scoring code, served from the same file the site imports, so it can't drift from what runs:

"""The open part of the product: every number on a page comes from here, deterministically."""
from __future__ import annotations

def max_drawdown(equity: list[float]) -> float:
    """Max peak-to-trough drawdown as a fraction (0.12 = 12%) over an equity curve."""
    peak, mdd = None, 0.0
    for v in equity:
        if v is None: continue
        if peak is None or v > peak: peak = v
        if peak and peak > 0:
            # An equity series that dips below zero (a wiped account with a negative point)
            # would make this exceed 1.0 — a 1,388% drawdown was on a record page. A fall
            # cannot exceed everything, so the floor of the curve is zero.
            dd = (peak - max(v, 0.0)) / peak
            if dd > mdd: mdd = dd
    return mdd

def equity_from_pnl(start_value: float, pnl_history: list[float]) -> list[float]:
    """Equity curve implied by cumulative PnL on a fixed starting value — deposits/withdrawals don't move it (time-weighted approximation)."""
    start = max(start_value, 1.0)
    return [start + p for p in pnl_history]

FLOORS = {"daily": 0.05, "weekly": 0.15, "realized": 0.10}
CAPS = {"daily": 3.0, "weekly": 1.0, "realized": 3.0}

def score(roi: float, mdd: float, trades_or_volume_ok: bool, band_label: str, resolution: str = "daily") -> float | None:
    """ELIGIBILITY GATE + risk-adjusted component (calmar). Returns None for ineligible records:
    - minimum equity under $1k in the window (lottery accounts can't qualify by pumping a tiny stake);
    - inactive / market-maker turnover;
    - anomalies: >100% return with <5% visible drawdown at daily/realized resolution — artifact until proven otherwise.
    The ratio itself uses per-resolution drawdown floors and return caps, because coarser data
    (weekly copy stats, realized-only curves) must not outrank finer data on missing information.
    The public 0-100 score is computed from this + return + consistency + track length in composite()."""
    if not trades_or_volume_ok or band_label == "<$1k": return None
    if resolution in ("daily", "realized") and roi > 1.0 and mdd < 0.05: return None
    return round(min(roi, CAPS.get(resolution, 3.0)) / max(mdd, FLOORS.get(resolution, 0.05)), 2)

def _pct_sorted(sv):
    """Mid-rank percentile function over an already sorted population (rank-based → robust to outliers and caps)."""
    import bisect
    n = max(len(sv), 1)
    return lambda v: (bisect.bisect_left(sv, v) + bisect.bisect_right(sv, v)) / 2 / n

def _pct_rank(values):
    return _pct_sorted(sorted(values))

W_RET, W_RISK, W_CON, W_TRK = 0.40, 0.25, 0.20, 0.15          # long windows (1y, max)
WEIGHTS_LONG = (W_RET, W_RISK, W_CON, W_TRK)
WEIGHTS_SHORT = (0.50, 0.20, 0.20, 0.10)                       # 7d + 30d: return leads harder, consistency keeps its anti-lottery weight

def _parts(i):
    """One record's four inputs: return, return per unit of drawdown, consistency, track length.
    Never rounded: the stored reference keeps full precision (JSON round-trips floats exactly), so a
    record scores the same against a reference in memory and against the same one read from disk."""
    return (i["ret"], i["ret"] / max(i["mdd"], FLOORS.get(i["res"], 0.05)), i["consist"], i["track"])

def _raw(parts, ref, weights):
    w_ret, w_risk, w_con, w_trk = weights
    return (w_ret * _pct_sorted(ref["ret"])(parts[0]) + w_risk * _pct_sorted(ref["cal"])(parts[1])
            + w_con * _pct_sorted(ref["con"])(parts[2]) + w_trk * _pct_sorted(ref["trk"])(parts[3]))

def reference(items: list[dict], weights: tuple = WEIGHTS_LONG) -> dict | None:
    """The population a score is measured against, frozen: the sorted values of each input and of
    the blended result. Scoring against a stored reference means a score moves when the record
    moves — not when another exchange's feed goes quiet and its traders drop out of the comparison."""
    if not items: return None
    parts = [_parts(i) for i in items]
    ref = {"ret": sorted(p[0] for p in parts), "cal": sorted(p[1] for p in parts),
           "con": sorted(p[2] for p in parts), "trk": sorted(p[3] for p in parts), "n": len(items)}
    ref["raw"] = sorted(_raw(p, ref, weights) for p in parts)
    return ref

def composite(items: list[dict], weights: tuple = WEIGHTS_LONG, ref: dict | None = None) -> list[dict]:
    """The public PnL Score, 0-100: literally 'beats X% of every scored trader on the board'.
    Weighted blend of percentiles — return, risk-adjusted return, consistency, track length —
    then re-percentiled so the number IS a percentile. Short windows weigh return at 50%
    (impressive weeks/months should look impressive); long windows keep 40/25/20/15 so
    history stays strict. items: [{"ret", "mdd", "res", "consist", "track"}] (already gated).
    ref: the reference population (see reference()); without one the items are their own."""
    if not items: return []
    ref = ref or reference(items, weights)
    p_ret, p_cal, p_con, p_trk, p_raw = (_pct_sorted(ref[k]) for k in ("ret", "cal", "con", "trk", "raw"))
    out = []
    for i in items:
        parts = _parts(i)
        out.append({"score": min(round(100 * p_raw(_raw(parts, ref, weights)), 1), 99.9),
                    "comps": {"ret": int(100 * p_ret(parts[0])), "risk": int(100 * p_cal(parts[1])),
                              "steady": int(100 * p_con(parts[2])), "track": int(100 * p_trk(parts[3]))}})
    return out

def return_on_average_equity(pnl_total: float, account_values: list[float]) -> float | None:
    """Our own 30-day return: total PnL over average equity in the window. Deposits raise the denominator, so they can't inflate it."""
    vals = [v for v in account_values if v and v > 0]
    if not vals: return None
    return pnl_total / (sum(vals) / len(vals))

def score_math(comps: dict | None) -> str:
    """The score's own components, for display next to it — every number shows its work.
    Returns "" (never None) when a source publishes only a single window: Jinja prints None
    as the word None, and it did, on ~1,900 record pages."""
    if not comps: return ""
    return f"return {comps.get('ret', '—')}p · risk {comps.get('risk', '—')}p · steadiness {comps.get('steady', '—')}p · track {comps.get('track', '—')}p"

def pct(x) -> str:
    try:
        v = float(x) * 100
        return "0.0%" if abs(v) < 0.05 else f"{v:+.1f}%"   # never "-0.0%"
    except Exception: return "—"

TIERS = [(99, "Diamond"), (95, "Gold"), (85, "Silver"), (60, "Bronze")]  # percentile cutoffs: Diamond = top 1%
def tier(score) -> str | None:
    if score is None: return None
    for lim, name in TIERS:
        if score >= lim: return name
    return None

def monthly_from_stamped(stamped, ms_hint=None):
    """Calendar-month returns from a dated (timestamp, index) series.

    Must run BEFORE any downsampling: connectors compress a multi-year curve to 60 points for
    display, at which point one point spans ~15 days and months are no longer recoverable. A
    month is only emitted if the series actually covers it — a gap stays a gap rather than
    being interpolated into a fabricated 0.0%.

    Returns [{"ym": "2026-08", "roi": 0.121, "mdd": 0.043, "pts": 22}, ...] oldest first.
    """
    import time as _t
    if not stamped or len(stamped) < 2:
        return []
    norm = []
    for ts, v in stamped:
        ts = float(ts)
        if ts > 1e11:            # milliseconds
            ts /= 1000.0
        if ts < 946_684_800 or v is None or v <= 0:   # before 2000, or a non-positive index
            continue
        norm.append((ts, float(v)))
    if len(norm) < 2:
        return []
    norm.sort()
    buckets = {}
    for ts, v in norm:
        ym = _t.strftime("%Y-%m", _t.gmtime(ts))
        buckets.setdefault(ym, []).append(v)
    out, prev_close = [], None
    for ym in sorted(buckets):
        vals = buckets[ym]
        opening = prev_close if prev_close is not None else vals[0]
        if opening <= 0:
            prev_close = vals[-1]; continue
        peak, dd = opening, 0.0
        for v in [opening] + vals:
            if v > peak: peak = v
            if peak > 0: dd = max(dd, (peak - v) / peak)
        # Store the EXACT ratio — no rounding, no floor. The month figures must telescope
        # back to the headline return, and across a near-zero pass (an account wiped to
        # 0.01% and rebuilt) the product is hypersensitive: rounding one month to four
        # places collapsed the compounding to zero, and flooring it at -99.9999% overstated
        # the recovery sixty-fold. The TWR already floors each STEP at -99.99%, so the index
        # never reaches exactly zero and the raw ratio is always finite. Round at display.
        out.append({"ym": ym,
                    "roi": vals[-1] / opening - 1,
                    "mdd": round(min(dd, 1.0), 4),
                    "pts": len(vals)})
        prev_close = vals[-1]
    return out


def monthly_grid(monthly):
    """Reshape monthly rows into calendar years for display.

    Absent months stay None — never 0.0%. A month the source could not cover and a month the
    trader was flat are different facts, and a registry may not blur them. Year figures are
    compounded from the months actually present, and say how many that was."""
    if not monthly:
        return []
    by_year = {}
    for m in monthly:
        try:
            y, mo = m["ym"].split("-"); y, mo = int(y), int(mo)
        except (KeyError, ValueError):
            continue
        by_year.setdefault(y, {})[mo] = m
    rows = []
    for y in sorted(by_year, reverse=True):
        cells = [by_year[y].get(i + 1) for i in range(12)]
        got = [c for c in cells if c]
        comp = 1.0
        for c in got:
            comp *= 1 + (c["roi"] or 0)
        rows.append({"year": y, "cells": cells,
                     "roi": round(comp - 1, 4),
                     "mdd": round(max((c["mdd"] or 0) for c in got), 4) if got else 0.0,
                     "n": len(got)})
    return rows


# ---- window reliability: one rule, used by scoring AND by every page that shows a window ----
IMPLAUSIBLE_GAIN = 100.0   # +10,000% over the window
CROSS_CHECK = 20.0         # ...and more than 20x what the exchange itself reports for the span
STALE_AFTER = 72 * 3600    # numbers older than this leave the board until they are refreshed


def assess_windows(ex: dict, venue_roi: dict) -> dict:
    """Mark, in place, every window whose numbers we can't stand behind. Returns the problems.
    crossed    — the reconstructed equity went below zero (a return under -100%): impossible, so
                 the whole record is unscored.
    implausible— a gain over +10,000% that is more than 20x the exchange's own figure for the
                 span, or that no exchange figure corroborates: deposits/withdrawals the curve
                 couldn't separate. The window is hidden and counts as neutral in the blend."""
    out = {"crossed": [], "implausible": []}
    for k, w in list((ex.get("windows") or {}).items()):
        if not isinstance(w, dict): continue
        w.pop("unreliable", None)
        roi = w.get("roi")
        pts = [p for p in (w.get("pts") or []) if isinstance(p, (int, float))]
        # at or past the floor: a curve pinned at exactly -100% (or the TWR's -99.99% step floor) is the
        # clamp talking, not a result — an account that really went to zero is under $1k and never scored
        if (roi is not None and roi <= -0.9999) or any(p <= -99.99 for p in pts):
            w["unreliable"] = "crossed zero"; out["crossed"].append(k); continue
        if roi is not None and roi > IMPLAUSIBLE_GAIN:
            ref = venue_roi.get(k)
            if ref is None or roi > CROSS_CHECK * max(ref, 0.05):
                w["unreliable"] = "implausible"; out["implausible"].append(k)
    return out


def stamp(win: dict | None, ts: list) -> dict | None:
    """Give a window its own time axis: the first and last point's time (seconds), plus every
    point's time when they aren't evenly spaced. Pages draw dates from these — never from today's
    date and an assumed spacing."""
    if not win or not ts or len(ts) != len(win.get("pts") or []): return win
    ts = [float(x) / 1000.0 if float(x) > 1e11 else float(x) for x in ts]
    t0, t1 = ts[0], ts[-1]
    if t0 < 946_684_800 or t1 < t0: return win
    win["t0"], win["t1"] = int(t0), int(t1)
    n = len(ts)
    if n > 2 and t1 > t0 and max(abs(x - (t0 + (t1 - t0) * i / (n - 1))) for i, x in enumerate(ts)) > 0.02 * (t1 - t0):
        win["t"] = [int(x) for x in ts]
    return win


def point_times(win) -> list | None:
    """The time (seconds) of every point of a window's curve, or None when the window carries no time
    axis. A duration on a page is measured on these — a count of points is not a count of days."""
    if not isinstance(win, dict): return None
    pts = win.get("pts") or []
    t = win.get("t")
    if isinstance(t, list) and len(t) == len(pts) and len(pts) >= 2: return [float(x) for x in t]
    t0, t1 = win.get("t0"), win.get("t1")
    if t0 and t1 and len(pts) >= 2 and t1 > t0:
        return [t0 + (t1 - t0) * i / (len(pts) - 1) for i in range(len(pts))]
    return None


def reliable(windows) -> dict:
    """The windows a page may chart, share or rank: everything not marked unreliable."""
    return {k: w for k, w in (windows or {}).items() if isinstance(w, dict) and not w.get("unreliable")}