"""T8, T9 y la corrida de extremo a extremo de run_study.py sobre datos sintéticos.

Nunca se lee data/: todo se construye en directorios temporales (ver synth.py).
"""

import ast
import csv
import hashlib
import json
import math
import os
import re
import shutil
import subprocess
import sys
from collections import Counter
from datetime import date
from pathlib import Path
from types import SimpleNamespace

import numpy as np
import pytest
import run_study
from estudio01 import data_h10, engine, metrics, rates, report
from synth import ANNEX, CANONICAL, PINS, make_data_dir, pins_of

STUDY = Path(__file__).resolve().parents[1]
SCRIPT = STUDY / "run_study.py"
RUN_UTC = "2026-10-23T00:00:00Z"
EXPECTED_FILES = sorted(
    [
        "metrics.csv",
        "dsr.csv",
        "pbo.csv",
        "tramo-final.csv",
        "atribucion.csv",
        "referencia.csv",
        "fomc-decisiones.csv",
        "ecb-dfr.csv",
        "run-log.json",
        *(f"equity-v{v}-{s}.csv" for v in (1, 2, 3, 4) for s in ("low", "base", "high")),
    ]
)
TEN_DECIMALS = re.compile(r"^-?\d+\.\d{10}$")
RAW_TEXT_COLUMNS = {"level_as_published", "obs_value_percent"}


def run_cli(*args: str) -> subprocess.CompletedProcess[str]:
    """La línea de comandos real: sin ninguna huella de prueba, solo PINNED_SHA256."""
    return subprocess.run(
        [sys.executable, "-B", str(SCRIPT), *args], capture_output=True, text=True, check=False
    )


CHILD = (
    "import json, sys; sys.path.insert(0, sys.argv[1]); import run_study; "
    "sys.exit(run_study.main(sys.argv[3:], pinned=json.loads(sys.argv[2])))"
)


def main(argv: list[str]) -> int:
    """run_study.main con las huellas que tenía el directorio sintético pedido al crearse."""
    return run_study.main(argv, pinned=PINS[argv[argv.index("--data-dir") + 1]])


def run_pinned_cli(
    data: Path, *args: str, timeout: float | None = None
) -> subprocess.CompletedProcess[str]:
    """Un proceso nuevo que llama a run_study.main con las huellas sintéticas de data."""
    command = [sys.executable, "-B", "-c", CHILD, str(STUDY), json.dumps(PINS[str(data)])]
    return subprocess.run(
        [*command, "--data-dir", str(data), *args],
        capture_output=True,
        text=True,
        check=False,
        timeout=timeout,
    )


def read_csv(path: Path) -> list[dict[str, str]]:
    with path.open(newline="", encoding="utf-8") as handle:
        return list(csv.DictReader(handle))


@pytest.fixture(scope="module")
def data_dir(tmp_path_factory) -> Path:
    return make_data_dir(tmp_path_factory.mktemp("synthetic") / "data")


@pytest.fixture(scope="module")
def full_run(data_dir, tmp_path_factory) -> Path:
    out = tmp_path_factory.mktemp("first") / "outputs"
    code = main(["--data-dir", str(data_dir), "--out-dir", str(out), "--run-utc", RUN_UTC])
    assert code == 0
    return out


# --- T8 ---------------------------------------------------------------------------------------


def alter_one_hex_digit_of_the_first_hash(data: Path) -> None:
    sums = data / "SHA256SUMS"
    text = sums.read_text(encoding="ascii")
    sums.write_text(text[:7] + ("0" if text[7] != "0" else "1") + text[8:], encoding="ascii")


def test_t8_altered_sha256sums_stops_with_nonzero_exit_and_no_outputs(tmp_path):
    """T8: un SHA256SUMS alterado en un byte -> código de salida distinto de cero y sin salidas."""
    data = make_data_dir(tmp_path / "data")
    alter_one_hex_digit_of_the_first_hash(data)
    out = tmp_path / "outputs"
    done = run_pinned_cli(data, "--out-dir", str(out))
    assert done.returncode != 0
    assert "DETENIDO" in done.stderr
    assert not out.exists()  # ni siquiera se crea el directorio


def flip_one_bit(target: Path) -> None:
    raw = bytearray(target.read_bytes())
    raw[len(raw) // 2] ^= 0x01  # un solo bit
    target.write_bytes(bytes(raw))


def rewrite_sha256sums_for(data: Path) -> None:
    """SHA256SUMS coherente con los archivos tal como están ahora (como lo dejaría quien altera)."""
    lines = [f"{hashlib.sha256((data / n).read_bytes()).hexdigest()}  {n}\n" for n in CANONICAL]
    (data / "SHA256SUMS").write_text("".join(lines), encoding="ascii")


@pytest.mark.parametrize("name", [*CANONICAL, ANNEX])
def test_t8_one_changed_bit_in_any_of_the_five_files_or_the_annex_stops(tmp_path, name):
    """Lo delatan la huella fijada y, en los cinco archivos de datos, también SHA256SUMS."""
    data = make_data_dir(tmp_path / "data")
    flip_one_bit(data / name)
    out = tmp_path / "outputs"
    done = run_pinned_cli(data, "--out-dir", str(out))
    assert done.returncode == run_study.EXIT_CHECKSUM
    assert name in done.stderr
    assert not out.exists()


@pytest.mark.parametrize("name", [*CANONICAL, ANNEX])
def test_t8_a_file_altered_together_with_its_sha256sums_line_still_stops(tmp_path, name):
    """La huella fijada en el código es lo que detiene un archivo con su SHA256SUMS reescrito."""
    data = make_data_dir(tmp_path / "data")
    flip_one_bit(data / name)
    rewrite_sha256sums_for(data)  # SHA256SUMS ya no delata nada: los cinco archivos coinciden
    out = tmp_path / "outputs"
    done = run_pinned_cli(data, "--out-dir", str(out))
    assert done.returncode == run_study.EXIT_CHECKSUM
    assert "huella fijada" in done.stderr and name in done.stderr
    assert not out.exists()


@pytest.mark.parametrize("name", [*CANONICAL, ANNEX])
def test_t8_a_missing_file_or_annex_stops(tmp_path, name):
    data = make_data_dir(tmp_path / "data")
    (data / name).unlink()
    out = tmp_path / "outputs"
    done = run_pinned_cli(data, "--out-dir", str(out))
    assert done.returncode == run_study.EXIT_CHECKSUM
    assert name in done.stderr
    assert not out.exists()


@pytest.mark.parametrize("name", [CANONICAL[0], ANNEX])
def test_t8_nothing_is_interpreted_or_computed_when_the_check_fails(tmp_path, monkeypatch, name):
    """Ni por las funciones de run_study ni por los analizadores, el motor o la escritura."""

    def forbidden(*_args, **_kwargs):
        raise AssertionError("se interpretó, calculó o escribió algo antes de verificar")

    for function in ("load_inputs", "compute", "write_all", "write_tables", "publish"):
        monkeypatch.setattr(run_study, function, forbidden)
    for module, function in (
        (rates, "parse_ecb"),
        (rates, "parse_fomc"),
        (data_h10, "parse_h10"),
        (data_h10, "load_bars"),
        (engine, "simulate"),
        (report, "write_csv"),
        (report, "write_json"),
        (run_study.tempfile, "mkdtemp"),
    ):
        monkeypatch.setattr(module, function, forbidden)
    data = make_data_dir(tmp_path / "data")
    flip_one_bit(data / name)  # un solo archivo difiere; los demás están intactos
    out = tmp_path / "outputs"
    assert main(["--data-dir", str(data), "--out-dir", str(out)]) == run_study.EXIT_CHECKSUM
    assert not out.exists()
    assert sorted(item.name for item in tmp_path.iterdir()) == ["data"]  # ni un directorio temporal


def test_the_real_command_line_refuses_data_that_is_not_the_pinned_data(tmp_path):
    """Sin huellas de prueba, datos coherentes con su SHA256SUMS pero distintos de los fijados."""
    data = make_data_dir(tmp_path / "data")
    out = tmp_path / "outputs"
    done = run_cli("--data-dir", str(data), "--out-dir", str(out))
    assert done.returncode == run_study.EXIT_CHECKSUM
    assert "DETENIDO" in done.stderr and "huella fijada" in done.stderr
    assert not out.exists()
    done = run_cli(
        "--data-dir", str(data), "--pinned", "{}"
    )  # no hay opción que sustituya las huellas
    assert done.returncode == 2 and "unrecognized arguments" in done.stderr


def exit_constants(path: Path) -> dict[str, int]:
    """Los EXIT_* enteros que define un módulo del estudio, leídos de su código fuente."""
    found = {}
    for node in ast.parse(path.read_text(encoding="utf-8")).body:
        if isinstance(node, ast.Assign) and isinstance(node.targets[0], ast.Name):
            name = node.targets[0].id
            if name.startswith("EXIT_") and isinstance(node.value, ast.Constant):
                found[name] = node.value.value
    return found


def test_exit_checksum_is_four_and_no_other_exit_code_of_the_study_shares_it():
    assert run_study.EXIT_CHECKSUM == 4
    others = {
        f"{module}:{name}": value
        for module in ("run_study.py", "download.py", "annex_counts.py")
        for name, value in exit_constants(STUDY / module).items()
        if (module, name) != ("run_study.py", "EXIT_CHECKSUM")
    }
    assert len(others) >= 4 and "run_study.py:EXIT_OUTPUTS_NOT_EMPTY" in others
    assert run_study.EXIT_CHECKSUM not in {0, 1, 2, *others.values()}


def test_the_real_command_line_exits_with_four_on_a_checksum_failure_and_two_on_a_usage_error(
    tmp_path,
):
    data = make_data_dir(tmp_path / "data")
    stopped = run_cli("--data-dir", str(data), "--out-dir", str(tmp_path / "outputs"))
    assert stopped.returncode == 4 and "DETENIDO" in stopped.stderr
    usage = run_cli("--no-such-option")
    assert usage.returncode == 2 and "usage:" in usage.stderr
    assert run_cli("--help").returncode == 0


def test_the_pinned_hashes_are_the_five_files_and_the_annex_and_nothing_else():
    pins = run_study.PINNED_SHA256
    assert ANNEX == run_study.ANNEX_NAME
    assert tuple(CANONICAL) == run_study.CANONICAL
    assert set(pins) == {*CANONICAL, ANNEX} and len(pins) == 6
    assert all(re.fullmatch(r"[0-9a-f]{64}", digest) for digest in pins.values())
    assert len(set(pins.values())) == 6  # seis archivos distintos, seis huellas distintas


def test_the_command_line_has_no_option_or_variable_that_replaces_the_pins():
    options = {opt for action in run_study.build_parser()._actions for opt in action.option_strings}
    assert options == {"-h", "--help", "--data-dir", "--out-dir", "--run-utc", "--tables-only"}
    source = SCRIPT.read_text(encoding="utf-8")
    assert "environ" not in source and "getenv" not in source
    assert source.count("PINNED_SHA256") >= 3  # la define, la usa por defecto y la documenta


def test_incomplete_pins_do_not_let_a_file_through(tmp_path):
    data = make_data_dir(tmp_path / "data")
    for name in (*CANONICAL, ANNEX):
        partial = {key: digest for key, digest in PINS[str(data)].items() if key != name}
        with pytest.raises(run_study.ChecksumError, match="huella fijada"):
            run_study.read_verified(data, partial)
    with pytest.raises(run_study.ChecksumError, match="huella fijada"):
        run_study.read_verified(data, {})


def test_the_bytes_that_were_hashed_are_the_bytes_that_are_interpreted(tmp_path):
    """Verificar y leer son un solo paso: borrar o cambiar el directorio después no afecta."""
    data = make_data_dir(tmp_path / "data")
    files = run_study.read_verified(data, PINS[str(data)])
    for name, raw in files.blobs.items():
        assert hashlib.sha256(raw).hexdigest() == files.sha256[name] == PINS[str(data)][name]
    shutil.rmtree(data)
    inputs = run_study.load_inputs(files)  # no abre ningún archivo
    assert len(inputs.bars.prices) > 5000 and inputs.decisions and inputs.ecb


def test_t8_a_repeated_line_in_sha256sums_is_rejected(tmp_path):
    """Si un nombre aparece dos veces, ganaría la última: se rechaza."""
    data = make_data_dir(tmp_path / "data")
    sums = data / "SHA256SUMS"
    lines = sums.read_text(encoding="ascii").splitlines(keepends=True)
    forged = lines[0].replace(lines[0][:8], "00000000", 1)  # primera línea con un hash falso
    sums.write_text("".join([forged, *lines]), encoding="ascii")
    out = tmp_path / "outputs"
    assert main(["--data-dir", str(data), "--out-dir", str(out)]) == run_study.EXIT_CHECKSUM
    assert not out.exists()


def test_t8_missing_or_unlisted_files_also_stop(tmp_path):
    data = make_data_dir(tmp_path / "data")
    sums = data / "SHA256SUMS"
    lines = sums.read_text(encoding="ascii").splitlines(keepends=True)
    sums.write_text("".join(lines[:-1]), encoding="ascii")  # falta una línea
    out = tmp_path / "outputs"
    assert main(["--data-dir", str(data), "--out-dir", str(out)]) == run_study.EXIT_CHECKSUM
    assert not out.exists()
    sums.write_text("".join(lines), encoding="ascii")
    (data / CANONICAL[2]).unlink()  # falta un archivo
    assert main(["--data-dir", str(data), "--out-dir", str(out)]) == run_study.EXIT_CHECKSUM
    assert not out.exists()
    sums.write_text("not a checksum line\n", encoding="ascii")
    assert main(["--data-dir", str(data), "--out-dir", str(out)]) == run_study.EXIT_CHECKSUM
    assert not out.exists()


# --- huellas fijadas: casos que una revisión independiente mostró que había que proteger -------


def sums_line_altered(data: Path, name: str) -> None:
    """Cambia el primer dígito de la línea de `name` en SHA256SUMS; archivos y huellas intactos."""
    sums = data / "SHA256SUMS"
    lines = sums.read_text(encoding="ascii").splitlines(keepends=True)
    lines = [
        (("0" if line[0] != "0" else "1") + line[1:]) if line.endswith(f"  {name}\n") else line
        for line in lines
    ]
    sums.write_text("".join(lines), encoding="ascii")


@pytest.mark.parametrize("name", CANONICAL)
def test_t8_a_wrong_sha256sums_line_for_any_of_the_five_files_stops(tmp_path, capsys, name):
    data = make_data_dir(tmp_path / "data")
    sums_line_altered(data, name)
    out = tmp_path / "outputs"
    assert main(["--data-dir", str(data), "--out-dir", str(out)]) == run_study.EXIT_CHECKSUM
    err = capsys.readouterr().err
    assert "SHA256SUMS" in err and name in err
    assert not out.exists()


def _extra_annex_line(data: Path) -> None:
    sums = data / "SHA256SUMS"
    extra = f"{PINS[str(data)][ANNEX]}  {ANNEX}\n"
    sums.write_text(sums.read_text(encoding="ascii") + extra, encoding="ascii")


def _extra_unknown_line(data: Path) -> None:
    sums = data / "SHA256SUMS"
    sums.write_text(sums.read_text(encoding="ascii") + f"{'0' * 64}  extra.csv\n", encoding="ascii")


@pytest.mark.parametrize(
    "damage",
    [
        lambda data: (data / "SHA256SUMS").unlink(),
        lambda data: (data / "SHA256SUMS").write_bytes(b""),
        lambda data: (data / "SHA256SUMS").write_bytes(b"\xff\xfe not ascii\n"),
        _extra_annex_line,
        _extra_unknown_line,
    ],
    ids=["absent", "empty", "not-ascii", "extra-annex-line", "extra-unknown-line"],
)
def test_t8_a_sha256sums_that_is_absent_empty_foreign_or_with_extra_lines_stops(
    tmp_path, capsys, damage
):
    data = make_data_dir(tmp_path / "data")
    damage(data)
    out = tmp_path / "outputs"
    assert main(["--data-dir", str(data), "--out-dir", str(out)]) == run_study.EXIT_CHECKSUM
    assert "SHA256SUMS" in capsys.readouterr().err
    assert not out.exists()


@pytest.mark.parametrize("name", [*CANONICAL, ANNEX])
@pytest.mark.parametrize("end", ["first", "last"])
def test_a_pinned_hash_that_differs_in_one_hex_digit_stops(tmp_path, name, end):
    """La comparación es de la huella completa: no basta con un prefijo ni con casi todo."""
    data = make_data_dir(tmp_path / "data")
    pins = dict(PINS[str(data)])
    digest = pins[name]
    index = 0 if end == "first" else -1
    flipped = "0" if digest[index] != "0" else "1"
    pins[name] = flipped + digest[1:] if end == "first" else digest[:-1] + flipped
    with pytest.raises(run_study.ChecksumError, match="huella fijada") as stop:
        run_study.read_verified(data, pins)
    others = [other for other in (*CANONICAL, ANNEX) if other != name]
    assert name in str(stop.value) and not any(other in str(stop.value) for other in others)


def test_t8_two_swapped_files_stop_even_with_a_coherent_sha256sums(tmp_path, capsys):
    """Cada archivo trae una huella válida, pero bajo el nombre equivocado."""
    data = make_data_dir(tmp_path / "data")
    first, second = data / CANONICAL[2], data / CANONICAL[3]
    swap = first.read_bytes()
    first.write_bytes(second.read_bytes())
    second.write_bytes(swap)
    rewrite_sha256sums_for(data)
    out = tmp_path / "outputs"
    assert main(["--data-dir", str(data), "--out-dir", str(out)]) == run_study.EXIT_CHECKSUM
    err = capsys.readouterr().err
    assert CANONICAL[2] in err and CANONICAL[3] in err and "huella fijada" in err
    assert not out.exists()


@pytest.mark.parametrize("name", [*CANONICAL, ANNEX])
def test_t8_a_directory_in_place_of_a_file_stops(tmp_path, capsys, name):
    data = make_data_dir(tmp_path / "data")
    (data / name).unlink()
    (data / name).mkdir()
    out = tmp_path / "outputs"
    assert main(["--data-dir", str(data), "--out-dir", str(out)]) == run_study.EXIT_CHECKSUM
    err = capsys.readouterr().err
    assert name in err and "archivo regular" in err
    assert not out.exists()


@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="sin FIFO en esta plataforma")
def test_t8_a_fifo_in_place_of_a_file_stops_instead_of_hanging(tmp_path):
    """Leer una FIFO sin escritor bloquea para siempre: se rechaza sin abrirla."""
    data = make_data_dir(tmp_path / "data")
    (data / CANONICAL[1]).unlink()
    os.mkfifo(data / CANONICAL[1])
    done = run_pinned_cli(data, "--out-dir", str(tmp_path / "outputs"), timeout=60)
    assert done.returncode == run_study.EXIT_CHECKSUM
    assert CANONICAL[1] in done.stderr and "archivo regular" in done.stderr


@pytest.mark.skipif(not hasattr(os, "geteuid") or os.geteuid() == 0, reason="root lee todo")
def test_t8_an_unreadable_file_stops(tmp_path, capsys):
    data = make_data_dir(tmp_path / "data")
    target = data / CANONICAL[4]
    target.chmod(0)
    try:
        out = tmp_path / "outputs"
        assert main(["--data-dir", str(data), "--out-dir", str(out)]) == run_study.EXIT_CHECKSUM
        err = capsys.readouterr().err
        assert "no se puede leer" in err and CANONICAL[4] in err
        assert not out.exists()
    finally:
        target.chmod(0o600)


def test_altered_data_are_reported_before_a_non_empty_outputs_directory(tmp_path):
    """La verificación va primero: con datos alterados y outputs/ ocupado, manda el de datos."""
    data = make_data_dir(tmp_path / "data")
    flip_one_bit(data / ANNEX)
    out = tmp_path / "outputs"
    out.mkdir()
    (out / "other.txt").write_text("x", encoding="utf-8")
    assert main(["--data-dir", str(data), "--out-dir", str(out)]) == run_study.EXIT_CHECKSUM
    assert sorted(item.name for item in out.iterdir()) == ["other.txt"]


@pytest.mark.parametrize("name", [CANONICAL[0], CANONICAL[4], ANNEX])
def test_the_tables_only_mode_verifies_too(tmp_path, name):
    data = make_data_dir(tmp_path / "data")
    flip_one_bit(data / name)
    out = tmp_path / "outputs"
    argv = ["--data-dir", str(data), "--out-dir", str(out), "--tables-only"]
    assert main(argv) == run_study.EXIT_CHECKSUM
    assert not out.exists()


def test_the_default_pins_are_the_module_constant_and_accept_matching_data(tmp_path, monkeypatch):
    """Sin `pinned` se usa PINNED_SHA256: con esa constante igual a los datos, la corrida pasa."""
    data = make_data_dir(tmp_path / "data")
    out = tmp_path / "outputs"
    argv = ["--data-dir", str(data), "--out-dir", str(out), "--tables-only"]
    assert run_study.main(argv) == run_study.EXIT_CHECKSUM  # la constante real no es la sintética
    assert not out.exists()
    monkeypatch.setattr(run_study, "PINNED_SHA256", dict(PINS[str(data)]))
    assert run_study.main(argv) == 0
    assert sorted(item.name for item in out.iterdir()) == ["ecb-dfr.csv", "fomc-decisiones.csv"]


@pytest.mark.parametrize("name", [CANONICAL[0], CANONICAL[2], CANONICAL[3], CANONICAL[4]])
def test_verified_bytes_that_are_not_utf8_stop_the_run_without_outputs(tmp_path, name):
    """UTF-8 estricto aunque la huella sea válida: nunca se reemplazan bytes en silencio."""
    data = make_data_dir(tmp_path / "data")
    with (data / name).open("ab") as handle:
        handle.write(b"\n\xff\xfe")
    rewrite_sha256sums_for(data)
    out = tmp_path / "outputs"
    argv = ["--data-dir", str(data), "--out-dir", str(out), "--tables-only"]
    with pytest.raises(UnicodeDecodeError):
        run_study.main(argv, pinned=pins_of(data))
    assert not out.exists()


def test_each_data_file_is_opened_exactly_once(tmp_path, monkeypatch):
    """Lo que se hashea es lo que se interpreta: ninguna segunda lectura de los seis archivos."""
    data = make_data_dir(tmp_path / "data")
    reads: Counter[str] = Counter()
    original = Path.read_bytes

    def counting(self: Path) -> bytes:
        reads[self.name] += 1
        return original(self)

    monkeypatch.setattr(Path, "read_bytes", counting)
    out = tmp_path / "outputs"
    assert main(["--data-dir", str(data), "--out-dir", str(out), "--tables-only"]) == 0
    assert {name: reads[name] for name in (*CANONICAL, ANNEX)} == dict.fromkeys(
        (*CANONICAL, ANNEX), 1
    )


def test_a_run_time_with_a_trailing_newline_is_rejected(data_dir, tmp_path):
    with pytest.raises(SystemExit) as stop:
        main(
            [
                "--data-dir",
                str(data_dir),
                "--out-dir",
                str(tmp_path / "o"),
                "--run-utc",
                RUN_UTC + "\n",
            ]
        )
    assert stop.value.code == 2


def test_an_outputs_path_that_is_a_file_or_holds_a_directory_named_like_a_table_is_refused(
    data_dir, tmp_path
):
    """Un nombre permitido (ecb-dfr.csv) que en realidad es un directorio no es una tabla previa."""
    as_file = tmp_path / "as_file"
    as_file.write_text("x", encoding="utf-8")
    argv = ["--data-dir", str(data_dir), "--out-dir", str(as_file)]
    assert main(argv) == run_study.EXIT_OUTPUTS_NOT_EMPTY
    assert as_file.read_text(encoding="utf-8") == "x"
    out = tmp_path / "outputs"
    (out / "ecb-dfr.csv").mkdir(parents=True)
    argv = ["--data-dir", str(data_dir), "--out-dir", str(out), "--run-utc", RUN_UTC]
    assert main(argv) == run_study.EXIT_OUTPUTS_NOT_EMPTY
    assert [item.name for item in out.iterdir()] == ["ecb-dfr.csv"]  # nada se movió a medias
    assert list((out / "ecb-dfr.csv").iterdir()) == []


# --- T9 ---------------------------------------------------------------------------------------


def tree(path: Path) -> dict[str, bytes]:
    return {item.name: item.read_bytes() for item in sorted(path.iterdir())}


def test_t9_two_runs_on_the_same_series_are_byte_identical(data_dir, full_run, tmp_path):
    """T9: dos corridas sobre la misma serie sintética dan salidas byte a byte iguales."""
    second = tmp_path / "outputs"
    argv = ["--data-dir", str(data_dir), "--out-dir", str(second), "--run-utc", RUN_UTC]
    assert main(argv) == 0
    assert tree(full_run) == tree(second)


def test_t9_a_fresh_process_reproduces_the_same_bytes(data_dir, full_run, tmp_path):
    """Un proceso nuevo (otra semilla de hash de Python) también reproduce los mismos bytes."""
    out = tmp_path / "outputs"
    done = run_pinned_cli(data_dir, "--out-dir", str(out), "--run-utc", RUN_UTC)
    assert done.returncode == 0, done.stderr
    assert tree(full_run) == tree(out)


# --- salidas ----------------------------------------------------------------------------------


def test_the_run_writes_exactly_the_declared_outputs(full_run):
    assert sorted(item.name for item in full_run.iterdir()) == EXPECTED_FILES
    assert not (full_run / "bitacora.md").exists()  # la bitácora se redacta a mano en la FASE 2b


def test_every_number_is_written_with_ten_decimals_and_no_negative_zero(full_run):
    for path in full_run.glob("*.csv"):
        raw = path.read_bytes()
        assert b"\r" not in raw and not raw.startswith(b"\xef\xbb\xbf")
        assert b"-0.0000000000" not in raw
        for row in read_csv(path):
            for key, cell in row.items():
                if key in RAW_TEXT_COLUMNS:
                    continue  # texto tal como lo publicó la fuente, no un número del estudio
                if "." in cell and re.fullmatch(r"-?\d+\.\d+", cell):
                    assert TEN_DECIMALS.match(cell), (path.name, key, cell)


def test_dsr_csv_has_the_three_scenarios_and_the_boolean_column(full_run):
    rows = read_csv(full_run / "dsr.csv")
    assert [row["scenario"] for row in rows] == ["low", "base", "high"]
    assert "DSR ≥ 0.95" in rows[0]
    assert {row["DSR ≥ 0.95"] for row in rows} <= {"true", "false"}
    assert all(row["N"] == "4" for row in rows)
    for row in rows:
        if row["DSR ≥ 0.95"] == "true":
            assert float(row["DSR"]) >= 0.95  # nunca «true» con un DSR menor a 0.95


def test_metrics_cover_four_variants_in_three_scenarios(full_run):
    rows = read_csv(full_run / "metrics.csv")
    assert len(rows) == 12
    assert {(row["variant"], row["ema_pair"]) for row in rows} == {
        ("1", "9/20"),
        ("2", "9/21"),
        ("3", "12/26"),
        ("4", "50/200"),
    }
    assert {row["scenario"] for row in rows} == {"low", "base", "high"}
    assert len({row["T"] for row in rows}) == 1  # todas sobre la misma ventana
    assert len(read_csv(full_run / "referencia.csv")) == 3
    assert len(read_csv(full_run / "tramo-final.csv")) == 12


def test_pbo_csv_reports_the_twelve_thousand_eight_hundred_seventy_combinations(full_run):
    rows = {
        row["item"]: row
        for row in read_csv(full_run / "pbo.csv")
        if row["item"] != "lambda_distribution"
    }
    assert rows["combinations"]["value"] == "12870"
    assert 0.0 <= float(rows["pbo"]["value"]) <= 1.0
    distribution = [
        row for row in read_csv(full_run / "pbo.csv") if row["item"] == "lambda_distribution"
    ]
    assert sum(int(row["count"]) for row in distribution) == 12870
    assert int(rows["n_dropped"]["value"]) + int(rows["n_used"]["value"]) == int(
        rows["n_returns"]["value"]
    )
    assert int(rows["n_used"]["value"]) % 16 == 0


def test_equity_files_are_consistent_with_the_metrics(full_run):
    metrics_rows = read_csv(full_run / "metrics.csv")
    for row in metrics_rows:
        equity = read_csv(full_run / f"equity-v{row['variant']}-{row['scenario']}.csv")
        assert len(equity) == int(row["T"])
        assert list(equity[0]) == ["date", "P", "d", "K_minus", "K_plus", "S", "c", "R"]
        assert equity[-1]["d"] == "0"  # la posición se cierra en la barra L
        assert float(equity[-1]["K_plus"]) - 1.0 == pytest.approx(
            float(row["total_return"]), abs=2e-9
        )
        assert float(equity[0]["K_minus"]) == 1.0
        assert equity[0]["date"] < equity[-1]["date"]


def test_attribution_removes_each_cost_from_variant_one_in_the_base_scenario(full_run):
    rows = {row["item"]: row for row in read_csv(full_run / "atribucion.csv")}
    assert set(rows) == {"no_spread", "no_swap"}
    base = next(
        r
        for r in read_csv(full_run / "metrics.csv")
        if (r["variant"], r["scenario"]) == ("1", "base")
    )
    for row in rows.values():
        assert row["total_return_base"] == base["total_return"]
        difference = float(row["total_return_base"]) - float(row["total_return_alt"])
        assert float(row["difference_base_minus_alt"]) == pytest.approx(difference, abs=2e-10)
    assert float(rows["no_spread"]["difference_base_minus_alt"]) < 0.0  # sin spread se gana más


def test_the_run_log_records_versions_hashes_and_the_given_time(full_run, data_dir):
    log = json.loads((full_run / "run-log.json").read_text(encoding="utf-8"))
    assert log["run_utc"] == RUN_UTC
    assert log["n_trials"] == 4
    assert log["zero_deviation_sharpe_reported_as_0"] == []  # ninguna corrida con desviación 0
    assert log["versions"]["python"] == sys.version.split()[0]
    assert set(log["versions"]) == {"python", "numpy", "pandas"}
    assert set(log["data_sha256"]) == set(CANONICAL)
    assert log["data_sha256"] == {name: PINS[str(data_dir)][name] for name in CANONICAL}
    assert log["annex_sha256"] == {
        ANNEX: PINS[str(data_dir)][ANNEX]
    }  # el anexo también se verificó
    names = sorted(item.name for item in full_run.iterdir() if item.name != "run-log.json")
    assert sorted(log["outputs_sha256"]) == names
    for name, digest in log["outputs_sha256"].items():
        assert hashlib.sha256((full_run / name).read_bytes()).hexdigest() == digest
    assert log["h10"]["bars"] > 5000 and isinstance(log["h10"]["skipped"], list)


def test_fomc_and_ecb_tables_are_written_as_the_code_parsed_them(full_run):
    fomc = read_csv(full_run / "fomc-decisiones.csv")
    assert list(fomc[0]) == [
        "date",
        "date_as_published",
        "level_as_published",
        "target_fraction",
        "note",
    ]
    assert [row["date"] for row in fomc] == sorted(row["date"] for row in fomc)
    by_date = {row["date"]: row for row in fomc}
    assert by_date["2008-12-16"]["level_as_published"] == "0-0.25"
    assert by_date["2008-12-16"]["target_fraction"] == "0.0012500000"
    assert by_date["1992-09-04"]["target_fraction"] == "0.0300000000"
    starred = by_date["2020-03-04"]  # la fila con nota al pie: se usa la fecha corregida
    assert (starred["date_as_published"], starred["level_as_published"]) == (
        "March 4*",
        "1.00-1.25",
    )
    assert starred["note"].startswith("On July 10, 2024, this date was corrected")
    assert all(row["note"] == "" for row in fomc if row["date"] != "2020-03-04")
    assert all(row["date_as_published"] for row in fomc)
    ecb = read_csv(full_run / "ecb-dfr.csv")
    assert {row["date"]: row["rate_fraction"] for row in ecb}["2014-06-11"] == "-0.0010000000"


# --- opciones ---------------------------------------------------------------------------------


def test_tables_only_writes_just_the_two_derived_tables(data_dir, tmp_path):
    out = tmp_path / "outputs"
    assert main(["--data-dir", str(data_dir), "--out-dir", str(out), "--tables-only"]) == 0
    assert sorted(item.name for item in out.iterdir()) == ["ecb-dfr.csv", "fomc-decisiones.csv"]
    first = tree(out)
    # la corrida completa puede seguir en el mismo directorio y deja las dos tablas idénticas
    assert main(["--data-dir", str(data_dir), "--out-dir", str(out), "--run-utc", RUN_UTC]) == 0
    assert sorted(item.name for item in out.iterdir()) == EXPECTED_FILES
    assert {name: tree(out)[name] for name in first} == first


def test_a_non_empty_outputs_directory_is_refused(data_dir, tmp_path):
    out = tmp_path / "outputs"
    out.mkdir()
    (out / "other.txt").write_text("x", encoding="utf-8")
    assert (
        main(["--data-dir", str(data_dir), "--out-dir", str(out)])
        == run_study.EXIT_OUTPUTS_NOT_EMPTY
    )
    assert sorted(item.name for item in out.iterdir()) == ["other.txt"]


def test_a_malformed_run_time_is_rejected(data_dir, tmp_path):
    with pytest.raises(SystemExit) as stop:
        main(["--data-dir", str(data_dir), "--out-dir", str(tmp_path / "o"), "--run-utc", "ayer"])
    assert stop.value.code == 2


# --- el cableado contra llamadas explícitas al motor ----------------------------------------------

PAIRS = {1: (9, 20), 2: (9, 21), 3: (12, 26), 4: (50, 200)}  # §5, escritos a mano
SCENARIOS = {"low": (0.0001, 0.005), "base": (0.0002, 0.01), "high": (0.0004, 0.02)}  # §4
SEGMENT = (date(2024, 10, 1), date(2026, 9, 30))  # §3


@pytest.fixture(scope="module")
def explicit(data_dir):
    """Las corridas hechas a mano con engine.simulate, sin pasar por run_study.compute."""
    bars = data_h10.load_bars(data_dir / "01-h10-daily-rates.csv")
    ecb = rates.parse_ecb((data_dir / "05-ecb-dfr.csv").read_text(encoding="utf-8"))
    pages = [
        rates.parse_fomc((data_dir / name).read_text(encoding="utf-8"))
        for name in ("03-fomc-openmarket.htm", "04-fomc-openmarket-archive.htm")
    ]
    r_eur, r_usd = rates.rates_for_dates(bars.dates, ecb, rates.merge_decisions(*pages))
    prices, days = list(bars.prices), list(bars.dates)

    def run(targets, spread, m, swap_on=True):
        return engine.simulate(prices, days, targets, r_eur, r_usd, spread, m, swap_on=swap_on)

    targets = {
        v: engine.pending_targets(engine.ema(prices, fast), engine.ema(prices, slow))
        for v, (fast, slow) in PAIRS.items()
    }
    runs = {(v, n): run(targets[v], sp, m) for v in PAIRS for n, (sp, m) in SCENARIOS.items()}
    refs = {
        n: run(engine.reference_targets(len(prices)), sp, m) for n, (sp, m) in SCENARIOS.items()
    }
    base_spread, base_m = SCENARIOS["base"]
    return {
        "runs": runs,
        "refs": refs,
        "no_spread": run(targets[1], 0.0, base_m),
        "no_swap": run(targets[1], base_spread, base_m, swap_on=False),
    }


def metric_cells(item: metrics.RunMetrics) -> list[str]:
    return [
        report.fmt(item.sharpe_annualized),
        report.fmt(item.sharpe_per_bar),
        report.fmt(item.total_return),
        report.fmt(item.max_drawdown),
        str(item.executions),
        report.fmt(100.0 * item.fraction_in_position),  # el porcentaje va de 0 a 100
        str(item.n_returns),
        report.fmt(item.fixes_per_year),
    ]


METRIC_COLUMNS = (
    "sharpe_annualized",
    "sharpe_per_bar",
    "total_return",
    "max_drawdown",
    "executions",
    "pct_in_position",
    "T",
    "fixes_per_year",
)


def test_metrics_csv_matches_explicit_runs_of_each_pair_and_scenario(full_run, explicit):
    rows = {(row["variant"], row["scenario"]): row for row in read_csv(full_run / "metrics.csv")}
    assert len(rows) == 12
    for (v, name), run in explicit["runs"].items():
        row = rows[(str(v), name)]
        assert row["ema_pair"] == f"{PAIRS[v][0]}/{PAIRS[v][1]}"
        assert [row[k] for k in METRIC_COLUMNS] == metric_cells(metrics.run_metrics(run))


def test_dsr_csv_uses_variant_one_and_the_variance_of_the_four_variants_of_its_scenario(
    full_run, explicit
):
    rows = {row["scenario"]: row for row in read_csv(full_run / "dsr.csv")}
    for name in SCENARIOS:
        returns = [explicit["runs"][(v, name)].ret for v in PAIRS]
        sharpes = [metrics.sharpe_per_bar(r) for r in returns]
        skew, kurt = metrics.skew_kurtosis(returns[0])
        want = metrics.deflated_sharpe(
            sharpes[0], len(returns[0]), skew, kurt, metrics.sample_variance(sharpes), 4
        )
        row = rows[name]
        assert [row[k] for k in ("N", "T", "sr_hat", "gamma3", "gamma4", "V")] == [
            "4",
            str(len(returns[0])),
            report.fmt(want.sr_hat),
            report.fmt(want.skew),
            report.fmt(want.kurt),
            report.fmt(want.variance),
        ]
        assert [row[k] for k in ("SR0", "DSR", "PSR0", "DSR ≥ 0.95")] == [
            report.fmt(want.sr0),
            report.fmt(want.dsr),
            report.fmt(want.psr0),
            report.fmt_bool(want.passes),
        ]


def test_final_segment_csv_uses_the_declared_dates(full_run, explicit):
    rows = {(r["variant"], r["scenario"]): r for r in read_csv(full_run / "tramo-final.csv")}
    assert len(rows) == 12
    for (v, name), run in explicit["runs"].items():
        want = metrics.run_metrics(run, *SEGMENT)
        row = rows[(str(v), name)]
        assert [row[k] for k in ("bars", "total_return", "sharpe_annualized", "max_drawdown")] == [
            str(want.n_returns),
            report.fmt(want.total_return),
            report.fmt(want.sharpe_annualized),
            report.fmt(want.max_drawdown),
        ]
        assert row["executions"] == str(want.executions)


def test_reference_csv_is_the_long_one_times_exposure_in_each_scenario(full_run, explicit):
    rows = {row["scenario"]: row for row in read_csv(full_run / "referencia.csv")}
    for name, run in explicit["refs"].items():
        row = rows[name]
        assert [row[k] for k in METRIC_COLUMNS] == metric_cells(metrics.run_metrics(run))
        final = metrics.run_metrics(run, *SEGMENT)
        assert [row[k] for k in ("final_bars", "final_total_return", "final_executions")] == [
            str(final.n_returns),
            report.fmt(final.total_return),
            str(final.executions),
        ]
        assert row["executions"] == "2"  # entrada en la barra 1,001 y cierre en L


def test_attribution_csv_removes_the_spread_or_the_swap_from_variant_one_in_the_base_scenario(
    full_run, explicit
):
    base = metrics.total_return(explicit["runs"][(1, "base")].ret)
    rows = {row["item"]: row for row in read_csv(full_run / "atribucion.csv")}
    for item, key in (("no_spread", "no_spread"), ("no_swap", "no_swap")):
        alt = metrics.total_return(explicit[key].ret)
        assert [rows[item][k] for k in ("total_return_base", "total_return_alt")] == [
            report.fmt(base),
            report.fmt(alt),
        ]
        assert rows[item]["difference_base_minus_alt"] == report.fmt(base - alt)


def test_pbo_csv_uses_the_four_variants_of_the_base_scenario_in_order(full_run, explicit):
    matrix = np.array([explicit["runs"][(v, "base")].ret for v in PAIRS]).T
    want = metrics.pbo_cscv(matrix)
    items = [r for r in read_csv(full_run / "pbo.csv") if r["item"] != "lambda_distribution"]
    got = {r["item"]: r["value"] for r in items}
    assert got == {
        "pbo": report.fmt(want.pbo),
        "combinations": str(want.n_combinations),
        "n_returns": str(want.n_returns),
        "n_dropped": str(want.n_dropped),
        "n_used": str(want.n_used),
        "block_length": str(want.block_length),
    }
    distribution = [r for r in read_csv(full_run / "pbo.csv") if r["item"] == "lambda_distribution"]
    assert [(r["omega"], r["lambda"], r["count"]) for r in distribution] == [
        (report.fmt(o), report.fmt(math.log(o / (1.0 - o))), str(c)) for o, c in want.omega_counts
    ]


def test_every_equity_row_matches_the_explicit_run(full_run, explicit):
    for (v, name), run in explicit["runs"].items():
        got = read_csv(full_run / f"equity-v{v}-{name}.csv")
        assert len(got) == len(run.dates)
        for i, row in enumerate(got):
            assert [row[k] for k in ("date", "P", "d", "K_minus", "K_plus", "S", "c", "R")] == [
                str(run.dates[i]),
                report.fmt(run.prices[i]),
                str(run.d[i]),
                report.fmt(run.k_minus[i]),
                report.fmt(run.k_plus[i]),
                report.fmt(run.swap[i]),
                report.fmt(run.cost[i]),
                report.fmt(run.ret[i]),
            ]


def test_position_percentage_is_zero_to_one_hundred_and_fixes_per_year_is_pinned(
    full_run, explicit
):
    """La referencia está en posición en todas las barras salvo la L: (T - 1) / T por 100."""
    for row in read_csv(full_run / "referencia.csv"):
        t = int(row["T"])
        assert row["pct_in_position"] == report.fmt(100.0 * (t - 1) / t)
        run = explicit["refs"][row["scenario"]]
        years = (run.dates[-1] - run.dates[0]).days / 365.25
        assert row["fixes_per_year"] == report.fmt(t / years)
    for row in read_csv(full_run / "metrics.csv"):
        assert 0.0 <= float(row["pct_in_position"]) <= 100.0


# --- robustez de run_study -------------------------------------------------------------------


def test_the_script_runs_in_isolated_mode():
    """python -I no agrega el directorio del script: run_study lo agrega por su cuenta."""
    done = subprocess.run(
        [sys.executable, "-I", "-B", str(SCRIPT), "--help"], capture_output=True, text=True
    )
    assert done.returncode == 0, done.stderr
    assert "--tables-only" in done.stdout


def test_the_script_starts_without_docstrings():
    """Con python -OO __doc__ es None: la ayuda no debe depender de él."""
    done = subprocess.run(
        [sys.executable, "-OO", "-B", str(SCRIPT), "--help"], capture_output=True, text=True
    )
    assert done.returncode == 0, done.stderr
    assert "--tables-only" in done.stdout


def test_a_failure_while_writing_leaves_outputs_untouched_and_a_retry_works(
    data_dir, tmp_path, monkeypatch
):
    out = tmp_path / "outputs"
    original = report.write_csv

    def failing(path, header, rows):
        if Path(path).name == "tramo-final.csv":
            raise OSError("disco lleno (simulado)")
        return original(path, header, rows)

    monkeypatch.setattr(report, "write_csv", failing)
    argv = ["--data-dir", str(data_dir), "--out-dir", str(out), "--run-utc", RUN_UTC]
    with pytest.raises(OSError, match="disco lleno"):
        main(argv)
    assert not out.exists()  # nada a medias en outputs/
    assert list(tmp_path.iterdir()) == []  # y el directorio temporal se limpió
    monkeypatch.undo()
    assert main(argv) == 0
    assert sorted(item.name for item in out.iterdir()) == EXPECTED_FILES
    assert list(tmp_path.iterdir()) == [out]


def test_a_variant_one_without_variance_stops_the_run_without_outputs(tmp_path):
    """Con precio constante nadie opera: la variante 1 no tiene varianza y el DSR no existe."""
    data = make_data_dir(tmp_path / "data", flat=True)
    out = tmp_path / "outputs"
    with pytest.raises(ValueError, match="varianza"):
        main(["--data-dir", str(data), "--out-dir", str(out)])
    assert not out.exists()
    assert sorted(item.name for item in tmp_path.iterdir()) == ["data"]


def test_zero_deviation_runs_are_declared_in_the_log_labels():
    """Una corrida con desviación 0 (Sharpe reportado como 0) se declara con su variante."""
    flat = metrics.RunMetrics(5, 0.0, 0.0, 0.0, 0.0, 0, 0.0, 1.0, True)
    live = metrics.RunMetrics(5, 0.1, 0.2, 3.0, 0.1, 2, 0.5, 1.0, False)
    fake = SimpleNamespace(
        window={(1, "base"): live, (2, "base"): flat, (4, "low"): flat},
        final={(3, "high"): flat, (1, "low"): live},
        ref_window={"base": live, "high": flat},
        ref_final={"low": flat, "base": live},
    )
    assert run_study.zero_deviation_labels(fake) == [
        "ref/high",
        "ref/low/tramo-final",
        "v2/base",
        "v3/high/tramo-final",
        "v4/low",
    ]