کد بک‌تست قابل بازتولید EMA 50/200 — BTC/USDT

#!/usr/bin/env python3
"""Nexito Quant Runner — EMA 50/200 BTC/USDT reproducible baseline.

No third-party packages are required. The runner downloads public Binance Spot
1d klines, freezes the normalized dataset to CSV, computes the baseline exactly
as documented in Nexito, and emits an import-ready JSON result.
"""
from __future__ import annotations

import argparse
import csv
import hashlib
import hmac
import io
import json
import math
import os
import statistics
import sys
import time
import urllib.parse
import urllib.request
import urllib.error
import zipfile
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Iterable, Optional

RUNNER_VERSION = "NXQR-1.1.0"
RESULT_SCHEMA = "nexito.quant.backtest-result.v1"
DAY_MS = 86_400_000
BASE_URLS = [
    "https://data-api.binance.vision",
    "https://api.binance.com",
    "https://api1.binance.com",
]
ARCHIVE_BASE = "https://data.binance.vision/data/spot"
CSV_FIELDS = [
    "open_time_ms", "open", "high", "low", "close", "volume", "close_time_ms",
    "quote_asset_volume", "number_of_trades", "taker_buy_base_volume",
    "taker_buy_quote_volume",
]


@dataclass(frozen=True)
class Bar:
    open_time_ms: int
    open: float
    high: float
    low: float
    close: float
    volume: float
    close_time_ms: int
    quote_asset_volume: float
    number_of_trades: int
    taker_buy_base_volume: float
    taker_buy_quote_volume: float

    @property
    def day(self) -> date:
        return datetime.fromtimestamp(self.open_time_ms / 1000, tz=timezone.utc).date()


def iso_z_from_ms(ms: int) -> str:
    return datetime.fromtimestamp(ms / 1000, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def canonical_json_bytes(obj: object) -> bytes:
    return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")


def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def api_get_json(base: str, path: str, params: dict, timeout: int = 25):
    url = base.rstrip("/") + path + "?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, headers={"User-Agent": f"NexitoQuantRunner/{RUNNER_VERSION}"})
    with urllib.request.urlopen(req, timeout=timeout) as response:
        if response.status != 200:
            raise RuntimeError(f"HTTP {response.status} from {base}")
        return json.loads(response.read().decode("utf-8"))


class ArchiveNotFound(RuntimeError):
    pass


def http_get_bytes(url: str, timeout: int = 35, attempts: int = 3) -> bytes:
    last_error: Optional[Exception] = None
    for attempt in range(1, attempts + 1):
        try:
            req = urllib.request.Request(url, headers={
                "User-Agent": f"NexitoQuantRunner/{RUNNER_VERSION}",
                "Accept": "*/*",
            })
            with urllib.request.urlopen(req, timeout=timeout) as response:
                if response.status != 200:
                    raise RuntimeError(f"HTTP {response.status} for {url}")
                return response.read()
        except urllib.error.HTTPError as exc:
            if exc.code == 404:
                raise ArchiveNotFound(f"Archive not found: {url}") from exc
            last_error = exc
        except Exception as exc:
            last_error = exc
        if attempt < attempts:
            time.sleep(min(4.0, 0.75 * (2 ** (attempt - 1))))
    raise RuntimeError(f"Download failed after {attempts} attempt(s): {url} | {last_error}")


def normalize_archive_timestamp(value: str | int) -> int:
    raw = int(value)
    # Binance Spot archive timestamps are milliseconds before 2025-01-01 and
    # microseconds from 2025-01-01 onward. Nexito freezes milliseconds.
    if raw >= 100_000_000_000_000:
        return raw // 1000
    return raw


def parse_archive_zip(data: bytes, expected_filename: str) -> list[Bar]:
    try:
        zf = zipfile.ZipFile(io.BytesIO(data))
    except zipfile.BadZipFile as exc:
        raise RuntimeError(f"Invalid Binance archive ZIP: {expected_filename}") from exc
    names = [n for n in zf.namelist() if not n.endswith('/')]
    csv_names = [n for n in names if n.lower().endswith('.csv')]
    if len(csv_names) != 1:
        raise RuntimeError(f"Expected exactly one CSV in {expected_filename}, found {len(csv_names)}")
    raw_csv = zf.read(csv_names[0]).decode('utf-8-sig')
    bars: list[Bar] = []
    for row_no, row in enumerate(csv.reader(io.StringIO(raw_csv)), start=1):
        if not row:
            continue
        if row_no == 1 and not row[0].strip().lstrip('-').isdigit():
            continue
        if len(row) < 11:
            raise RuntimeError(f"Malformed Binance archive row {row_no} in {expected_filename}")
        bars.append(Bar(
            open_time_ms=normalize_archive_timestamp(row[0]),
            open=float(row[1]), high=float(row[2]), low=float(row[3]), close=float(row[4]),
            volume=float(row[5]), close_time_ms=normalize_archive_timestamp(row[6]),
            quote_asset_volume=float(row[7]), number_of_trades=int(row[8]),
            taker_buy_base_volume=float(row[9]), taker_buy_quote_volume=float(row[10]),
        ))
    return bars


def verify_archive_checksum(zip_bytes: bytes, checksum_bytes: bytes, filename: str) -> str:
    text = checksum_bytes.decode('utf-8', errors='replace').strip()
    token = text.split()[0].lower() if text else ''
    if not (len(token) == 64 and all(c in '0123456789abcdef' for c in token)):
        raise RuntimeError(f"Invalid Binance CHECKSUM for {filename}")
    actual = sha256_bytes(zip_bytes)
    if not hmac.compare_digest(token, actual):
        raise RuntimeError(f"Binance CHECKSUM mismatch for {filename}")
    return actual


def month_floor(d: date) -> date:
    return date(d.year, d.month, 1)


def next_month(d: date) -> date:
    return date(d.year + (1 if d.month == 12 else 0), 1 if d.month == 12 else d.month + 1, 1)


def month_last_day(d: date) -> date:
    return next_month(month_floor(d)) - timedelta(days=1)


def archive_file_urls(symbol: str, interval: str, kind: str, stamp: str, archive_base: str) -> tuple[str, str, str]:
    if kind == 'monthly':
        filename = f"{symbol}-{interval}-{stamp}.zip"
        url = f"{archive_base.rstrip('/')}/monthly/klines/{symbol}/{interval}/{filename}"
    elif kind == 'daily':
        filename = f"{symbol}-{interval}-{stamp}.zip"
        url = f"{archive_base.rstrip('/')}/daily/klines/{symbol}/{interval}/{filename}"
    else:
        raise ValueError('unknown archive kind')
    return filename, url, url + '.CHECKSUM'


def download_archive_file(symbol: str, interval: str, kind: str, stamp: str, archive_base: str) -> tuple[list[Bar], dict]:
    filename, zip_url, checksum_url = archive_file_urls(symbol, interval, kind, stamp, archive_base)
    print(f"[archive] {kind:7s} {stamp} · download + checksum", flush=True)
    zip_bytes = http_get_bytes(zip_url)
    checksum_bytes = http_get_bytes(checksum_url)
    zip_sha = verify_archive_checksum(zip_bytes, checksum_bytes, filename)
    bars = parse_archive_zip(zip_bytes, filename)
    return bars, {
        'kind': kind,
        'file': filename,
        'sha256': zip_sha,
        'rows': len(bars),
        'url': zip_url,
    }


def fetch_klines_archive(symbol: str, interval: str, source_start: date, end: date, archive_base: str = ARCHIVE_BASE) -> tuple[list[Bar], str, list[dict]]:
    if interval != '1d':
        raise RuntimeError('Archive mode for this baseline currently supports interval=1d only')
    if end < source_start:
        raise RuntimeError('Archive end is before source start')

    all_bars: list[Bar] = []
    manifest: list[dict] = []
    cursor = month_floor(source_start)
    end_month = month_floor(end)

    while cursor < end_month:
        stamp = cursor.strftime('%Y-%m')
        try:
            bars, meta = download_archive_file(symbol, interval, 'monthly', stamp, archive_base)
            all_bars.extend(bars)
            manifest.append(meta)
        except ArchiveNotFound:
            # A missing monthly archive is recovered from official daily archives.
            day = max(cursor, source_start)
            last = min(month_last_day(cursor), end)
            recovered = 0
            while day <= last:
                try:
                    bars, meta = download_archive_file(symbol, interval, 'daily', day.isoformat(), archive_base)
                except ArchiveNotFound as exc:
                    raise RuntimeError(f"Missing Binance archive for required day {day.isoformat()}") from exc
                all_bars.extend(bars)
                manifest.append(meta)
                recovered += 1
                day += timedelta(days=1)
            if recovered == 0:
                raise RuntimeError(f"No Binance archive rows recovered for {stamp}")
        cursor = next_month(cursor)

    # The ending month may be incomplete and therefore may not have a monthly file yet.
    day = max(end_month, source_start)
    while day <= end:
        try:
            bars, meta = download_archive_file(symbol, interval, 'daily', day.isoformat(), archive_base)
        except ArchiveNotFound as exc:
            raise RuntimeError(
                f"Binance daily archive is not available for {day.isoformat()}. "
                "Daily archives are normally published after the UTC day closes; retry later or choose an earlier --end."
            ) from exc
        all_bars.extend(bars)
        manifest.append(meta)
        day += timedelta(days=1)

    dedup: dict[int, Bar] = {}
    for b in all_bars:
        if source_start <= b.day <= end:
            dedup[b.open_time_ms] = b
    bars = sorted(dedup.values(), key=lambda b: b.open_time_ms)
    if not bars:
        raise RuntimeError('Binance Public Data Archive returned no bars in the requested range')
    return bars, archive_base, manifest


def fetch_klines(symbol: str, interval: str, start_ms: int, end_ms: int) -> tuple[list[Bar], str]:
    last_error: Optional[Exception] = None
    for base in BASE_URLS:
        try:
            rows: list[list] = []
            cursor = start_ms
            while cursor <= end_ms:
                payload = api_get_json(base, "/api/v3/klines", {
                    "symbol": symbol,
                    "interval": interval,
                    "startTime": cursor,
                    "endTime": end_ms,
                    "limit": 1000,
                    "timeZone": "0",
                })
                if not isinstance(payload, list):
                    raise RuntimeError(f"Unexpected Binance response: {payload!r}")
                if not payload:
                    break
                rows.extend(payload)
                next_cursor = int(payload[-1][0]) + 1
                if next_cursor <= cursor:
                    raise RuntimeError("Binance pagination cursor did not advance")
                cursor = next_cursor
                if len(payload) < 1000:
                    break
                time.sleep(0.12)
            bars = [Bar(
                open_time_ms=int(r[0]), open=float(r[1]), high=float(r[2]), low=float(r[3]), close=float(r[4]),
                volume=float(r[5]), close_time_ms=int(r[6]), quote_asset_volume=float(r[7]),
                number_of_trades=int(r[8]), taker_buy_base_volume=float(r[9]), taker_buy_quote_volume=float(r[10]),
            ) for r in rows]
            bars.sort(key=lambda b: b.open_time_ms)
            dedup: dict[int, Bar] = {b.open_time_ms: b for b in bars}
            return list(dedup.values()), base
        except Exception as exc:
            last_error = exc
    raise RuntimeError(f"All Binance endpoints failed. Last error: {last_error}")


def write_dataset_csv(path: Path, bars: Iterable[Bar]) -> str:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", newline="", encoding="utf-8") as f:
        w = csv.writer(f, lineterminator="\n")
        w.writerow(CSV_FIELDS)
        for b in bars:
            w.writerow([
                b.open_time_ms, f"{b.open:.8f}", f"{b.high:.8f}", f"{b.low:.8f}", f"{b.close:.8f}",
                f"{b.volume:.8f}", b.close_time_ms, f"{b.quote_asset_volume:.8f}", b.number_of_trades,
                f"{b.taker_buy_base_volume:.8f}", f"{b.taker_buy_quote_volume:.8f}",
            ])
    return sha256_bytes(path.read_bytes())


def read_dataset_csv(path: Path) -> list[Bar]:
    out: list[Bar] = []
    with path.open(newline="", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            out.append(Bar(
                open_time_ms=int(row["open_time_ms"]), open=float(row["open"]), high=float(row["high"]),
                low=float(row["low"]), close=float(row["close"]), volume=float(row["volume"]),
                close_time_ms=int(row["close_time_ms"]), quote_asset_volume=float(row["quote_asset_volume"]),
                number_of_trades=int(row["number_of_trades"]), taker_buy_base_volume=float(row["taker_buy_base_volume"]),
                taker_buy_quote_volume=float(row["taker_buy_quote_volume"]),
            ))
    out.sort(key=lambda b: b.open_time_ms)
    return out


def ema(values: list[float], period: int) -> list[Optional[float]]:
    if period <= 0:
        raise ValueError("EMA period must be positive")
    out: list[Optional[float]] = [None] * len(values)
    if len(values) < period:
        return out
    seed = sum(values[:period]) / period
    out[period - 1] = seed
    alpha = 2.0 / (period + 1.0)
    prev = seed
    for i in range(period, len(values)):
        prev = alpha * values[i] + (1.0 - alpha) * prev
        out[i] = prev
    return out


def pct(v: Optional[float]) -> Optional[float]:
    return None if v is None or not math.isfinite(v) else round(v, 8)


def safe_ratio(a: float, b: float) -> Optional[float]:
    if b == 0:
        return None
    v = a / b
    return v if math.isfinite(v) else None


def std_sample(values: list[float]) -> float:
    return statistics.stdev(values) if len(values) >= 2 else 0.0


def backtest(bars: list[Bar], start: date, end: date, initial_capital: float, fee_pct: float, slippage_pct: float) -> dict:
    if not bars:
        raise RuntimeError("Dataset is empty")
    by_day = {b.day: i for i, b in enumerate(bars)}
    expected_days = [start + timedelta(days=i) for i in range((end - start).days + 1)]
    missing = [d.isoformat() for d in expected_days if d not in by_day]
    if missing:
        raise RuntimeError(f"Dataset gap inside test period ({len(missing)} missing day(s)); first: {missing[:5]}")

    closes = [b.close for b in bars]
    ema50 = ema(closes, 50)
    ema200 = ema(closes, 200)
    test_indices = [by_day[d] for d in expected_days]
    first_idx, last_idx = test_indices[0], test_indices[-1]
    if ema200[first_idx] is None:
        raise RuntimeError("Insufficient warm-up data for EMA200 at test start")

    fee = fee_pct / 100.0
    slip = slippage_pct / 100.0
    cash = initial_capital
    qty = 0.0
    entry_cash = 0.0
    entry_price = 0.0
    entry_fee = 0.0
    entry_time_ms = 0
    in_position = False
    trades: list[dict] = []
    equity_series: list[list] = []
    benchmark_series: list[list] = []
    drawdown_series: list[list] = []

    # Benchmark: buy at first test candle open, include entry friction and mark-to-market with exit friction.
    b0 = bars[first_idx]
    bench_entry_px = b0.open * (1.0 + slip)
    bench_qty = initial_capital / (bench_entry_px * (1.0 + fee))

    peak = initial_capital
    exposure_days = 0

    for idx in test_indices:
        bar = bars[idx]
        prev = idx - 1

        # Execute signal generated on previous candle close at today's open.
        if prev >= 1 and ema50[prev] is not None and ema200[prev] is not None and ema50[prev - 1] is not None and ema200[prev - 1] is not None:
            cross_up = ema50[prev] > ema200[prev] and ema50[prev - 1] <= ema200[prev - 1]
            cross_down = ema50[prev] < ema200[prev] and ema50[prev - 1] >= ema200[prev - 1]
            if cross_up and not in_position:
                entry_cash = cash
                entry_price = bar.open * (1.0 + slip)
                qty = cash / (entry_price * (1.0 + fee))
                notional = qty * entry_price
                entry_fee = notional * fee
                cash = 0.0
                entry_time_ms = bar.open_time_ms
                in_position = True
            elif cross_down and in_position:
                exit_price = bar.open * (1.0 - slip)
                gross = qty * exit_price
                exit_fee = gross * fee
                cash = gross - exit_fee
                pnl = cash - entry_cash
                trades.append({
                    "trade_no": len(trades) + 1,
                    "entry_time": iso_z_from_ms(entry_time_ms),
                    "entry_price": round(entry_price, 8),
                    "exit_time": iso_z_from_ms(bar.open_time_ms),
                    "exit_price": round(exit_price, 8),
                    "side": "LONG",
                    "position_size": round(qty, 12),
                    "fee": round(entry_fee + exit_fee, 8),
                    "pnl": round(pnl, 8),
                    "pnl_pct": round((pnl / entry_cash) * 100.0, 8),
                    "exit_reason": "EMA50 cross below EMA200",
                })
                qty = 0.0
                in_position = False

        if in_position:
            exposure_days += 1
            liquidation_px = bar.close * (1.0 - slip)
            equity = qty * liquidation_px * (1.0 - fee)
        else:
            equity = cash

        bench_liq = bench_qty * bar.close * (1.0 - slip) * (1.0 - fee)
        peak = max(peak, equity)
        dd = ((equity / peak) - 1.0) * 100.0 if peak > 0 else 0.0
        stamp = bar.day.isoformat()
        equity_series.append([stamp, round(equity, 8)])
        benchmark_series.append([stamp, round(bench_liq, 8)])
        drawdown_series.append([stamp, round(dd, 8)])

    # Finite-window valuation: liquidate an open position at final close, clearly labeled.
    if in_position:
        bar = bars[last_idx]
        exit_price = bar.close * (1.0 - slip)
        gross = qty * exit_price
        exit_fee = gross * fee
        cash = gross - exit_fee
        pnl = cash - entry_cash
        trades.append({
            "trade_no": len(trades) + 1,
            "entry_time": iso_z_from_ms(entry_time_ms),
            "entry_price": round(entry_price, 8),
            "exit_time": datetime.combine(bar.day, datetime.max.time(), tzinfo=timezone.utc).strftime("%Y-%m-%dT23:59:59Z"),
            "exit_price": round(exit_price, 8),
            "side": "LONG",
            "position_size": round(qty, 12),
            "fee": round(entry_fee + exit_fee, 8),
            "pnl": round(pnl, 8),
            "pnl_pct": round((pnl / entry_cash) * 100.0, 8),
            "exit_reason": "End-of-test liquidation for finite-window valuation",
        })
        equity_series[-1][1] = round(cash, 8)
        peak_before_last = max(v[1] for v in equity_series)
        drawdown_series[-1][1] = round(((cash / peak_before_last) - 1.0) * 100.0 if peak_before_last else 0.0, 8)

    final_equity = float(equity_series[-1][1])
    final_benchmark = float(benchmark_series[-1][1])
    test_days = len(test_indices)
    elapsed_days = max(1, (end - start).days)
    years = elapsed_days / 365.2425
    total_return = (final_equity / initial_capital - 1.0) * 100.0
    annualized = ((final_equity / initial_capital) ** (1.0 / years) - 1.0) * 100.0 if final_equity > 0 else None
    benchmark_return = (final_benchmark / initial_capital - 1.0) * 100.0

    daily_returns: list[float] = []
    eq = [float(x[1]) for x in equity_series]
    for a, b in zip(eq, eq[1:]):
        if a > 0:
            daily_returns.append(b / a - 1.0)
    mean_daily = statistics.fmean(daily_returns) if daily_returns else 0.0
    stdev_daily = std_sample(daily_returns)
    downside = [min(0.0, r) for r in daily_returns]
    downside_dev = math.sqrt(statistics.fmean([r * r for r in downside])) if downside else 0.0
    sharpe = safe_ratio(mean_daily * math.sqrt(365.0), stdev_daily)
    sortino = safe_ratio(mean_daily * math.sqrt(365.0), downside_dev)
    max_dd = min(float(x[1]) for x in drawdown_series)
    calmar = safe_ratio(annualized if annualized is not None else 0.0, abs(max_dd)) if max_dd != 0 else None
    volatility = stdev_daily * math.sqrt(365.0) * 100.0

    wins = [t for t in trades if t["pnl"] > 0]
    losses = [t for t in trades if t["pnl"] < 0]
    gross_profit = sum(float(t["pnl"]) for t in wins)
    gross_loss = abs(sum(float(t["pnl"]) for t in losses))
    profit_factor = safe_ratio(gross_profit, gross_loss)
    win_rate = (len(wins) / len(trades) * 100.0) if trades else None
    loss_rate = (len(losses) / len(trades) * 100.0) if trades else None
    expectancy = statistics.fmean([float(t["pnl_pct"]) for t in trades]) if trades else None
    avg_win = statistics.fmean([float(t["pnl_pct"]) for t in wins]) if wins else None
    avg_loss = statistics.fmean([float(t["pnl_pct"]) for t in losses]) if losses else None
    largest_win = max((float(t["pnl_pct"]) for t in trades), default=None)
    largest_loss = min((float(t["pnl_pct"]) for t in trades), default=None)

    max_cw = max_cl = cw = cl = 0
    for t in trades:
        if t["pnl"] > 0:
            cw += 1; cl = 0; max_cw = max(max_cw, cw)
        elif t["pnl"] < 0:
            cl += 1; cw = 0; max_cl = max(max_cl, cl)
        else:
            cw = cl = 0

    dd_values = [float(x[1]) for x in drawdown_series]
    avg_dd = statistics.fmean([x for x in dd_values if x < 0]) if any(x < 0 for x in dd_values) else 0.0
    longest_dd = cur_dd = 0
    for x in dd_values:
        if x < 0:
            cur_dd += 1; longest_dd = max(longest_dd, cur_dd)
        else:
            cur_dd = 0

    metrics = {
        "total_return": pct(total_return),
        "annualized_return": pct(annualized),
        "benchmark_return": pct(benchmark_return),
        "number_of_trades": len(trades),
        "win_rate": pct(win_rate),
        "profit_factor": pct(profit_factor),
        "max_drawdown": pct(max_dd),
        "sharpe_ratio": pct(sharpe),
        "sortino_ratio": pct(sortino),
        "calmar_ratio": pct(calmar),
        "expectancy": pct(expectancy),
    }
    advanced = {
        "winning_trades": {"value": len(wins), "unit": "count"},
        "losing_trades": {"value": len(losses), "unit": "count"},
        "loss_rate": {"value": pct(loss_rate), "unit": "%"},
        "average_win": {"value": pct(avg_win), "unit": "%"},
        "average_loss": {"value": pct(avg_loss), "unit": "%"},
        "average_drawdown": {"value": pct(avg_dd), "unit": "%"},
        "volatility": {"value": pct(volatility), "unit": "% annualized"},
        "exposure": {"value": pct(exposure_days / test_days * 100.0), "unit": "%"},
        "largest_win": {"value": pct(largest_win), "unit": "%"},
        "largest_loss": {"value": pct(largest_loss), "unit": "%"},
        "consecutive_wins": {"value": max_cw, "unit": "count"},
        "consecutive_losses": {"value": max_cl, "unit": "count"},
        "max_drawdown_duration_days": {"value": longest_dd, "unit": "days"},
        "final_equity": {"value": round(final_equity, 8), "unit": "USDT"},
        "benchmark_final_equity": {"value": round(final_benchmark, 8), "unit": "USDT"},
    }
    return {
        "metrics": metrics,
        "advanced_metrics": advanced,
        "series": {
            "strategy_equity": equity_series,
            "benchmark_equity": benchmark_series,
            "drawdown": drawdown_series,
        },
        "trades": trades,
    }


def build_result(args, bars: list[Bar], source_base: str, dataset_path: Path, dataset_sha: str, archive_manifest: Optional[list[dict]] = None) -> dict:
    start = date.fromisoformat(args.start)
    end = date.fromisoformat(args.end)
    result = backtest(bars, start, end, args.initial_capital, args.fee_pct, args.slippage_pct)
    archive_manifest = archive_manifest or []
    archive_manifest_hash = sha256_bytes(canonical_json_bytes([
        {"file": x.get("file"), "sha256": x.get("sha256"), "rows": x.get("rows")} for x in archive_manifest
    ])) if archive_manifest else None
    using_archive = bool(archive_manifest)
    provenance = {
        "runner_version": RUNNER_VERSION,
        "source": "Binance Public Data Archive / spot/klines" if using_archive else "Binance Spot REST /api/v3/klines",
        "source_base": source_base,
        "source_timezone": "UTC",
        "symbol": args.symbol,
        "interval": args.interval,
        "dataset_file": dataset_path.name,
        "dataset_sha256": dataset_sha,
        "dataset_rows": len(bars),
        "source_first_bar": bars[0].day.isoformat(),
        "source_last_bar": bars[-1].day.isoformat(),
        "execution_model": "signal on daily close; execute next daily open; long-only; fixed fee/slippage; end-of-test liquidation",
        "generated_at_utc": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
        "python": sys.version.split()[0],
        "archive_files": len(archive_manifest),
        "archive_manifest_sha256": archive_manifest_hash,
        "archive_checksum_verified": bool(archive_manifest),
    }
    advanced = result["advanced_metrics"]
    advanced.update({
        "dataset_sha256": {"text": dataset_sha},
        "runner_version": {"text": RUNNER_VERSION},
        "dataset_rows": {"value": len(bars), "unit": "rows"},
        "source_first_bar": {"text": provenance["source_first_bar"]},
        "source_last_bar": {"text": provenance["source_last_bar"]},
        "archive_files": {"value": len(archive_manifest), "unit": "files"},
        "archive_manifest_sha256": {"text": archive_manifest_hash} if archive_manifest_hash else {"text": ""},
        "archive_checksum_verified": {"text": "yes" if archive_manifest else "n/a"},
    })
    payload = {
        "schema": RESULT_SCHEMA,
        "mode": "apply_result",
        "target": {
            "backtest_slug": "ema-50-200-btc-1d-baseline",
            "strategy_slug": "ema-50-200-btc",
            "strategy_version": "1.0",
        },
        "config": {
            "symbol": "BTC/USDT",
            "market": "Crypto",
            "exchange": "Binance Spot",
            "timeframe": "1D",
            "start_date": args.start,
            "end_date": args.end,
            "initial_capital": args.initial_capital,
            "fee_pct": args.fee_pct,
            "slippage_pct": args.slippage_pct,
            "leverage": 1.0,
            "benchmark": "BTC Buy & Hold",
            "data_source": "Binance Public Data Archive / spot/klines" if using_archive else "Binance Spot /api/v3/klines",
            "data_resolution": "1D UTC",
            "dataset_version": f"binance-BTCUSDT-1d-{provenance['source_first_bar']}_{args.end}@{dataset_sha[:12]}",
            "code_version": f"EMA-CROSS-BTC-V1.0+{RUNNER_VERSION}",
            "parameters": {
                "ema_fast": 50,
                "ema_slow": 200,
                "signal_confirmation": "daily_close",
                "execution": "next_candle_open",
                "direction": "long_only",
                "pyramiding": False,
                "end_of_test": "liquidate_at_last_close_for_finite_window_valuation",
            },
        },
        "run_status": "completed",
        "publish_status": "draft",
        "is_demo": False,
        "executed_at": provenance["generated_at_utc"],
        "metrics": result["metrics"],
        "advanced_metrics": advanced,
        "series": result["series"],
        "trades": result["trades"],
        "provenance": provenance,
    }
    hash_view = dict(payload)
    hash_view["provenance"] = dict(provenance)
    result_sha = sha256_bytes(canonical_json_bytes(hash_view))
    payload["provenance"]["result_payload_sha256"] = result_sha
    payload["advanced_metrics"]["result_payload_sha256"] = {"text": result_sha}
    return payload


def main() -> int:
    p = argparse.ArgumentParser(description="Nexito EMA 50/200 BTC/USDT reproducible baseline runner")
    p.add_argument("--symbol", default="BTCUSDT")
    p.add_argument("--interval", default="1d")
    p.add_argument("--source-start", default="2017-08-17", help="warm-up/source fetch start (UTC)")
    p.add_argument("--source", choices=["archive", "api", "auto"], default="archive", help="market-data source; archive is the reproducible default")
    p.add_argument("--archive-base", default=ARCHIVE_BASE, help="Binance Public Data Archive base URL")
    p.add_argument("--start", default="2020-01-01")
    p.add_argument("--end", default="2026-08-16")
    p.add_argument("--initial-capital", type=float, default=10000.0)
    p.add_argument("--fee-pct", type=float, default=0.10)
    p.add_argument("--slippage-pct", type=float, default=0.05)
    p.add_argument("--dataset", help="Use an existing frozen dataset CSV instead of downloading")
    p.add_argument("--output-dir", default="quant-run-output")
    args = p.parse_args()

    start = date.fromisoformat(args.start)
    end = date.fromisoformat(args.end)
    source_start = date.fromisoformat(args.source_start)
    if end < start:
        raise SystemExit("end date must be >= start date")
    if end >= datetime.now(timezone.utc).date():
        raise SystemExit("end date must be a fully closed UTC day")

    out_dir = Path(args.output_dir).resolve()
    out_dir.mkdir(parents=True, exist_ok=True)
    dataset_path = Path(args.dataset).resolve() if args.dataset else out_dir / "NEXITO-FIRST-LAB-RUN-dataset.csv"

    source_base = "frozen-local-dataset"
    archive_manifest: list[dict] = []
    if args.dataset:
        bars = read_dataset_csv(dataset_path)
        dataset_sha = sha256_bytes(dataset_path.read_bytes())
    else:
        if args.source in ("archive", "auto"):
            try:
                bars, source_base, archive_manifest = fetch_klines_archive(args.symbol, args.interval, source_start, end, args.archive_base)
            except Exception as archive_exc:
                if args.source == "archive":
                    raise
                print(f"WARN: Binance archive failed: {archive_exc}; trying REST API fallback", file=sys.stderr)
                start_ms = int(datetime.combine(source_start, datetime.min.time(), tzinfo=timezone.utc).timestamp() * 1000)
                end_ms = int(datetime.combine(end, datetime.max.time(), tzinfo=timezone.utc).timestamp() * 1000)
                bars, source_base = fetch_klines(args.symbol, args.interval, start_ms, end_ms)
        else:
            start_ms = int(datetime.combine(source_start, datetime.min.time(), tzinfo=timezone.utc).timestamp() * 1000)
            end_ms = int(datetime.combine(end, datetime.max.time(), tzinfo=timezone.utc).timestamp() * 1000)
            bars, source_base = fetch_klines(args.symbol, args.interval, start_ms, end_ms)
        if not bars:
            raise SystemExit("Binance returned no bars")
        dataset_sha = write_dataset_csv(dataset_path, bars)

    payload = build_result(args, bars, source_base, dataset_path, dataset_sha, archive_manifest)
    result_path = out_dir / "NEXITO-FIRST-LAB-RUN-result.json"
    result_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8")
    manifest = {
        "runner_version": RUNNER_VERSION,
        "result_schema": RESULT_SCHEMA,
        "dataset_file": dataset_path.name,
        "dataset_sha256": dataset_sha,
        "result_file": result_path.name,
        "result_file_sha256": sha256_bytes(result_path.read_bytes()),
        "result_payload_sha256": payload["provenance"]["result_payload_sha256"],
        "test_period": [args.start, args.end],
        "metrics": payload["metrics"],
        "visual_rows": {k: len(v) for k, v in payload["series"].items()},
        "trades": len(payload["trades"]),
    }
    manifest_path = out_dir / "NEXITO-FIRST-LAB-RUN-manifest.json"
    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8")
    sums = [
        f"{dataset_sha}  {dataset_path.name}",
        f"{sha256_bytes(result_path.read_bytes())}  {result_path.name}",
        f"{sha256_bytes(manifest_path.read_bytes())}  {manifest_path.name}",
    ]
    (out_dir / "SHA256SUMS.txt").write_text("\n".join(sums) + "\n", encoding="utf-8")

    print("NEXITO FIRST LAB RUN COMPLETE")
    print(f"Dataset : {dataset_path}")
    print(f"SHA256  : {dataset_sha}")
    print(f"Result  : {result_path}")
    print(f"Trades  : {len(payload['trades'])}")
    for key, value in payload["metrics"].items():
        print(f"{key:20s}: {value}")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except KeyboardInterrupt:
        raise SystemExit(130)
    except Exception as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(1)