crypto_monitor/tests/test_strategy_snapshot_dedup.py

"""策略快照：同一计划同结果不重复写入。"""
from __future__ import annotations

import json
import sqlite3
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from strategy_snapshot_lib import (  # noqa: E402
    STRATEGY_TREND,
    dedupe_strategy_snapshots,
    init_strategy_snapshot_table,
    list_strategy_snapshots,
    save_trend_plan_snapshot,
)


def _mem_conn() -> sqlite3.Connection:
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    init_strategy_snapshot_table(conn)
    return conn


def test_save_trend_plan_snapshot_skips_duplicate_result():
    conn = _mem_conn()
    plan = {
        "id": 42,
        "symbol": "ONDO/USDT",
        "exchange_symbol": "ONDO/USDT:USDT",
        "direction": "short",
        "status": "active",
        "opened_at": "2026-06-08 08:00:00",
        "legs_done": 4,
        "dca_legs": 4,
        "first_order_done": 1,
        "grid_prices_json": "[]",
        "leg_amounts_json": "[]",
    }
    cfg = {"app_module": type("M", (), {"app_now_str": staticmethod(lambda: "2026-06-08 08:41:00")})()}
    save_trend_plan_snapshot(cfg, conn, plan, result_label="止损", pnl_amount=-2.3)
    save_trend_plan_snapshot(cfg, conn, plan, result_label="止损", pnl_amount=-2.4)
    conn.commit()
    rows = conn.execute(
        "SELECT COUNT(*) AS c FROM strategy_trade_snapshots WHERE source_id=? AND result_label=?",
        (42, "止损"),
    ).fetchone()
    assert int(rows["c"]) == 1


def test_dedupe_strategy_snapshots_handles_many_duplicates():
    conn = _mem_conn()
    payload = json.dumps({"symbol": "ONDO/USDT"}, ensure_ascii=False)
    for snap_id in range(1, 46):
        conn.execute(
            """INSERT INTO strategy_trade_snapshots (
                id, strategy_type, source_id, symbol, result_label, snapshot_json, closed_at, created_at, pnl_amount
            ) VALUES (?,?,?,?,?,?,?,?,?)""",
            (
                snap_id,
                STRATEGY_TREND,
                99,
                "ONDO/USDT",
                "止损",
                payload,
                "2026-06-08 08:41:00",
                "2026-06-08 08:41:00",
                -2.2,
            ),
        )
    conn.commit()
    removed = dedupe_strategy_snapshots(conn)
    conn.commit()
    assert removed == 44
    row = conn.execute(
        "SELECT COUNT(*) AS c FROM strategy_trade_snapshots WHERE source_id=?",
        (99,),
    ).fetchone()
    assert int(row["c"]) == 1


def test_dedupe_strategy_snapshots_keeps_latest_id():
    conn = _mem_conn()
    payload = json.dumps({"symbol": "ONDO/USDT"}, ensure_ascii=False)
    for snap_id, pnl in ((1, -2.23), (2, -2.31), (3, -2.38)):
        conn.execute(
            """INSERT INTO strategy_trade_snapshots (
                id, strategy_type, source_id, symbol, result_label, snapshot_json, closed_at, created_at, pnl_amount
            ) VALUES (?,?,?,?,?,?,?,?,?)""",
            (
                snap_id,
                STRATEGY_TREND,
                5,
                "ONDO/USDT",
                "止损",
                payload,
                "2026-06-08 08:41:00",
                "2026-06-08 08:41:00",
                pnl,
            ),
        )
    conn.commit()
    removed = dedupe_strategy_snapshots(conn)
    conn.commit()
    assert removed == 2
    row = conn.execute(
        "SELECT id, pnl_amount FROM strategy_trade_snapshots WHERE source_id=?",
        (5,),
    ).fetchone()
    assert int(row["id"]) == 3
    assert abs(float(row["pnl_amount"]) - (-2.38)) < 1e-6


def test_list_strategy_snapshots_hides_duplicate_keys():
    conn = _mem_conn()
    payload = json.dumps({"symbol": "ONDO/USDT", "dca_levels": []}, ensure_ascii=False)
    for snap_id in (10, 11, 12):
        conn.execute(
            """INSERT INTO strategy_trade_snapshots (
                id, strategy_type, source_id, symbol, direction, result_label,
                snapshot_json, closed_at, created_at, pnl_amount
            ) VALUES (?,?,?,?,?,?,?,?,?,?)""",
            (
                snap_id,
                STRATEGY_TREND,
                7,
                "ONDO/USDT",
                "short",
                "止损",
                payload,
                "2026-06-08 08:41:00",
                "2026-06-08 08:41:00",
                -2.2,
            ),
        )
    conn.commit()
    rows = list_strategy_snapshots(conn, limit=50)
    stop_rows = [r for r in rows if int(r.get("source_id") or 0) == 7]
    assert len(stop_rows) == 1
    assert int(stop_rows[0]["id"]) == 12


def test_dedupe_keeps_manual_over_stop_loss():
    conn = _mem_conn()
    payload = json.dumps({"symbol": "ONDO/USDT"}, ensure_ascii=False)
    for snap_id, label in ((10, "止损"), (11, "手动平仓")):
        conn.execute(
            """INSERT INTO strategy_trade_snapshots (
                id, strategy_type, source_id, symbol, result_label, snapshot_json, closed_at, created_at, pnl_amount
            ) VALUES (?,?,?,?,?,?,?,?,?)""",
            (
                snap_id,
                STRATEGY_TREND,
                7,
                "ONDO/USDT",
                label,
                payload,
                "2026-06-08 08:44:00",
                "2026-06-08 08:44:00",
                -2.23,
            ),
        )
    conn.commit()
    removed = dedupe_strategy_snapshots(conn)
    conn.commit()
    assert removed == 1
    row = conn.execute(
        "SELECT result_label FROM strategy_trade_snapshots WHERE source_id=?",
        (7,),
    ).fetchone()
    assert row["result_label"] == "手动平仓"


if __name__ == "__main__":
    test_save_trend_plan_snapshot_skips_duplicate_result()
    test_dedupe_strategy_snapshots_handles_many_duplicates()
    test_dedupe_strategy_snapshots_keeps_latest_id()
    test_list_strategy_snapshots_hides_duplicate_keys()
    test_dedupe_keeps_manual_over_stop_loss()
    print("all ok")