#!/usr/bin/env python3
"""Build 1-minute bid/ask bars (UTC) from the MetaTrader 5 tick exports in ~/data/ticks/.

Usage (needs ~/.venv with pandas):
    ~/.venv/bin/python bars_mt5.py XAUUSD.r ~/data/bars/XAUUSD_mt5_m1.csv

Input: ~/data/ticks/<SYMBOL>/TickExport_<SYMBOL>_YYYYMM.csv.gz, columns
time_msc,bid,ask,last,volume,flags. `time_msc` is broker SERVER time written as if it were UTC.
This broker's server clock is EET/EEST (UTC+2 in winter, UTC+3 in EU summer time), which is
verified below against Dukascopy ticks on the same instant: if the bids at 2026-01-02 00:00 UTC
disagree by more than a few dollars the offset is wrong and the script stops.

Output columns are the same as bars.py so backtest.py can read either:
minute (ISO, UTC), bid_open, bid_high, bid_low, bid_close, ask_open, ask_high, ask_low, ask_close, ticks
"""
import glob
import os
import sys
from datetime import datetime, timezone

import pandas as pd

DATA = os.path.expanduser("~/data")

# EU summer time 2026: last Sunday of March 01:00 UTC to last Sunday of October 01:00 UTC.
DST_START_MS = int(datetime(2026, 3, 29, 1, tzinfo=timezone.utc).timestamp() * 1000)
DST_END_MS = int(datetime(2026, 10, 25, 1, tzinfo=timezone.utc).timestamp() * 1000)
HOUR_MS = 3_600_000


def to_utc_ms(server_ms):
    """Server time (UTC+2/+3) expressed as epoch ms -> true UTC epoch ms, vectorised."""
    utc = server_ms - 2 * HOUR_MS
    summer = (utc - HOUR_MS >= DST_START_MS) & (utc - HOUR_MS < DST_END_MS)
    return utc - summer.astype("int64") * HOUR_MS


def bars_for(path):
    out = []
    for chunk in pd.read_csv(path, usecols=["time_msc", "bid", "ask"], dtype={"time_msc": "int64"},
                             chunksize=2_000_000):
        chunk["m"] = to_utc_ms(chunk["time_msc"].to_numpy()) // 60_000
        g = chunk.groupby("m", sort=True)
        b = g["bid"].agg(["first", "max", "min", "last"])
        a = g["ask"].agg(["first", "max", "min", "last"])
        n = g.size()
        df = pd.concat([b, a, n], axis=1)
        df.columns = ["bid_open", "bid_high", "bid_low", "bid_close", "ask_open", "ask_high", "ask_low", "ask_close", "ticks"]
        out.append(df)
    df = pd.concat(out)
    # a minute can straddle two chunks: merge duplicates
    if df.index.has_duplicates:
        g = df.groupby(level=0)
        df = pd.DataFrame({
            "bid_open": g["bid_open"].first(), "bid_high": g["bid_high"].max(), "bid_low": g["bid_low"].min(),
            "bid_close": g["bid_close"].last(), "ask_open": g["ask_open"].first(), "ask_high": g["ask_high"].max(),
            "ask_low": g["ask_low"].min(), "ask_close": g["ask_close"].last(), "ticks": g["ticks"].sum()})
    return df


def verify_offset(df):
    """Compare with the Dukascopy bar at 2026-01-02 00:00 UTC (bid 4347.798 in the cached file)."""
    m = int(datetime(2026, 1, 2, 0, 0, tzinfo=timezone.utc).timestamp()) // 60
    if m not in df.index:
        print("verify: no MT5 bar at 2026-01-02 00:00 UTC; cannot check offset", file=sys.stderr)
        return
    diff = abs(df.loc[m, "bid_open"] - 4347.798)
    print(f"verify: MT5 bid_open at 2026-01-02 00:00 UTC = {df.loc[m, 'bid_open']:.3f}, Dukascopy 4347.798, diff {diff:.3f}")
    if diff > 5:
        sys.exit("offset check failed: server-time offset assumption is wrong")


def main():
    symbol, out = sys.argv[1], sys.argv[2]
    files = sorted(glob.glob(os.path.join(DATA, "ticks", symbol, f"TickExport_{symbol}_*.csv.gz")))
    frames = []
    for p in files:
        df = bars_for(p)
        print(f"{os.path.basename(p)}: {len(df)} bars, {int(df['ticks'].sum())} ticks", flush=True)
        frames.append(df)
    df = pd.concat(frames).sort_index()
    verify_offset(df)
    df.insert(0, "minute", [datetime.fromtimestamp(m * 60, tz=timezone.utc).isoformat() for m in df.index])
    df.to_csv(out, index=False, float_format="%.3f")
    print(f"{len(df)} bars -> {out}")


if __name__ == "__main__":
    main()