"""Build a SMALL, downloadable fill-analysis dataset from mm_live_legs.csv.

WHY THIS EXISTS. mm_live_legs.csv is ~31MB and growing ~10MB/day, so the
dashboard has to gzip it to embed it, and a .gz is exactly what an outside
analysis environment tends to refuse. This writes a plain CSV small enough to
embed and hand to anything.

TWO THINGS IT FIXES BESIDES SIZE:

1. ONE ROW PER RESTING EPISODE, NOT PER RE-POST. The legs log writes a row every
   time a leg is re-posted, so a leg that rested for an hour contributes ~50 rows
   and one that filled in 90s contributes one. Treating those as independent
   observations is a trap: it weights the sample toward legs that DID NOT fill,
   which is the outcome being predicted. Collapsing to (ticker, price) episodes
   makes each resting order one observation and moves exposure into its own
   column, where it belongs as a feature.

2. FEATURES ONLY EXIST FROM 2026-08-18. ask/bid/mid/pin_fair/league logging was
   added mid-stream: 0% of 08-16 rows have them, 21.7% of 08-17, 100% from 08-18.
   Rows without them cannot support the analysis and are dropped rather than
   silently becoming empty cells that a model reads as zeros.

The outcome column is `filled`, joined from mm_live_realfills.csv on
(ticker, price_cents).

Usage:  python research\\mm_fill_dataset.py
Writes: research\\mm_fill_dataset.csv
"""
import csv
import io
import datetime as dt
from collections import defaultdict

LEGS = r'E:\soccer\research\mm_live_legs.csv'
FILLS = r'E:\soccer\research\mm_live_realfills.csv'
OUT = r'E:\soccer\research\mm_fill_dataset.csv'

# Before this date the feature columns are empty or partial (see docstring).
FEATURES_FROM = '2026-08-18'

# ── FIELD DEFINITIONS ───────────────────────────────────────────────────
# One place, used by the dashboard to render the glossary next to the download.
# Keep in step with OUT_COLS: a column with no entry here ships undocumented.
#
# THE UNIT OF OBSERVATION IS A RESTING ORDER, not a re-post and not a leg. A
# leg re-posted 50 times over an hour is ONE row with minutes_exposed=60 and
# n_posts=50. Treating re-posts as independent rows biases the sample toward
# legs that did not fill, which is the outcome being predicted.
FIELD_DOCS = [
    ('ts_first', 'When this resting order first went on the book (local time).'),
    ('minutes_exposed', 'How long it stayed on the book across all re-posts. '
                        'A feature, not an outcome: longer exposure is more '
                        'chance to fill.'),
    ('n_posts', 'How many re-posts were collapsed into this one row. Orders '
                'carry a 90s TTL and are refreshed at 45s, so this is roughly '
                'minutes_exposed/0.75.'),
    ('league', 'Division code (E0, MLS, AR1, ...).'),
    ('event', 'Kalshi event ticker -- the match.'),
    ('ticker', 'Kalshi market ticker -- the individual leg being sold.'),
    ('side', 'Leg abbreviation: a team code, or TIE for the draw.'),
    ('price_cents', 'Our ask, in cents. This is the price the order rested at.'),
    ('ask_at_post', 'Best ask on the market when we posted.'),
    ('bid_at_post', 'Best bid when we posted.'),
    ('mid_at_post', 'Mid when we posted.'),
    ('spread_c', 'ask_at_post - bid_at_post, in cents. Mostly 1c: the venue '
                 'prices on a 1-cent grid, so this is usually at the tick floor.'),
    ('pin_fair', 'Pinnacle fair probability for this outcome (overround removed). '
                 'The independent estimate of what the leg is worth.'),
    ('ask_vs_pin_c', 'Our ask minus pin_fair, in cents. Positive = we are asking '
                     'above what Pinnacle thinks it is worth, which is the '
                     'maker edge on that leg.'),
    ('above_touch_c', 'Cents ABOVE the best ask we quoted. 0 = at the touch. '
                      'Non-zero comes from the budget rule: once part of a book '
                      'has filled, the rest may only be sold at prices that '
                      'still clear the payout, so a stranded book quotes high.'),
    ('queue_ahead', 'Contracts already resting at our price when we posted. The '
                    'number the whole experiment exists to calibrate -- a fill '
                    'behind a large queue means the queue cleared, not that we '
                    'were competitive. Median is ~4,800.'),
    ('bid_size', 'Depth on the bid at post.'),
    ('volume', 'Contracts traded on this leg.'),
    ('volume_24h', 'Contracts traded on this leg in the last 24h.'),
    ('book_volume', 'Contracts traded across ALL THREE legs of the match. The '
                    'strongest single predictor of book completion found so far '
                    '(+0.091 AUC); min-leg volume was null.'),
    ('open_interest', 'Open interest on this leg.'),
    ('hrs_to_ko', 'Hours until kickoff at post. Negative would be in play, but '
                  'new books are never opened in play.'),
    ('collected', 'What this BOOK had already banked from earlier fills, per '
                  'contract, when this leg was posted. 0 = nothing filled yet.'),
    ('target', 'What the book must collect in total to clear the $1.00 payout '
               'plus fees plus margin. Roughly 1.017.'),
    ('scale', 'target-minus-collected divided by the remaining legs at the '
              'touch. >1 means the touches do NOT cover what the book still '
              'needs -- which gates a new book and forces a committed one to '
              'quote above the touch.'),
    ('fresh', '1 if this was the first time this leg had ever been offered, '
              '0 if it was already in the offered set from an earlier session.'),
    ('filled', 'OUTCOME. 1 if the order was filled, 0 if it expired or was '
               'cancelled. Joined from mm_live_realfills.csv on '
               '(ticker, price_cents).'),
    ('fill_qty', 'Contracts filled, blank if never filled. Kalshi can fill '
                 'fractionally.'),
    ('seconds_to_fill', 'Seconds from post to fill, blank if never filled.'),
    ('is_xlow', '1 if the frozen xlow rule ALSO fired on this exact fixture and '
                'side, 0 if not. Both signal rules are home/away only, so a '
                'draw leg is always 0. BLANK means the fixture could not be '
                'resolved (unmapped team abbreviation) -- unknown, not "no".'),
    ('is_xhigh', '1 if the frozen xhigh rule ALSO fired on this exact fixture '
                 'and side, 0 if not. Same blank/0 convention as is_xlow. '
                 'xhigh overlaps xlow heavily -- both fire on Kalshi CLV in the '
                 'same band -- so the two columns are not independent.'),
]

_MONTHS = {'JAN': 1, 'FEB': 2, 'MAR': 3, 'APR': 4, 'MAY': 5, 'JUN': 6,
           'JUL': 7, 'AUG': 8, 'SEP': 9, 'OCT': 10, 'NOV': 11, 'DEC': 12}


def _parse_event(event):
    """'KXEPLGAME-26AUG23MCIBOU' -> ('2026-08-23', 'MCIBOU'), or (None, None).

    The Kalshi event code packs the date and BOTH team abbreviations with no
    separator, and the abbreviations are 2-4 characters, so the pair cannot be
    split on width. It does not need to be: the HOME abbr is a prefix of the
    pair and the AWAY abbr is a suffix, which is enough to place a leg.
    """
    import re
    tail = (event or '').rsplit('-', 1)[-1]
    m = re.match(r'^(\d{2})([A-Z]{3})(\d{2})([A-Z0-9]+)$', tail)
    if not m or m.group(2) not in _MONTHS:
        return None, None
    yy, mon, dd, teams = m.group(1), m.group(2), m.group(3), m.group(4)
    return f'20{yy}-{_MONTHS[mon]:02d}-{dd}', teams


def _leg_side(teams, leg):
    """'home' | 'draw' | 'away' for a leg abbreviation, or None if ambiguous."""
    if not teams or not leg:
        return None
    if leg == 'TIE':
        return 'draw'
    starts, ends = teams.startswith(leg), teams.endswith(leg)
    if starts and not ends:
        return 'home'
    if ends and not starts:
        return 'away'
    return None            # e.g. both ends match -- refuse to guess


def signal_hits():
    """{(game_date, league, teams_code, side)} for xlow and for xhigh.

    WHY THIS JOIN EXISTS. The maker quotes all three legs of a book blind to
    what the signal rules think. Tagging each leg with whether xlow or xhigh
    also fired on that exact fixture AND side makes it possible to ask whether
    a leg the signal rules call mispriced behaves differently -- fills faster,
    completes more often -- which is a question neither dataset can answer
    alone.

    Keyed on the FIXTURE, not on (date, league): several matches in one league
    share a date, and a coarser key would tag the wrong game. The Kalshi
    abbreviations come from team_map via maps.kalshi_abr, so a team missing
    there simply produces no key and the row is left blank rather than a
    confident 0.

    Returns ({}, {}) rather than raising if anything is unavailable -- this is
    an annotation on a dataset, and it must never stop the dataset building.
    """
    try:
        import forward_test_unique_book as F
        import maps
    except Exception as e:
        print(f'  [warn] signal join unavailable ({type(e).__name__}: {e}); '
              f'is_xlow/is_xhigh will be blank')
        return None, None

    xlow, xhigh = set(), set()
    try:
        for fx in F.load_fixtures(require_result=False):
            r0 = fx['pre'][0] if fx['pre'] else {}
            h, a = r0.get('home'), r0.get('away')
            if not (h and a):
                continue
            ha, aa = maps.kalshi_abr(h), maps.kalshi_abr(a)
            if not (ha and aa):
                continue
            key = (fx['gd'], fx['league'], f'{ha}{aa}'.upper())
            b = F.fire_xlow(fx)
            if b and b.get('side'):
                xlow.add(key + (b['side'],))
            d = F.fire_xhigh_detail(fx)
            if d and d.get('side'):
                xhigh.add(key + (d['side'],))
    except Exception as e:
        print(f'  [warn] signal join failed ({type(e).__name__}: {e}); '
              f'is_xlow/is_xhigh will be blank')
        return None, None
    return xlow, xhigh


OUT_COLS = [
    'ts_first', 'minutes_exposed', 'n_posts',
    'league', 'event', 'ticker', 'side',
    'price_cents', 'ask_at_post', 'bid_at_post', 'mid_at_post',
    'spread_c', 'pin_fair', 'ask_vs_pin_c', 'above_touch_c',
    'queue_ahead', 'bid_size', 'volume', 'volume_24h', 'book_volume',
    'open_interest', 'hrs_to_ko', 'collected', 'target', 'scale', 'fresh',
    'filled', 'fill_qty', 'seconds_to_fill',
    'is_xlow', 'is_xhigh',
]


def _read(path):
    """Read a CSV that may contain NUL bytes from an interrupted write."""
    raw = open(path, 'rb').read()
    if b'\x00' in raw:
        raw = raw.replace(b'\x00', b'')
    return list(csv.DictReader(io.StringIO(raw.decode('utf-8', errors='replace'))))


def _f(v):
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


def _hit(hits, a, ticker):
    """1 / 0 / '' for whether a signal rule fired on this exact leg.

    BLANK, NOT ZERO, when the fixture cannot be resolved -- an unmatched team
    abbreviation or an unparseable event code is 'unknown', and writing 0 there
    would tell a model the rule declined this leg when in fact nobody asked.
    Both rules are home/away only, so a draw leg is a genuine 0.
    """
    if hits is None:
        return ''
    gd, teams = _parse_event(a.get('event', ''))
    leg = (ticker or '').rsplit('-', 1)[-1]
    side = _leg_side(teams, leg)
    if not gd or not teams or side is None:
        return ''
    if side == 'draw':
        return 0
    return 1 if (gd, a.get('league', ''), teams, side) in hits else 0


def main():
    xlow_hits, xhigh_hits = signal_hits()
    fills = defaultdict(list)
    for r in _read(FILLS):
        try:
            key = (r['ticker'], int(float(r['price_cents'])))
        except (KeyError, TypeError, ValueError):
            continue
        fills[key].append(r)

    legs = _read(LEGS)
    print(f'legs rows read      : {len(legs):,}')

    kept = [r for r in legs if (r.get('ts') or '')[:10] >= FEATURES_FROM]
    print(f'rows >= {FEATURES_FROM}  : {len(kept):,}')

    # Collapse to one episode per (ticker, price). The ratchet holds a leg's
    # price flat while it rests, so this is almost always one contiguous run.
    ep = {}
    for r in kept:
        try:
            pc = int(float(r['price_cents']))
        except (TypeError, ValueError):
            continue
        key = (r['ticker'], pc)
        e = ep.get(key)
        if e is None:
            ep[key] = {'first': r, 'last': r, 'n': 1}
        else:
            e['n'] += 1
            if r['ts'] > e['last']['ts']:
                e['last'] = r
            if r['ts'] < e['first']['ts']:
                e['first'] = r

    print(f'resting episodes    : {len(ep):,}')

    rows = []
    n_filled = 0
    for (tkr, pc), e in ep.items():
        a, b = e['first'], e['last']
        t0 = t1 = None
        mins = None
        try:
            t0 = dt.datetime.fromisoformat(a['ts'])
            t1 = dt.datetime.fromisoformat(b['ts'])
            mins = (t1 - t0).total_seconds() / 60.0
        except (TypeError, ValueError):
            pass

        ask, bid = _f(a.get('ask_at_post')), _f(a.get('bid_at_post'))
        pin = _f(a.get('pin_fair'))
        spread = (ask - bid) * 100 if (ask is not None and bid is not None) else None
        ask_vs_pin = (pc / 100.0 - pin) * 100 if pin is not None else None

        # TIME-AWARE MATCH, ON posted_at NOT ts.
        #
        # Matching on (ticker, price) alone would attach a fill to any episode
        # that ever quoted that price, including one that started after the fill.
        # But matching on the fill's `ts` is also wrong: that is when the maker
        # DETECTED the fill on a later poll, not when it happened. Cycles run
        # ~70s (and were 300s during the 2026-08-19 stall), so detection lands
        # 120-150s after the last re-post and a naive window drops real fills --
        # it discarded 12 of 45 when first written, all of them genuine.
        #
        # `posted_at` is the placement time of the order that actually filled, so
        # it falls inside the episode's own re-post window by construction. Small
        # tolerance for clock skew only. Falls back to `ts` with a generous
        # allowance when posted_at is missing.
        cand = fills.get((tkr, pc)) or []
        fl = []
        for x in (cand if (t0 and t1) else []):
            stamp = x.get('posted_at') or ''
            slack = dt.timedelta(seconds=60)
            if not stamp:
                stamp = x.get('ts') or ''
                slack = dt.timedelta(seconds=420)   # detection lag, incl. stalls
            if not stamp:
                continue
            try:
                xd = dt.datetime.fromisoformat(stamp)
            except ValueError:
                continue
            if t0 - slack <= xd <= t1 + slack:
                fl.append(x)
        if fl:
            n_filled += 1
        qty = sum(_f(x.get('count')) or 0 for x in fl)
        stf = _f(fl[0].get('seconds_to_fill')) if fl else None

        rows.append({
            'ts_first': a['ts'],
            'minutes_exposed': f'{mins:.2f}' if mins is not None else '',
            'n_posts': e['n'],
            'league': a.get('league', ''),
            'event': a.get('event', ''),
            'ticker': tkr,
            'side': a.get('side', ''),
            'price_cents': pc,
            'ask_at_post': a.get('ask_at_post', ''),
            'bid_at_post': a.get('bid_at_post', ''),
            'mid_at_post': a.get('mid_at_post', ''),
            'spread_c': f'{spread:.2f}' if spread is not None else '',
            'pin_fair': a.get('pin_fair', ''),
            'ask_vs_pin_c': f'{ask_vs_pin:.2f}' if ask_vs_pin is not None else '',
            'above_touch_c': a.get('above_touch_c', ''),
            'queue_ahead': a.get('queue_ahead', ''),
            'bid_size': a.get('bid_size', ''),
            'volume': a.get('volume', ''),
            'volume_24h': a.get('volume_24h', ''),
            'book_volume': a.get('book_volume', ''),
            'open_interest': a.get('open_interest', ''),
            'hrs_to_ko': a.get('hrs_to_ko', ''),
            'collected': a.get('collected', ''),
            'target': a.get('target', ''),
            'scale': a.get('scale', ''),
            'fresh': a.get('fresh', ''),
            'filled': 1 if fl else 0,
            'fill_qty': f'{qty:.2f}' if fl else '',
            'seconds_to_fill': f'{stf:.1f}' if stf is not None else '',
            'is_xlow': _hit(xlow_hits, a, tkr),
            'is_xhigh': _hit(xhigh_hits, a, tkr),
        })

    rows.sort(key=lambda r: r['ts_first'])
    with open(OUT, 'w', newline='', encoding='utf-8') as fh:
        w = csv.DictWriter(fh, fieldnames=OUT_COLS)
        w.writeheader()
        w.writerows(rows)

    import os
    size = os.path.getsize(OUT)
    print()
    print(f'wrote {OUT}')
    print(f'  rows          : {len(rows):,}')
    print(f'  filled        : {n_filled:,}  ({100.0*n_filled/max(len(rows),1):.1f}%)')
    print(f'  size          : {size/1e6:.2f} MB')
    print(f'  base64 embed  : {size*1.34/1e6:.2f} MB')


if __name__ == '__main__':
    main()