"""振幅统计核心逻辑单元测试(不打交易所)."""
from __future__ import annotations

import unittest
from datetime import date, datetime
from zoneinfo import ZoneInfo

from lib.hub.amp_stats_lib import (
    build_export_csv,
    compute_amp_stats,
    compute_day_row,
    list_settlement_dates,
    summarize_rows,
    window_bounds_for_settlement,
)

import httpx

TZ = ZoneInfo("Asia/Shanghai")


def _bar(ts_ms: int, o: float, h: float, l: float, c: float) -> dict:
    return {"ts": ts_ms, "o": o, "h": h, "l": l, "c": c}


class AmpStatsLibTests(unittest.TestCase):
    def test_window_cross_day_22_to_16(self):
        start, end = window_bounds_for_settlement(date(2026, 7, 22), 22)
        self.assertEqual(start.strftime("%Y-%m-%d %H:%M"), "2026-07-21 22:00")
        self.assertEqual(end.strftime("%Y-%m-%d %H:%M"), "2026-07-22 16:00")

    def test_window_same_day_8_to_16(self):
        start, end = window_bounds_for_settlement(date(2026, 7, 22), 8)
        self.assertEqual(start.strftime("%Y-%m-%d %H:%M"), "2026-07-22 08:00")
        self.assertEqual(end.strftime("%Y-%m-%d %H:%M"), "2026-07-22 16:00")

    def test_window_two_day_16_to_16(self):
        # 结算 27 日 → 两日窗 25日16:00 → 27日16:00
        start, end = window_bounds_for_settlement(date(2026, 7, 27), 16, span_days=2)
        self.assertEqual(start.strftime("%Y-%m-%d %H:%M"), "2026-07-25 16:00")
        self.assertEqual(end.strftime("%Y-%m-%d %H:%M"), "2026-07-27 16:00")
        one_start, _ = window_bounds_for_settlement(date(2026, 7, 27), 16, span_days=1)
        self.assertEqual(one_start.strftime("%Y-%m-%d %H:%M"), "2026-07-26 16:00")

    def test_settlement_excludes_incomplete_today(self):
        now = datetime(2026, 7, 22, 10, 0, tzinfo=TZ)
        days = list_settlement_dates(sample_days=3, now=now)
        self.assertEqual(days[0].isoformat(), "2026-07-21")
        self.assertEqual(len(days), 3)

    def test_settlement_includes_today_after_1600(self):
        now = datetime(2026, 7, 22, 16, 0, tzinfo=TZ)
        days = list_settlement_dates(sample_days=1, now=now)
        self.assertEqual(days[0].isoformat(), "2026-07-22")

    def test_day_row_points(self):
        # 22:00 D-1 → 16:00 D; O=2000 H=2500 L=1800 C=2100 → up500 down200 amp700
        settlement = date(2026, 7, 22)
        start, end = window_bounds_for_settlement(settlement, 22)
        bar_map = {}
        t = int(start.timestamp() * 1000)
        last = int((end.replace(hour=15)).timestamp() * 1000)
        # first bar
        bar_map[t] = {"o": 2000.0, "h": 2100.0, "l": 1950.0, "c": 2050.0}
        cur = t + 3600 * 1000
        while cur < last:
            bar_map[cur] = {"o": 2050.0, "h": 2200.0, "l": 1900.0, "c": 2100.0}
            cur += 3600 * 1000
        # peak and trough somewhere
        mid = t + 5 * 3600 * 1000
        bar_map[mid] = {"o": 2100.0, "h": 2500.0, "l": 1800.0, "c": 2000.0}
        bar_map[last] = {"o": 2000.0, "h": 2150.0, "l": 1990.0, "c": 2100.0}
        # fill any missing hours with flat
        cur = t
        while cur <= last:
            if cur not in bar_map:
                bar_map[cur] = {"o": 2000.0, "h": 2000.0, "l": 2000.0, "c": 2000.0}
            cur += 3600 * 1000
        row = compute_day_row(settlement, 22, bar_map)
        self.assertIsNotNone(row)
        self.assertEqual(row["open"], 2000.0)
        self.assertEqual(row["high"], 2500.0)
        self.assertEqual(row["low"], 1800.0)
        self.assertEqual(row["up_points"], 500.0)
        self.assertEqual(row["down_points"], 200.0)
        self.assertEqual(row["amplitude"], 700.0)
        self.assertEqual(row["change"], 100.0)

    def test_summary_max_amplitude(self):
        rows = [
            {"amplitude": 100, "up_points": 40, "down_points": 60, "change": 10, "settlement_day": "2026-07-01"},
            {"amplitude": 700, "up_points": 500, "down_points": 200, "change": -5, "settlement_day": "2026-07-02"},
            {"amplitude": 200, "up_points": 50, "down_points": 150, "change": 20, "settlement_day": "2026-07-03"},
        ]
        s = summarize_rows(rows)
        self.assertEqual(s["max_amplitude"], 700)
        self.assertEqual(s["max_amplitude_day"], "2026-07-02")
        self.assertEqual(s["max_up_points"], 500)
        self.assertEqual(s["max_down_points"], 200)
        self.assertIsNone(s["move_points_stats"])

    def test_move_points_amp_ratio(self):
        rows = [
            {"amplitude": 100, "up_points": 40, "down_points": 60, "change": 10, "settlement_day": "2026-07-01"},
            {"amplitude": 40, "up_points": 10, "down_points": 30, "change": -5, "settlement_day": "2026-07-02"},
            {"amplitude": 50, "up_points": 50, "down_points": 0, "change": 20, "settlement_day": "2026-07-03"},
        ]
        s = summarize_rows(rows, move_points=50)
        ms = s["move_points_stats"]
        self.assertIsNotNone(ms)
        self.assertEqual(ms["move_points"], 50)
        self.assertEqual(ms["amp_hit_days"], 2)  # 100, 50
        self.assertEqual(ms["amp_hit_ratio"], round(2 / 3, 4))
        self.assertEqual(ms["up_hit_days"], 1)  # 50
        self.assertEqual(ms["down_hit_days"], 1)  # 60
        csv_text = build_export_csv(
            {
                "exchange": "okx",
                "symbol_label": "ETH",
                "summary": s,
                "rows": rows,
                "start_hour": 16,
                "end_hour": 16,
            }
        )
        self.assertIn("振幅占比", csv_text)
        self.assertIn("振幅达标", csv_text)

    def test_weekend_and_reframe_move_points(self):
        from lib.hub.amp_stats_lib import enrich_rows, filter_weekend_rows, reframe_amp_stats

        rows = [
            {
                "settlement_day": "2026-07-18",
                "is_weekend": True,
                "weekday_label": "六",
                "up_points": 100,
                "down_points": 10,
                "change": -5,
                "amplitude": 110,
            },
            {
                "settlement_day": "2026-07-19",
                "is_weekend": True,
                "weekday_label": "日",
                "up_points": 20,
                "down_points": 15,
                "change": 12,
                "amplitude": 35,
            },
            {
                "settlement_day": "2026-07-20",
                "is_weekend": False,
                "weekday_label": "",
                "up_points": 50,
                "down_points": 40,
                "change": 8,
                "amplitude": 90,
            },
        ]
        excl = filter_weekend_rows(rows, "exclude")
        self.assertEqual(len(excl), 1)
        self.assertEqual(excl[0]["settlement_day"], "2026-07-20")
        only = filter_weekend_rows(rows, "only")
        self.assertEqual(len(only), 2)

        enriched = enrich_rows(rows, move_points=80)
        self.assertTrue(enriched[0]["amp_hit"])
        self.assertFalse(enriched[1]["amp_hit"])
        self.assertTrue(enriched[2]["amp_hit"])

        reframed = reframe_amp_stats(
            rows_all=rows,
            symbol="eth",
            weekend_filter="exclude",
            move_points=80,
        )
        self.assertEqual(reframed["summary"]["sample_count"], 1)
        self.assertTrue(reframed["rows"][0]["amp_hit"])
        self.assertIn("振幅占比", build_export_csv(reframed))

    def test_fetch_switches_to_history_endpoint(self):
        """近期接口到头后应切 history 续拉."""
        from lib.hub.amp_stats_lib import fetch_okx_candles

        calls: list[str] = []

        class FakeResp:
            def __init__(self, data, url="https://x", status_code=200):
                self._data = data
                self.status_code = status_code
                self.url = url
                self.request = httpx.Request("GET", url)

            def raise_for_status(self):
                if self.status_code >= 400:
                    raise httpx.HTTPStatusError(
                        "err", request=self.request, response=self
                    )

            def json(self):
                return {"code": "0", "data": self._data}

        class FakeClient:
            def get(self, url, params=None):
                calls.append(url)
                after = (params or {}).get("after")
                # recent: only 2 pages then empty; history continues
                if "history" not in url:
                    if after is None:
                        return FakeResp([["2000", "1", "2", "0.5", "1.5"], ["1900", "1", "2", "0.5", "1.5"]], url=url)
                    if after == "1900":
                        return FakeResp([], url=url)  # recent exhausted
                    return FakeResp([], url=url)
                # history
                if after == "1900":
                    return FakeResp([["1800", "1", "2", "0.5", "1.5"], ["1000", "1", "2", "0.5", "1.5"]], url=url)
                return FakeResp([], url=url)

            def close(self):
                return None

        bars = fetch_okx_candles(
            url="https://www.okx.com/api/v5/market/index-candles",
            history_url="https://www.okx.com/api/v5/market/history-index-candles",
            inst_id="ETH-USD",
            since_ms=1000,
            until_ms=3000,
            client=FakeClient(),
            max_pages=10,
            page_pause_sec=0,
            history_page_pause_sec=0,
        )
        self.assertTrue(any("history-index-candles" in u for u in calls))
        self.assertGreaterEqual(len(bars), 3)
        self.assertEqual(bars[0]["ts"], 1000)

    def test_fetch_retries_on_429(self):
        from lib.hub.amp_stats_lib import fetch_okx_candles
        import httpx as _httpx

        hits = {"n": 0}

        class FakeResp:
            def __init__(self, status_code, data=None):
                self.status_code = status_code
                self.url = "https://www.okx.com/api/v5/market/history-candles"
                self.request = _httpx.Request("GET", self.url)
                self._data = data or []

            def raise_for_status(self):
                if self.status_code >= 400:
                    raise _httpx.HTTPStatusError("429", request=self.request, response=self)

            def json(self):
                return {"code": "0", "data": self._data}

        class FakeClient:
            def get(self, url, params=None):
                hits["n"] += 1
                if hits["n"] < 3:
                    return FakeResp(429)
                return FakeResp(200, [["1000", "1", "2", "0.5", "1.5"]])

            def close(self):
                return None

        bars = fetch_okx_candles(
            url="https://www.okx.com/api/v5/market/candles",
            history_url=None,
            inst_id="ETH-USDT-SWAP",
            since_ms=1000,
            until_ms=2000,
            client=FakeClient(),
            max_pages=3,
            page_pause_sec=0,
            history_page_pause_sec=0,
        )
        self.assertGreaterEqual(hits["n"], 3)
        self.assertEqual(len(bars), 1)

    def test_compute_with_mock_fetch(self):
        now = datetime(2026, 7, 22, 18, 0, tzinfo=TZ)

        def fetch_fn(*, inst_id, since_ms, until_ms):
            bars = []
            t = since_ms - (since_ms % (3600 * 1000))
            while t <= until_ms:
                # synthetic: open 2000, one spike day
                o = 2000.0
                h = 2500.0 if t == since_ms + 5 * 3600 * 1000 else 2050.0
                l = 1800.0 if t == since_ms + 5 * 3600 * 1000 else 1950.0
                c = 2020.0
                bars.append(_bar(t, o, h, l, c))
                t += 3600 * 1000
            return bars

        result = compute_amp_stats(
            symbol="eth",
            start_hour=16,
            period="custom",
            custom_days=7,
            now=now,
            fetch_fn=fetch_fn,
        )
        self.assertTrue(result["ok"])
        self.assertEqual(result["exchange"], "okx")
        self.assertGreaterEqual(result["summary"]["sample_count"], 1)
        csv_text = build_export_csv(result)
        self.assertIn("最大振幅", csv_text)
        self.assertIn("日表明细", csv_text)


if __name__ == "__main__":
    unittest.main()