72de8d534e
vol_breakout 멀티종목 tick, vol_live HTML 모니터, 마감 봉만 저장하는 캔들 다운로드, 텔레그램 체결 알림, cron/watch 감시 스크립트 및 테스트를 포함한다. Co-authored-by: Cursor <cursoragent@cursor.com>
73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
"""watch_ops vol_breakout 분기."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from bithumb.operations.watch_ops import (
|
|
WatchIssue,
|
|
WatchReport,
|
|
inspect_vol_watch,
|
|
is_vol_breakout_ops,
|
|
remediate_vol_watch,
|
|
)
|
|
|
|
|
|
def test_is_vol_breakout_ops(tmp_path: Path) -> None:
|
|
"""vol state 파일이 있으면 vol 감시 모드."""
|
|
state_path = tmp_path / "vol_breakout_state.json"
|
|
state_path.write_text(
|
|
json.dumps({"strategy": "vol_breakout_15m_spot_long", "symbols": {}}),
|
|
encoding="utf-8",
|
|
)
|
|
settings = SimpleNamespace(vol_state_json=state_path)
|
|
assert is_vol_breakout_ops(settings) is True
|
|
|
|
|
|
def test_is_vol_breakout_ops_missing_file(tmp_path: Path) -> None:
|
|
"""state 없으면 fractal 감시."""
|
|
settings = SimpleNamespace(vol_state_json=tmp_path / "missing.json")
|
|
assert is_vol_breakout_ops(settings) is False
|
|
|
|
|
|
def test_inspect_vol_watch_recent_tick(tmp_path: Path) -> None:
|
|
"""최근 tick이면 이슈 없음."""
|
|
state_path = tmp_path / "vol.json"
|
|
state_path.write_text(
|
|
json.dumps({
|
|
"strategy": "vol_breakout_15m_spot_long",
|
|
"symbols": {},
|
|
"last_run_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
settings = SimpleNamespace(
|
|
vol_state_json=state_path,
|
|
ops_watch_tick_stale_min=12,
|
|
)
|
|
report = inspect_vol_watch(settings)
|
|
assert report.issues == []
|
|
|
|
|
|
def test_remediate_vol_dry_run() -> None:
|
|
"""vol dry-run은 tick 없음."""
|
|
report = WatchReport(
|
|
checked_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
issues=[
|
|
WatchIssue(kind="tick_stale", severity="critical", message="stale"),
|
|
],
|
|
)
|
|
settings = SimpleNamespace(
|
|
ops_telegram_enabled=False,
|
|
telegram_bot_token="",
|
|
telegram_chat_id="",
|
|
ops_watch_auto_remediate=True,
|
|
ops_mode="live",
|
|
)
|
|
result = remediate_vol_watch(settings, report, dry_run=True)
|
|
assert "dry_run" in result.actions
|
|
assert result.tick_report is None
|