373 lines
12 KiB
Python
373 lines
12 KiB
Python
"""Unit tests for live mode functionality."""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from openclaw.trading.live_mode import (
|
|
LiveModeConfig,
|
|
LiveModeManager,
|
|
LiveTradeLogEntry,
|
|
TradingMode,
|
|
)
|
|
|
|
|
|
class TestLiveModeConfig:
|
|
"""Tests for LiveModeConfig."""
|
|
|
|
def test_default_config(self):
|
|
"""Test default configuration."""
|
|
config = LiveModeConfig()
|
|
assert config.enabled is False
|
|
assert config.daily_trade_limit_usd == 10000.0
|
|
assert config.max_position_pct == 0.2
|
|
assert config.require_confirmation is True
|
|
assert config.confirmation_timeout_seconds == 30
|
|
|
|
def test_custom_config(self):
|
|
"""Test custom configuration."""
|
|
config = LiveModeConfig(
|
|
enabled=True,
|
|
daily_trade_limit_usd=50000.0,
|
|
max_position_pct=0.5,
|
|
require_confirmation=False,
|
|
)
|
|
assert config.enabled is True
|
|
assert config.daily_trade_limit_usd == 50000.0
|
|
assert config.max_position_pct == 0.5
|
|
assert config.require_confirmation is False
|
|
|
|
def test_invalid_position_pct(self):
|
|
"""Test invalid position percentage validation."""
|
|
with pytest.raises(ValueError):
|
|
LiveModeConfig(max_position_pct=1.5)
|
|
|
|
def test_invalid_webhook_url(self):
|
|
"""Test invalid webhook URL validation."""
|
|
with pytest.raises(ValueError):
|
|
LiveModeConfig(alert_webhook_url="invalid-url")
|
|
|
|
|
|
class TestLiveModeManager:
|
|
"""Tests for LiveModeManager."""
|
|
|
|
def test_default_mode_is_simulated(self):
|
|
"""Test that default mode is simulated."""
|
|
manager = LiveModeManager()
|
|
assert manager.is_simulated_mode is True
|
|
assert manager.is_live_mode is False
|
|
assert "SIMULATED" in manager.mode_indicator
|
|
|
|
def test_live_mode_enabled(self):
|
|
"""Test live mode when enabled."""
|
|
config = LiveModeConfig(enabled=True)
|
|
manager = LiveModeManager(config=config)
|
|
assert manager.is_live_mode is True
|
|
assert manager.is_simulated_mode is False
|
|
assert "LIVE" in manager.mode_indicator
|
|
|
|
def test_daily_limit(self):
|
|
"""Test daily limit retrieval."""
|
|
config = LiveModeConfig(daily_trade_limit_usd=5000.0)
|
|
manager = LiveModeManager(config=config)
|
|
assert manager.get_daily_limit() == 5000.0
|
|
assert manager.get_daily_limit_remaining() == 5000.0
|
|
|
|
def test_validate_live_trade_in_simulated_mode(self):
|
|
"""Test validation fails in simulated mode."""
|
|
manager = LiveModeManager() # Simulated by default
|
|
is_valid, reason = manager.validate_live_trade(
|
|
symbol="BTC/USDT",
|
|
amount=1.0,
|
|
price=50000.0,
|
|
current_balance=100000.0,
|
|
)
|
|
assert is_valid is False
|
|
assert "Not in live trading mode" in reason
|
|
|
|
def test_validate_live_trade_daily_limit(self):
|
|
"""Test validation fails when daily limit exceeded."""
|
|
config = LiveModeConfig(enabled=True, daily_trade_limit_usd=1000.0)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
is_valid, reason = manager.validate_live_trade(
|
|
symbol="BTC/USDT",
|
|
amount=1.0,
|
|
price=5000.0, # Exceeds $1000 daily limit
|
|
current_balance=100000.0,
|
|
)
|
|
assert is_valid is False
|
|
assert "Daily limit exceeded" in reason
|
|
|
|
def test_validate_live_trade_position_size(self):
|
|
"""Test validation fails when position size too large."""
|
|
config = LiveModeConfig(
|
|
enabled=True,
|
|
max_position_pct=0.1,
|
|
daily_trade_limit_usd=100000.0, # High limit so it doesn't trigger first
|
|
)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
is_valid, reason = manager.validate_live_trade(
|
|
symbol="BTC/USDT",
|
|
amount=1.0,
|
|
price=50000.0, # $50k trade
|
|
current_balance=100000.0, # 50% of balance, but limit is 10%
|
|
)
|
|
assert is_valid is False
|
|
assert "Position size exceeds limit" in reason
|
|
|
|
def test_validate_live_trade_insufficient_balance(self):
|
|
"""Test validation fails with insufficient balance."""
|
|
config = LiveModeConfig(
|
|
enabled=True,
|
|
daily_trade_limit_usd=100000.0, # High limit
|
|
max_position_pct=1.0, # 100% to not trigger position size limit
|
|
)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
is_valid, reason = manager.validate_live_trade(
|
|
symbol="BTC/USDT",
|
|
amount=1.0,
|
|
price=50000.0,
|
|
current_balance=60000.0, # Has 1.2x but needs 1.5x buffer
|
|
)
|
|
assert is_valid is False
|
|
assert "Insufficient balance" in reason
|
|
|
|
def test_validate_live_trade_passes(self):
|
|
"""Test validation passes with valid parameters."""
|
|
config = LiveModeConfig(enabled=True, daily_trade_limit_usd=100000.0)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
is_valid, reason = manager.validate_live_trade(
|
|
symbol="BTC/USDT",
|
|
amount=0.1,
|
|
price=50000.0, # $5k trade
|
|
current_balance=100000.0, # Has 1.5x buffer
|
|
)
|
|
assert is_valid is True
|
|
assert "Validation passed" in reason
|
|
|
|
def test_confirmation_request(self):
|
|
"""Test confirmation request."""
|
|
config = LiveModeConfig(enabled=True, require_confirmation=False)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
confirmed, code = manager.request_confirmation(
|
|
symbol="BTC/USDT",
|
|
side="buy",
|
|
amount=0.1,
|
|
price=50000.0,
|
|
)
|
|
assert confirmed is True
|
|
assert code == "AUTO_CONFIRMED"
|
|
|
|
def test_log_live_trade(self):
|
|
"""Test logging a live trade."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
log_path = Path(tmpdir) / "trades.jsonl"
|
|
config = LiveModeConfig(
|
|
enabled=True,
|
|
audit_log_path=str(log_path),
|
|
)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
manager.log_live_trade(
|
|
symbol="BTC/USDT",
|
|
side="buy",
|
|
amount=0.1,
|
|
price=50000.0,
|
|
order_id="test-123",
|
|
confirmation_code="CONF-ABC",
|
|
risk_checks_passed=True,
|
|
metadata={"strategy": "test"},
|
|
)
|
|
|
|
# Check audit log in memory
|
|
assert len(manager._audit_log) == 1
|
|
entry = manager._audit_log[0]
|
|
assert entry.symbol == "BTC/USDT"
|
|
assert entry.order_id == "test-123"
|
|
|
|
# Check file was written
|
|
assert log_path.exists()
|
|
content = log_path.read_text()
|
|
assert "BTC/USDT" in content
|
|
assert "test-123" in content
|
|
|
|
def test_daily_limit_tracking(self):
|
|
"""Test daily limit is tracked correctly."""
|
|
config = LiveModeConfig(enabled=True, daily_trade_limit_usd=10000.0)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
assert manager.get_daily_limit_remaining() == 10000.0
|
|
|
|
manager.log_live_trade(
|
|
symbol="BTC/USDT",
|
|
side="buy",
|
|
amount=0.1,
|
|
price=50000.0,
|
|
order_id="test-1",
|
|
confirmation_code="CONF-1",
|
|
risk_checks_passed=True,
|
|
)
|
|
|
|
assert manager.get_daily_limit_remaining() == 5000.0
|
|
assert manager._trade_count_today == 1
|
|
|
|
def test_get_live_stats(self):
|
|
"""Test getting live stats."""
|
|
config = LiveModeConfig(
|
|
enabled=True,
|
|
daily_trade_limit_usd=50000.0,
|
|
max_position_pct=0.3,
|
|
)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
stats = manager.get_live_stats()
|
|
assert stats["mode"] == "live"
|
|
assert stats["is_live"] is True
|
|
assert stats["daily_limit_usd"] == 50000.0
|
|
assert stats["max_position_pct"] == 0.3
|
|
assert stats["confirmation_required"] is True
|
|
|
|
def test_switch_mode(self):
|
|
"""Test mode switching."""
|
|
config = LiveModeConfig(enabled=True)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
assert manager.is_live_mode is True
|
|
|
|
# Switch to simulated
|
|
result = manager.switch_mode(TradingMode.SIMULATED)
|
|
assert result is True
|
|
assert manager.is_simulated_mode is True
|
|
|
|
# Switch back to live
|
|
result = manager.switch_mode(TradingMode.LIVE)
|
|
assert result is True
|
|
assert manager.is_live_mode is True
|
|
|
|
def test_cannot_switch_to_live_if_disabled(self):
|
|
"""Test cannot switch to live if not enabled in config."""
|
|
config = LiveModeConfig(enabled=False)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
assert manager.is_simulated_mode is True
|
|
|
|
result = manager.switch_mode(TradingMode.LIVE)
|
|
assert result is False
|
|
assert manager.is_simulated_mode is True
|
|
|
|
def test_enable_disable_live_mode(self):
|
|
"""Test enable/disable methods."""
|
|
config = LiveModeConfig(enabled=False)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
assert manager.is_simulated_mode is True
|
|
|
|
# Enable live mode
|
|
result = manager.enable_live_mode()
|
|
assert result is True
|
|
assert manager.is_live_mode is True
|
|
|
|
# Disable live mode
|
|
result = manager.disable_live_mode()
|
|
assert result is True
|
|
assert manager.is_simulated_mode is True
|
|
|
|
def test_get_audit_log(self):
|
|
"""Test retrieving audit log."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
log_path = Path(tmpdir) / "trades.jsonl"
|
|
config = LiveModeConfig(
|
|
enabled=True,
|
|
audit_log_path=str(log_path),
|
|
)
|
|
manager = LiveModeManager(config=config)
|
|
|
|
# Log a trade
|
|
manager.log_live_trade(
|
|
symbol="BTC/USDT",
|
|
side="buy",
|
|
amount=0.1,
|
|
price=50000.0,
|
|
order_id="test-1",
|
|
confirmation_code="CONF-1",
|
|
risk_checks_passed=True,
|
|
)
|
|
|
|
# Retrieve log
|
|
entries = manager.get_audit_log()
|
|
assert len(entries) == 1
|
|
assert entries[0].symbol == "BTC/USDT"
|
|
|
|
|
|
class TestLiveTradeLogEntry:
|
|
"""Tests for LiveTradeLogEntry model."""
|
|
|
|
def test_entry_creation(self):
|
|
"""Test log entry creation."""
|
|
entry = LiveTradeLogEntry(
|
|
timestamp=datetime.now().isoformat(),
|
|
symbol="BTC/USDT",
|
|
side="buy",
|
|
amount=0.1,
|
|
price=50000.0,
|
|
order_id="test-123",
|
|
confirmation_code="CONF-ABC",
|
|
risk_checks_passed=True,
|
|
daily_limit_before=10000.0,
|
|
daily_limit_after=5000.0,
|
|
)
|
|
assert entry.symbol == "BTC/USDT"
|
|
assert entry.side == "buy"
|
|
assert entry.amount == 0.1
|
|
assert entry.risk_checks_passed is True
|
|
|
|
def test_entry_with_metadata(self):
|
|
"""Test log entry with metadata."""
|
|
entry = LiveTradeLogEntry(
|
|
timestamp=datetime.now().isoformat(),
|
|
symbol="ETH/USDT",
|
|
side="sell",
|
|
amount=1.0,
|
|
price=3000.0,
|
|
order_id="test-456",
|
|
confirmation_code="CONF-DEF",
|
|
risk_checks_passed=True,
|
|
daily_limit_before=5000.0,
|
|
daily_limit_after=2000.0,
|
|
metadata={"strategy": "momentum", "signal_strength": 0.85},
|
|
)
|
|
assert entry.metadata["strategy"] == "momentum"
|
|
assert entry.metadata["signal_strength"] == 0.85
|
|
|
|
def test_entry_json_serialization(self):
|
|
"""Test JSON serialization."""
|
|
entry = LiveTradeLogEntry(
|
|
timestamp="2024-01-15T10:30:00",
|
|
symbol="BTC/USDT",
|
|
side="buy",
|
|
amount=0.1,
|
|
price=50000.0,
|
|
order_id="test-123",
|
|
confirmation_code="CONF-ABC",
|
|
risk_checks_passed=True,
|
|
daily_limit_before=10000.0,
|
|
daily_limit_after=5000.0,
|
|
)
|
|
json_str = entry.model_dump_json()
|
|
assert "BTC/USDT" in json_str
|
|
assert "test-123" in json_str
|
|
|
|
# Parse back and verify
|
|
data = json.loads(json_str)
|
|
assert data["symbol"] == "BTC/USDT"
|
|
assert data["amount"] == 0.1
|