crypto_monitor/tests/test_options_review_lib.py

"""期权复盘(含对冲)单元测试:导入去重、双计防护、复盘不被覆盖、统计."""
from __future__ import annotations

import sqlite3
import tempfile
import unittest
from pathlib import Path

from lib.hedge_plan.hedge_plan_db import init_hedge_plan_tables, insert_leg, insert_plan
from lib.options.options_review_db import SOURCE_OPTION, SOURCE_PERP_OPTIONS, init_options_review_tables
from lib.options.options_review_images_lib import (
    build_options_review_slot_filename,
    is_valid_options_review_file,
    options_review_upload_dir,
    save_options_review_slot_file,
)
from lib.options.options_review_lib import (
    compute_review_stats,
    list_review_trades,
    save_review_entry,
    sync_hedge_plans_closed,
    sync_options_from_exchange,
    upsert_option_history_row,
)


def _conn() -> sqlite3.Connection:
    c = sqlite3.connect(":memory:")
    c.row_factory = sqlite3.Row
    init_options_review_tables(c)
    init_hedge_plan_tables(c)
    return c


class _FakeFile:
    def __init__(self, name: str, data: bytes = b"img"):
        self.filename = name
        self._data = data

    def save(self, path: str) -> None:
        Path(path).write_bytes(self._data)


class OptionsReviewTests(unittest.TestCase):
    def test_option_upsert_idempotent(self):
        conn = _conn()
        row = {
            "history_key": "ex:pos1",
            "pos_id": "pos1",
            "inst_id": "ETH-USD-260328-2000-C",
            "underlying": "ETH",
            "opt_type": "C",
            "strike": 2000,
            "sheets": 10,
            "open_avg_px": 0.01,
            "close_avg_px": 0.02,
            "premium_paid": 1.0,
            "realized_pnl": 5.5,
            "created_at": "2026-03-01 10:00:00",
            "closed_at": "2026-03-01 12:00:00",
            "status_label": "已平",
        }
        self.assertEqual(upsert_option_history_row(conn, row), "inserted")
        row["realized_pnl"] = 6.0
        self.assertEqual(upsert_option_history_row(conn, row), "updated")
        n = conn.execute("SELECT COUNT(*) AS c FROM options_review_trades").fetchone()["c"]
        self.assertEqual(n, 1)
        pnl = conn.execute(
            "SELECT realized_pnl_total FROM options_review_trades WHERE history_key='ex:pos1'"
        ).fetchone()["realized_pnl_total"]
        self.assertEqual(float(pnl), 6.0)

    def test_entry_not_overwritten_by_resync(self):
        conn = _conn()
        upsert_option_history_row(
            conn,
            {
                "history_key": "ex:p2",
                "inst_id": "ETH-USD-260328-1800-P",
                "underlying": "ETH",
                "opt_type": "P",
                "realized_pnl": 1.0,
                "created_at": "2026-03-02 10:00:00",
                "closed_at": "2026-03-02 11:00:00",
            },
        )
        tid = conn.execute("SELECT id FROM options_review_trades").fetchone()["id"]
        save_review_entry(
            conn,
            tid,
            {"strategy_tag": "突破追涨", "note": "keep-me", "images": []},
        )
        upsert_option_history_row(
            conn,
            {
                "history_key": "ex:p2",
                "inst_id": "ETH-USD-260328-1800-P",
                "underlying": "ETH",
                "opt_type": "P",
                "realized_pnl": 2.0,
                "created_at": "2026-03-02 10:00:00",
                "closed_at": "2026-03-02 11:00:00",
            },
        )
        note = conn.execute(
            "SELECT note, strategy_tag FROM options_review_entries WHERE trade_id=?",
            (tid,),
        ).fetchone()
        self.assertEqual(note["note"], "keep-me")
        self.assertEqual(note["strategy_tag"], "突破追涨")
        pnl = conn.execute(
            "SELECT realized_pnl_total FROM options_review_trades WHERE id=?", (tid,)
        ).fetchone()["realized_pnl_total"]
        self.assertEqual(float(pnl), 2.0)

    def test_hedge_import_and_double_count_guard(self):
        conn = _conn()
        upsert_option_history_row(
            conn,
            {
                "history_key": "ex:leg1",
                "inst_id": "ETH-USD-260328-2000-C",
                "underlying": "ETH",
                "opt_type": "C",
                "realized_pnl": -3.0,
                "created_at": "2026-03-03 09:00:00",
                "closed_at": "2026-03-03 18:00:00",
            },
        )
        plan_id = insert_plan(
            conn,
            {
                "plan_type": SOURCE_PERP_OPTIONS,
                "status": "closed",
                "underlying": "ETH",
                "direction": "long",
                "realized_pnl_perp": 20.0,
                "realized_pnl_options": -3.0,
                "realized_pnl_total": 17.0,
                "close_reason": "tp",
                "opened_at": "2026-03-03 09:00:00",
                "closed_at": "2026-03-03 18:00:00",
                "premium_total": 3.0,
            },
        )
        insert_leg(
            conn,
            {
                "plan_id": plan_id,
                "leg_role": "perp",
                "symbol": "ETH-USDT-SWAP",
                "status": "closed",
                "realized_pnl": 20.0,
            },
        )
        insert_leg(
            conn,
            {
                "plan_id": plan_id,
                "leg_role": "option_hedge",
                "inst_id": "ETH-USD-260328-2000-C",
                "opt_type": "C",
                "status": "closed",
                "realized_pnl": -3.0,
            },
        )
        out = sync_hedge_plans_closed(conn)
        self.assertTrue(out["ok"])
        self.assertEqual(out["inserted"], 1)

        listed = list_review_trades(conn, include_hedge_legs=False)
        types = {r["source_type"] for r in listed}
        self.assertIn(SOURCE_PERP_OPTIONS, types)
        self.assertNotIn(SOURCE_OPTION, types)

        listed_all = list_review_trades(conn, include_hedge_legs=True)
        self.assertEqual(len(listed_all), 2)

        stats = compute_review_stats(conn, include_hedge_legs=False)
        self.assertEqual(stats["kpi"]["total"], 1)
        self.assertEqual(stats["kpi"]["pnl_sum"], 17.0)

    def test_sync_options_from_mock_exchange(self):
        conn = _conn()

        def fetch(_ex, limit=500):
            return [
                {
                    "instId": "ETH-USD-260328-2100-C",
                    "posId": "mock1",
                    "openAvgPx": "0.01",
                    "closeAvgPx": "0.02",
                    "closeTotalPos": "5",
                    "realizedPnl": "1.23",
                    "type": "2",
                    "cTime": "1700000000000",
                    "uTime": "1700003600000",
                    "uly": "ETH-USD",
                }
            ]

        def fmt(raw, tick_sz=None, ct_mult=0.01):
            return {
                "history_key": f"ex:{raw['posId']}",
                "pos_id": raw["posId"],
                "inst_id": raw["instId"],
                "underlying": "ETH",
                "opt_type": "C",
                "sheets": 5,
                "open_avg_px": 0.01,
                "close_avg_px": 0.02,
                "premium_paid": 0.5,
                "realized_pnl": float(raw["realizedPnl"]),
                "created_at": "2026-01-01 00:00:00",
                "closed_at": "2026-01-01 01:00:00",
                "status_label": "已平",
            }

        result = sync_options_from_exchange(
            conn, object(), limit=10, fetch_fn=fetch, format_fn=fmt
        )
        self.assertTrue(result["ok"])
        self.assertEqual(result["inserted"], 1)
        row = conn.execute(
            "SELECT realized_pnl_total FROM options_review_trades WHERE history_key='ex:mock1'"
        ).fetchone()
        self.assertEqual(float(row["realized_pnl_total"]), 1.23)

    def test_hide_trade_persists_across_local_sync(self):
        conn = _conn()
        from lib.options.options_db import init_options_tables
        from lib.options.options_review_lib import (
            hide_review_trade,
            sync_options_from_local_trades,
        )

        init_options_tables(conn)
        conn.execute(
            """
            INSERT INTO options_trades
            (inst_id, underlying, opt_type, strike, sheets, eth_amount,
             open_quote, premium_paid, status, realized_pnl, created_at, closed_at)
            VALUES ('ETH-USD-1-C','ETH','C',2000,1,0.01,0.01,0.2,'closed',1.0,
                    '2026-03-01 10:00:00','2026-03-01 11:00:00')
            """
        )
        sync_options_from_local_trades(conn)
        tid = conn.execute("SELECT id FROM options_review_trades").fetchone()["id"]
        out = hide_review_trade(conn, tid)
        self.assertTrue(out["ok"])
        self.assertEqual(
            conn.execute("SELECT COUNT(*) AS c FROM options_review_trades").fetchone()["c"],
            0,
        )
        sync_options_from_local_trades(conn)
        self.assertEqual(
            conn.execute("SELECT COUNT(*) AS c FROM options_review_trades").fetchone()["c"],
            0,
        )

    def test_local_options_trades_import(self):
        conn = _conn()
        from lib.options.options_db import init_options_tables

        init_options_tables(conn)
        conn.execute(
            """
            INSERT INTO options_trades
            (inst_id, underlying, opt_type, strike, sheets, eth_amount,
             open_quote, premium_paid, status, realized_pnl, created_at, closed_at)
            VALUES ('ETH-USD-260328-2000-C','ETH','C',2000,2,0.02,0.01,0.5,'closed',3.2,
                    '2026-03-01 10:00:00','2026-03-01 12:00:00')
            """
        )
        from lib.options.options_review_lib import sync_options_from_local_trades

        out = sync_options_from_local_trades(conn)
        self.assertTrue(out["ok"])
        self.assertEqual(out["inserted"], 1)
        row = conn.execute(
            "SELECT history_key, realized_pnl_total, source_type FROM options_review_trades"
        ).fetchone()
        self.assertTrue(str(row["history_key"]).startswith("local_opt:"))
        self.assertEqual(float(row["realized_pnl_total"]), 3.2)
        self.assertEqual(row["source_type"], SOURCE_OPTION)

    def test_image_paths_resolve_legacy_journal_root(self):
        with tempfile.TemporaryDirectory() as td:
            root = Path(td)
            # 误存到 UPLOAD 根目录的 journal_*
            legacy = root / "journal_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa_5m.png"
            legacy.write_bytes(b"img")
            # 正常 options_journal 子目录
            sub = root / "options_journal"
            sub.mkdir()
            modern = sub / "options_journal_bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb_5m.png"
            modern.write_bytes(b"img2")
            from lib.options.options_review_images_lib import options_review_image_paths

            class Row:
                images_json = (
                    '[{"tf":"5m","file":"journal_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa_5m.png"},'
                    '{"tf":"5m","file":"options_journal_bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb_5m.png"}]'
                )
                image = None

            paths = options_review_image_paths(Row(), str(root))
            self.assertEqual(len(paths), 2)
            self.assertTrue(any(p.endswith(legacy.name) for p in paths))
            self.assertTrue(any(p.endswith(modern.name) for p in paths))

    def test_image_namespace(self):
        with tempfile.TemporaryDirectory() as tmp:
            folder = options_review_upload_dir(tmp)
            fname = build_options_review_slot_filename(
                "a" * 32, "5m", ".png", secure_filename_fn=lambda x: x
            )
            self.assertTrue(fname.startswith("options_journal_"))
            self.assertTrue(is_valid_options_review_file(fname, "a" * 32, "5m"))
            item = save_options_review_slot_file(
                _FakeFile("x.png"),
                "a" * 32,
                "5m",
                folder,
                secure_filename_fn=lambda x: x,
            )
            self.assertIsNotNone(item)
            self.assertTrue((Path(folder) / item["file"]).is_file())

    def test_strategy_stats_only_tagged(self):
        conn = _conn()
        upsert_option_history_row(
            conn,
            {
                "history_key": "ex:a",
                "inst_id": "ETH-USD-1-C",
                "underlying": "ETH",
                "opt_type": "C",
                "realized_pnl": 10,
                "created_at": "2026-01-01 00:00:00",
                "closed_at": "2026-01-01 02:00:00",
            },
        )
        upsert_option_history_row(
            conn,
            {
                "history_key": "ex:b",
                "inst_id": "ETH-USD-2-P",
                "underlying": "ETH",
                "opt_type": "P",
                "realized_pnl": -4,
                "created_at": "2026-01-01 00:00:00",
                "closed_at": "2026-01-01 05:00:00",
            },
        )
        tid = conn.execute(
            "SELECT id FROM options_review_trades WHERE history_key='ex:a'"
        ).fetchone()["id"]
        save_review_entry(conn, tid, {"strategy_tag": "假破", "images": []})
        stats = compute_review_stats(conn)
        self.assertEqual(len(stats["by_strategy"]), 1)
        self.assertEqual(stats["by_strategy"][0]["key"], "假破")
        self.assertEqual(stats["kpi"]["total"], 2)

    def test_q_search_btcusdt_matches_btc_pending(self):
        from lib.options.options_review_lib import count_review_trades

        conn = _conn()
        upsert_option_history_row(
            conn,
            {
                "history_key": "ex:btc1",
                "inst_id": "BTC-USD-260328-90000-C",
                "underlying": "BTC",
                "opt_type": "C",
                "realized_pnl": 1.2,
                "created_at": "2026-01-01 00:00:00",
                "closed_at": "2026-01-01 02:00:00",
            },
        )
        upsert_option_history_row(
            conn,
            {
                "history_key": "ex:eth1",
                "inst_id": "ETH-USD-260328-2000-C",
                "underlying": "ETH",
                "opt_type": "C",
                "realized_pnl": 2.0,
                "created_at": "2026-01-01 00:00:00",
                "closed_at": "2026-01-01 03:00:00",
            },
        )
        # 旧精确 strategy_tag 会把待复盘滤成空
        self.assertEqual(
            count_review_trades(conn, strategy_tag="BTCUSDT", reviewed="0"),
            0,
        )
        self.assertEqual(
            count_review_trades(conn, q="BTCUSDT", reviewed="0"),
            1,
        )
        rows = list_review_trades(conn, q="BTCUSDT", reviewed="0")
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["underlying"], "BTC")


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