import numpy as np
import pandas as pd
from src.live.rotation_worker import RotationWorker


def _df(n=200, slope=1.0):
    c = np.linspace(100, 100 + slope * n, n)
    ts = (pd.date_range("2023-01-01", periods=n, freq="1D", tz="UTC").astype("int64") // 10**6)
    return pd.DataFrame({"timestamp": ts, "open": c, "high": c, "low": c, "close": c, "volume": 1.0})


def test_rotation_picks_top_momentum_when_risk_on(tmp_path):
    w = RotationWorker(universe=["BTC", "AAA", "BBB"], top_k=2, gross=0.45, data_dir=tmp_path)
    data = {"BTC": _df(slope=1.0), "AAA": _df(slope=3.0), "BBB": _df(slope=0.1)}
    w.tick(data)
    assert w.weights["AAA"] > 0
    assert abs(sum(w.weights.values()) - 0.45) < 1e-9


def test_rotation_flat_when_risk_off(tmp_path):
    # BTC in downtrend -> risk_off -> nessuna posizione
    w = RotationWorker(universe=["BTC", "AAA"], top_k=1, gross=0.45, data_dir=tmp_path)
    data = {"BTC": _df(slope=-1.0), "AAA": _df(slope=3.0)}
    w.tick(data)
    assert sum(w.weights.values()) == 0.0


def test_rotation_persists_and_resumes(tmp_path):
    w = RotationWorker(universe=["BTC", "AAA"], top_k=1, gross=0.45, data_dir=tmp_path)
    w.tick({"BTC": _df(slope=1.0), "AAA": _df(slope=3.0)})
    w2 = RotationWorker(universe=["BTC", "AAA"], top_k=1, gross=0.45, data_dir=tmp_path)
    assert w2.weights == w.weights


def test_rotation_warns_once_on_short_panel(tmp_path, monkeypatch):
    # worker GIA' operativo + panel troncato -> WARN (una sola notifica per episodio)
    calls = []
    monkeypatch.setattr("src.live.telegram_notifier.notify_event",
                        lambda e, d=None: calls.append(e))
    w = RotationWorker(universe=["BTC", "AAA"], data_dir=tmp_path)
    w.last_bar_ts = 123                       # era gia' operativo
    short = {"BTC": _df(n=20), "AAA": _df(n=20)}
    w.tick(short)
    w.tick(short)                              # dedup: niente seconda notifica
    assert calls == ["PANEL_SHORT"]
    w.tick({"BTC": _df(n=200), "AAA": _df(n=200)})   # recovery resetta il dedup
    assert w._panel_warned is False


def test_panel_drops_forming_daily_bar(tmp_path):
    # serie 1d che termina ADESSO -> l'ultima riga e' la candela 1d in formazione
    import time
    from src.live.rotation_worker import _panel
    n = 200
    now = int(time.time() * 1000)
    ts = now - 86_400_000 * np.arange(n - 1, -1, -1)
    def mk(slope):
        c = np.linspace(100, 100 + slope * n, n)
        return pd.DataFrame({"timestamp": ts, "open": c, "high": c, "low": c,
                             "close": c, "volume": 1.0})
    panel, cols = _panel({"BTC": mk(1.0), "AAA": mk(2.0)}, ["BTC", "AAA"])
    assert len(panel) == n - 1          # barra in formazione scartata
    assert int(panel["timestamp"].iloc[-1]) == int(ts[-2])


def test_rotation_no_warn_on_cold_start(tmp_path, monkeypatch):
    calls = []
    monkeypatch.setattr("src.live.telegram_notifier.notify_event",
                        lambda e, d=None: calls.append(e))
    w = RotationWorker(universe=["BTC", "AAA"], data_dir=tmp_path)   # last_bar_ts=0
    w.tick({"BTC": _df(n=20), "AAA": _df(n=20)})
    assert calls == []