import os
import subprocess
import sys

import pandas as pd

from tools.compare_tv_trades import (
    apply_pine_time_gate,
    check_tv_entry_markers,
    run_python_strategy,
    strategy_stem,
    tv_score_column,
)


REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
DATA_1D = os.path.join(REPO, "data", "mlp", "COINBASE_BTCUSD, 1D.csv")
MLP_PARAMS_1D = os.path.join(
    REPO,
    "results",
    "winners",
    "optimization_winner_strategy_mlp_scores_COINBASE_BTCUSD_1D.csv",
)


def test_mlp_parity_accepts_unix_seconds_and_iso_dates(tmp_path):
    from tools.check_mlp_parity import _auto_resolve, load_params, run_parity

    data, weights, params_path = _auto_resolve("COINBASE_BTCUSD", "6H")
    frame = pd.read_csv(data).iloc[-20:-1].copy()
    unit = "s" if pd.api.types.is_numeric_dtype(frame.time) else None
    times = pd.to_datetime(frame.time, unit=unit, utc=True)
    unix_path, iso_path = tmp_path / "unix.csv", tmp_path / "iso.csv"
    frame["time"] = times.dt.as_unit("ns").astype("int64") // 10**9
    frame.to_csv(unix_path, index=False)
    frame["time"] = times.map(lambda time: time.isoformat())
    frame.to_csv(iso_path, index=False)
    params = load_params(params_path)
    unix = run_parity(str(unix_path), weights, params)
    iso = run_parity(str(iso_path), weights, params)
    assert len(unix) == 19
    assert unix.tv_score.notna().any()
    unix["time"] = unix.time.dt.as_unit("ns")
    iso["time"] = iso.time.dt.as_unit("ns")
    pd.testing.assert_frame_equal(unix, iso)


def test_strategy_stem_supports_activation_and_mlp_files():
    assert strategy_stem("strategy_activation_scores.py") == "activation_scores"
    assert strategy_stem("strategy_mlp_scores.py") == "mlp_scores"


def test_trade_loader_accepts_return_percent_header(tmp_path):
    from tools.compare_tv_trades import load_tv_trades

    frame = pd.DataFrame({"Trade number": [1, 1], "Type": ["Exit long", "Entry long"],
        "Date and time": ["2026-09-24 12:00", "2026-09-23 12:00"],
        "Price USD": [110., 100.], "Return %": [8.9, 8.9],
        "Signal": ["Close entry(s) order long", "long"]})
    new, old = tmp_path / "new.csv", tmp_path / "old.csv"
    frame.to_csv(new, index=False, encoding="utf-8-sig")
    frame.rename(columns={"Return %": "Net P&L %"}).to_csv(old, index=False)
    parsed = load_tv_trades(str(new))
    assert len(parsed) == 1
    assert parsed[0]["net_pnl_pct"] == 8.9
    assert parsed == load_tv_trades(str(old))


def test_completed_tv_totals_excludes_open_trade_mark(tmp_path):
    from tools.compare_tv_trades import completed_tv_totals

    path = tmp_path / "trades.csv"
    pd.DataFrame({"Trade number": [2, 1, 2, 1],
        "Type": ["Exit long", "Exit long", "Entry long", "Entry long"],
        "Date and time": ["Open", "2026-09-01", "2026-09-02", "2026-08-01"],
        "Cumulative PnL %": [50., 10., 50., 10.]}).to_csv(path, index=False)
    totals = completed_tv_totals(str(path), [{'trade_num': 1}])
    assert totals == {'closed_cumulative_pnl_pct': 10., 'open_trade_count': 1}


def test_tv_score_column_prefers_available_export_score():
    assert tv_score_column(pd.DataFrame({"activation_score_poc": [1.0]})) == "activation_score_poc"
    assert tv_score_column(pd.DataFrame({"mlp_score": [1.0]})) == "mlp_score"
    assert tv_score_column(pd.DataFrame({"close": [1.0]})) is None


def test_tv_entry_marker_check_detects_export_mismatch():
    signals = pd.DataFrame({
        "time": pd.to_datetime([
            "2020-01-01 00:00",
            "2020-01-01 06:00",
            "2020-01-01 12:00",
        ]),
        "entry (standard)": [1.0, 0.0, 1.0],
        "missed entry": [0.0, 0.0, 1.0],
    })
    tv_trades = [{
        "trade_num": 1,
        "entry_dt": pd.Timestamp("2020-01-01 12:00"),
        "is_margin": False,
    }]

    result = check_tv_entry_markers(
        tv_trades,
        signals,
        "6H",
        pd.Timestamp("2020-01-01 00:00"),
        pd.Timestamp("2020-01-01 12:00"),
    )

    assert result["tv_with_marker"] == 0
    assert len(result["tv_without_marker"]) == 1
    assert result["chart_exec_count"] == 1
    assert result["chart_without_tv"] == [pd.Timestamp("2020-01-01 00:00")]


def test_pine_time_gate_starts_flat_at_score_start():
    signals = pd.DataFrame({
        "time": pd.to_datetime([
            "2017-11-29 00:00",
            "2017-11-30 00:00",
            "2017-12-01 00:00",
            "2017-12-01 12:00",
            "2017-12-02 00:00",
        ]),
        "activation_score": [100.0, -250.0, 100.0, -250.0, -300.0],
        "close": [100.0, 90.0, 100.0, 110.0, 120.0],
        "high": [100.0, 90.0, 100.0, 110.0, 120.0],
    })

    replayed = apply_pine_time_gate(
        signals,
        {
            "i_long_entry_activation_threshold": -200.0,
            "i_long_exit_activation_threshold": 50.0,
            "i_long_exit_activation_confirmation_threshold": -230.0,
            "i_use_long_exit_confirmation": True,
            "i_use_long_entry_confirmation": False,
            "i_trailing_stop_threshold": 0.0,
        },
        pd.Timestamp("2017-12-01"),
    )

    assert replayed["execute_entry"].tolist() == [False, False, False, True, False]
    assert replayed["execute_exit"].tolist() == [False, False, False, False, True]


def test_pine_time_gate_preserves_mlp_entry_smoothing():
    signals = pd.DataFrame({
        "time": pd.to_datetime([
            "2020-01-01 00:00",
            "2020-01-01 08:00",
            "2020-01-01 16:00",
            "2020-01-02 00:00",
        ]),
        "activation_score": [200.0, 200.0, 0.0, 0.0],
        "close": [100.0, 101.0, 102.0, 103.0],
        "high": [100.0, 101.0, 102.0, 103.0],
        "stoch_peak_norm": [0.0, 0.0, 0.0, 0.0],
    })
    signals.attrs["strategy_stem"] = "mlp_scores"

    replayed = apply_pine_time_gate(
        signals,
        {
            "i_long_entry_activation_threshold": 100.0,
            "i_long_exit_activation_threshold": 900.0,
            "i_long_exit_activation_confirmation_threshold": -900.0,
            "i_use_long_exit_confirmation": True,
            "i_use_long_entry_confirmation": False,
            "i_entry_score_window": 3,
            "i_exit_score_window": 3,
            "i_trailing_stop_threshold": 0.0,
        },
        pd.Timestamp("2020-01-01 16:00"),
    )

    assert replayed["execute_entry"].tolist() == [False, False, False, True]


def test_compare_tv_trades_can_run_mlp_strategy():
    params = pd.read_csv(MLP_PARAMS_1D).iloc[0].to_dict()
    signals = run_python_strategy(
        DATA_1D,
        params,
        strategy_file="strategy_mlp_scores.py",
    )

    assert "activation_score" in signals.columns
    assert "execute_entry" in signals.columns
    assert "execute_exit" in signals.columns
    assert signals["execute_entry"].sum() > 0
    assert signals["time"].max() > pd.Timestamp("2020-01-01")


def test_validate_strategy_prints_canonical_metrics_section():
    result = subprocess.run(
        [
            sys.executable,
            "strategies/validate_strategy.py",
            "--data",
            DATA_1D,
            "--strategy_file",
            "strategy_mlp_scores.py",
            "--params_file",
            MLP_PARAMS_1D,
        ],
        cwd=REPO,
        capture_output=True,
        text=True,
        timeout=120,
        check=False,
    )

    assert result.returncode == 0, result.stderr
    assert "Trade-walk diagnostics" in result.stdout
    assert "Canonical optimizer metrics" in result.stdout
    assert "Source: strategy.calculate_metrics(signals, score_start=...)" in result.stdout