tests/test_autotune.py

import sqlite3
import json
from autotune import init_tuning_table, save_tuning, load_tuning

def _params(**overrides):
    base = dict(
        lookback=200, pred_len=30, stride=30,
        sample_count=1, max_anchors=20,
        start_date="", end_date="",
    )
    base.update(overrides)
    return base

def test_init_tuning_table_creates_schema(tmp_db):
    init_tuning_table(tmp_db)
    with sqlite3.connect(tmp_db) as c:
        cols = [r[1] for r in c.execute("PRAGMA table_info(tuning_results)").fetchall()]
    assert "symbol" in cols
    assert "interval" in cols
    assert "best_T" in cols
    assert "best_top_p" in cols
    assert "best_pnl_pct" in cols
    assert "grid_json" in cols

def test_init_tuning_table_unique_constraint(tmp_db):
    init_tuning_table(tmp_db)
    with sqlite3.connect(tmp_db) as c:
        c.execute("INSERT INTO tuning_results (ts, symbol, interval, best_T, best_top_p, best_pnl_pct, best_hit_rate, best_rmse, grid_json) "
                  "VALUES ('t1','SPY','5min',1.0,0.9,5.0,0.6,1.2,'[]')")
        c.execute("INSERT INTO tuning_results (ts, symbol, interval, best_T, best_top_p, best_pnl_pct, best_hit_rate, best_rmse, grid_json) "
                  "VALUES ('t2','SPY','5min',1.5,0.85,7.0,0.7,1.0,'[]')")
        rows = c.execute("SELECT best_T, best_top_p FROM tuning_results WHERE symbol='SPY' AND interval='5min'").fetchall()
    assert rows == [(1.5, 0.85)]

def test_save_and_load_roundtrip(tmp_db):
    init_tuning_table(tmp_db)
    best = {"T": 1.0, "top_p": 0.85, "total_return_pct": 4.2,
            "hit_rate": 0.6, "mean_rmse": 1.1}
    results = [{"T": 1.0, "top_p": 0.85, "total_return_pct": 4.2}]
    save_tuning(tmp_db, "SPY", "5min", best, results, _params())
    got = load_tuning(tmp_db, "SPY", "5min")
    assert got is not None
    assert got["best_T"] == 1.0
    assert got["best_top_p"] == 0.85
    assert got["best_pnl_pct"] == 4.2
    assert json.loads(got["grid_json"]) == results

def test_load_tuning_missing_returns_none(tmp_db):
    init_tuning_table(tmp_db)
    assert load_tuning(tmp_db, "ZZZ", "1min") is None

def test_save_tuning_replaces_on_conflict(tmp_db):
    init_tuning_table(tmp_db)
    save_tuning(tmp_db, "SPY", "5min",
                {"T": 1.0, "top_p": 0.85, "total_return_pct": 1.0,
                 "hit_rate": 0.5, "mean_rmse": 2.0},
                [], _params())
    save_tuning(tmp_db, "SPY", "5min",
                {"T": 1.5, "top_p": 0.7, "total_return_pct": 9.0,
                 "hit_rate": 0.8, "mean_rmse": 0.5},
                [], _params())
    got = load_tuning(tmp_db, "SPY", "5min")
    assert got["best_T"] == 1.5
    assert got["best_pnl_pct"] == 9.0
    with sqlite3.connect(tmp_db) as c:
        n = c.execute("SELECT COUNT(*) FROM tuning_results WHERE symbol='SPY' AND interval='5min'").fetchone()[0]
    assert n == 1


import numpy as np
import pandas as pd
from autotune import backtest_core

def _make_full(rows):
    ts = pd.date_range("2026-01-01", periods=rows, freq="5min")
    closes = np.linspace(100.0, 110.0, rows)
    df = pd.DataFrame({
        "timestamps": ts,
        "open":   closes, "high": closes + 0.1, "low": closes - 0.1,
        "close":  closes, "volume": 1000.0, "amount": closes * 1000.0,
    })
    return df

def _stub_predict(rising):
    """Returns a predict fn that always forecasts a monotone path."""
    def _f(df, x_timestamp, y_timestamp, pred_len, T, top_p, sample_count, verbose):
        last = float(df["close"].iloc[-1])
        delta = 1.0 if rising else -1.0
        path = np.linspace(last + delta, last + delta * pred_len, pred_len)
        return pd.DataFrame({
            "open": path, "high": path + 0.1, "low": path - 0.1,
            "close": path, "volume": 1000.0, "amount": path * 1000.0,
        }, index=y_timestamp)
    return _f

def test_backtest_core_perfect_long_when_price_rises_and_forecast_rises():
    full = _make_full(rows=400)  # rising
    fetch_fn = lambda symbol, interval, n_bars: full.copy()
    predict_fn = _stub_predict(rising=True)
    out = backtest_core(predict_fn, fetch_fn,
                        symbol="X", interval="5min",
                        start_date="", end_date="",
                        lookback=100, pred_len=10, stride=10,
                        T=1.0, top_p=0.9,
                        sample_count=1, max_anchors=5)
    assert out["anchors"] == 5
    assert out["hit_rate"] == 1.0  # forecast up, realized up — every anchor hits
    assert out["total_return_pct"] > 0
    assert isinstance(out["per_anchor"], pd.DataFrame)
    assert len(out["per_anchor"]) == 5

def test_backtest_core_short_when_forecast_down_but_price_up():
    full = _make_full(rows=400)
    fetch_fn = lambda *a, **k: full.copy()
    predict_fn = _stub_predict(rising=False)  # forecast says down
    out = backtest_core(predict_fn, fetch_fn,
                        symbol="X", interval="5min",
                        start_date="", end_date="",
                        lookback=100, pred_len=10, stride=10,
                        T=1.0, top_p=0.9,
                        sample_count=1, max_anchors=5)
    assert out["hit_rate"] == 0.0  # always wrong
    assert out["total_return_pct"] < 0  # short while market rises = loss

def test_backtest_core_raises_when_lookback_plus_predlen_too_big():
    full = _make_full(rows=400)
    fetch_fn = lambda *a, **k: full.copy()
    predict_fn = _stub_predict(rising=True)
    import pytest
    with pytest.raises(ValueError, match="≤ 512"):
        backtest_core(predict_fn, fetch_fn,
                      symbol="X", interval="5min",
                      start_date="", end_date="",
                      lookback=400, pred_len=200, stride=50,
                      T=1.0, top_p=0.9,
                      sample_count=1, max_anchors=5)


from unittest.mock import MagicMock
from autotune import run_autotune, GRID_T, GRID_TOP_P

def test_run_autotune_iterates_full_grid_and_picks_max(tmp_db, monkeypatch):
    init_tuning_table(tmp_db)
    calls = []

    def fake_core(predict_fn, fetch_fn, *, symbol, interval, start_date, end_date,
                  lookback, pred_len, stride, T, top_p, sample_count, max_anchors):
        calls.append((T, top_p))
        # Make T=1.0, top_p=0.85 win clearly
        score = 9.0 if (T == 1.0 and top_p == 0.85) else 1.0
        return {
            "anchors": 5, "mean_rmse": 1.0,
            "hit_rate": 0.5,
            "total_return_pct": score,
            "max_dd_pct": -1.0, "sharpe": 0.5,
            "per_anchor": pd.DataFrame(),
        }

    monkeypatch.setattr("autotune.backtest_core", fake_core)
    fig, summary, grid = run_autotune(
        predict_fn=MagicMock(), fetch_fn=MagicMock(),
        db_path=tmp_db, symbol="SPY", interval="5min",
        start_date="", end_date="",
        lookback=200, pred_len=30, stride=30,
        sample_count=1, max_anchors=5,
    )
    assert sorted(calls) == sorted([(t, p) for t in GRID_T for p in GRID_TOP_P])
    assert len(grid) == 9
    best_row = summary[summary["Field"] == "Best T"].iloc[0]["Value"]
    assert float(best_row) == 1.0

    persisted = load_tuning(tmp_db, "SPY", "5min")
    assert persisted["best_T"] == 1.0
    assert persisted["best_top_p"] == 0.85
    assert persisted["best_pnl_pct"] == 9.0

def test_run_autotune_skips_failed_cells(tmp_db, monkeypatch):
    init_tuning_table(tmp_db)

    def flaky_core(predict_fn, fetch_fn, *, T, top_p, **kw):
        if T == 0.5:
            raise RuntimeError("fmp blew up")
        return {"anchors": 5, "mean_rmse": 1.0, "hit_rate": 0.5,
                "total_return_pct": 2.0 + top_p,
                "max_dd_pct": -1.0, "sharpe": 0.5,
                "per_anchor": pd.DataFrame()}

    monkeypatch.setattr("autotune.backtest_core", flaky_core)
    fig, summary, grid = run_autotune(
        predict_fn=MagicMock(), fetch_fn=MagicMock(),
        db_path=tmp_db, symbol="X", interval="5min",
        start_date="", end_date="",
        lookback=200, pred_len=30, stride=30,
        sample_count=1, max_anchors=5,
    )
    errored = grid[grid["error"].notna()]
    assert len(errored) == 3                      # all T=0.5 rows
    valid = grid[grid["error"].isna()]
    assert len(valid) == 6
    persisted = load_tuning(tmp_db, "X", "5min")
    assert persisted["best_T"] != 0.5            # never the failing T

def test_run_autotune_raises_when_all_cells_fail(tmp_db, monkeypatch):
    init_tuning_table(tmp_db)
    def always_fail(*a, **kw):
        raise RuntimeError("everything is broken")
    monkeypatch.setattr("autotune.backtest_core", always_fail)

    import pytest
    with pytest.raises(RuntimeError, match="All grid cells failed"):
        run_autotune(
            predict_fn=MagicMock(), fetch_fn=MagicMock(),
            db_path=tmp_db, symbol="X", interval="5min",
            start_date="", end_date="",
            lookback=200, pred_len=30, stride=30,
            sample_count=1, max_anchors=5,
        )
    assert load_tuning(tmp_db, "X", "5min") is None

def test_apply_tuning_returns_values_when_cached_and_enabled(tmp_db):
    from autotune import apply_tuning
    init_tuning_table(tmp_db)
    save_tuning(tmp_db, "SPY", "5min",
                {"T": 1.5, "top_p": 0.7, "total_return_pct": 7.0,
                 "hit_rate": 0.6, "mean_rmse": 0.9},
                [], {"lookback": 200, "pred_len": 30, "stride": 30,
                     "sample_count": 1, "max_anchors": 20,
                     "start_date": "", "end_date": ""})
    t_val, top_p_val = apply_tuning(tmp_db, "SPY", "5min", True)
    assert t_val == 1.5
    assert top_p_val == 0.7

def test_apply_tuning_returns_none_when_disabled(tmp_db):
    from autotune import apply_tuning
    init_tuning_table(tmp_db)
    save_tuning(tmp_db, "SPY", "5min",
                {"T": 1.5, "top_p": 0.7, "total_return_pct": 7.0,
                 "hit_rate": 0.6, "mean_rmse": 0.9},
                [], {"lookback": 200, "pred_len": 30, "stride": 30,
                     "sample_count": 1, "max_anchors": 20,
                     "start_date": "", "end_date": ""})
    assert apply_tuning(tmp_db, "SPY", "5min", False) == (None, None)

def test_apply_tuning_returns_none_when_no_cache(tmp_db):
    from autotune import apply_tuning
    init_tuning_table(tmp_db)
    assert apply_tuning(tmp_db, "ZZZ", "1min", True) == (None, None)