Bithumb/scripts/test_buy_sell_rehearsal.py

#!/usr/bin/env python3
"""
40만 원 기준 매수·매도 리허설 (DB 없이 synthetic + hybrid replay).
"""
from __future__ import annotations

import runpy
import sys
from pathlib import Path

runpy.run_path(str(Path(__file__).resolve().parent / "_bootstrap.py"))

import pandas as pd

from config import GT_INITIAL_CASH_KRW, TRADING_FEE_RATE
from deepcoin.ops.hybrid_sim_execution import (
    HybridSimPortfolio,
    hit_key,
    plan_live_hit,
    replay_hybrid_signals,
    sort_hits_sim_order,
)


def _mini_ohlc() -> pd.DataFrame:
    """drawdown 계산용 최소 3m OHLC."""
    idx = pd.date_range("2026-06-01 09:00:00", periods=200, freq="3min")
    close = pd.Series([500.0 - i * 0.1 for i in range(200)], index=idx, dtype=float)
    return pd.DataFrame(
        {"Open": close, "High": close + 2, "Low": close - 2, "Close": close},
        index=idx,
    )


def test_sort_buy_before_sell() -> None:
    """동일 시각 buy·sell → buy 먼저."""
    hits = [
        {"dt": "2026-06-01 12:00:00", "rule_id": "sell_mtf_cross_all_tf", "side": "sell", "close": 500.0},
        {"dt": "2026-06-01 12:00:00", "rule_id": "buy_compound_tight", "side": "buy", "close": 500.0},
    ]
    ordered = sort_hits_sim_order(hits)
    assert ordered[0]["side"] == "buy", ordered
    print("  [OK] 동일 시각 buy → sell 순서")


def test_sell_without_holdings() -> None:
    """보유 없이 매도만 → 모의 보유 없음."""
    ohlc = _mini_ohlc()
    approved = None
    hist = [
        {
            "dt": "2026-06-01 12:00:00",
            "rule_id": "sell_mtf_cross_all_tf",
            "side": "sell",
            "close": 500.0,
        }
    ]
    portfolio, results = replay_hybrid_signals(hist, ohlc, approved_buy_rules=approved)
    key = hit_key(hist[0])
    res = results[key]
    assert not res.ok and "보유 없음" in res.message, res
    assert portfolio.qty < 1e-9 and portfolio.cash_krw == float(GT_INITIAL_CASH_KRW)
    print("  [OK] 보유 없음 매도 스킵")


def test_buy_then_partial_sell() -> None:
    """매수 후 분할 매도 1회."""
    ohlc = _mini_ohlc()
    approved = None
    dt_buy = str(ohlc.index[50])
    dt_sell = str(ohlc.index[80])
    price_buy = float(ohlc.loc[ohlc.index[50], "Close"])
    price_sell = float(ohlc.loc[ohlc.index[80], "Close"])
    hist = [
        {
            "dt": dt_buy,
            "rule_id": "buy_compound_tight",
            "side": "buy",
            "close": price_buy,
        },
        {
            "dt": dt_sell,
            "rule_id": "sell_mtf_cross_all_tf",
            "side": "sell",
            "close": price_sell,
        },
    ]
    portfolio, results = replay_hybrid_signals(hist, ohlc, approved_buy_rules=approved)
    buy_res = results[hit_key(hist[0])]
    sell_res = results[hit_key(hist[1])]
    assert buy_res.ok, buy_res.message
    assert portfolio.qty > 0 or sell_res.ok, (portfolio.qty, sell_res)
    if sell_res.ok:
        assert sell_res.sell_qty > 0 and sell_res.amount_krw > 0
        assert portfolio.cash_krw > float(GT_INITIAL_CASH_KRW) * 0.5
    print(
        f"  [OK] 매수 ₩{buy_res.amount_krw:,.0f} → 매도 "
        f"ok={sell_res.ok} qty={sell_res.sell_qty:.4f} 현금=₩{portfolio.cash_krw:,.0f}"
    )


def test_unapproved_buy_excluded_when_filter_set() -> None:
    """approved_buy_rules 지정 시에만 미포함 매수 제외 (시뮬 기본은 필터 없음)."""
    ohlc = _mini_ohlc()
    hist = [
        {
            "dt": str(ohlc.index[60]),
            "rule_id": "buy_fake_rule",
            "side": "buy",
            "close": 500.0,
        },
    ]
    approved = {"buy_compound_tight"}
    sized_hist = replay_hybrid_signals(hist, ohlc, approved_buy_rules=approved)[0]
    assert sized_hist.qty < 1e-9
    print("  [OK] approved_buy_rules 지정 시 미승인 매수 제외")


def test_plan_live_matches_replay() -> None:
    """plan_live_hit == replay 마지막 건."""
    ohlc = _mini_ohlc()
    approved = None
    hist = []
    hit = {
        "dt": str(ohlc.index[70]),
        "rule_id": "buy_compound_tight",
        "side": "buy",
        "close": float(ohlc["Close"].iloc[70]),
    }
    plan = plan_live_hit(hist, hit, ohlc, approved_buy_rules=approved)
    hist.append(hit)
    _, results = replay_hybrid_signals(hist, ohlc, approved_buy_rules=approved)
    replay_res = results[hit_key(hit)]
    assert plan.amount_krw == replay_res.amount_krw, (plan, replay_res)
    assert plan.ok == replay_res.ok
    print(f"  [OK] plan_live_hit ≡ replay (₩{plan.amount_krw:,.0f})")


def test_initial_cash_400k_large_buy() -> None:
    """40만·대형 DD 시 매수액 ≤ 가용현금."""
    ohlc = _mini_ohlc()
    approved = None
    hit = {
        "dt": str(ohlc.index[100]),
        "rule_id": "buy_compound_tight",
        "side": "buy",
        "close": float(ohlc["Close"].iloc[100]),
    }
    plan = plan_live_hit([], hit, ohlc, approved_buy_rules=approved)
    assert plan.ok
    assert 0 < plan.amount_krw <= GT_INITIAL_CASH_KRW
    fee = plan.amount_krw * TRADING_FEE_RATE
    assert plan.amount_krw + fee <= GT_INITIAL_CASH_KRW + 1
    print(f"  [OK] 40만 대형 tier 매수 ₩{plan.amount_krw:,.0f} (≤{GT_INITIAL_CASH_KRW:,})")


def test_apply_buy_insufficient_cash() -> None:
    """현금 부족 시 apply_buy 실패."""
    p = HybridSimPortfolio()
    p.cash_krw = 10_000.0
    ok = p.apply_buy(50_000, 500.0, leg_id=1)
    assert not ok
    print("  [OK] 현금 부족 매수 거부")


def main() -> int:
    """리허설 실행."""
    print(f"[리허설] GT_INITIAL_CASH_KRW=₩{GT_INITIAL_CASH_KRW:,}")
    print("  approved_buy_rules: None (sim_causal_hybrid 동일)")
    fails = 0
    tests = [
        test_sort_buy_before_sell,
        test_sell_without_holdings,
        test_buy_then_partial_sell,
        test_unapproved_buy_excluded_when_filter_set,
        test_plan_live_matches_replay,
        test_initial_cash_400k_large_buy,
        test_apply_buy_insufficient_cash,
    ]
    for fn in tests:
        try:
            fn()
        except AssertionError as e:
            print(f"  [FAIL] {fn.__name__}: {e}")
            fails += 1
        except Exception as e:
            print(f"  [ERROR] {fn.__name__}: {e}")
            fails += 1
    print(f"\n[결과] {'PASS' if fails == 0 else f'FAIL ({fails})'}")
    return 1 if fails else 0


if __name__ == "__main__":
    raise SystemExit(main())