crypto_monitor/tests/test_hub_trades_lib.py

"""hub_trades_lib 单元测试。"""
from __future__ import annotations

import sqlite3
import unittest
from datetime import datetime

from hub_trades_lib import (
    fetch_trades_for_trading_day,
    summarize_trades,
    trading_day_from_dt,
    trading_day_window_bounds,
)


class HubTradesLibTest(unittest.TestCase):
    def test_trading_day_reset(self):
        dt = datetime(2026, 6, 6, 7, 30, 0)
        self.assertEqual(trading_day_from_dt(dt, 8), "2026-06-05")
        dt2 = datetime(2026, 6, 6, 8, 0, 0)
        self.assertEqual(trading_day_from_dt(dt2, 8), "2026-06-06")

    def test_trading_day_window_bounds(self):
        start, end = trading_day_window_bounds("2026-06-06", 8)
        self.assertEqual(start, "2026-06-06 08:00:00")
        self.assertEqual(end, "2026-06-07 07:59:59")

    def test_fetch_and_summarize(self):
        conn = sqlite3.connect(":memory:")
        conn.row_factory = sqlite3.Row
        conn.execute(
            """CREATE TABLE trade_records (
                symbol TEXT, direction TEXT, result TEXT, reviewed_result TEXT,
                pnl_amount REAL, reviewed_pnl_amount REAL, exchange_realized_pnl REAL,
                closed_at TEXT, reviewed_closed_at TEXT, opened_at TEXT, reviewed_opened_at TEXT,
                created_at TEXT, monitor_type TEXT, actual_rr REAL, planned_rr REAL,
                trade_style TEXT, entry_reason TEXT, reviewed_at TEXT
            )"""
        )
        conn.execute(
            "INSERT INTO trade_records VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                "ONDO/USDT",
                "short",
                "止损",
                None,
                -0.5,
                None,
                None,
                "2026-06-06 10:00:00",
                None,
                "2026-06-06 09:00:00",
                None,
                "2026-06-06 10:00:00",
                "趋势回调",
                None,
                None,
                "trend",
                "",
                None,
            ),
        )
        conn.commit()
        rows = fetch_trades_for_trading_day(conn, "2026-06-06")
        self.assertEqual(len(rows), 1)
        stats = summarize_trades(rows)
        self.assertEqual(stats["closed_count"], 1)
        self.assertEqual(stats["loss_count"], 1)
        self.assertAlmostEqual(stats["total_pnl_u"], -0.5)
        conn.close()

    def test_early_morning_belongs_prev_trading_day(self):
        conn = sqlite3.connect(":memory:")
        conn.row_factory = sqlite3.Row
        conn.execute(
            """CREATE TABLE trade_records (
                symbol TEXT, direction TEXT, result TEXT, reviewed_result TEXT,
                pnl_amount REAL, reviewed_pnl_amount REAL, exchange_realized_pnl REAL,
                closed_at TEXT, reviewed_closed_at TEXT, opened_at TEXT, reviewed_opened_at TEXT,
                created_at TEXT, monitor_type TEXT, actual_rr REAL, planned_rr REAL,
                trade_style TEXT, entry_reason TEXT, reviewed_at TEXT
            )"""
        )
        conn.execute(
            "INSERT INTO trade_records VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                "BTC/USDT",
                "long",
                "止盈",
                None,
                1.2,
                None,
                None,
                "2026-06-07 07:30:00",
                None,
                "2026-06-07 06:00:00",
                None,
                "2026-06-07 07:30:00",
                "关键位",
                None,
                None,
                "trend",
                "",
                None,
            ),
        )
        conn.commit()
        self.assertEqual(len(fetch_trades_for_trading_day(conn, "2026-06-07")), 0)
        self.assertEqual(len(fetch_trades_for_trading_day(conn, "2026-06-06")), 1)
        conn.close()

    def test_reviewed_fields_preferred(self):
        conn = sqlite3.connect(":memory:")
        conn.row_factory = sqlite3.Row
        conn.execute(
            """CREATE TABLE trade_records (
                symbol TEXT, direction TEXT, result TEXT, reviewed_result TEXT,
                pnl_amount REAL, reviewed_pnl_amount REAL, exchange_realized_pnl REAL,
                closed_at TEXT, reviewed_closed_at TEXT, opened_at TEXT, reviewed_opened_at TEXT,
                created_at TEXT, monitor_type TEXT, actual_rr REAL, planned_rr REAL,
                trade_style TEXT, entry_reason TEXT, reviewed_at TEXT
            )"""
        )
        conn.execute(
            "INSERT INTO trade_records VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                "ETH/USDT",
                "long",
                "止损",
                "止盈",
                -0.5,
                2.0,
                None,
                "2026-06-06 09:00:00",
                "2026-06-06 11:00:00",
                "2026-06-06 08:00:00",
                None,
                "2026-06-06 11:00:00",
                "趋势回调",
                None,
                None,
                "trend",
                "",
                "2026-06-06 12:00:00",
            ),
        )
        conn.commit()
        rows = fetch_trades_for_trading_day(conn, "2026-06-06")
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["result"], "止盈")
        self.assertAlmostEqual(rows[0]["pnl_amount"], 2.0)
        self.assertTrue(rows[0]["reviewed"])
        conn.close()

    def test_time_close_result_included(self):
        conn = sqlite3.connect(":memory:")
        conn.row_factory = sqlite3.Row
        conn.execute(
            """CREATE TABLE trade_records (
                symbol TEXT, direction TEXT, result TEXT, reviewed_result TEXT,
                pnl_amount REAL, reviewed_pnl_amount REAL, exchange_realized_pnl REAL,
                closed_at TEXT, reviewed_closed_at TEXT, opened_at TEXT, reviewed_opened_at TEXT,
                created_at TEXT, monitor_type TEXT, actual_rr REAL, planned_rr REAL,
                trade_style TEXT, entry_reason TEXT, reviewed_at TEXT
            )"""
        )
        conn.execute(
            "INSERT INTO trade_records VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                "BTC/USDT",
                "long",
                "时间平仓",
                None,
                1.2,
                None,
                None,
                "2026-06-06 12:00:00",
                None,
                "2026-06-06 08:00:00",
                None,
                "2026-06-06 12:00:00",
                "趋势回调",
                None,
                None,
                "trend",
                "",
                None,
            ),
        )
        conn.commit()
        rows = fetch_trades_for_trading_day(conn, "2026-06-06")
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["result"], "时间平仓")
        conn.close()


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