"""hub_trades_lib 单元测试.""" from __future__ import annotations import sqlite3 import unittest from datetime import datetime from lib.hub.hub_trades_lib import ( attach_journal_mood_tags, fetch_trades_for_trading_day, journal_trade_match_key, 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() def test_attach_journal_mood_tags_marks_sick(self): conn = sqlite3.connect(":memory:") conn.row_factory = sqlite3.Row conn.execute( """CREATE TABLE journal_entries ( coin TEXT, open_datetime TEXT, close_datetime TEXT, mood_issues TEXT, created_at TEXT )""" ) conn.execute( "INSERT INTO journal_entries VALUES (?,?,?,?,?)", ("ETH", "2026-07-06 21:51", "2026-07-07 00:00", "报复开仓,扛单", "2026-07-07 00:05"), ) conn.commit() trades = [ { "id": 42, "symbol": "ETH/USDT", "opened_at": "2026-07-06 21:51:00", "closed_at": "2026-07-07 00:00:00", } ] attach_journal_mood_tags(conn, trades, cutoff_s="2026-01-01 00:00:00") self.assertTrue(trades[0]["journal_mood_sick"]) self.assertEqual(trades[0]["behavior_tag"], "sick") self.assertEqual(trades[0]["journal_mood_issues"], ["报复开仓", "扛单"]) key = journal_trade_match_key("ETH/USDT", "2026-07-06 21:51:00", "2026-07-07 00:00:00") self.assertEqual(key, ("ETH", "2026-07-06 21:51", "2026-07-07 00:00")) conn.close() if __name__ == "__main__": unittest.main()