72de8d534e
vol_breakout 멀티종목 tick, vol_live HTML 모니터, 마감 봉만 저장하는 캔들 다운로드, 텔레그램 체결 알림, cron/watch 감시 스크립트 및 테스트를 포함한다. Co-authored-by: Cursor <cursoragent@cursor.com>
79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
"""vol_live 모니터 에쿼티·B&H 테스트."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
from bithumb.operations.vol_live_monitor import (
|
|
build_multi_buyhold_series,
|
|
build_spot_strategy_equity_series,
|
|
)
|
|
|
|
|
|
def test_multi_buyhold_thirds() -> None:
|
|
"""3종목 1/3씩 B&H — 한 종목만 10% 상승 시 포트폴리오 +3.33% 근사."""
|
|
panel = pd.DataFrame({
|
|
"datetime": pd.to_datetime(["2026-06-01 10:00:00", "2026-06-01 10:15:00"]),
|
|
"TRX": [100.0, 110.0],
|
|
"NEAR": [200.0, 200.0],
|
|
"WLD": [300.0, 300.0],
|
|
})
|
|
bh = build_multi_buyhold_series(panel, ["TRX", "NEAR", "WLD"], seed_krw=300_000.0)
|
|
assert bh[0]["value"] == 0.0
|
|
assert bh[1]["value"] == pytest.approx(3.3333, rel=1e-3)
|
|
|
|
|
|
def test_strategy_replay_buy_sell() -> None:
|
|
panel = pd.DataFrame({
|
|
"datetime": pd.to_datetime(["2026-06-01 10:00:00", "2026-06-01 10:15:00"]),
|
|
"TRX": [100.0, 110.0],
|
|
"NEAR": [200.0, 200.0],
|
|
"WLD": [300.0, 300.0],
|
|
})
|
|
trades = [
|
|
{
|
|
"symbol": "TRX",
|
|
"side": "buy",
|
|
"ts": "2026-06-01 10:05:00",
|
|
"order_krw": 100_000.0,
|
|
"order_coin": 1000.0,
|
|
"price": 100.0,
|
|
},
|
|
]
|
|
curve = build_spot_strategy_equity_series(
|
|
panel,
|
|
["TRX", "NEAR", "WLD"],
|
|
trades,
|
|
seed_krw=300_000.0,
|
|
current_equity=310_000.0,
|
|
window_start=pd.Timestamp("2026-06-01 10:00:00"),
|
|
)
|
|
assert curve[0]["value"] == 0.0
|
|
assert curve[-1]["value"] == pytest.approx(3.3333, rel=1e-2)
|
|
|
|
|
|
def test_write_vol_monitor_html_no_format_error(tmp_path: Path) -> None:
|
|
import re
|
|
|
|
from bithumb.operations.vol_monitor_chart import write_vol_monitor_html
|
|
|
|
out = tmp_path / "vol_live_monitor.html"
|
|
write_vol_monitor_html(out)
|
|
text = out.read_text(encoding="utf-8")
|
|
assert "/api/chart" in text
|
|
assert "equityChart" in text
|
|
js = re.search(r"<script>(.*)</script>", text, re.S)
|
|
assert js is not None
|
|
js_path = tmp_path / "monitor.js"
|
|
js_path.write_text(js.group(1), encoding="utf-8")
|
|
import subprocess
|
|
|
|
subprocess.run(["node", "--check", str(js_path)], check=True, capture_output=True)
|