"""When did the machines arrive on en.wikipedia? The pre-registered run.

The rule was fixed and published before this script existed, at
https://untilnextsession.com/experiments/when-the-machines-arrived/ — in short:

  * the machine series is `spider + automated` summed, for the whole decade,
    never `automated` alone;
  * 2020-04 (the month the `automated` agent type first appears) and
    2025-09 -> 2025-10 (the classifier change identified in the human study)
    are named in advance as instrument events: a break in either neighbourhood
    is scored as the instrument and never as an arrival;
  * a step counts only if the month-on-month change falls outside the range of
    that same calendar-month transition in every other year of the series, and
    the twelve months that follow never return to the pre-break range;
  * if the only surviving breaks are the two instrument months, the finding is
    that this series records Wikimedia's classifier rather than the machines.

Reads the JSON already fetched by fetch.py into ~/data/wikipedia-traffic.
Writes machines.md and machines.csv beside this file.

    python research/wikipedia-traffic/machines.py
"""

import csv
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from analyse import series, prev, yoy, w, BN, HERE          # noqa: E402

#: Named in the kill rule, before the run. A break whose month is in one of
#: these windows is the instrument changing, not the world changing.
INSTRUMENT = {
    "2020-04": "the `automated` agent type first appears; before this month "
               "Wikimedia published no such class at all",
    "2025-10": "the reclassification already identified in the human study, "
               "when the non-human count halved from one month to the next",
}
#: One month either side, because a classifier change need not land tidily on
#: the boundary of the month it was announced in.
WINDOW = 1


def months_around(month, k=WINDOW):
    """The months within k of `month`, as a set of 'YYYY-MM'."""
    y, m = int(month[:4]), int(month[5:])
    out = set()
    for d in range(-k, k + 1):
        mm = m + d
        yy = y + (mm - 1) // 12
        mm = (mm - 1) % 12 + 1
        out.add("%04d-%02d" % (yy, mm))
    return out


INSTRUMENT_MONTHS = {}
for _m, _why in INSTRUMENT.items():
    for _n in months_around(_m):
        INSTRUMENT_MONTHS[_n] = (_m, _why)


def step(month):
    """The month before `month`."""
    y, m = int(month[:4]), int(month[5:])
    return "%04d-%02d" % (y - 1, 12) if m == 1 else "%04d-%02d" % (y, m - 1)


def mom(s, month):
    """Month-on-month ratio minus one, or None if either month is missing."""
    p = step(month)
    if p not in s or month not in s or not s[p]:
        return None
    return s[month] / s[p] - 1.0


def calendar_peers(s, month):
    """The same calendar transition (e.g. Mar->Apr) in every other year."""
    out = {}
    for m in sorted(s):
        if m[5:] == month[5:] and m != month:
            v = mom(s, m)
            if v is not None:
                out[m] = v
    return out


def outside(value, peers):
    """Is `value` outside the closed range of its peers? Returns '' or 'up'/'down'."""
    if not peers:
        return ""
    lo, hi = min(peers.values()), max(peers.values())
    if value > hi:
        return "up"
    if value < lo:
        return "down"
    return ""


def persists(s, month, direction, horizon=12):
    """Do the `horizon` months from `month` on stay outside the pre-break range?

    Pre-break range is the twelve months before `month`. A step up must leave
    every following month above the pre-break maximum; a step down, below the
    pre-break minimum. This is the "never returns" half of the rule, and it is
    what separates a step from a spike.
    """
    before = []
    m = step(month)
    for _ in range(12):
        if m in s:
            before.append(s[m])
        m = step(m)
    if len(before) < 12:
        return None                                   # not enough history to judge
    lo, hi = min(before), max(before)
    after, m = [], month
    for _ in range(horizon):
        if m not in s:
            break
        after.append(s[m])
        y, mm = int(m[:4]), int(m[5:])
        m = "%04d-%02d" % (y + 1, 1) if mm == 12 else "%04d-%02d" % (y, mm + 1)
    if len(after) < horizon:
        return None                                   # the run is still open
    if direction == "up":
        return all(v > hi for v in after)
    return all(v < lo for v in after)


def scan(s, first="2017-01"):
    """Every month whose month-on-month change is outside its calendar peers."""
    found = []
    for m in sorted(s):
        if m < first:
            continue
        v = mom(s, m)
        if v is None:
            continue
        peers = calendar_peers(s, m)
        if len(peers) < 4:                            # too few years to have a range
            continue
        d = outside(v, peers)
        if d:
            found.append({
                "month": m,
                "change": v,
                "direction": d,
                "peer_lo": min(peers.values()),
                "peer_hi": max(peers.values()),
                "peers": len(peers),
                "persists": persists(s, m, d),
            })
    return found


def inject(s, month, size):
    """A copy of `s` with a multiplicative step of `size` from `month` onward."""
    return {m: (v * (1 + size) if m >= month else v) for m, v in s.items()}


def positive_control(s, month="2022-06", sizes=None):
    """Can this test see a step at all, and how big must one be?

    A negative result needs a positive control: a scan that finds nothing is
    only worth reading if it would have found something. This injects a step
    of known size at a quiet month and reruns the identical scan, including the
    null case (size 0), which must not be detected.
    """
    if sizes is None:
        sizes = [0.0] + [round(0.05 * i, 2) for i in range(1, 21)]
    out = []
    for size in sizes:
        found = [b for b in scan(inject(s, month, size))
                 if b["month"] == month and b["persists"] is True]
        out.append((size, bool(found)))
    return month, out


def main():
    spider = series("en.wikipedia", "all-access", "spider")
    automated = series("en.wikipedia", "all-access", "automated")
    user = series("en.wikipedia", "all-access", "user")

    months = sorted(set(spider) | set(automated))
    machine = {m: spider.get(m, 0) + automated.get(m, 0) for m in months}
    latest = months[-1]

    # The table, for anyone who wants to redo this from the numbers alone.
    with open(os.path.join(HERE, "machines.csv"), "w", newline="") as fh:
        wr = csv.writer(fh)
        wr.writerow(["month", "spider", "automated", "machine", "user",
                     "machine_share", "machine_yoy", "machine_mom"])
        for m in months:
            tot = machine[m] + user.get(m, 0)
            wr.writerow([
                m, spider.get(m, ""), automated.get(m, ""), machine[m],
                user.get(m, ""),
                "%.4f" % (machine[m] / tot) if tot else "",
                "" if yoy(machine, m) is None else "%.4f" % yoy(machine, m),
                "" if mom(machine, m) is None else "%.4f" % mom(machine, m),
            ])

    out = []
    add = out.append
    add("# When did the machines arrive on en.wikipedia?")
    add("")
    add("Run by `machines.py` on %d months, %s to %s. The question and the rule "
        "were committed before this ran: "
        "https://untilnextsession.com/experiments/when-the-machines-arrived/"
        % (len(months), months[0], latest))
    add("")

    add("## 0. The trap, shown rather than described")
    add("")
    first_auto = min((m for m in months if automated.get(m)), default=None)
    add("`automated` is blank in every month before **%s** and non-zero from it "
        "onward. Its first twelve months, in millions of pageviews:" % first_auto)
    add("")
    add("| month | automated | spider | machine (the series used here) |")
    add("|---|---|---|---|")
    i = months.index(first_auto)
    for m in months[max(0, i - 2):i + 4]:
        add("| %s | %s | %.0f | %.0f |" % (
            m,
            "—" if not automated.get(m) else "%.0f" % (automated[m] / 1e6),
            spider.get(m, 0) / 1e6, machine[m] / 1e6))
    add("")
    add("A series built on `automated` alone arrives from nothing in %s, which is "
        "a category being created and not a machine being built. That is why the "
        "rule says `spider + automated`, and why %s is named as an instrument "
        "month before the scan runs." % (first_auto, first_auto))
    add("")

    add("## 1. The scan")
    add("")
    add("Every month from 2017-01 whose month-on-month change falls outside the "
        "range of the same calendar transition in every other year. `persists` "
        "is the second half of the rule: do the twelve months from the break on "
        "stay outside the range of the twelve months before it.")
    add("")
    add("| month | change | peer range | direction | persists | verdict |")
    add("|---|---|---|---|---|---|")
    breaks = scan(machine)
    surviving = []
    for b in breaks:
        inst = INSTRUMENT_MONTHS.get(b["month"])
        if inst:
            verdict = "instrument (%s)" % inst[0]
        elif b["persists"] is True:
            verdict = "**step**"
            surviving.append(b)
        elif b["persists"] is None:
            verdict = "too recent to judge"
        else:
            verdict = "spike, returns"
        add("| %s | %+.1f %% | %+.1f %% .. %+.1f %% | %s | %s | %s |" % (
            b["month"], 100 * b["change"], 100 * b["peer_lo"], 100 * b["peer_hi"],
            b["direction"],
            {True: "yes", False: "no", None: "n/a"}[b["persists"]],
            verdict))
    add("")
    add("%d months are outside their peer range; %d of them are in an instrument "
        "window named before the run; %d survive both halves of the rule."
        % (len(breaks), sum(1 for b in breaks if b["month"] in INSTRUMENT_MONTHS),
           len(surviving)))
    add("")

    add("## 2. The positive control: what a step would have had to look like")
    add("")
    add("The same scan, on the same series, with a multiplicative step of known "
        "size injected at a quiet month and every later month raised by it. The "
        "null case must not be detected; the smallest detected size is the floor "
        "of what this test can see. Three injection months, because the "
        "seasonal band is not the same width in every part of the year.")
    add("")
    where = ("2018-06", "2021-03", "2022-06", "2024-09")
    add("| injected step | " + " | ".join("at %s" % m for m in where) + " |")
    add("|" + "---|" * (len(where) + 1))
    controls = [positive_control(machine, m) for m in where]
    for i, (size, _) in enumerate(controls[0][1]):
        add("| %s | %s |" % (
            "none (null case)" if size == 0 else "%+.0f %%" % (100 * size),
            " | ".join("yes" if c[1][i][1] else "no" for c in controls)))
    floors = [next((s for s, hit in c[1] if hit), None) for c in controls]
    add("")
    if all(f is None for f in floors):
        add("**Nothing in the tested range is detected**, so the scan above is "
            "worthless as evidence and the result of this study is that the "
            "instrument, not the world, is the finding.")
    else:
        seen = [f for f in floors if f is not None]
        add("The null case is not detected at any of the four: %s. So the scan "
            "can see a sudden change of that size arriving in one month and "
            "nothing smaller — and the floor is not one number, it is %.0f %% "
            "to %.0f %% depending on where the step lands, because the bar a "
            "step has to clear is the highest month of the year before it. "
            "That spread is the honest limit on reading nothing from "
            "section 1." % (", ".join(
                ("%+.0f %% at %s" % (100 * f, c[0])) if f is not None
                else "nothing in range at %s" % c[0]
                for f, c in zip(floors, controls)),
                100 * min(seen), 100 * max(seen)))
    add("")

    add("## 3. The level, for scale")
    add("")
    add("| year | machine (bn) | year on year | human (bn) | machine share |")
    add("|---|---|---|---|---|")
    prev_mv = None
    for y in range(int(months[0][:4]), int(latest[:4]) + 1):
        ms = [m for m in months if m[:4] == str(y)]
        if len(ms) < 12:
            continue
        mv, uv = sum(machine[m] for m in ms), sum(user.get(m, 0) for m in ms)
        add("| %d | %.2f | %s | %.2f | %.1f %% |" % (
            y, mv / BN,
            "—" if prev_mv is None else "%+.1f %%" % (100 * (mv / prev_mv - 1)),
            uv / BN, 100 * mv / (mv + uv)))
        prev_mv = mv
    add("")
    add("Two of those years are big — +46.1 %% in 2020 and +35.0 %% in 2023 — "
        "and neither is a step. A year's growth is not a month's: %+.0f %% spread "
        "over twelve months moves the level by about %.1f %% a month, which never "
        "leaves the seasonal band and so is invisible to the scan by construction. "
        "2020's is also inside the instrument window named before the run and "
        "cannot be read as an arrival at all." % (46.1, (1.461 ** (1 / 12) - 1) * 100))
    add("")

    text = "\n".join(out) + "\n"
    with open(os.path.join(HERE, "machines.md"), "w") as fh:
        fh.write(text)
    print(text)


if __name__ == "__main__":
    main()