"""
overnight_share.py - how much daily variance arrives while a market is shut.

The realized-variance estimator used by the forecast engine is

    daily variance = overnight squared + Garman-Klass

The overnight term exists because a range estimator computed on one session
cannot see the gap from the previous close to the next open. On SPY that gap
carries 40 to 46 percent of daily variance. This script measures whether the
same is true of the instruments a MetaTrader user actually charts, which
trade close to around the clock.

The implied variance comes from options on exchange-traded funds with a
six-and-a-half hour session, while the MQL5 engine measures realized variance
on the broker's own bars. If the two venues split their variance differently,
the realized side has to be measured locally rather than imported.

So the share is measured twice:

  the four ETFs            GLD, SPY, USO, QQQ      session traded, the feed side
  four near-24h futures    GC, ES, CL, NQ          the closest free proxy for a
                                                   broker CFD, the terminal side

Assumes yfinance daily bars carry the session open rather than a synthetic one,
which is what makes the prior-close-to-open gap meaningful.

    python overnight_share.py

Reads only. Never trades.
"""
import math
import sys

import numpy as np
import pandas as pd

try:
    import yfinance as yf
except ImportError:
    sys.exit("yfinance is required:  pip install yfinance")

#--- the feed side: exchange-traded funds, roughly a 6.5 hour session
ETFS = [("GLD", "gold ETF"),
        ("SPY", "S&P 500 ETF"),
        ("USO", "crude oil ETF"),
        ("QQQ", "Nasdaq 100 ETF")]

#--- the terminal side: futures trade close to 23 hours, which is the nearest
#--- freely available stand-in for the broker CFDs a reader will actually chart
FUTURES = [("GC=F", "gold futures", "XAUUSD"),
           ("ES=F", "S&P 500 futures", "US500"),
           ("CL=F", "crude oil futures", "USOIL"),
           ("NQ=F", "Nasdaq 100 futures", "US100")]

START = "2019-01-01"

#--- SPY is the control. Its overnight share was measured independently at 42
#--- to 44 percent, so a run that does not land in this range is a broken run
#--- rather than a finding
SPY_EXPECTED = (0.40, 0.46)

#--- Garman-Klass can go slightly negative on a bar whose open and close sit
#--- outside a very narrow high-low range, which is a quote artifact rather
#--- than a measurement. The count is reported instead of being hidden
GK_CONST = 2.0 * math.log(2.0) - 1.0


def garman_klass(o, h, l, c):
    """
    Garman-Klass variance for one bar, from its open, high, low and close.

    Assumes all four prices are positive and drawn from the same session.
    """
    hl = np.log(h / l)
    co = np.log(c / o)
    return 0.5 * hl * hl - GK_CONST * co * co


def overnight_share(ticker, start=START):
    """
    Split daily variance into the part that arrives overnight and the rest.

    The overnight piece is the squared log gap from the previous close to this
    open. The intraday piece is Garman-Klass on the bar itself. The share is
    the aggregate ratio, not the mean of daily ratios, because a mean over
    days gives a quiet day the same weight as a violent one.

    Returns None when the download yields nothing usable, so the caller can
    report the absence rather than treating it as a zero.
    """
    try:
        df = yf.download(ticker, start=start, progress=False, auto_adjust=False)
    except Exception as reason:
        print("    download failed for %s: %s" % (ticker, reason))
        return None
    if df is None or df.empty:
        return None
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = df.columns.get_level_values(0)

    need = ("Open", "High", "Low", "Close")
    if any(col not in df.columns for col in need):
        return None
    df = df[list(need)].dropna()
    if len(df) < 200:
        return None

    o = df["Open"].to_numpy(dtype=float)
    h = df["High"].to_numpy(dtype=float)
    l = df["Low"].to_numpy(dtype=float)
    c = df["Close"].to_numpy(dtype=float)
    ok = (o > 0) & (h > 0) & (l > 0) & (c > 0)
    o, h, l, c = o[ok], h[ok], l[ok], c[ok]

    #--- the gap needs a previous close, so the first bar is dropped
    gap = np.log(o[1:] / c[:-1])
    on = gap * gap
    gk = garman_klass(o[1:], h[1:], l[1:], c[1:])

    gk_negative = int(np.sum(gk < 0.0))
    gk = np.clip(gk, 0.0, None)

    total = on + gk
    keep = total > 0.0
    on, gk, total = on[keep], gk[keep], total[keep]
    if len(total) < 200:
        return None

    aggregate = float(np.sum(on) / np.sum(total))
    per_day = on / total
    return {"days": int(len(total)),
            "aggregate": aggregate,
            "median_day": float(np.median(per_day)),
            "gk_negative": gk_negative,
            "ann_vol": float(np.sqrt(np.mean(total) * 252.0) * 100.0)}


def report(rows, heading):
    """Print one block of results. Assumes rows is a list of (label, result)."""
    print()
    print(heading)
    print("-" * 76)
    print("  %-22s %7s %12s %12s %10s" %
          ("instrument", "days", "overnight %", "median day %", "ann vol %"))
    for label, r in rows:
        if r is None:
            print("  %-22s %7s %12s %12s %10s" % (label, "-", "no data", "-", "-"))
            continue
        print("  %-22s %7d %11.1f%% %11.1f%% %9.1f%%" %
              (label, r["days"], 100.0 * r["aggregate"],
               100.0 * r["median_day"], r["ann_vol"]))


def main():
    """
    Measure both groups, check the SPY control, and state what the result means.

    Assumes network access to yfinance. Returns 1 when the control fails, so a
    broken download is never mistaken for a measurement.
    """
    print("overnight variance share, aggregate over the whole sample")
    print("estimator: overnight squared, against overnight squared plus Garman-Klass")
    print("sample starts %s" % START)

    etf_rows = []
    for ticker, label in ETFS:
        etf_rows.append(("%s  (%s)" % (ticker, label), overnight_share(ticker)))
    report(etf_rows, "THE FEED SIDE: exchange-traded funds, 6.5 hour session")

    fut_rows = []
    for ticker, label, local in FUTURES:
        fut_rows.append(("%s  (-> %s)" % (ticker, local), overnight_share(ticker)))
    report(fut_rows, "THE TERMINAL SIDE: futures, close to 23 hours")

    #--- the control has to reproduce before anything else is believed
    spy = dict(etf_rows)["SPY  (S&P 500 ETF)"]
    print()
    print("control")
    print("-" * 76)
    if spy is None:
        print("  SPY did not download. Nothing here can be trusted; rerun.")
        return 1
    lo, hi = SPY_EXPECTED
    got = spy["aggregate"]
    if lo <= got <= hi:
        print("  SPY overnight share %.1f%%, inside the expected %.0f to %.0f%%."
              % (100.0 * got, 100.0 * lo, 100.0 * hi))
        print("  The measurement reproduces, so the futures numbers above mean something.")
    else:
        print("  SPY overnight share %.1f%%, OUTSIDE the expected %.0f to %.0f%%."
              % (100.0 * got, 100.0 * lo, 100.0 * hi))
        print("  Something changed in the data or the code. Do not use this run.")
        return 1

    #--- the decision the measurement exists to make, in one line
    print()
    print("what this decides")
    print("-" * 76)
    usable = [(lab, r) for lab, r in fut_rows if r is not None]
    if not usable:
        print("  No futures data. The question is still open.")
        return 1
    shares = [r["aggregate"] for _, r in usable]
    print("  Futures overnight share runs %.1f%% to %.1f%% across %d instruments."
          % (100.0 * min(shares), 100.0 * max(shares), len(usable)))
    if max(shares) < 0.10:
        print("  Small everywhere. On a near-24h instrument the estimator is")
        print("  effectively Garman-Klass alone, and the SPY share does not")
        print("  carry over to it.")
    elif min(shares) > 0.25:
        print("  Large everywhere. The overnight term carries real weight on")
        print("  these instruments too, and the estimator needs no branch.")
    else:
        print("  Mixed. The share is instrument-dependent, so it has to be")
        print("  published per instrument rather than stated once.")
    return 0


if __name__ == "__main__":
    sys.exit(main())